#include <compressive_storage.hpp>

Inheritance diagram for jubatus::core::clustering::compressive_storage:

Collaboration diagram for jubatus::core::clustering::compressive_storage:

Public Member Functions
void	add (const weighted_point &point)

	compressive_storage (const std::string &name, const clustering_config &config)

wplist	get_mine () const

void	set_compressor (jubatus::util::lang::shared_ptr< compressor::compressor > compressor)

Public Member Functions inherited from jubatus::core::clustering::storage
void	clear ()

virtual wplist	get_all () const

virtual wplist	get_common () const

void	get_diff (diff_t &d) const

size_t	get_revision ()

core::storage::version	get_version () const

void	mix (const diff_t &, diff_t &ret)

	MSGPACK_DEFINE (revision_, name_, config_, common_)

void	pack (framework::packer &packer) const

bool	put_diff (const diff_t &d)

	storage (const std::string &name, const clustering_config &config)

void	unpack (msgpack::object o)

virtual	~storage ()

Public Member Functions inherited from jubatus::core::clustering::event_dispatcher< storage_event_type, wplist >
void	add_event_listener (const storage_event_type &type, callback_t callback)

virtual	~event_dispatcher ()

Private Member Functions
void	carry_up (size_t r)

void	clear_impl_ ()

void	forget_weight (wplist &points)

bool	is_next_bucket_full (size_t bucket_number)

void	pack_impl_ (framework::packer &packer) const

bool	reach_forgetting_threshold (size_t bucket_number)

void	unpack_impl_ (msgpack::object o)

Private Attributes
jubatus::util::lang::shared_ptr< compressor::compressor >	compressor_

std::vector< wplist >	mine_

uint64_t	status_

Additional Inherited Members
Protected Member Functions inherited from jubatus::core::clustering::storage
void	increment_revision ()

Protected Member Functions inherited from jubatus::core::clustering::event_dispatcher< storage_event_type, wplist >
void	dispatch (const storage_event_type &type, const wplist &data) const

Protected Attributes inherited from jubatus::core::clustering::storage
std::vector< std::pair< std::string, wplist > >	common_

clustering_config	config_

std::string	name_

size_t	revision_

Protected Attributes inherited from jubatus::core::clustering::event_dispatcher< storage_event_type, wplist >
event_type_table	events_

Detailed Description

Definition at line 31 of file compressive_storage.hpp.

Constructor & Destructor Documentation

jubatus::core::clustering::compressive_storage::compressive_storage	(	const std::string &	name,
		const clustering_config &	config
	)

Definition at line 29 of file compressive_storage.cpp.

References mine_.

     : storage(name, config),
       status_(0) {
   mine_.push_back(wplist());
 }

Member Function Documentation

void jubatus::core::clustering::compressive_storage::add ( const weighted_point & point )

virtual

Implements jubatus::core::clustering::storage.

Definition at line 42 of file compressive_storage.cpp.

References jubatus::core::clustering::clustering_config::bicriteria_base_size, jubatus::core::clustering::clustering_config::bucket_size, carry_up(), jubatus::core::clustering::clustering_config::compressed_bucket_size, compressor_, jubatus::core::clustering::storage::config_, jubatus::core::clustering::storage::increment_revision(), mine_, and status_.

                                                          {
   wplist& c0 = mine_[0];
   c0.push_back(point);
   if (c0.size() >= static_cast<size_t>(config_.bucket_size)) {
     wplist cr;
     compressor_->compress(
         c0, config_.bicriteria_base_size, config_.compressed_bucket_size, cr);
     c0.swap(cr);
     status_ += 1;
     carry_up(0);
 
     increment_revision();
   }
 }

Here is the call graph for this function:

void jubatus::core::clustering::compressive_storage::carry_up ( size_t r )

private

Definition at line 88 of file compressive_storage.cpp.

References jubatus::core::clustering::clustering_config::bicriteria_base_size, jubatus::core::clustering::clustering_config::compressed_bucket_size, compressor_, jubatus::core::clustering::concat(), jubatus::core::clustering::storage::config_, forget_weight(), get_mine(), is_next_bucket_full(), mine_, and reach_forgetting_threshold().

