foedus::storage::hash::ComposedBinsMergedStream Struct Reference

Packages all ComposedBinsBuffer to easily extract bins of current interest. More...

Detailed Description

Packages all ComposedBinsBuffer to easily extract bins of current interest.

This is the highest-level of the object hierarchy defined in this file.

Definition at line 205 of file hash_composed_bins_impl.hpp.

#include <hash_composed_bins_impl.hpp>

Collaboration diagram for foedus::storage::hash::ComposedBinsMergedStream:

Public Types
typedef HashIntermediatePage *	PagePtr

Public Member Functions
ErrorStack	init (const HashRootInfoPage *const *inputs, uint32_t input_count, PagePtr root_page, uint16_t root_child_index, memory::AlignedMemory *read_buffer, cache::SnapshotFileSet *fileset, snapshot::SnapshotWriter *writer, uint32_t *writer_buffer_pos, uint32_t *writer_higher_buffer_pos, bool *had_any_change)
ErrorCode	process_a_bin (uint32_t *installed_count, HashBin *next_lowest_bin)
	Consumes inputs for the cur_path_[0] page and install snapshot pointers there. More...
ErrorCode	switch_path (HashBin lowest_bin, cache::SnapshotFileSet *fileset, snapshot::SnapshotWriter *writer, uint32_t *writer_buffer_pos, uint32_t *writer_higher_buffer_pos)
	Moves cur_path_ to a page that cotains the specified bin. More...
ErrorCode	open_path (HashBin bin, uint8_t fixed_upto_level, cache::SnapshotFileSet *fileset, snapshot::SnapshotWriter *writer, uint32_t *writer_buffer_pos, uint32_t *writer_higher_buffer_pos)
	Recursively opens pages down from fixed_upto_level. More...
ErrorCode	assure_writer_buffer (snapshot::SnapshotWriter *writer, uint32_t *writer_buffer_pos, uint32_t writer_higher_buffer_pos)
	Subroutine to flush the writer if needed to make sure it has enough room. More...

Public Attributes
std::unique_ptr< ComposedBinsBuffer[] >	inputs_memory_
	just for auto release More...
ComposedBinsBuffer *	inputs_
uint32_t	input_count_
	total number of buffers in inputs_ More...
snapshot::SnapshotId	snapshot_id_
uint8_t	levels_
uint8_t	padding_
PagePtr	cur_path_ [kHashMaxLevels]
	The pages we are now composing. More...

Member Typedef Documentation

typedef HashIntermediatePage* foedus::storage::hash::ComposedBinsMergedStream::PagePtr

Definition at line 206 of file hash_composed_bins_impl.hpp.

Member Function Documentation

ErrorCode foedus::storage::hash::ComposedBinsMergedStream::assure_writer_buffer	(	snapshot::SnapshotWriter *	writer,
		uint32_t *	writer_buffer_pos,
		uint32_t	writer_higher_buffer_pos
	)

Subroutine to flush the writer if needed to make sure it has enough room.

Definition at line 290 of file hash_composed_bins_impl.cpp.

References ASSERT_ND, CHECK_ERROR_CODE, foedus::snapshot::SnapshotWriter::dump_pages(), foedus::snapshot::SnapshotWriter::expand_intermediate_memory(), foedus::snapshot::SnapshotWriter::get_intermediate_size(), foedus::snapshot::SnapshotWriter::get_page_size(), foedus::kErrorCodeOk, levels_, and UNLIKELY.

