foedus::snapshot::MergeSort Class Reference

Receives an arbitrary number of sorted buffers and emits one fully sorted stream of logs. More...

Detailed Description

Receives an arbitrary number of sorted buffers and emits one fully sorted stream of logs.

Where this is used, and why

During snapshot, log reducer merges the current in-memory buffer with zero or more dumped sorted runs on disk. We initially used a tuple-based algorithm to select a stream that has the smallest key for each iteration, but it was not desirable for overheads and code complexity. Batching the sorting is much faster than even tournament-tree, and also separates sorting logic from the composer. Hence this class.

Batching, Batching, Batching

All frequently used code path must be batched, at least in 1000s. An exception is the code that is invoked only occasionally, eg once per block. There are a few logics that must be customized for each storage type (eg handling of keys). Such customization is done in granularity of batch, too. No switch for individual logs.

Algorithm at a glance

• a) Pick a chunk (1000s of logs) from an input whose last key (called chunk-last-key) is the smallest among inputs.
• b) Repeat a) until some input must move on to next window or we have a large number of logs.
• c) The smallest chunk key among all inputs is the batch-threshold. All tuples in all input's chunks whose key is strictly smaller than the threshold are sorted.
• d) Composer consumes the sorted inputs.
• e) Advance each input's current place, move their windows if needed (step a-d never moves windows), and repeat a) until all inputs are done.
• Some special condition: chunk-last-key survives to next iteration except when the chunk is the last chunk of last window. In that case, the batch-threshold is chosen ignoring the input. The logs are merged if key < threshold, marking the input as ended if all logs are merged. If all inputs are in the last chunk (last iteration), we include all remainings from all inputs.
• Another special condition: Obviously, this class does nothing when input_count is 1, skipping all the overheads. This should hopefully happen often if reducers have large buffers.

Modularity

This class has no dependency to other modules. It receives buffers of logs, that's it. We must keep this class in that way for easier testing and tuning.

Definition at line 77 of file merge_sort.hpp.

#include <merge_sort.hpp>

Inheritance diagram for foedus::snapshot::MergeSort:

Collaboration diagram for foedus::snapshot::MergeSort:

Classes
struct	AdjustComparatorMasstree
	Used in batch_sort_adjust_sort if the storage is a masstree storage. More...
struct	GroupifyResult
	Represents a group of consecutive logs in the current batch. More...
struct	InputStatus
	Current status of each input. More...
struct	PositionEntry
	Provides additional information for each entry we are sorting. More...
struct	SortEntry
	Entries we actually sort. More...

Public Types
enum	Constants {   kMaxMergedPosition = 1 << 23, kLogChunk = 1 << 10, kDefaultChunkBatch = 1 << 8, kMaxLevels = 32,   kWindowChunkReserveBytes = 1 << 16 }
typedef uint32_t	MergedPosition
	Position in MergeSort's buffer. More...
typedef uint16_t	InputIndex
	Index in input streams. More...

Public Member Functions
	MergeSort (storage::StorageId id, storage::StorageType type, Epoch base_epoch, SortedBuffer *const *inputs, uint16_t inputs_count, uint16_t max_original_pages, memory::AlignedMemory *const work_memory, uint16_t chunk_batch_size=kDefaultChunkBatch)
ErrorStack	next_batch ()
	Executes merge-sort on several thousands of logs and provides the result as a batch. More...
uint32_t	fetch_logs (uint32_t sort_pos, uint32_t count, log::RecordLogType const **out) const
	To reduce L1 cache miss stall, we prefetch some number of position entries and the pointed log entries in parallel. More...
GroupifyResult	groupify (uint32_t begin, uint32_t limit=0) const __attribute__((always_inline))
	Find a group of consecutive logs from the given position that have either a common log type or a common key. More...
void	assert_sorted ()
	For debug/test only. More...
