foedus::snapshot::LogMapper Class Reference

A log mapper, which reads log files from one logger and sends them to corresponding log reducers. More...

Detailed Description

A log mapper, which reads log files from one logger and sends them to corresponding log reducers.

Overview

Mappers read logs per epoch. As log files are guaranteed to be strictly ordered by epoch (see Logger code), we can simply read log files sequentially to achieve this.

Mappers send logs to partitions as follows:

• Engine-wide and Storage-wide logs (eg DROP STORAGE) are centrally processed at the end of epoch. So, mappers just buffer them and send all of them back to LogGleaner, which will process all of them.
• Record-wise logs always have storage-id. Mappers bucketize logs by storage ID to do the following in a batched fashion.
• For each storage batch, mappers check the partitioning information, creating one if not exists (see LogGleaner).
• Mappers send logs to corresponding reducers with a compact metadata for each storage.

Possible Optimization

The log gleaner so far simply reads from log files. We have a plan to optimize its behavior when we have a large amount of DRAM by directly reading from the log buffer if it is not blown away yet. ThreadLogBuffer has an additional marker "head" for this purpose, but so far we don't use it to simplify the implementation.

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.

Definition at line 73 of file log_mapper_impl.hpp.

#include <log_mapper_impl.hpp>

Inheritance diagram for foedus::snapshot::LogMapper:

Collaboration diagram for foedus::snapshot::LogMapper:

Public Member Functions
	LogMapper (Engine *engine, uint16_t local_ordinal)
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 LogMapper &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::LogMapper::LogMapper	(	Engine *	engine,
		uint16_t	local_ordinal
	)

Definition at line 58 of file log_mapper_impl.cpp.

  : MapReduceBase(engine, calculate_logger_id(engine, local_ordinal)),
    processed_log_count_(0) {
  clear_storage_buckets();
}

Member Function Documentation

ErrorStack foedus::snapshot::LogMapper::handle_process ( )

overrideprotectedvirtual

Implements the specific logics in derived class.

Implements foedus::snapshot::MapReduceBase.

Definition at line 140 of file log_mapper_impl.cpp.

