foedus::snapshot::LogReducer Class Reference

A log reducer, which receives log entries sent from mappers and applies them to construct new snapshot files. More...

Detailed Description

A log reducer, which receives log entries sent from mappers and applies them to construct new snapshot files.

Overview

Reducers receive log entries from mappers and apply them to new snapshot files.

Sorting

The log entries are sorted in a few steps to be processed efficiently and simply. Sorting starts when one of the reducer's buffer becomes full. Reducer never starts sorting until that to maximize the benefits of batch-processing (this design might be revisited later, though). Reducers maintain two buffers to let mappers keep sending data while reducers are sorting and dumping to temporary files.

Storage Sorting

The first step is to sort log entries by storage, which is done in mappers. We process all log entries of one storage together. This has a benefit of code simplicity and less D-cache misses. We don't actually sort in this case because we don't care the order between storages. Thus, we use hashmap-like structure in mappers to sort based on storage-id.

Upon receiving a chunk of data from mappers, the reducer has to collect all of them to do the following (otherwise the sorting is incomplete). This is done by simply reading all block headers utilizing the block_length_ property. Assuming each block is sufficiently large, this jumping cost on DRAM should be negligible. If each block is not sufficiently large, there are anyway other performance issues.

Key and Ordinal Sorting

Then, in each storage, we sort logs by keys and then by ordinal (*). The algorithm to do this sorting depends on the storage type (eg Array, Masstree) because some storage has a VERY efficient way to do this. We exploit the fact that this sorting occurs only per storage, just passing the whole log entries for the storage to storage-specific logic defined in foedus::storage::Partitioner. This is another reason to sort by storage first.

(*) We do need to sort by ordinal. Otherwise correct result is not guaranteed. For example, imagine the following case:

• UPDATE rec-1 to A. Log-ordinal 1.
• UPDATE rec-1 to B. Log-ordinal 2. Ordinal-1 must be processed before ordinal 2.

For more details, see foedus::storage::Partitioner.

Dumping Logs and and Merging

After the sorting, the reducer dumps the buffer to a file. When all logs are received, the reducer does merge-sort on top of the sorted run files.

Compacting Logs

In some cases, we can delete log entries for the same keys. For example, when we have two logs for the same key like the example above, we can safely omit the first log with ordinal 1 AS FAR AS both logs appear in the same reducer buffer and updated byte positions in the record are the same. Another example is updates followed by a deletion.

This compaction is especially useful for a record that is repeatedly updated/inserted/deleted, such as TPC-C's WAREHOUSE/DISTRICT records, where several thousands of overwrite-logs in each reducer buffer will be compacted into just one log.

See foedus::storage::Partitioner::sort_batch() for more details.

Data Pages

One tricky thing in reducer is how it manages data pages to read previous snapshot pages and apply the new logs. So far, we assume each reducer allocates a sufficient amount of DRAM to hold all pages it read/write during one snapshotting. If this doesn't hold, we might directly allocate pages on NVRAM and read/write there.

Note

This is a private implementation-details of Snapshot Manager, thus file name ends with _impl. Do not include this header from a client program. There is no case client program needs to access this internal class.

Todo:

This class got a bit bloated and hard to do a whitebox test because of dependencies to other modules. Dump-part and merge-part should be separated into its own classes in a way testcases can independently test them. Maybe reducer should be its own package?

Definition at line 293 of file log_reducer_impl.hpp.

#include <log_reducer_impl.hpp>

Inheritance diagram for foedus::snapshot::LogReducer:

Collaboration diagram for foedus::snapshot::LogReducer:

Public Member Functions
	LogReducer (Engine *engine)
ErrorStack	initialize_once () override
ErrorStack	uninitialize_once () override
std::string	to_string () const override
	Expects "LogReducer-x", "LogMapper-y" etc. More...
Public Member Functions inherited from foedus::snapshot::MapReduceBase
	MapReduceBase (Engine *engine, uint16_t id)
	MapReduceBase ()=delete
	MapReduceBase (const MapReduceBase &other)=delete
MapReduceBase &	operator= (const MapReduceBase &other)=delete
LogGleanerRef *	get_parent ()
uint16_t	get_id () const
uint16_t	get_numa_node () const
void	launch_thread ()
	Start executing. More...
void	join_thread ()
Public Member Functions inherited from foedus::DefaultInitializable
	DefaultInitializable ()
virtual	~DefaultInitializable ()
	DefaultInitializable (const DefaultInitializable &)=delete
DefaultInitializable &	operator= (const DefaultInitializable &)=delete
ErrorStack	initialize () override final
	Typical implementation of Initializable::initialize() that provides initialize-once semantics. More...
ErrorStack	uninitialize () override final
	Typical implementation of Initializable::uninitialize() that provides uninitialize-once semantics. More...
bool	is_initialized () const override final
	Returns whether the object has been already initialized or not. More...
Public Member Functions inherited from foedus::Initializable
virtual	~Initializable ()