ErrorStack	initialize_once () override
ErrorStack	uninitialize_once () override
bool	is_no_merging () const __attribute__((always_inline))
bool	is_ended_all () const
storage::StorageId	get_storage_id () const __attribute__((always_inline))
Epoch	get_base_epoch () const __attribute__((always_inline))
MergedPosition	get_current_count () const __attribute__((always_inline))
InputIndex	get_inputs_count () const __attribute__((always_inline))
const PositionEntry *	get_position_entries () const __attribute__((always_inline))
log::LogCode	get_log_type_from_sort_position (uint32_t sort_pos) const __attribute__((always_inline))
const SortEntry *	get_sort_entries () const __attribute__((always_inline))
const log::RecordLogType *	resolve_merged_position (MergedPosition pos) const __attribute__((always_inline))
const log::RecordLogType *	resolve_sort_position (uint32_t sort_pos) const __attribute__((always_inline))
void	change_log_type_at (uint32_t sort_pos, log::LogCode before, log::LogCode after)
	used to switch between two compatible log types. More...
storage::Page *	get_original_pages () const __attribute__((always_inline))
bool	are_all_single_layer_logs () const __attribute__((always_inline))
uint16_t	get_longest_key_length () const __attribute__((always_inline))
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 ()

Member Typedef Documentation

typedef uint16_t foedus::snapshot::MergeSort::InputIndex

Index in input streams.

Definition at line 85 of file merge_sort.hpp.

typedef uint32_t foedus::snapshot::MergeSort::MergedPosition

Position in MergeSort's buffer.

We never sort more than 2^23 entries at once because we merge by chunks.

Definition at line 83 of file merge_sort.hpp.

Member Enumeration Documentation

enum foedus::snapshot::MergeSort::Constants

Enumerator
kMaxMergedPosition	This is a theoretical max. additionally it must be less than buffer_capacity_
kLogChunk	1024 logs per chunk
kDefaultChunkBatch	Suppose each log is 50 bytes: 1k*256*50b=12.5 MB worth logs to sort per batch.
kMaxLevels	We assume the path wouldn't be this deep.
kWindowChunkReserveBytes	To avoid handling the case where a log spans an end of window, chunks leave at least this many bytes in each window. No single log can be more than this size, so it simplifies the logic. If the input is in the last window, this value has no effects.

Definition at line 87 of file merge_sort.hpp.

                 {
    kMaxMergedPosition = 1 << 23,
    kLogChunk = 1 << 10,
    kDefaultChunkBatch = 1 << 8,
    kMaxLevels = 32,
    kWindowChunkReserveBytes = 1 << 16,
  };

Constructor & Destructor Documentation

foedus::snapshot::MergeSort::MergeSort	(	storage::StorageId	id,
		storage::StorageType	type,
		Epoch	base_epoch,
		SortedBuffer *const *	inputs,
		uint16_t	inputs_count,
		uint16_t	max_original_pages,
		memory::AlignedMemory *const	work_memory,
		uint16_t	chunk_batch_size = kDefaultChunkBatch
	)

Definition at line 63 of file merge_sort.cpp.

References ASSERT_ND.

  : DefaultInitializable(),
    id_(id),
    type_(type),
    base_epoch_(base_epoch),
    shortest_key_length_(extract_shortest_key_length(inputs, inputs_count)),
    longest_key_length_(extract_longest_key_length(inputs, inputs_count)),
    inputs_(inputs),
    inputs_count_(inputs_count),
    max_original_pages_(max_original_pages),
    chunk_batch_size_(chunk_batch_size),
    work_memory_(work_memory) {
  ASSERT_ND(shortest_key_length_ <= longest_key_length_);
  ASSERT_ND(shortest_key_length_ > 0);
  ASSERT_ND(chunk_batch_size_ > 0);
  current_count_ = 0;
  sort_entries_ = nullptr;
  position_entries_ = nullptr;
  original_pages_ = nullptr;
  inputs_status_ = nullptr;
}

Member Function Documentation

bool foedus::snapshot::MergeSort::are_all_single_layer_logs ( ) const

inline

Definition at line 411 of file merge_sort.hpp.

{ return longest_key_length_ <= 8U; }

void foedus::snapshot::MergeSort::assert_sorted

(

)

For debug/test only.

Checks if sort_entries_ are indeed fully sorted. This method is wiped out in release mode.

