Collaboration diagram for jubatus::core::burst::aggregator::impl_:

Public Member Functions
bool	add_document (int d, int r, double pos)

int	flush_results (double scaling_param, double gamma, double costcut_threshold, int max_reuse_batches, result_storage &stored)

	impl_ (int window_batch_size, double batch_interval, int max_stored)

	MSGPACK_DEFINE (inputs_, window_batch_size_, batch_interval_, max_stored_)

Private Member Functions
input_window	make_new_window_ (double pos, const input_window &prev) const

Private Attributes
double	batch_interval_

std::deque< input_window >	inputs_

int	max_stored_

int	window_batch_size_

Detailed Description

Definition at line 31 of file aggregator.cpp.

Constructor & Destructor Documentation

jubatus::core::burst::aggregator::impl_::impl_	(	int	window_batch_size,
		double	batch_interval,
		int	max_stored
	)

inline

Definition at line 33 of file aggregator.cpp.

       : window_batch_size_(window_batch_size),
         batch_interval_(batch_interval),
         max_stored_(max_stored) {
   }

Member Function Documentation

bool jubatus::core::burst::aggregator::impl_::add_document	(	int	d,
		int	r,
		double	pos
	)

inline

Definition at line 39 of file aggregator.cpp.

References batch_interval_, inputs_, make_new_window_(), and max_stored_.

                                               {
     typedef std::deque<input_window>::iterator iterator_t;
 
     if (inputs_.empty()) {
       inputs_.push_front(
           make_new_window_(pos, input_window(0, batch_interval_, 0)));
     } else if (inputs_.front().get_end_pos() <= pos) {
       input_window new_window = make_new_window_(pos, inputs_[0]);
       inputs_.push_front(input_window());
       inputs_.front().swap(new_window);  // move semantics
       while (inputs_.size() > static_cast<size_t>(max_stored_)) {
         inputs_.pop_back();
       }
     }
 
     bool added = false;
     for (iterator_t iter = inputs_.begin(), end = inputs_.end();
          iter != end; ++iter) {
       if (iter->add_document(d, r, pos)) {
         added = true;
       } else {
         return added;
       }
     }
     return added;
   }

Here is the call graph for this function:

int jubatus::core::burst::aggregator::impl_::flush_results	(	double	scaling_param,
		double	gamma,
		double	costcut_threshold,
		int	max_reuse_batches,
		result_storage &	stored
	)

inline

Definition at line 66 of file aggregator.cpp.

References batch_interval_, jubatus::core::burst::get_intersection(), jubatus::core::burst::result_storage::get_result_at(), inputs_, JUBATUS_ASSERT_GT, jubatus::core::burst::result_storage::store(), and jubatus::core::burst::burst_result::swap().

                                             {
     if (inputs_.empty()) {
       return 0;
     }
 
     burst_result prev = stored.get_result_at(
         inputs_.back().get_start_pos() - batch_interval_/2);
 
     typedef std::deque<input_window>::reverse_iterator iterator_t;
 
     for (iterator_t iter = inputs_.rbegin(), end = inputs_.rend();
          iter != end; ++iter) {
       burst_result r(*iter, scaling_param, gamma, costcut_threshold,
                      prev, max_reuse_batches);
       stored.store(r);
       prev.swap(r);  // more efficient than prev = r; use move when C++11/14
     }
 
     // erase inputs which will no longer be modified by add_document
     int n = 0;
     for (;;) {
       JUBATUS_ASSERT_GT(inputs_.size(), 0, "");
 
       std::pair<int, int> intersection =
           get_intersection(inputs_.back(), inputs_.front());
       if (intersection.first != intersection.second) {
         break;  // break if intersection exists
       }
 
       inputs_.pop_back();
       ++n;
     }
     // return erased count
     return n;
   }

Here is the call graph for this function:

input_window jubatus::core::burst::aggregator::impl_::make_new_window_	(	double	pos,
		const input_window &	prev
	)		const

inlineprivate

Definition at line 114 of file aggregator.cpp.

References jubatus::core::burst::basic_window< Batch >::get_batch_by_index(), jubatus::core::burst::get_intersection(), jubatus::core::burst::basic_window< Batch >::get_start_pos(), and JUBATUS_ASSERT_LT.

Referenced by add_document().

                                                                             {
     double prev_start_pos = prev.get_start_pos();
     int i = static_cast<int>(
               std::floor((pos - prev_start_pos) / batch_interval_));
     int j = i - window_batch_size_/2;
     double new_start_pos = prev_start_pos + batch_interval_ * j;
 
     input_window new_window(
         new_start_pos, batch_interval_, window_batch_size_);
 
     // fill new_window's d&r vector
     std::pair<int, int> intersection = get_intersection(prev, new_window);
     for (int i = 0, j = intersection.first;
          j < intersection.second;
          ++i, ++j) {
       JUBATUS_ASSERT_LT(i, window_batch_size_, "");
       new_window.get_batch_by_index(i) = prev.get_batch_by_index(j);
     }
 
     return new_window;  // NRVO
   }