A remote view of LogReducer from all engines. More...

Detailed Description

A remote view of LogReducer from all engines.

Definition at line 40 of file log_reducer_ref.hpp.

#include <log_reducer_ref.hpp>

Collaboration diagram for foedus::snapshot::LogReducerRef:

[legend]

Public Member Functions
	LogReducerRef ()

	LogReducerRef (Engine *engine, uint16_t node)

uint16_t	get_id () const

std::string	to_string () const

void	clear ()

uint32_t	get_total_storage_count () const

storage::Page *	get_root_info_pages ()

void	append_log_chunk (storage::StorageId storage_id, const char *send_buffer, uint32_t log_count, uint64_t send_buffer_size, uint32_t shortest_key_length, uint32_t longest_key_length)
	Append the log entries of one storage in the given buffer to this reducer's buffer. More...

Protected Member Functions
const Snapshot &	get_cur_snapshot () const

uint64_t	get_buffer_size () const

uint32_t	get_current_buffer_index_atomic () const

void *	get_buffer (uint32_t index) const

bool	verify_log_chunk (storage::StorageId storage_id, const char *send_buffer, uint32_t log_count, uint64_t send_buffer_size) const
	used only in debug mode More...

Protected Attributes
Engine *	engine_

LogReducerControlBlock *	control_block_

void *	buffers_ [2]

storage::Page *	root_info_pages_

Friends
std::ostream &	operator<< (std::ostream &o, const LogReducerRef &v)

Constructor & Destructor Documentation

foedus::snapshot::LogReducerRef::LogReducerRef ( )

inline

Definition at line 42 of file log_reducer_ref.hpp.

References buffers_, control_block_, CXX11_NULLPTR, engine_, and root_info_pages_.

                   {
     engine_ = CXX11_NULLPTR;
     control_block_ = CXX11_NULLPTR;
     buffers_[0] = CXX11_NULLPTR;
     buffers_[1] = CXX11_NULLPTR;
     root_info_pages_ = CXX11_NULLPTR;
   }

foedus::snapshot::LogReducerRef::LogReducerRef	(	Engine *	engine,
		uint16_t	node
	)

Definition at line 36 of file log_reducer_ref.cpp.

References buffers_, control_block_, engine_, foedus::soc::SharedMemoryRepo::get_node_memory_anchors(), foedus::soc::SocManager::get_shared_memory_repo(), foedus::Engine::get_soc_manager(), foedus::soc::NodeMemoryAnchors::log_reducer_buffers_, foedus::soc::NodeMemoryAnchors::log_reducer_memory_, foedus::soc::NodeMemoryAnchors::log_reducer_root_info_pages_, and root_info_pages_.

                                                           {
   engine_ = engine;
   soc::NodeMemoryAnchors* anchors
     = engine_->get_soc_manager()->get_shared_memory_repo()->get_node_memory_anchors(node);
   control_block_ = anchors->log_reducer_memory_;
   buffers_[0] = anchors->log_reducer_buffers_[0];
   buffers_[1] = anchors->log_reducer_buffers_[1];
   root_info_pages_ = anchors->log_reducer_root_info_pages_;
 }

Here is the call graph for this function:

Member Function Documentation

void foedus::snapshot::LogReducerRef::append_log_chunk	(	storage::StorageId	storage_id,
		const char *	send_buffer,
		uint32_t	log_count,
		uint64_t	send_buffer_size,
		uint32_t	shortest_key_length,
		uint32_t	longest_key_length
	)

Append the log entries of one storage in the given buffer to this reducer's buffer.

Parameters

[in]	storage_id	all log entries are of this storage
[in]	send_buffer	contains log entries to copy
[in]	log_count	number of log entries to copy
[in]	send_buffer_size	byte count to copy
[in]	shortest_key_length	[masstree/hash] shortest key length in the log entries
[in]	longest_key_length	[masstree/hash] longest key length in the log entries

This is the interface via which mappers send log entries to reducers. Internally, this atomically changes the status of the current reducer buffer to reserve a contiguous space and then copy without blocking other mappers. If this methods hits a situation where the current buffer becomes full, this methods wakes up the reducer and lets it switch the current buffer. All log entries are contiguously copied. One block doesn't span two buffers.