Definition at line 695 of file merge_sort.cpp.

References ASSERT_ND, foedus::storage::hash::HashCommonLogType::assert_type(), foedus::snapshot::MergeSort::SortEntry::data_, foedus::xct::XctId::get_epoch(), foedus::xct::XctId::get_ordinal(), foedus::snapshot::MergeSort::SortEntry::get_position(), foedus::log::BaseLogType::header_, foedus::snapshot::MergeSort::PositionEntry::input_index_, foedus::storage::kArrayStorage, foedus::storage::kHashStorage, foedus::storage::kMasstreeStorage, foedus::snapshot::MergeSort::SortEntry::set(), foedus::Epoch::subtract(), and foedus::log::LogHeader::xct_id_.

                              {
#ifndef NDEBUG
  for (MergedPosition i = 0; i < current_count_; ++i) {
    MergedPosition cur_pos = sort_entries_[i].get_position();
    const log::RecordLogType* cur = inputs_status_[position_entries_[cur_pos].input_index_].
      from_compact_pos(position_entries_[cur_pos].input_position_);

    // does it point to a correct log?
    Epoch epoch = cur->header_.xct_id_.get_epoch();
    uint16_t compressed_epoch = epoch.subtract(base_epoch_);
    SortEntry dummy;
    if (type_ == storage::kArrayStorage) {
      const auto* casted = reinterpret_cast<const storage::array::ArrayCommonUpdateLogType*>(cur);
      dummy.set(
        casted->offset_,
        compressed_epoch,
        cur->header_.xct_id_.get_ordinal(),
        false,
        cur_pos);
    } else if (type_ == storage::kMasstreeStorage) {
      const auto* casted = reinterpret_cast<const storage::masstree::MasstreeCommonLogType*>(cur);
      dummy.set(
        casted->get_first_slice(),
        compressed_epoch,
        cur->header_.xct_id_.get_ordinal(),
        casted->key_length_ != sizeof(storage::masstree::KeySlice),
        cur_pos);
    } else {
      ASSERT_ND(type_ == storage::kHashStorage);
      const auto* casted = reinterpret_cast<const storage::hash::HashCommonLogType*>(cur);
      casted->assert_type();
      storage::hash::HashBin bin = casted->hash_ >> (64U - casted->bin_bits_);
      dummy.set(
        bin,
        compressed_epoch,
        cur->header_.xct_id_.get_ordinal(),
        false,  // hash doesn't need further sorting so far.
        cur_pos);
    }
    ASSERT_ND(dummy.data_ == sort_entries_[i].data_);
    if (i == 0) {
      continue;
    }

    // compare with previous
    MergedPosition prev_pos = sort_entries_[i - 1].get_position();
    ASSERT_ND(prev_pos != cur_pos);
    const log::RecordLogType* prev = inputs_status_[position_entries_[prev_pos].input_index_].
      from_compact_pos(position_entries_[prev_pos].input_position_);
    int cmp = compare_logs(prev, cur);
    ASSERT_ND(cmp <= 0);
    if (cmp == 0) {
      // the last of sort order is position.
      ASSERT_ND(prev_pos < cur_pos);
    }
  }
#endif  // NDEBUG
}

Here is the call graph for this function:

void foedus::snapshot::MergeSort::change_log_type_at ( uint32_t  sort_pos, log::LogCode  before, log::LogCode  after  )

inline

used to switch between two compatible log types.

It's a special-purpose method, use it with caution.

Definition at line 393 of file merge_sort.hpp.