References foedus::snapshot::align_io_ceil(), foedus::snapshot::align_io_floor(), ASSERT_ND, foedus::log::LogRange::begin_file_ordinal, foedus::log::LogRange::begin_offset, foedus::snapshot::MapReduceBase::check_cancelled(), CHECK_ERROR, foedus::log::LogOptions::construct_suffixed_log_path(), foedus::debugging::StopWatch::elapsed_sec(), foedus::snapshot::SnapshotOptions::emulation_, foedus::log::LogRange::end_file_ordinal, foedus::log::LogRange::end_offset, foedus::snapshot::MapReduceBase::engine_, foedus::fs::file_size(), foedus::snapshot::LogGleanerRef::get_base_epoch(), foedus::Engine::get_log_manager(), foedus::log::LoggerRef::get_log_range(), foedus::log::LogManager::get_logger(), foedus::Engine::get_options(), foedus::memory::AlignedMemory::get_size(), foedus::snapshot::LogGleanerRef::get_valid_until_epoch(), foedus::snapshot::MapReduceBase::id_, foedus::log::LogRange::is_empty(), foedus::fs::DirectIoFile::kDirectIoSeekSet, foedus::kRetOk, foedus::EngineOptions::log_, foedus::snapshot::MapReduceBase::numa_node_, foedus::snapshot::MapReduceBase::parent_, foedus::EngineOptions::snapshot_, foedus::fs::status(), foedus::debugging::StopWatch::stop(), to_string(), and WRAP_ERROR_CODE.

                                     {
  const Epoch base_epoch = parent_.get_base_epoch();
  const Epoch until_epoch = parent_.get_valid_until_epoch();
  log::LoggerRef logger = engine_->get_log_manager()->get_logger(id_);
  const log::LogRange log_range = logger.get_log_range(base_epoch, until_epoch);
  // uint64_t cur_offset = log_range.begin_offset;
  if (log_range.is_empty()) {
    LOG(INFO) << to_string() << " has no logs to process";
    report_completion(0);
    return kRetOk;
  }

  // open the file and seek to there. be careful on page boundary.
  // as we use direct I/O, all I/O must be 4kb-aligned. when the read range is not
  // a multiply of 4kb, we read a little bit more (at most 4kb per read, so negligible).
  // to clarify, here we use the following suffixes
  //   "infile"/"inbuf" : the offset is an offset in entire file/IO buffer
  //   "aligned" : the offset is 4kb-aligned (careful on floor vs ceil)
  // Lengthy, but otherwise it's so confusing.
  processed_log_count_ = 0;
  IoBufStatus status;
  status.size_inbuf_aligned_ = io_buffer_.get_size();
  status.cur_file_ordinal_ = log_range.begin_file_ordinal;
  status.ended_ = false;
  status.first_read_ = true;
  debugging::StopWatch watch;
  while (!status.ended_) {  // loop for log file switch
    fs::Path path(engine_->get_options().log_.construct_suffixed_log_path(
      numa_node_,
      id_,
      status.cur_file_ordinal_));
    uint64_t file_size = fs::file_size(path);
    if (file_size % kIoAlignment != 0) {
      LOG(WARNING) << to_string() << " Interesting, non-aligned file size, which probably means"
        << " previous writes didn't flush. file path=" << path << ", file size=" << file_size;
      file_size = align_io_floor(file_size);
    }
    ASSERT_ND(file_size % kIoAlignment == 0);
    status.size_infile_aligned_ = file_size;

    // If this is the first file to read, we might be reading from non-zero position.
    // In that case, be careful on alignment.
    if (status.cur_file_ordinal_ == log_range.begin_file_ordinal) {
      status.next_infile_ = log_range.begin_offset;
    } else {
      status.next_infile_ = 0;
    }

    if (status.cur_file_ordinal_ == log_range.end_file_ordinal) {
      ASSERT_ND(log_range.end_offset <= file_size);
      status.end_infile_ = log_range.end_offset;
    } else {
      status.end_infile_ = file_size;
    }

    DVLOG(1) << to_string() << " file path=" << path << ", file size=" << assorted::Hex(file_size)
      << ", read_end=" << assorted::Hex(status.end_infile_);
    fs::DirectIoFile file(path, engine_->get_options().snapshot_.emulation_);
    WRAP_ERROR_CODE(file.open(true, false, false, false));
    DVLOG(1) << to_string() << "opened log file " << file;

    while (true) {
      WRAP_ERROR_CODE(check_cancelled());  // check per each read
      status.buf_infile_aligned_ = align_io_floor(status.next_infile_);
      WRAP_ERROR_CODE(file.seek(status.buf_infile_aligned_, fs::DirectIoFile::kDirectIoSeekSet));
      DVLOG(1) << to_string() << " seeked to: " << assorted::Hex(status.buf_infile_aligned_);
      status.end_inbuf_aligned_ = std::min(
        io_buffer_.get_size(),
        align_io_ceil(status.end_infile_ - status.buf_infile_aligned_));
      ASSERT_ND(status.end_inbuf_aligned_ % kIoAlignment == 0);
      WRAP_ERROR_CODE(file.read(status.end_inbuf_aligned_, &io_buffer_));

      status.cur_inbuf_ = 0;
      if (status.next_infile_ != status.buf_infile_aligned_) {
        ASSERT_ND(status.next_infile_ > status.buf_infile_aligned_);
        status.cur_inbuf_ = status.next_infile_ - status.buf_infile_aligned_;
        status.cur_inbuf_ = status.next_infile_ - status.buf_infile_aligned_;
        DVLOG(1) << to_string() << " skipped " << status.cur_inbuf_ << " bytes for aligned read";
      }

      CHECK_ERROR(handle_process_buffer(file, &status));
      if (status.more_in_the_file_) {
        ASSERT_ND(status.next_infile_ > status.buf_infile_aligned_);
      } else {
        if (log_range.end_file_ordinal == status.cur_file_ordinal_) {
          status.ended_ = true;
          break;
        } else {
          ++status.cur_file_ordinal_;
          status.next_infile_ = 0;
          LOG(INFO) << to_string()
            << " moved on to next log file ordinal " << status.cur_file_ordinal_;
        }
      }
    }
    file.close();
  }
  watch.stop();
  LOG(INFO) << to_string() << " processed " << processed_log_count_ << " log entries in "
    << watch.elapsed_sec() << "s";
  report_completion(watch.elapsed_sec());
  return kRetOk;
}

Here is the call graph for this function:

ErrorStack foedus::snapshot::LogMapper::initialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 64 of file log_mapper_impl.cpp.