Definition at line 121 of file log_reducer_ref.cpp.

                                {
   DVLOG(1) << "Appending a block of " << send_buffer_size << " bytes (" << log_count
     << " entries) to " << to_string() << "'s buffer for storage-" << storage_id;
   ASSERT_ND(verify_log_chunk(storage_id, send_buffer, log_count, send_buffer_size));
 
   debugging::RdtscWatch stop_watch;
 
   const uint64_t required_size = send_buffer_size + sizeof(FullBlockHeader);
   uint32_t buffer_index = 0;
   uint64_t begin_position = 0;
   while (true) {
     buffer_index = get_current_buffer_index_atomic();
     std::atomic<uint64_t>* status_address = control_block_->get_buffer_status_address(buffer_index);
 
     // If even the current buffer is marked as no more writers, the reducer is getting behind.
     // Mappers have to wait, potentially for a long time. So, let's just sleep.
     ReducerBufferStatus cur_status = control_block_->get_buffer_status_atomic(buffer_index);
     if (cur_status.components.flags_ & kFlagNoMoreWriters) {
       VLOG(0) << "Both buffers full in" << to_string() << ". I'll sleep for a while..";
       while (get_current_buffer_index_atomic() == buffer_index) {
         std::this_thread::sleep_for(std::chrono::milliseconds(1));
       }
       VLOG(0) << "Buffer switched in" << to_string() << " after sleep. Let's resume.";
       continue;
     }
 
     // the buffer is now full. let's mark this buffer full and
     // then wake up reducer to do switch.
     uint64_t buffer_size = get_buffer_size();
     ASSERT_ND(buffer_size
       == engine_->get_options().snapshot_.log_reducer_buffer_mb_ * (1ULL << 19));
     ASSERT_ND(from_buffer_position(cur_status.components.tail_position_) <= buffer_size);
     ASSERT_ND(reinterpret_cast<char*>(buffers_[0]) + buffer_size
       == reinterpret_cast<char*>(buffers_[1]));
     if (from_buffer_position(cur_status.components.tail_position_) + required_size > buffer_size) {
       ReducerBufferStatus new_status = cur_status;
       new_status.components.flags_ |= kFlagNoMoreWriters;
       if (!status_address->compare_exchange_strong(cur_status.word, new_status.word)) {
         // if CAS fails, someone else might have already done it. retry
         continue;
       }
 
       // Buffer switch won't be that often, so we simply wait without waking up the reducer.
       continue;
     }
 
     // okay, "looks like" we can append our log. make it sure with atomic CAS
     ReducerBufferStatus new_status = cur_status;
     ++new_status.components.active_writers_;
     new_status.components.tail_position_ += to_buffer_position(required_size);
     ASSERT_ND(from_buffer_position(cur_status.components.tail_position_) <= buffer_size);
     if (!status_address->compare_exchange_strong(cur_status.word, new_status.word)) {
       // someone else did something. retry
       continue;
     }
 
     // okay, we atomically reserved the space.
     begin_position = from_buffer_position(cur_status.components.tail_position_);
     break;
   }
 
   // now start copying. this might take a few tens of microseconds if it's 1MB and on another
   // NUMA node.
   void* buffer = get_buffer(buffer_index);
   debugging::RdtscWatch copy_watch;
   char* destination = reinterpret_cast<char*>(buffer) + begin_position;
   ASSERT_ND(begin_position + sizeof(FullBlockHeader) + send_buffer_size
     <= engine_->get_options().snapshot_.log_reducer_buffer_mb_ * (1ULL << 19));
   FullBlockHeader header;
   header.storage_id_ = storage_id;
   header.log_count_ = log_count;
   header.block_length_ = to_buffer_position(required_size);
   header.magic_word_ = BlockHeaderBase::kFullBlockHeaderMagicWord;
   header.shortest_key_length_ = shortest_key_length;
   header.longest_key_length_ = longest_key_length;
   std::memcpy(destination, &header, sizeof(FullBlockHeader));
   std::memcpy(destination + sizeof(FullBlockHeader), send_buffer, send_buffer_size);
   copy_watch.stop();
   DVLOG(1) << "memcpy of " << send_buffer_size << " bytes took "
     << copy_watch.elapsed() << " cycles";
 
   // done, let's decrement the active_writers_ to declare we are done.
   while (true) {
     std::atomic<uint64_t>* status_address = control_block_->get_buffer_status_address(buffer_index);
     ReducerBufferStatus cur_status = control_block_->get_buffer_status_atomic(buffer_index);
     ReducerBufferStatus new_status = cur_status;
     ASSERT_ND(new_status.components.active_writers_ > 0);
     --new_status.components.active_writers_;
     if (!status_address->compare_exchange_strong(cur_status.word, new_status.word)) {
       // if CAS fails, someone else might have already done it. retry
       continue;
     }
 
     // okay, decremented. let's exit.
 
     // Disabled. for now the reducer does spin. so no need for wakeup
     // if (new_status.components.active_writers_ == 0
     //   && (new_status.components.flags_ & kFlagNoMoreWriters)) {
     //   // if this was the last writer and the buffer was already closed for new writers,
     //   // the reducer might be waiting for us. let's wake her up
     //   thread_.wakeup();
     // }
     break;
   }
 
   stop_watch.stop();
   DVLOG(1) << "Completed appending a block of " << send_buffer_size << " bytes to " << to_string()
     << "'s buffer for storage-" << storage_id << " in " << stop_watch.elapsed() << " cycles";
   ASSERT_ND(reinterpret_cast<FullBlockHeader*>(destination)->is_full_block());
   ASSERT_ND(reinterpret_cast<FullBlockHeader*>(destination)->storage_id_ == storage_id);
   ASSERT_ND(reinterpret_cast<FullBlockHeader*>(destination)->block_length_
     == to_buffer_position(required_size));
 }