References ASSERT_ND, foedus::snapshot::MergeSort::PositionEntry::get_log_type(), get_log_type_from_sort_position(), foedus::snapshot::MergeSort::SortEntry::get_position(), foedus::log::LogHeader::get_type(), foedus::log::BaseLogType::header_, foedus::snapshot::MergeSort::PositionEntry::log_type_, foedus::log::LogHeader::log_type_code_, and resolve_merged_position().

                                                                                  {
    // we have to change two places; PositionEntry and header of the log itself.
    ASSERT_ND(sort_pos < current_count_);
    ASSERT_ND(get_log_type_from_sort_position(sort_pos) == before);
    MergedPosition pos = sort_entries_[sort_pos].get_position();
    PositionEntry& entry = position_entries_[pos];
    ASSERT_ND(entry.get_log_type() == before);
    entry.log_type_ = after;
    ASSERT_ND(entry.get_log_type() == after);
    ASSERT_ND(get_log_type_from_sort_position(sort_pos) == after);

    log::RecordLogType* record = const_cast<log::RecordLogType*>(resolve_merged_position(pos));
    ASSERT_ND(record->header_.get_type() == before);
    record->header_.log_type_code_ = after;
    ASSERT_ND(record->header_.get_type() == after);
  }

Here is the call graph for this function:

uint32_t foedus::snapshot::MergeSort::fetch_logs	(	uint32_t	sort_pos,
		uint32_t	count,
		log::RecordLogType const **	out
	)		const

To reduce L1 cache miss stall, we prefetch some number of position entries and the pointed log entries in parallel.

Parameters

[in]	sort_pos	we prefetch from this sort_entries_ -> position_entries_ -> logs
[in]	count	we prefetch upto sort_entries_[sort_pos + count - 1]. Should be at least 2 or 4 to make parallel prefetch effective.
[out]	out	Fetched logs.

Precondition

sort_pos <= current_count_

Returns

fetched count. unless sort_pos+count>current_count, same as count.

Definition at line 273 of file merge_sort.cpp.

References ASSERT_ND, foedus::snapshot::MergeSort::InputStatus::from_compact_pos(), foedus::snapshot::MergeSort::SortEntry::get_position(), foedus::snapshot::MergeSort::PositionEntry::input_index_, is_no_merging(), and foedus::assorted::prefetch_cacheline().

                                      {
  ASSERT_ND(sort_pos <= current_count_);
  uint32_t fetched_count = count;
  if (sort_pos + count > current_count_) {
    fetched_count = current_count_ - sort_pos;
  }

  if (is_no_merging()) {
    // no merge sort.
#ifndef NDEBUG
    for (uint32_t i = 0; i < fetched_count; ++i) {
      ASSERT_ND(sort_entries_[sort_pos + i].get_position() == sort_pos + i);
    }
#endif  // NDEBUG
    // in this case, the pointed logs are also contiguous. no point to do prefetching.
    for (uint32_t i = 0; i < fetched_count; ++i) {
      MergedPosition pos = sort_pos + i;
      ASSERT_ND(position_entries_[pos].input_index_ == 0);
      out[i] = inputs_status_[0].from_compact_pos(position_entries_[pos].input_position_);
    }
    return fetched_count;
  }

  // prefetch position entries
  for (uint32_t i = 0; i < fetched_count; ++i) {
    MergedPosition pos = sort_entries_[sort_pos + i].get_position();
    assorted::prefetch_cacheline(position_entries_ + pos);
  }
  // prefetch and fetch logs
  for (uint32_t i = 0; i < fetched_count; ++i) {
    MergedPosition pos = sort_entries_[sort_pos + i].get_position();
    InputIndex input = position_entries_[pos].input_index_;
    out[i] = inputs_status_[input].from_compact_pos(position_entries_[pos].input_position_);
    assorted::prefetch_cacheline(out[i]);
  }
  return fetched_count;
}

Here is the call graph for this function:

Epoch foedus::snapshot::MergeSort::get_base_epoch ( ) const

inline

Definition at line 368 of file merge_sort.hpp.

{ return base_epoch_; }

MergedPosition foedus::snapshot::MergeSort::get_current_count ( ) const

inline

Definition at line 369 of file merge_sort.hpp.

Referenced by foedus::storage::array::ArrayComposeContext::execute(), foedus::storage::hash::HashComposeContext::execute(), and foedus::storage::masstree::MasstreeComposeContext::execute().

{ return current_count_; }

Here is the caller graph for this function:

InputIndex foedus::snapshot::MergeSort::get_inputs_count ( ) const

inline

Definition at line 370 of file merge_sort.hpp.

{ return inputs_count_; }