Referenced by open_path().

                                     {
  ASSERT_ND(*writer_buffer_pos <= writer->get_page_size());
  ASSERT_ND(writer_higher_buffer_pos <= writer->get_intermediate_size());
  if (UNLIKELY(*writer_buffer_pos == writer->get_page_size())) {
    CHECK_ERROR_CODE(writer->dump_pages(0, *writer_buffer_pos));
    *writer_buffer_pos = 0;
  }
  if (UNLIKELY(writer_higher_buffer_pos + levels_ >= writer->get_intermediate_size())) {
    CHECK_ERROR_CODE(writer->expand_intermediate_memory(writer_higher_buffer_pos + levels_, true));
    ASSERT_ND(writer_higher_buffer_pos + levels_ < writer->get_intermediate_size());
  }
  return kErrorCodeOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::storage::hash::ComposedBinsMergedStream::init	(	const HashRootInfoPage *const *	inputs,
		uint32_t	input_count,
		PagePtr	root_page,
		uint16_t	root_child_index,
		memory::AlignedMemory *	read_buffer,
		cache::SnapshotFileSet *	fileset,
		snapshot::SnapshotWriter *	writer,
		uint32_t *	writer_buffer_pos,
		uint32_t *	writer_higher_buffer_pos,
		bool *	had_any_change
	)

Definition at line 87 of file hash_composed_bins_impl.cpp.

References ASSERT_ND, foedus::storage::hash::ComposedBinsBuffer::assure_read_buffer_size(), foedus::storage::hash::ComposedBin::bin_, cur_path_, foedus::memory::AlignedMemory::get_block(), foedus::storage::hash::ComposedBinsBuffer::get_cur_bin(), foedus::storage::hash::HashIntermediatePage::get_level(), foedus::storage::hash::HashIntermediatePage::get_pointer(), foedus::memory::AlignedMemory::get_size(), foedus::snapshot::SnapshotWriter::get_snapshot_id(), foedus::storage::hash::ComposedBinsBuffer::has_more(), foedus::storage::hash::ComposedBinsBuffer::init(), input_count_, inputs_, inputs_memory_, foedus::storage::hash::kInvalidHashBin, foedus::storage::kPageSize, foedus::kRetOk, levels_, open_path(), snapshot_id_, foedus::storage::DualPagePointer::snapshot_pointer_, foedus::storage::DualPagePointer::volatile_pointer_, foedus::storage::VolatilePagePointer::word, and WRAP_ERROR_CODE.

                                            {
  ASSERT_ND(root_page->get_level() >= 1U);
  snapshot_id_ = writer->get_snapshot_id();
  std::memset(cur_path_, 0, sizeof(cur_path_));

  inputs_memory_.reset(new ComposedBinsBuffer[input_count]);
  inputs_ = inputs_memory_.get();
  input_count_ = input_count;

  ComposedBinsBuffer::assure_read_buffer_size(read_buffer, input_count);
  uint64_t buffer_piece_size = read_buffer->get_size() / kPageSize / input_count;
  for (uint32_t i = 0; i < input_count_; ++i) {
    const DualPagePointer& root_child = inputs[i]->get_pointer(root_child_index);
    SnapshotPagePointer head_page_id = root_child.snapshot_pointer_;
    uint32_t total_pages = root_child.volatile_pointer_.word;
    ASSERT_ND(total_pages > 0);
    HashComposedBinsPage* read_buffer_casted
      = reinterpret_cast<HashComposedBinsPage*>(read_buffer->get_block());
    HashComposedBinsPage* buffer_piece = read_buffer_casted + buffer_piece_size * i;
    inputs_[i].init(fileset, head_page_id, total_pages, buffer_piece_size, buffer_piece);
    if (inputs_[i].has_more()) {
      WRAP_ERROR_CODE(inputs_[i].next_pages());
    }
  }

  levels_ = root_page->get_level() + 1U;
  cur_path_[root_page->get_level()] = root_page;

  // what's the bin we initially seek to?
  HashBin initial_bin = kInvalidHashBin;
  for (uint32_t i = 0; i < input_count_; ++i) {
    ComposedBinsBuffer* input = inputs_ + i;
    if (input->has_more()) {
      initial_bin = std::min<HashBin>(initial_bin, input->get_cur_bin().bin_);
    }
  }

  if (initial_bin == kInvalidHashBin) {
    *had_any_change = false;
  } else {
    *had_any_change = true;
    WRAP_ERROR_CODE(open_path(
      initial_bin,
      root_page->get_level(),
      fileset,
      writer,
      writer_buffer_pos,
      writer_higher_buffer_pos));
  }

  return kRetOk;
}

Here is the call graph for this function:

ErrorCode foedus::storage::hash::ComposedBinsMergedStream::open_path	(	HashBin	bin,
		uint8_t	fixed_upto_level,
		cache::SnapshotFileSet *	fileset,
		snapshot::SnapshotWriter *	writer,
		uint32_t *	writer_buffer_pos,
		uint32_t *	writer_higher_buffer_pos
	)

Recursively opens pages down from fixed_upto_level.

Definition at line 225 of file hash_composed_bins_impl.cpp.

References ASSERT_ND, assure_writer_buffer(), foedus::storage::hash::HashBinRange::begin_, CHECK_ERROR_CODE, foedus::storage::hash::IntermediateRoute::construct(), foedus::storage::hash::HashBinRange::contains(), cur_path_, foedus::storage::extract_snapshot_id_from_snapshot_pointer(), foedus::storage::hash::HashIntermediatePage::get_bin_range(), foedus::snapshot::SnapshotWriter::get_intermediate_base(), foedus::snapshot::SnapshotWriter::get_intermediate_size(), foedus::storage::hash::HashIntermediatePage::get_level(), foedus::snapshot::SnapshotWriter::get_next_page_id(), foedus::snapshot::SnapshotWriter::get_page_base(), foedus::snapshot::SnapshotWriter::get_page_size(), foedus::storage::hash::HashIntermediatePage::get_pointer(), foedus::storage::hash::HashIntermediatePage::header(), foedus::storage::hash::HashIntermediatePage::initialize_snapshot_page(), foedus::kErrorCodeOk, foedus::storage::hash::kHashMaxBins, foedus::storage::hash::kInvalidHashBin, foedus::storage::hash::HashBinRange::length(), levels_, foedus::storage::PageHeader::page_id_, foedus::cache::SnapshotFileSet::read_page(), foedus::storage::hash::IntermediateRoute::route, snapshot_id_, foedus::storage::DualPagePointer::snapshot_pointer_, and foedus::storage::PageHeader::storage_id_.

Referenced by init(), and switch_path().

                                      {
  ASSERT_ND(bin != kInvalidHashBin);
  ASSERT_ND(cur_path_[fixed_upto_level]->get_bin_range().contains(bin));
  ASSERT_ND(fixed_upto_level < levels_);

  // we need a room for one-page in main buffer, and levels pages in higher-level buffer.
  CHECK_ERROR_CODE(assure_writer_buffer(writer, writer_buffer_pos, *writer_higher_buffer_pos));
  PagePtr higher_base = reinterpret_cast<PagePtr>(writer->get_intermediate_base());
  PagePtr main_base = reinterpret_cast<PagePtr>(writer->get_page_base());

  IntermediateRoute route = IntermediateRoute::construct(bin);
  ASSERT_ND(route.route[levels_] == 0);
  const StorageId storage_id = cur_path_[levels_ - 1U]->header().storage_id_;

  // let's open the page that contains the bin. higher levels first.
  for (uint8_t parent_level = fixed_upto_level; parent_level > 0; --parent_level) {
    uint8_t level = parent_level - 1U;
    PagePtr parent = cur_path_[parent_level];
    ASSERT_ND(parent->get_level() == parent_level);
    ASSERT_ND(parent->get_bin_range().contains(bin));

    const uint16_t index = route.route[parent_level];
    HashBin range_begin = parent->get_bin_range().begin_ + index * kHashMaxBins[parent_level];
    ASSERT_ND(parent->get_bin_range().length() == kHashMaxBins[parent_level + 1U]);

    SnapshotPagePointer old_page_id = parent->get_pointer(index).snapshot_pointer_;
    SnapshotPagePointer new_page_id;
    PagePtr new_page;
    if (level > 0) {
      new_page_id = *writer_higher_buffer_pos;
      new_page = higher_base + new_page_id;
      ++(*writer_higher_buffer_pos);
      ASSERT_ND((*writer_higher_buffer_pos) <= writer->get_intermediate_size());
    } else {
      // Unlike higher-levels, we can finalize the page ID for level-0 pages.
      new_page_id = writer->get_next_page_id() + (*writer_buffer_pos);
      new_page = main_base + *writer_buffer_pos;
      ++(*writer_buffer_pos);
      ASSERT_ND((*writer_buffer_pos) <= writer->get_page_size());
    }
    cur_path_[level] = new_page;

    if (old_page_id == 0) {
      // the page didn't exist before, create a new one.
      new_page->initialize_snapshot_page(storage_id, new_page_id, level, range_begin);
    } else {
      // otherwise, start from the previous page image.
      ASSERT_ND(extract_snapshot_id_from_snapshot_pointer(old_page_id) != snapshot_id_);
      CHECK_ERROR_CODE(fileset->read_page(old_page_id, new_page));
      ASSERT_ND(new_page->get_bin_range().begin_ == range_begin);
      ASSERT_ND(new_page->get_level() == level);
      new_page->header().page_id_ = new_page_id;
    }
    parent->get_pointer(index).snapshot_pointer_ = new_page_id;
  }

  return kErrorCodeOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorCode foedus::storage::hash::ComposedBinsMergedStream::process_a_bin	(	uint32_t *	installed_count,
		HashBin *	next_lowest_bin
	)

Consumes inputs for the cur_path_[0] page and install snapshot pointers there.

Parameters

[in,out]	installed_count	Increments the number of installed pointers
[out]	next_lowest_bin	The smallest bin out of all remaining inputs. kInvalidHashBin if all of them are depleted.

Definition at line 150 of file hash_composed_bins_impl.cpp.

References ASSERT_ND, foedus::storage::hash::HashBinRange::begin_, foedus::storage::hash::ComposedBin::bin_, CHECK_ERROR_CODE, cur_path_, foedus::storage::hash::HashBinRange::end_, foedus::storage::extract_snapshot_id_from_snapshot_pointer(), foedus::storage::hash::HashIntermediatePage::get_bin_range(), foedus::storage::hash::ComposedBinsBuffer::get_cur_bin(), foedus::storage::hash::HashIntermediatePage::get_level(), foedus::storage::hash::HashIntermediatePage::get_pointer_address(), foedus::storage::hash::ComposedBinsBuffer::has_more(), foedus::storage::hash::HashIntermediatePage::header(), input_count_, inputs_, foedus::storage::VolatilePagePointer::is_null(), foedus::kErrorCodeOk, foedus::storage::hash::kHashIntermediatePageFanout, foedus::storage::hash::kInvalidHashBin, foedus::storage::hash::ComposedBinsBuffer::next_bin(), foedus::storage::hash::ComposedBin::page_id_, foedus::storage::PageHeader::snapshot_, snapshot_id_, foedus::storage::DualPagePointer::snapshot_pointer_, and foedus::storage::DualPagePointer::volatile_pointer_.

                            {
  HashIntermediatePage* const page = cur_path_[0];
  ASSERT_ND(page);
  const HashBinRange& cur_range = page->get_bin_range();
  ASSERT_ND(page->get_level() == 0);
  ASSERT_ND(page->header().snapshot_);
  *next_lowest_bin = kInvalidHashBin;
  uint32_t installed_diff = 0;

  for (uint32_t i = 0; i < input_count_; ++i) {
    ComposedBinsBuffer* input = inputs_ + i;
    while (input->has_more()) {
      const ComposedBin& entry = input->get_cur_bin();
      ASSERT_ND(cur_range.begin_ <= entry.bin_);
      ASSERT_ND(entry.page_id_ != 0);
      ASSERT_ND(extract_snapshot_id_from_snapshot_pointer(entry.page_id_) == snapshot_id_);
      if (entry.bin_ >= cur_range.end_) {
        // no more bins for this page from this input
        *next_lowest_bin = std::min<HashBin>(*next_lowest_bin, entry.bin_);
        break;
      }

      uint16_t index = entry.bin_ - cur_range.begin_;
      ASSERT_ND(index < kHashIntermediatePageFanout);
      DualPagePointer* target = page->get_pointer_address(index);
      ASSERT_ND(target->volatile_pointer_.is_null());
      ASSERT_ND(extract_snapshot_id_from_snapshot_pointer(target->snapshot_pointer_)
        != snapshot_id_);
      target->snapshot_pointer_ = entry.page_id_;
      ++installed_diff;

      CHECK_ERROR_CODE(input->next_bin());
    }
  }

  *installed_count += installed_diff;
  return kErrorCodeOk;
}

Here is the call graph for this function:

ErrorCode foedus::storage::hash::ComposedBinsMergedStream::switch_path	(	HashBin	lowest_bin,
		cache::SnapshotFileSet *	fileset,
		snapshot::SnapshotWriter *	writer,
		uint32_t *	writer_buffer_pos,
		uint32_t *	writer_higher_buffer_pos
	)

Moves cur_path_ to a page that cotains the specified bin.

It opens higher-level pages if needed.

Definition at line 191 of file hash_composed_bins_impl.cpp.

References ASSERT_ND, cur_path_, foedus::storage::hash::kInvalidHashBin, levels_, and open_path().

                                      {
  ASSERT_ND(lowest_bin != kInvalidHashBin);
  // because a path in this sub-tree must be switched, this is at least 3-levels.
  // levels_==1 : single-level hash, separately processed in construct_root
  // levels_==2 : then each sub-tree has only one level-0 page. switch_path is never called.
  ASSERT_ND(levels_ >= 3U);
  ASSERT_ND(!cur_path_[0]->get_bin_range().contains(lowest_bin));
  ASSERT_ND(!cur_path_[levels_ - 1U]->get_bin_range().contains(lowest_bin));  // root page
  ASSERT_ND(!cur_path_[levels_ - 2U]->get_bin_range().contains(lowest_bin));  // root_child page

  // where do we have to switch?
  uint8_t valid_upto;
  for (valid_upto = 1U; valid_upto + 2U < levels_; ++valid_upto) {
    if (cur_path_[valid_upto]->get_bin_range().contains(lowest_bin)) {
      break;
    }
  }
  ASSERT_ND(cur_path_[valid_upto]->get_bin_range().contains(lowest_bin));
  ASSERT_ND(!cur_path_[valid_upto - 1U]->get_bin_range().contains(lowest_bin));

  return open_path(
    lowest_bin,
    valid_upto,
    fileset,
    writer,
    writer_buffer_pos,
    writer_higher_buffer_pos);
}

Here is the call graph for this function:

Member Data Documentation

PagePtr foedus::storage::hash::ComposedBinsMergedStream::cur_path_[kHashMaxLevels]

The pages we are now composing.

cur_path_[n] is the level-n intermediate page.

Definition at line 219 of file hash_composed_bins_impl.hpp.

Referenced by init(), open_path(), process_a_bin(), and switch_path().

uint32_t foedus::storage::hash::ComposedBinsMergedStream::input_count_

total number of buffers in inputs_

Definition at line 213 of file hash_composed_bins_impl.hpp.

Referenced by init(), and process_a_bin().

ComposedBinsBuffer* foedus::storage::hash::ComposedBinsMergedStream::inputs_

Definition at line 211 of file hash_composed_bins_impl.hpp.

Referenced by init(), and process_a_bin().

std::unique_ptr< ComposedBinsBuffer[] > foedus::storage::hash::ComposedBinsMergedStream::inputs_memory_

just for auto release

Definition at line 209 of file hash_composed_bins_impl.hpp.

Referenced by init().

uint8_t foedus::storage::hash::ComposedBinsMergedStream::levels_

Definition at line 215 of file hash_composed_bins_impl.hpp.

Referenced by assure_writer_buffer(), init(), open_path(), and switch_path().

uint8_t foedus::storage::hash::ComposedBinsMergedStream::padding_

Definition at line 216 of file hash_composed_bins_impl.hpp.

snapshot::SnapshotId foedus::storage::hash::ComposedBinsMergedStream::snapshot_id_

Definition at line 214 of file hash_composed_bins_impl.hpp.

Referenced by init(), open_path(), and process_a_bin().

---

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

• /home/shino/foedus_code/foedus-core/include/foedus/storage/hash/hash_composed_bins_impl.hpp
• /home/shino/foedus_code/foedus-core/src/foedus/storage/hash/hash_composed_bins_impl.cpp