#include <mixable_versioned_table.hpp>

Inheritance diagram for jubatus::core::framework::mixable_versioned_table:

Collaboration diagram for jubatus::core::framework::mixable_versioned_table:

Public Types
typedef jubatus::util::lang::shared_ptr< storage::column_table >	model_ptr

Public Member Functions
void	clear ()

framework::diff_object	convert_diff_object (const msgpack::object &) const

void	get_argument (framework::packer &pk) const

void	get_diff (framework::packer &) const

model_ptr	get_model () const

storage::version	get_version () const

void	mix (const msgpack::object &obj, framework::diff_object) const

void	pull (const msgpack::object &arg, framework::packer &pk) const

void	push (const msgpack::object &diff)

bool	put_diff (const framework::diff_object &obj)

void	set_model (model_ptr m)

void	set_unlearner (jubatus::util::lang::shared_ptr< unlearner::unlearner_base > unlearner)

Public Member Functions inherited from jubatus::core::framework::push_mixable
	push_mixable ()

virtual	~push_mixable ()

Public Member Functions inherited from jubatus::core::framework::mixable
	mixable ()

	mixable (const std::string &name)

virtual std::set< std::string >	mixables () const

virtual	~mixable ()

Public Member Functions inherited from jubatus::core::framework::linear_mixable
	linear_mixable ()

virtual	~linear_mixable ()

Private Member Functions
void	pull_impl (const version_clock &vc, framework::packer &) const

void	push_impl (const msgpack::object &)

void	update_version (const storage::column_table::version_t &version)

Private Attributes
model_ptr	model_

jubatus::util::lang::shared_ptr< unlearner::unlearner_base >	unlearner_

version_clock	vc_

Additional Inherited Members
Protected Attributes inherited from jubatus::core::framework::mixable
std::set< std::string >	mixables_

Detailed Description

Definition at line 35 of file mixable_versioned_table.hpp.

Member Typedef Documentation

typedef jubatus::util::lang::shared_ptr<storage::column_table> jubatus::core::framework::mixable_versioned_table::model_ptr

Definition at line 38 of file mixable_versioned_table.hpp.

Member Function Documentation

void jubatus::core::framework::mixable_versioned_table::clear ( )

inline

Definition at line 68 of file mixable_versioned_table.hpp.

68 {}

framework::diff_object jubatus::core::framework::mixable_versioned_table::convert_diff_object ( const msgpack::object & obj ) const

virtual

Implements jubatus::core::framework::linear_mixable.

Definition at line 60 of file mixable_versioned_table.cpp.

References JUBATUS_EXCEPTION.

                                     {
   if (obj.type != msgpack::type::RAW) {
     throw JUBATUS_EXCEPTION(
         core::common::exception::runtime_error("bad diff_object"));
   }
   internal_diff* diff = new internal_diff;
   diff_object diff_obj(diff);
 
   msgpack::unpacked msg;
   msgpack::unpack(&msg, obj.via.raw.ptr, obj.via.raw.size);
 
   msg.get().convert(&diff->objs);
   if (!diff->objs.empty()) {
     shared_ptr<msgpack::zone> owner(msg.zone().release());
     diff->zones.push_back(owner);
   }
   return diff_obj;
 }

void jubatus::core::framework::mixable_versioned_table::get_argument ( framework::packer & pk ) const

virtual

Implements jubatus::core::framework::push_mixable.

Definition at line 139 of file mixable_versioned_table.cpp.

References vc_.

                                                                     {
   pk.pack(vc_);
 }

void jubatus::core::framework::mixable_versioned_table::get_diff ( framework::packer & pk ) const

virtual

Implements jubatus::core::framework::linear_mixable.

Definition at line 80 of file mixable_versioned_table.cpp.

References pull_impl(), and vc_.

                                                                 {
   msgpack::sbuffer data;
   core::framework::stream_writer<msgpack::sbuffer> st(data);
   core::framework::jubatus_packer jp(st);
   core::framework::packer p(jp);
   pull_impl(vc_, p);
 
   // Wrap msgpack binary more for holding msgpack::zone in internal diff_object.
   pk.pack_raw(data.size());
   pk.pack_raw_body(data.data(), data.size());
 }

Here is the call graph for this function:

model_ptr jubatus::core::framework::mixable_versioned_table::get_model ( ) const

inline

Definition at line 64 of file mixable_versioned_table.hpp.

References model_.

Referenced by pull_impl(), and push_impl().

                               {
     return model_;
   }

Here is the caller graph for this function:

storage::version jubatus::core::framework::mixable_versioned_table::get_version ( ) const

inlinevirtual

Reimplemented from jubatus::core::framework::mixable.

Definition at line 56 of file mixable_versioned_table.hpp.

                                      {
     return storage::version();
   }

void jubatus::core::framework::mixable_versioned_table::mix	(	const msgpack::object &	obj,
		framework::diff_object	ptr
	)		const

virtual

Implements jubatus::core::framework::linear_mixable.

Definition at line 109 of file mixable_versioned_table.cpp.