log::LogCode foedus::snapshot::MergeSort::get_log_type_from_sort_position ( uint32_t  sort_pos ) const

inline

Definition at line 374 of file merge_sort.hpp.

References foedus::snapshot::MergeSort::PositionEntry::get_log_type(), and foedus::snapshot::MergeSort::SortEntry::get_position().

Referenced by change_log_type_at().

                                                                                           {
    MergedPosition pos = sort_entries_[sort_pos].get_position();
    return position_entries_[pos].get_log_type();
  }

Here is the call graph for this function:

Here is the caller graph for this function:

uint16_t foedus::snapshot::MergeSort::get_longest_key_length ( ) const

inline

Definition at line 412 of file merge_sort.hpp.

{ return longest_key_length_; }

storage::Page* foedus::snapshot::MergeSort::get_original_pages ( ) const

inline

Definition at line 410 of file merge_sort.hpp.

{ return original_pages_; }

const PositionEntry* foedus::snapshot::MergeSort::get_position_entries ( ) const

inline

Definition at line 371 of file merge_sort.hpp.

                                                                         {
    return position_entries_;
  }

const SortEntry* foedus::snapshot::MergeSort::get_sort_entries ( ) const

inline

Definition at line 378 of file merge_sort.hpp.

Referenced by foedus::storage::array::ArrayComposeContext::execute(), and foedus::storage::hash::HashComposeContext::execute().

{ return sort_entries_; }

Here is the caller graph for this function:

storage::StorageId foedus::snapshot::MergeSort::get_storage_id ( ) const

inline

Definition at line 367 of file merge_sort.hpp.

{ return id_; }

MergeSort::GroupifyResult foedus::snapshot::MergeSort::groupify ( uint32_t  begin, uint32_t  limit = 0  ) const

inline

Find a group of consecutive logs from the given position that have either a common log type or a common key.

Parameters

[in]	begin	starting position in the sorted array.
[in]	limit	maximum number of logs to check. If not specified, all logs in the batch.

Returns

the group found in the current batch.

Some composer uses this optional method to batch-process the logs. In an ideal case, there always is a big group, and composer calls this method per hundreds of logs. But, it's also quite possible that there isn't a good group. Hence, these methods are called very frequently (potentially for each log). Worth explicit inlining.

Definition at line 534 of file merge_sort.hpp.

References ASSERT_ND, foedus::snapshot::MergeSort::GroupifyResult::count_, foedus::snapshot::MergeSort::SortEntry::get_position(), foedus::snapshot::MergeSort::GroupifyResult::has_common_key_, foedus::snapshot::MergeSort::GroupifyResult::has_common_log_code_, foedus::log::kLogCodeInvalid, foedus::snapshot::MergeSort::GroupifyResult::log_code_, foedus::snapshot::MergeSort::PositionEntry::log_type_, and UNLIKELY.

