ArrayComposer's compose() implementation separated from the class itself. More...

Detailed Description

ArrayComposer's compose() implementation separated from the class itself.

It's a complicated method, so worth being its own class. This defines all the variables maintained during one compose() call.

Definition at line 109 of file array_composer_impl.hpp.

#include <array_composer_impl.hpp>

Public Member Functions
	ArrayComposeContext (Engine engine, snapshot::MergeSort merge_sort, snapshot::SnapshotWriter snapshot_writer, cache::SnapshotFileSet previous_snapshot_files, Page *root_info_page)
	ArrayComposeContext methods. More...

ErrorStack	execute ()

Constructor & Destructor Documentation

foedus::storage::array::ArrayComposeContext::ArrayComposeContext	(	Engine *	engine,
		snapshot::MergeSort *	merge_sort,
		snapshot::SnapshotWriter *	snapshot_writer,
		cache::SnapshotFileSet *	previous_snapshot_files,
		Page *	root_info_page
	)

ArrayComposeContext methods.

Definition at line 191 of file array_composer_impl.cpp.

References foedus::storage::array::ArrayPartitionerData::array_levels_, foedus::storage::array::ArrayPartitionerData::array_size_, ASSERT_ND, foedus::storage::array::ArrayStorage::get_array_size(), foedus::snapshot::SnapshotWriter::get_intermediate_base(), foedus::snapshot::SnapshotWriter::get_intermediate_size(), foedus::storage::PartitionerMetadata::get_metadata(), foedus::snapshot::SnapshotWriter::get_page_base(), foedus::snapshot::SnapshotWriter::get_page_size(), foedus::storage::array::LookupRouteFinder::get_records_in_leaf(), foedus::storage::array::kInteriorFanout, foedus::storage::PartitionerMetadata::locate_data(), and foedus::storage::PartitionerMetadata::valid_.

   : engine_(engine),
     merge_sort_(merge_sort),
     system_initial_epoch_(engine->get_savepoint_manager()->get_initial_durable_epoch()),
     storage_id_(merge_sort_->get_storage_id()),
     snapshot_id_(snapshot_writer->get_snapshot_id()),
     storage_(engine, storage_id_),
     snapshot_writer_(snapshot_writer),
     previous_snapshot_files_(previous_snapshot_files),
     root_info_page_(reinterpret_cast<ArrayRootInfoPage*>(root_info_page)),
     payload_size_(storage_.get_payload_size()),
     levels_(storage_.get_levels()),
     previous_root_page_pointer_(storage_.get_metadata()->root_snapshot_page_id_) {
   LookupRouteFinder route_finder(levels_, payload_size_);
   offset_intervals_[0] = route_finder.get_records_in_leaf();
   for (uint8_t level = 1; level < levels_; ++level) {
     offset_intervals_[level] = offset_intervals_[level - 1] * kInteriorFanout;
   }
   std::memset(cur_path_, 0, sizeof(cur_path_));
 
   allocated_pages_ = 0;
   allocated_intermediates_ = 0;
   page_base_ = reinterpret_cast<ArrayPage*>(snapshot_writer_->get_page_base());
   max_pages_ = snapshot_writer_->get_page_size();
   intermediate_base_ = reinterpret_cast<ArrayPage*>(snapshot_writer_->get_intermediate_base());
   max_intermediates_ = snapshot_writer_->get_intermediate_size();
 
   PartitionerMetadata* metadata = PartitionerMetadata::get_metadata(engine_, storage_id_);
   ASSERT_ND(metadata->valid_);  // otherwise composer invoked without partitioner. maybe testcase?
   partitioning_data_ = reinterpret_cast<ArrayPartitionerData*>(metadata->locate_data(engine_));
   ASSERT_ND(levels_ == partitioning_data_->array_levels_);
   ASSERT_ND(storage_.get_array_size() == partitioning_data_->array_size_);
 }

Here is the call graph for this function:

Member Function Documentation

ErrorStack foedus::storage::array::ArrayComposeContext::execute ( )

Definition at line 230 of file array_composer_impl.cpp.

                                         {
   std::memset(root_info_page_, 0, kPageSize);
   root_info_page_->header_.storage_id_ = storage_id_;
 
   if (levels_ <= 1U) {
     // this storage has only one page. This is very special and trivial.
     // we process this case separately.
     return execute_single_level_array();
   }
 
   // This implementation is batch-based to make it significantly more efficient.
   // We receive a fully-sorted/integrated stream of logs from merge_sort_, and merge them with
   // previous snapshot on page-basis. Most pages have many logs, so this achieves a tight loop
   // without expensive cost to switch pages.
   bool processed_any = false;
   while (true) {
     CHECK_ERROR(merge_sort_->next_batch());
     uint64_t count = merge_sort_->get_current_count();
     if (count == 0 && merge_sort_->is_ended_all()) {
       break;
     }
     const snapshot::MergeSort::SortEntry* sort_entries = merge_sort_->get_sort_entries();
     if (!processed_any) {
       // this is the first log. let's initialize cur_path to this log.
       processed_any = true;
       ArrayOffset initial_offset = sort_entries[0].get_key();
 
       CHECK_ERROR(initialize(initial_offset));
     }
 
     uint64_t cur = 0;
     while (cur < count) {
       const ArrayCommonUpdateLogType* head = reinterpret_cast<const ArrayCommonUpdateLogType*>(
         merge_sort_->resolve_sort_position(cur));
       ArrayOffset head_offset = head->offset_;
       ASSERT_ND(head_offset == sort_entries[cur].get_key());
       // switch to a page containing this offset
       WRAP_ERROR_CODE(update_cur_path(head_offset));
       ArrayRange page_range = cur_path_[0]->get_array_range();
       ASSERT_ND(page_range.contains(head_offset));
 
       // grab a range of logs that are in the same page.
       uint64_t next;
       for (next = cur + 1U; LIKELY(next < count); ++next) {
         // this check uses sort_entries which are nicely contiguous.
         ArrayOffset offset = sort_entries[next].get_key();
         ASSERT_ND(offset >= page_range.begin_);
         if (UNLIKELY(offset >= page_range.end_)) {
           break;
         }
       }
 
       apply_batch(cur, next);
       cur = next;
     }
     ASSERT_ND(cur == count);
   }
 
   if (processed_any) {
     CHECK_ERROR(finalize());
   } else {
     LOG(ERROR) << "wtf? no logs? storage-" << storage_id_;
   }
 
 #ifndef NDEBUG
   uint16_t children = get_root_children();
   PartitionId partition = snapshot_writer_->get_numa_node();
   ASSERT_ND(partitioning_data_);
   for (uint16_t i = 0; i < children; ++i) {
     if (!partitioning_data_->partitionable_ || partitioning_data_->bucket_owners_[i] == partition) {
       ASSERT_ND(root_info_page_->pointers_[i] != 0);
     } else {
       ASSERT_ND((!is_initial_snapshot() && root_info_page_->pointers_[i] != 0)
         || (is_initial_snapshot() && root_info_page_->pointers_[i] == 0));
     }
   }
   for (uint16_t i = children; i < kInteriorFanout; ++i) {
     ASSERT_ND(root_info_page_->pointers_[i] == 0);
   }
 #endif  // NDEBUG
 
   return kRetOk;
 }

Here is the call graph for this function:

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

/home/shino/foedus_code/foedus-core/include/foedus/storage/array/array_composer_impl.hpp
/home/shino/foedus_code/foedus-core/src/foedus/storage/array/array_composer_impl.cpp

Detailed Description

Public Member Functions

Constructor & Destructor Documentation

Member Function Documentation