References foedus::memory::AlignedMemory::alloc(), ASSERT_ND, foedus::snapshot::MapReduceBase::engine_, foedus::Engine::get_options(), foedus::memory::AlignedMemory::is_null(), foedus::memory::kHugepageSize, foedus::memory::AlignedMemory::kNumaAllocOnnode, foedus::kRetOk, foedus::snapshot::SnapshotOptions::log_mapper_bucket_kb_, foedus::snapshot::SnapshotOptions::log_mapper_io_buffer_mb_, foedus::snapshot::MapReduceBase::numa_node_, and foedus::EngineOptions::snapshot_.

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

  uint64_t io_buffer_size = static_cast<uint64_t>(option.log_mapper_io_buffer_mb_) << 20;
  io_buffer_size = assorted::align<uint64_t, memory::kHugepageSize>(io_buffer_size);
  io_buffer_.alloc(
    io_buffer_size,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node_);
  ASSERT_ND(!io_buffer_.is_null());

  uint64_t bucket_size = static_cast<uint64_t>(option.log_mapper_bucket_kb_) << 10;
  buckets_memory_.alloc(
    bucket_size,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node_);
  ASSERT_ND(!buckets_memory_.is_null());

  const uint64_t tmp_memory_size = memory::kHugepageSize;
  tmp_memory_.alloc(
    tmp_memory_size,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node_);
  ASSERT_ND(!tmp_memory_.is_null());

  // these automatically expand
  presort_buffer_.alloc(
    1U << 21,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node_);
  presort_ouputs_.alloc(
    1U << 21,
    memory::kHugepageSize,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node_);

  uint64_t tmp_offset = 0;
  tmp_send_buffer_slice_ = memory::AlignedMemorySlice(&tmp_memory_, tmp_offset, kSendBufferSize);
  tmp_offset += kSendBufferSize;
  tmp_position_array_slice_ = memory::AlignedMemorySlice(&tmp_memory_, tmp_offset, kBucketSize);
  tmp_offset += kBucketSize;
  tmp_sort_array_slice_ = memory::AlignedMemorySlice(&tmp_memory_, tmp_offset, kBucketSize << 1);
  tmp_offset += kBucketSize << 1;
  const uint64_t hashlist_bytesize = kBucketHashListMaxCount * sizeof(BucketHashList);
  tmp_hashlist_buffer_slice_ = memory::AlignedMemorySlice(
    &tmp_memory_, tmp_offset, hashlist_bytesize);
  tmp_offset += hashlist_bytesize;
  tmp_partition_array_slice_ = memory::AlignedMemorySlice(&tmp_memory_, tmp_offset, kBucketSize);
  tmp_offset += kBucketSize;
  if (tmp_memory_size < tmp_offset) {
    LOG(FATAL) << "tmp_memory_size is too small. some contant values are messed up";
  }

  processed_log_count_ = 0;
  clear_storage_buckets();

  return kRetOk;
}

Here is the call graph for this function:

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

inlineoverridevirtual

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

Used only for logging/debugging.

Implements foedus::snapshot::MapReduceBase.

Definition at line 80 of file log_mapper_impl.hpp.

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

Referenced by handle_process().

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

Here is the caller graph for this function:

ErrorStack foedus::snapshot::LogMapper::uninitialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 127 of file log_mapper_impl.cpp.

References foedus::memory::AlignedMemory::release_block(), and SUMMARIZE_ERROR_BATCH.

                                        {
  ErrorStackBatch batch;
  io_buffer_.release_block();
  buckets_memory_.release_block();
  tmp_memory_.release_block();
  clear_storage_buckets();
  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 LogMapper &  v  )

friend

Definition at line 700 of file log_mapper_impl.cpp.

                                                          {
  o << "<LogMapper>"
    << "<id_>" << v.id_ << "</id_>"
    << "<numa_node_>" << static_cast<int>(v.numa_node_) << "</numa_node_>"
    << "<buckets_allocated_count_>" << v.buckets_allocated_count_ << "</buckets_allocated_count_>"
    << "<hashlist_allocated_count>" << v.hashlist_allocated_count_ << "</hashlist_allocated_count>"
    << "<processed_log_count_>" << v.processed_log_count_ << "</processed_log_count_>"
    << "</LogMapper>";
  return o;
}

---

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

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