Referenced by add().

                                            {
   if (r >= mine_.size() - 1) {
     mine_.push_back(wplist());
   }
   forget_weight(mine_[r]);
   if (!is_next_bucket_full(r)) {
     if (!reach_forgetting_threshold(r + 1) ||
         mine_[r].size() == get_mine().size()) {
       concat(mine_[r], mine_[r + 1]);
       mine_[r].clear();
     } else {
       mine_[r + 1].swap(mine_[r]);
       mine_[r].clear();
     }
   } else {
     wplist cr = mine_[r];
     wplist crr = mine_[r + 1];
     mine_[r].clear();
     mine_[r + 1].clear();
     concat(cr, crr);
     size_t dstsize = (r == 0) ? config_.compressed_bucket_size :
         2 * r * r * config_.compressed_bucket_size;
     compressor_->compress(crr, config_.bicriteria_base_size,
                           dstsize, mine_[r + 1]);
     carry_up(r + 1);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void jubatus::core::clustering::compressive_storage::clear_impl_ ( )

privatevirtual

Reimplemented from jubatus::core::clustering::storage.

Definition at line 136 of file compressive_storage.cpp.

References jubatus::core::clustering::storage::clear_impl_(), mine_, and status_.

                                       {
   storage::clear_impl_();
   mine_.clear();
   mine_.push_back(wplist());
   status_ = 0;
 }

Here is the call graph for this function:

void jubatus::core::clustering::compressive_storage::forget_weight ( wplist & points )

private

Definition at line 66 of file compressive_storage.cpp.

References jubatus::core::clustering::storage::config_, and jubatus::core::clustering::clustering_config::forgetting_factor.

Referenced by carry_up().

                                                       {
   double factor = std::exp(-config_.forgetting_factor);
   typedef wplist::iterator iter;
   for (iter it = points.begin(); it != points.end(); ++it) {
     it->weight *= factor;
   }
 }

Here is the caller graph for this function:

wplist jubatus::core::clustering::compressive_storage::get_mine ( ) const

virtual

Implements jubatus::core::clustering::storage.

Definition at line 57 of file compressive_storage.cpp.

References jubatus::core::clustering::concat(), and mine_.

Referenced by carry_up().

                                            {
   wplist ret;
   for (std::vector<wplist>::const_iterator it = mine_.begin();
       it != mine_.end(); ++it) {
     concat(*it, ret);
   }
   return ret;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool jubatus::core::clustering::compressive_storage::is_next_bucket_full ( size_t bucket_number )

private

Definition at line 83 of file compressive_storage.cpp.

References jubatus::core::clustering::clustering_config::bucket_length, jubatus::core::clustering::storage::config_, jubatus::core::clustering::digit(), and status_.

Referenced by carry_up().

                                                                   {
   return digit(status_ - 1, bucket_number, config_.bucket_length) ==
       config_.bucket_length - 1;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void jubatus::core::clustering::compressive_storage::pack_impl_ ( framework::packer & packer ) const

privatevirtual

Reimplemented from jubatus::core::clustering::storage.

Definition at line 116 of file compressive_storage.cpp.

References compressor_, mine_, jubatus::core::clustering::storage::pack_impl_(), and status_.

                                                                   {
   packer.pack_array(4);
   storage::pack_impl_(packer);
   packer.pack(mine_);
   packer.pack(status_);
   packer.pack(*compressor_);
 }

Here is the call graph for this function:

bool jubatus::core::clustering::compressive_storage::reach_forgetting_threshold ( size_t bucket_number )

private

Definition at line 74 of file compressive_storage.cpp.

References jubatus::core::clustering::storage::config_, jubatus::core::clustering::clustering_config::forgetting_factor, and jubatus::core::clustering::clustering_config::forgetting_threshold.

Referenced by carry_up().

                                                                          {
   double C = config_.forgetting_threshold;
   double lam = config_.forgetting_factor;
   if (std::exp(-lam * bucket_number) < C) {
     return true;
   }
   return false;
 }

Here is the caller graph for this function:

void jubatus::core::clustering::compressive_storage::set_compressor ( jubatus::util::lang::shared_ptr< compressor::compressor > compressor )

Definition at line 37 of file compressive_storage.cpp.

References compressor_.

Referenced by jubatus::core::clustering::storage_factory::create().

                                                                 {
   compressor_ = compressor;
 }

Here is the caller graph for this function:

void jubatus::core::clustering::compressive_storage::unpack_impl_ ( msgpack::object o )

privatevirtual

Reimplemented from jubatus::core::clustering::storage.

Definition at line 124 of file compressive_storage.cpp.

References compressor_, mine_, status_, and jubatus::core::clustering::storage::unpack_impl_().

                                                       {
   std::vector<msgpack::object> mems;
   o.convert(&mems);
   if (mems.size() != 4) {
     throw msgpack::type_error();
   }
   storage::unpack_impl_(mems[0]);
   mems[1].convert(&mine_);
   mems[2].convert(&status_);
   mems[3].convert(compressor_.get());
 }