Protected Member Functions
ErrorStack	handle_process () override
	Implements the specific logics in derived class. More...
Protected Member Functions inherited from foedus::snapshot::MapReduceBase
ErrorCode	check_cancelled () const
	Derived class's handle_process() should occasionally call this to exit if it's cancelled. More...

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

Additional Inherited Members
Protected Attributes inherited from foedus::snapshot::MapReduceBase
Engine *const	engine_
LogGleanerRef	parent_
const uint16_t	id_
	Unique ID of this mapper or reducer. More...
const uint16_t	numa_node_
std::atomic< bool >	running_
	only for sanity check More...
std::thread	thread_

Constructor & Destructor Documentation

foedus::snapshot::LogReducer::LogReducer ( Engine *  engine )

explicit

Definition at line 52 of file log_reducer_impl.cpp.

References ASSERT_ND, foedus::snapshot::MapReduceBase::engine_, foedus::Engine::get_options(), foedus::soc::SocManager::get_shared_memory_repo(), foedus::Engine::get_soc_manager(), foedus::snapshot::SnapshotOptions::log_reducer_buffer_mb_, foedus::soc::NodeMemoryAnchors::log_reducer_buffers_, foedus::soc::NodeMemoryAnchors::log_reducer_memory_, foedus::soc::NodeMemoryAnchors::log_reducer_root_info_pages_, foedus::snapshot::MapReduceBase::numa_node_, and foedus::EngineOptions::snapshot_.

: MapReduceBase(engine, engine->get_soc_id()),
  previous_snapshot_files_(engine_),
  sorted_runs_(0) {
  soc::NodeMemoryAnchors* anchors = engine->get_soc_manager()->get_shared_memory_repo()->
    get_node_memory_anchors(numa_node_);
  control_block_ = anchors->log_reducer_memory_;
  buffers_[0] = anchors->log_reducer_buffers_[0];
  buffers_[1] = anchors->log_reducer_buffers_[1];
  const SnapshotOptions& option = engine_->get_options().snapshot_;
  buffer_half_size_bytes_ = static_cast<uint64_t>(option.log_reducer_buffer_mb_) << 19;
  ASSERT_ND(reinterpret_cast<char*>(buffers_[0]) + buffer_half_size_bytes_
    == reinterpret_cast<char*>(buffers_[1]));
  root_info_pages_ = anchors->log_reducer_root_info_pages_;
}

Here is the call graph for this function:

Member Function Documentation

ErrorStack foedus::snapshot::LogReducer::handle_process ( )

overrideprotectedvirtual

Implements the specific logics in derived class.

Implements foedus::snapshot::MapReduceBase.

Definition at line 134 of file log_reducer_impl.cpp.

References ASSERT_ND, foedus::snapshot::MapReduceBase::check_cancelled(), CHECK_ERROR, foedus::snapshot::LogReducerControlBlock::current_buffer_, foedus::snapshot::LogReducerControlBlock::get_current_buffer_status(), foedus::snapshot::LogReducerControlBlock::get_non_current_buffer_status(), foedus::snapshot::LogGleanerRef::is_all_mappers_completed(), foedus::snapshot::ReducerBufferStatus::is_clear(), foedus::snapshot::ReducerBufferStatus::is_no_more_writers(), foedus::kRetOk, foedus::snapshot::MapReduceBase::parent_, to_string(), and WRAP_ERROR_CODE.

                                      {
  while (true) {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    WRAP_ERROR_CODE(check_cancelled());
    if (parent_.is_all_mappers_completed()) {
      break;
    }
    // should I switch the current buffer?
    // this is while, not if, in case the new current buffer becomes full while this reducer is
    // dumping the old current buffer.
    while (control_block_->get_current_buffer_status().is_no_more_writers()) {
      WRAP_ERROR_CODE(check_cancelled());
      // okay, let's switch now. As this thread dumps the buffer as soon as this happens,
      // only one of the buffers can be full.
      if (!control_block_->get_non_current_buffer_status().is_clear()) {
        LOG(FATAL) << to_string() << " wtf. both buffers are in use, can't happen";
      }
      LOG(INFO) << to_string() << " switching buffer. current_buffer_="
        << control_block_->current_buffer_;
      control_block_->current_buffer_.fetch_add(1U);
      ASSERT_ND(sorted_runs_ + 1U == control_block_->current_buffer_);
      // Then, immediately start dumping the full buffer.
      CHECK_ERROR(dump_buffer());
    }
  }

  LOG(INFO) << to_string() << " all mappers are done, this reducer starts the merge-sort phase.";
  ASSERT_ND(parent_.is_all_mappers_completed());
  WRAP_ERROR_CODE(check_cancelled());
  CHECK_ERROR(merge_sort());

  LOG(INFO) << to_string() << " all done.";
  return kRetOk;
}

Here is the call graph for this function:

ErrorStack foedus::snapshot::LogReducer::initialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 68 of file log_reducer_impl.cpp.