References JUBATUS_EXCEPTION.

                                     {
   internal_diff* diff_obj = dynamic_cast<internal_diff*>(ptr.get());
   if (!diff_obj) {
     throw JUBATUS_EXCEPTION(
         core::common::exception::runtime_error("bad diff_object"));
   }
 
   if (obj.type != msgpack::type::RAW) {
     throw JUBATUS_EXCEPTION(
         core::common::exception::runtime_error("bad diff_object"));
   }
   msgpack::unpacked msg;
   msgpack::unpack(&msg, obj.via.raw.ptr, obj.via.raw.size);
   msgpack::object o = msg.get();
   if (o.type != msgpack::type::ARRAY) {
     throw JUBATUS_EXCEPTION(
         core::common::exception::runtime_error("bad diff_object"));
   }
   for (size_t i = 0; i < o.via.array.size; i++) {
     diff_obj->objs.push_back(o.via.array.ptr[i]);
   }
 
   if (o.via.array.size > 0) {
     shared_ptr<msgpack::zone> owner(msg.zone().release());
     diff_obj->zones.push_back(owner);
   }
 }

void jubatus::core::framework::mixable_versioned_table::pull	(	const msgpack::object &	arg,
		framework::packer &	pk
	)		const

virtual

Implements jubatus::core::framework::push_mixable.

Definition at line 143 of file mixable_versioned_table.cpp.

References pull_impl().

                                {
   version_clock vc;
   arg.convert(&vc);
   pull_impl(vc, pk);
 }

Here is the call graph for this function:

void jubatus::core::framework::mixable_versioned_table::pull_impl	(	const version_clock &	vc,
		framework::packer &	pk
	)		const

private

Definition at line 155 of file mixable_versioned_table.cpp.

References get_model().

Referenced by get_diff(), and pull().

                                                         {
 
   model_ptr table = get_model();
   const uint64_t table_size = table->size();
 
   size_t pack_size = 0;
   for (uint64_t i = 0; i < table_size; ++i) {
     const version_t version = table->get_version(i);
     version_clock::const_iterator it = vc.find(version.first);
     if (it == vc.end() || it->second < version.second) {
       pack_size++;
     }
   }
   pk.pack_array(pack_size);
 
   for (uint64_t i = 0; i < table_size; ++i) {
     const version_t version = table->get_version(i);
     version_clock::const_iterator it = vc.find(version.first);
     if (it == vc.end() || it->second < version.second) {
       table->get_row(i, pk);
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void jubatus::core::framework::mixable_versioned_table::push ( const msgpack::object & diff )

virtual

Implements jubatus::core::framework::push_mixable.

Definition at line 151 of file mixable_versioned_table.cpp.

References push_impl().

                                                             {
   push_impl(diff);
 }

Here is the call graph for this function:

void jubatus::core::framework::mixable_versioned_table::push_impl ( const msgpack::object & o )

private

Definition at line 180 of file mixable_versioned_table.cpp.

References get_model(), JUBATUS_EXCEPTION, and update_version().

Referenced by push(), and put_diff().

                             {
   model_ptr table = get_model();
   if (o.type != msgpack::type::ARRAY) {
     throw JUBATUS_EXCEPTION(
         core::common::exception::runtime_error("bad diff_object"));
   }
   for (uint64_t i = 0; i < o.via.array.size; ++i) {
     const version_t version = table->set_row(o.via.array.ptr[i]);
     update_version(version);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool jubatus::core::framework::mixable_versioned_table::put_diff ( const framework::diff_object & obj )

virtual

Implements jubatus::core::framework::linear_mixable.

Definition at line 92 of file mixable_versioned_table.cpp.

References JUBATUS_EXCEPTION, and push_impl().

                                                                       {
   internal_diff* diff_obj = dynamic_cast<internal_diff*>(ptr.get());
   if (!diff_obj) {
     throw JUBATUS_EXCEPTION(
         core::common::exception::runtime_error("bad diff_object"));
   }
 
   msgpack::object obj;
   obj.type = msgpack::type::ARRAY;
   obj.via.array.ptr = &diff_obj->objs[0];
   obj.via.array.size = diff_obj->objs.size();
 
   push_impl(obj);
 
   return true;
 }

Here is the call graph for this function:

void jubatus::core::framework::mixable_versioned_table::set_model ( model_ptr m )

inline

Definition at line 60 of file mixable_versioned_table.hpp.

References model_.

                               {
     model_ = m;
   }

void jubatus::core::framework::mixable_versioned_table::set_unlearner ( jubatus::util::lang::shared_ptr< unlearner::unlearner_base > unlearner )

inline

Definition at line 40 of file mixable_versioned_table.hpp.

References unlearner, and unlearner_.

                                                                     {
     unlearner_ = unlearner;
   }

void jubatus::core::framework::mixable_versioned_table::update_version ( const storage::column_table::version_t & version )

private

Definition at line 193 of file mixable_versioned_table.cpp.

References vc_.

Referenced by push_impl().

                                                                      {
   version_clock::iterator it = vc_.find(version.first);
   if (it == vc_.end()) {
     vc_.insert(version);
   } else if (it->second < version.second) {
     it->second = version.second;
   }
 }