Here is the call graph for this function:

void foedus::snapshot::LogReducerRef::clear ( )

Definition at line 53 of file log_reducer_ref.cpp.

References foedus::snapshot::LogReducerControlBlock::clear(), and control_block_.

                           {
   control_block_->clear();
 }

Here is the call graph for this function:

void * foedus::snapshot::LogReducerRef::get_buffer ( uint32_t index ) const

protected

Definition at line 71 of file log_reducer_ref.cpp.

References engine_, get_id(), foedus::soc::SharedMemoryRepo::get_node_memory_anchors(), foedus::soc::SocManager::get_shared_memory_repo(), foedus::Engine::get_soc_manager(), and foedus::soc::NodeMemoryAnchors::log_reducer_buffers_.

Referenced by append_log_chunk().

                                                           {
   soc::NodeMemoryAnchors* anchors
     = engine_->get_soc_manager()->get_shared_memory_repo()->get_node_memory_anchors(get_id());
   return anchors->log_reducer_buffers_[index % 2];
 }

Here is the call graph for this function:

Here is the caller graph for this function:

uint64_t foedus::snapshot::LogReducerRef::get_buffer_size ( ) const

protected

Definition at line 65 of file log_reducer_ref.cpp.

References engine_, foedus::Engine::get_options(), foedus::snapshot::SnapshotOptions::log_reducer_buffer_mb_, and foedus::EngineOptions::snapshot_.