Referenced by foedus::storage::masstree::MasstreeComposeContext::execute().

                                                                                       {
  ASSERT_ND(begin < current_count_);
  GroupifyResult result;
  result.count_ = 1;
  result.has_common_key_ = false;
  result.has_common_log_code_ = false;
  result.log_code_ = log::kLogCodeInvalid;
  if (UNLIKELY(begin + 1U == current_count_)) {
    return result;
  }

  // first, check common keys
  uint32_t cur;
  if (shortest_key_length_ == 8U && longest_key_length_ == 8U) {
    cur = groupify_find_common_keys_8b(begin, limit);
  } else {
    cur = groupify_find_common_keys_general(begin, limit);
  }

  ASSERT_ND(cur >= begin);
  ASSERT_ND(cur < current_count_);
  if (UNLIKELY(cur != begin)) {
    result.has_common_key_ = true;
    result.count_ = cur - begin + 1U;
    return result;
  }

  // if keys are different, check common log types.
  cur = groupify_find_common_log_type(begin, limit);
  ASSERT_ND(cur >= begin);
  ASSERT_ND(cur < current_count_);
  if (UNLIKELY(cur != begin)) {
    result.has_common_log_code_ = true;
    result.log_code_ = position_entries_[sort_entries_[cur].get_position()].log_type_;
    result.count_ = cur - begin + 1U;
    return result;
  } else {
    return result;
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::MergeSort::initialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 93 of file merge_sort.cpp.

References foedus::snapshot::SortedBuffer::assert_checks(), foedus::snapshot::MergeSort::InputStatus::assert_consistent(), ASSERT_ND, foedus::memory::AlignedMemory::assure_capacity(), foedus::snapshot::MergeSort::InputStatus::chunk_relative_pos_, foedus::snapshot::MergeSort::InputStatus::cur_relative_pos_, foedus::snapshot::MergeSort::InputStatus::end_absolute_pos_, foedus::memory::AlignedMemory::get_block(), foedus::snapshot::SortedBuffer::get_buffer(), foedus::snapshot::SortedBuffer::get_buffer_size(), foedus::snapshot::SortedBuffer::get_cur_block_abosulte_begin(), foedus::snapshot::SortedBuffer::get_cur_block_abosulte_end(), foedus::snapshot::SortedBuffer::get_offset(), foedus::memory::AlignedMemory::get_size(), kLogChunk, foedus::storage::kPageSize, foedus::kRetOk, foedus::snapshot::MergeSort::InputStatus::previous_chunk_relative_pos_, foedus::fs::status(), foedus::snapshot::MergeSort::InputStatus::window_, foedus::snapshot::MergeSort::InputStatus::window_offset_, foedus::snapshot::MergeSort::InputStatus::window_size_, and WRAP_ERROR_CODE.

                                      {
  // in each batch, we might include tuples from an input even if we didn't fully pick a chunk from
  // it (at most kLogChunk-1 such tuples). so, conservatively chunk_batch_size_ + inputs_count_.
  uint32_t buffer_capacity = kLogChunk * (chunk_batch_size_ + inputs_count_);
  buffer_capacity_ = assorted::align<uint32_t, 512U>(buffer_capacity);
  uint64_t byte_size = buffer_capacity_ * (sizeof(SortEntry) + sizeof(PositionEntry));
  ASSERT_ND(byte_size % 4096U == 0);
  byte_size += storage::kPageSize * (max_original_pages_ + 1U);
  byte_size += sizeof(InputStatus) * inputs_count_;
  WRAP_ERROR_CODE(work_memory_->assure_capacity(byte_size));

  // assign pointers
  char* block = reinterpret_cast<char*>(work_memory_->get_block());
#ifndef NDEBUG
  std::memset(block, 0xDA, work_memory_->get_size());
#endif  // NDEBUG
  uint64_t offset = 0;
  sort_entries_ = reinterpret_cast<SortEntry*>(block + offset);
  offset += sizeof(SortEntry) * buffer_capacity;
  position_entries_ = reinterpret_cast<PositionEntry*>(block + offset);
  offset += sizeof(PositionEntry) * buffer_capacity;
  original_pages_ = reinterpret_cast<storage::Page*>(block + offset);
  offset += sizeof(storage::Page) * (max_original_pages_ + 1U);
  inputs_status_ = reinterpret_cast<InputStatus*>(block + offset);
  offset += sizeof(InputStatus) * inputs_count_;
  ASSERT_ND(offset == byte_size);

  // initialize inputs_status_
  for (InputIndex i = 0; i < inputs_count_; ++i) {
    InputStatus* status = inputs_status_ + i;
    SortedBuffer* input = inputs_[i];
    input->assert_checks();
    status->window_ = input->get_buffer();
    status->window_offset_ = input->get_offset();
    status->window_size_ = input->get_buffer_size();
    uint64_t cur_abs = input->get_cur_block_abosulte_begin();
    // this is the initial read of this block, so we are sure cur_block_abosulte_begin is the window
    ASSERT_ND(cur_abs >= status->window_offset_);
    status->cur_relative_pos_ = cur_abs - status->window_offset_;
    status->chunk_relative_pos_ = status->cur_relative_pos_;  // hence the chunk has only 1 log
    status->previous_chunk_relative_pos_ = status->chunk_relative_pos_;

    uint64_t end_abs = input->get_cur_block_abosulte_end();
    status->end_absolute_pos_ = end_abs;

    status->assert_consistent();
  }
  return kRetOk;
}

Here is the call graph for this function:

bool foedus::snapshot::MergeSort::is_ended_all ( ) const

inline

Definition at line 358 of file merge_sort.hpp.

Referenced by foedus::storage::array::ArrayComposeContext::execute(), foedus::storage::hash::HashComposeContext::execute(), foedus::storage::masstree::MasstreeComposeContext::execute(), and next_batch().

                                   {
    for (InputIndex i = 0; i < inputs_count_; ++i) {
      if (!inputs_status_[i].is_ended()) {
        return false;
      }
    }
    return true;
  }

Here is the caller graph for this function:

bool foedus::snapshot::MergeSort::is_no_merging ( ) const

inline

Definition at line 356 of file merge_sort.hpp.

Referenced by fetch_logs(), and next_batch().

{ return inputs_count_ == 1U; }

Here is the caller graph for this function:

ErrorStack foedus::snapshot::MergeSort::next_batch

(

)

Executes merge-sort on several thousands of logs and provides the result as a batch.

Precondition

is_initialized (call initialize() first!)

Composer repeatedly invokes this method until it returns false. For each invokation (batch), composer consumes the entries in sorted order, in other words sort_entries_[0] to sort_entries_[current_count_ - 1], which points to position_entries_. This level of indirection might cause more L1 cache misses (only if we could fit everything into 16 bytes!), so we do prefetching to ameriolate it.

See also

fetch_logs(), which does the prefetching

get_current_count(), which will return the number of logs in the generated batch.

Definition at line 143 of file merge_sort.cpp.

References ASSERT_ND, CHECK_ERROR, is_ended_all(), foedus::DefaultInitializable::is_initialized(), is_no_merging(), and foedus::kRetOk.

Referenced by foedus::storage::array::ArrayComposeContext::execute(), foedus::storage::hash::HashComposeContext::execute(), and foedus::storage::masstree::MasstreeComposeContext::execute().

                                 {
  ASSERT_ND(is_initialized());
  current_count_ = 0;
  CHECK_ERROR(advance_window());

  if (is_ended_all()) {
    return kRetOk;
  }

  if (is_no_merging()) {
    next_batch_one_input();
  } else {
    InputIndex min_input = pick_chunks();
    batch_sort(min_input);
  }
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

const log::RecordLogType* foedus::snapshot::MergeSort::resolve_merged_position ( MergedPosition  pos ) const

inline

Definition at line 379 of file merge_sort.hpp.

References ASSERT_ND, foedus::snapshot::MergeSort::InputStatus::from_compact_pos(), foedus::snapshot::MergeSort::PositionEntry::input_index_, and foedus::snapshot::MergeSort::PositionEntry::input_position_.

Referenced by change_log_type_at(), and resolve_sort_position().

                                                                                                 {
    ASSERT_ND(pos < current_count_);
    const PositionEntry& entry = position_entries_[pos];
    return inputs_status_[entry.input_index_].from_compact_pos(entry.input_position_);
  }

Here is the call graph for this function:

Here is the caller graph for this function:

const log::RecordLogType* foedus::snapshot::MergeSort::resolve_sort_position ( uint32_t  sort_pos ) const

inline

Definition at line 384 of file merge_sort.hpp.

References ASSERT_ND, foedus::snapshot::MergeSort::SortEntry::get_position(), and resolve_merged_position().

Referenced by foedus::storage::array::ArrayComposeContext::execute(), and foedus::storage::masstree::MasstreeComposeContext::execute().

                                                                                              {
    ASSERT_ND(sort_pos < current_count_);
    MergedPosition pos = sort_entries_[sort_pos].get_position();
    return resolve_merged_position(pos);
  }

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::MergeSort::uninitialize_once ( )

inlineoverridevirtual

Implements foedus::DefaultInitializable.

Definition at line 354 of file merge_sort.hpp.

References foedus::kRetOk.

{ return kRetOk; }

---

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

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