References foedus::memory::AlignedMemory::alloc(), ASSERT_ND, CHECK_ERROR, foedus::snapshot::MapReduceBase::engine_, foedus::snapshot::MapReduceBase::get_numa_node(), foedus::Engine::get_options(), foedus::memory::AlignedMemory::get_size(), foedus::Engine::get_soc_id(), foedus::snapshot::LogReducerControlBlock::id_, foedus::snapshot::LogReducerControlBlock::initialize(), foedus::DefaultInitializable::initialize(), foedus::memory::AlignedMemory::is_null(), foedus::memory::kHugepageSize, foedus::memory::AlignedMemory::kNumaAllocOnnode, foedus::kRetOk, foedus::snapshot::SnapshotOptions::log_reducer_dump_io_buffer_mb_, foedus::snapshot::MapReduceBase::numa_node_, foedus::EngineOptions::snapshot_, foedus::snapshot::SnapshotOptions::snapshot_writer_intermediate_pool_size_mb_, and foedus::snapshot::SnapshotOptions::snapshot_writer_page_pool_size_mb_.

                                       {
  control_block_->initialize();
  control_block_->id_ = engine_->get_soc_id();

  const SnapshotOptions& option = engine_->get_options().snapshot_;

  uint64_t dump_buffer_size = static_cast<uint64_t>(option.log_reducer_dump_io_buffer_mb_) << 20;
  dump_io_buffer_.alloc(
    dump_buffer_size,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    get_numa_node());
  ASSERT_ND(!dump_io_buffer_.is_null());

  // start from 1/16 of the main buffer. Should be big enough.
  sort_buffer_.alloc(
    buffer_half_size_bytes_ >> 5,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    get_numa_node());

  // start from 1/16 of the main buffer. Should be big enough.
  positions_buffers_.alloc(
    buffer_half_size_bytes_ >> 5,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    get_numa_node());
  input_positions_slice_ = memory::AlignedMemorySlice(
    &positions_buffers_,
    0,
    positions_buffers_.get_size() >> 1);
  output_positions_slice_ = memory::AlignedMemorySlice(
    &positions_buffers_,
    positions_buffers_.get_size() >> 1,
    positions_buffers_.get_size() >> 1);

  writer_pool_memory_.alloc(
    static_cast<uint64_t>(option.snapshot_writer_page_pool_size_mb_) << 20,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node_);

  writer_intermediate_memory_.alloc(
    static_cast<uint64_t>(option.snapshot_writer_intermediate_pool_size_mb_) << 20,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node_),

  sorted_runs_ = 0;

  CHECK_ERROR(previous_snapshot_files_.initialize());
  return kRetOk;
}

Here is the call graph for this function:

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

inlineoverridevirtual

Expects "LogReducer-x", "LogMapper-y" etc.

Used only for logging/debugging.

Implements foedus::snapshot::MapReduceBase.

Definition at line 300 of file log_reducer_impl.hpp.

References foedus::snapshot::MapReduceBase::id_.

Referenced by handle_process().

{ return std::string("Reducer-") + std::to_string(id_); }

Here is the caller graph for this function:

ErrorStack foedus::snapshot::LogReducer::uninitialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 122 of file log_reducer_impl.cpp.

References foedus::ErrorStackBatch::emprace_back(), foedus::memory::AlignedMemory::release_block(), SUMMARIZE_ERROR_BATCH, foedus::snapshot::LogReducerControlBlock::uninitialize(), and foedus::DefaultInitializable::uninitialize().

                                         {
  ErrorStackBatch batch;
  batch.emprace_back(previous_snapshot_files_.uninitialize());
  writer_intermediate_memory_.release_block();
  writer_pool_memory_.release_block();
  dump_io_buffer_.release_block();
  sort_buffer_.release_block();
  positions_buffers_.release_block();
  control_block_->uninitialize();
  return SUMMARIZE_ERROR_BATCH(batch);
}

Here is the call graph for this function:

Friends And Related Function Documentation

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

friend

Definition at line 912 of file log_reducer_impl.cpp.

                                                           {
  o << "<LogReducer>"
    << "<id_>" << v.get_id() << "</id_>"
    << "<numa_node>" << v.get_numa_node() << "</numa_node>"
    << "<total_storage_count>" << v.control_block_->total_storage_count_ << "</total_storage_count>"
    << "<sort_buffer_>" << v.sort_buffer_ << "</sort_buffer_>"
    << "<positions_buffers_>" << v.positions_buffers_ << "</positions_buffers_>"
    << "<current_buffer_>" << v.control_block_->current_buffer_ << "</current_buffer_>"
    << "<sorted_runs_>" << v.sorted_runs_ << "</sorted_runs_>"
    << "</LogReducer>";
  return o;
}

---

The documentation for this class was generated from the following files:

• /home/shino/foedus_code/foedus-core/include/foedus/snapshot/log_reducer_impl.hpp
• /home/shino/foedus_code/foedus-core/src/foedus/snapshot/log_reducer_impl.cpp