Referenced by append_log_chunk().

                                                  {
   // the value is in total of two buffers. (<< 20) / 2 == << 19
   return static_cast<uint64_t>(engine_->get_options().snapshot_.log_reducer_buffer_mb_) << 19;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const Snapshot & foedus::snapshot::LogReducerRef::get_cur_snapshot ( ) const

protected

Definition at line 77 of file log_reducer_ref.cpp.

References engine_, foedus::soc::SharedMemoryRepo::get_global_memory_anchors(), foedus::soc::SocManager::get_shared_memory_repo(), and foedus::Engine::get_soc_manager().

Referenced by verify_log_chunk().

                                                       {
   return engine_->get_soc_manager()->get_shared_memory_repo()->get_global_memory_anchors()->
     snapshot_manager_memory_->gleaner_.cur_snapshot_;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

uint32_t foedus::snapshot::LogReducerRef::get_current_buffer_index_atomic ( ) const

protected

Definition at line 61 of file log_reducer_ref.cpp.

References control_block_, and foedus::snapshot::LogReducerControlBlock::current_buffer_.

Referenced by append_log_chunk().

                                                                  {
   return control_block_->current_buffer_;
 }

Here is the caller graph for this function:

uint16_t foedus::snapshot::LogReducerRef::get_id ( ) const

Definition at line 46 of file log_reducer_ref.cpp.

References control_block_, and foedus::snapshot::LogReducerControlBlock::id_.

Referenced by get_buffer(), and to_string().

                                         {
   return control_block_->id_;
 }

Here is the caller graph for this function:

storage::Page* foedus::snapshot::LogReducerRef::get_root_info_pages ( )

inline

Definition at line 55 of file log_reducer_ref.hpp.

References root_info_pages_.

55 { return root_info_pages_; }

foedus::snapshot::LogReducerRef::root_info_pages_

storage::Page * root_info_pages_

Definition: log_reducer_ref.hpp:97

uint32_t foedus::snapshot::LogReducerRef::get_total_storage_count ( ) const

Definition at line 57 of file log_reducer_ref.cpp.

References control_block_, and foedus::snapshot::LogReducerControlBlock::total_storage_count_.

                                                       {
   return control_block_->total_storage_count_;
 }

std::string foedus::snapshot::LogReducerRef::to_string ( ) const

Definition at line 49 of file log_reducer_ref.cpp.

References get_id().

Referenced by append_log_chunk().

                                          {
   return std::string("LogReducer-") + std::to_string(get_id());
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool foedus::snapshot::LogReducerRef::verify_log_chunk	(	storage::StorageId	storage_id,
		const char *	send_buffer,
		uint32_t	log_count,
		uint64_t	send_buffer_size
	)		const

protected

used only in debug mode

Definition at line 82 of file log_reducer_ref.cpp.

References ASSERT_ND, foedus::snapshot::Snapshot::base_epoch_, get_cur_snapshot(), foedus::xct::XctId::get_epoch(), foedus::log::LogHeader::get_kind(), foedus::xct::XctId::get_ordinal(), foedus::log::LogHeader::get_type(), foedus::log::BaseLogType::header_, foedus::snapshot::Snapshot::id_, foedus::Epoch::is_valid(), foedus::xct::XctId::is_valid(), foedus::log::kLogCodeEpochMarker, foedus::log::kLogCodeFiller, foedus::log::kLogCodeInvalid, foedus::snapshot::kNullSnapshotId, foedus::log::kRecordLogs, foedus::log::LogHeader::log_length_, foedus::snapshot::Snapshot::max_storage_id_, foedus::log::LogHeader::storage_id_, foedus::snapshot::Snapshot::valid_until_epoch_, and foedus::log::LogHeader::xct_id_.

Referenced by append_log_chunk().

                                    {
   uint32_t real_log_count = 0;
   uint64_t cur = 0;
   const Snapshot& cur_snapshot = get_cur_snapshot();
   ASSERT_ND(cur_snapshot.id_ != kNullSnapshotId);
   ASSERT_ND(cur_snapshot.valid_until_epoch_.is_valid());
   ASSERT_ND(cur_snapshot.max_storage_id_ > 0);
   ASSERT_ND(cur_snapshot.max_storage_id_ >= storage_id);
   while (cur < send_buffer_size) {
     const log::BaseLogType* entry = reinterpret_cast<const log::BaseLogType*>(send_buffer + cur);
     log::LogCode type = entry->header_.get_type();
     log::LogCodeKind kind = entry->header_.get_kind();
     ASSERT_ND(type != log::kLogCodeInvalid);
     ASSERT_ND(type != log::kLogCodeEpochMarker);  // should have been skipped in mapper.
     ASSERT_ND(entry->header_.log_length_ > 0);
     ASSERT_ND(entry->header_.log_length_ % 8 == 0);
     cur += entry->header_.log_length_;
     ++real_log_count;
     if (type == log::kLogCodeFiller) {
       continue;
     } else {
       ASSERT_ND(entry->header_.storage_id_ == storage_id);
       ASSERT_ND(entry->header_.xct_id_.is_valid());
       ASSERT_ND(entry->header_.xct_id_.get_ordinal() > 0);
       ASSERT_ND(kind == log::kRecordLogs);
       Epoch epoch = entry->header_.xct_id_.get_epoch();
       ASSERT_ND(!cur_snapshot.base_epoch_.is_valid() || epoch > cur_snapshot.base_epoch_);
       ASSERT_ND(epoch <= cur_snapshot.valid_until_epoch_);
     }
   }
   ASSERT_ND(real_log_count == log_count);
   ASSERT_ND(cur == send_buffer_size);
   return true;
 }