Composer for an sequential storage. More...

Detailed Description

Composer for an sequential storage.

Like partitioner, this does a quite simple stuff. We don't need to do any merge-sort as there is no order. We just sequentially add them all.

Page allcation in compose(): This composer sequentially writes out data pages until the main buffer in snapshot_writer_ becomes full. Whenever it does, it writes out all the pages and treat the first page as one head page. So, this compose() can output more than one head pages. By doing this, we don't have to worry about any of the intermediate pages and pointer installations. Sooooo simple. The limit is of course 500 pointers (4kb), but surely it will fit. If it doesn't, we must consider allowing variable-sized root info page.

Note: This is a private implementation-details of Sequential Storage, 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 60 of file sequential_composer_impl.hpp.

#include <sequential_composer_impl.hpp>

Classes
struct	RootInfoPage
	Output of one compose() call, which are then combined in construct_root(). More...

Public Member Functions
	SequentialComposer (Composer *parent)

std::string	to_string () const

ErrorStack	compose (const Composer::ComposeArguments &args)

ErrorStack	construct_root (const Composer::ConstructRootArguments &args)

Composer::DropResult	drop_volatiles (const Composer::DropVolatilesArguments &args)

Constructor & Destructor Documentation

foedus::storage::sequential::SequentialComposer::SequentialComposer ( Composer * parent )

explicit

Definition at line 45 of file sequential_composer_impl.cpp.

46 : engine_(parent->get_engine()), storage_id_(parent->get_storage_id()) {

47 }

Member Function Documentation

ErrorStack foedus::storage::sequential::SequentialComposer::compose ( const Composer::ComposeArguments & args )

Definition at line 154 of file sequential_composer_impl.cpp.

Referenced by foedus::storage::Composer::compose().

                                                                            {
   debugging::StopWatch stop_watch;
 
   // Everytime it's full, we write out all pages. much simpler than other storage types.
   // No intermediate pages to track any information.
   snapshot::SnapshotWriter* snapshot_writer = args.snapshot_writer_;
   SequentialPage* base = reinterpret_cast<SequentialPage*>(snapshot_writer->get_page_base());
 
   SequentialPage* cur_page = compose_new_head(snapshot_writer);
   SnapshotPagePointer head_page_id = cur_page->header().page_id_;
   uint32_t allocated_pages = 1;
   uint64_t total_pages = 0;
   const uint32_t max_pages = snapshot_writer->get_page_size();
   VLOG(0) << to_string() << " composing with " << args.log_streams_count_ << " streams.";
   Epoch min_epoch;
   Epoch max_epoch;
   for (uint32_t i = 0; i < args.log_streams_count_; ++i) {
     StreamStatus status;
     WRAP_ERROR_CODE(status.init(args.log_streams_[i]));
     while (!status.ended_) {
       const SequentialAppendLogType* entry = status.get_entry();
       Epoch epoch = entry->header_.xct_id_.get_epoch();
       min_epoch.store_min(epoch);
       max_epoch.store_max(epoch);
 
       // need to allocate a new page?
       if (!cur_page->can_insert_record(entry->payload_count_)) {
         // need to flush the buffer?
         if (allocated_pages >= max_pages) {
           // dump everything and allocate a new head page
           CHECK_ERROR(dump_pages(snapshot_writer, false, allocated_pages, &total_pages));
           cur_page = compose_new_head(snapshot_writer);
           allocated_pages = 1;
         } else {
           // sequential storage is a bit special. As every page is written-once, we need only
           // snapshot pointer. No dual page pointers.
           SequentialPage* next_page = base + allocated_pages;
           ++allocated_pages;
           next_page->initialize_snapshot_page(storage_id_, cur_page->header().page_id_ + 1ULL);
           cur_page->next_page().snapshot_pointer_ = next_page->header().page_id_;
           cur_page = next_page;
           ASSERT_ND(extract_numa_node_from_snapshot_pointer(cur_page->header().page_id_)
               == snapshot_writer->get_numa_node());
           ASSERT_ND(extract_snapshot_id_from_snapshot_pointer(cur_page->header().page_id_)
               == snapshot_writer->get_snapshot_id());
         }
       }
 
       ASSERT_ND(cur_page->can_insert_record(entry->payload_count_));
       cur_page->append_record_nosync(
         status.cur_owner_id_,
         entry->payload_count_,
         status.cur_payload_);
 
       // then, read next
       WRAP_ERROR_CODE(status.next());
     }
   }
   // dump everything
   CHECK_ERROR(dump_pages(snapshot_writer, true, allocated_pages, &total_pages));
 
   // this compose() emits just one pointer to the head page.
   RootInfoPage* root_info_page_casted = reinterpret_cast<RootInfoPage*>(args.root_info_page_);
   root_info_page_casted->header_.storage_id_ = storage_id_;
   root_info_page_casted->pointer_.page_id_ = head_page_id;
   root_info_page_casted->pointer_.from_epoch_ = min_epoch;
   root_info_page_casted->pointer_.to_epoch_ = max_epoch.one_more();  // to make it exclusive
   root_info_page_casted->pointer_.page_count_ = total_pages;
 
   stop_watch.stop();
   LOG(INFO) << to_string() << " compose() done in " << stop_watch.elapsed_ms() << "ms. # pages="
     << total_pages << ", head_page=" << assorted::Hex(head_page_id, 16)
       << ", min_epoch=" << min_epoch << ", max_epoch=" << max_epoch;
   return kRetOk;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::storage::sequential::SequentialComposer::construct_root ( const Composer::ConstructRootArguments & args )

Definition at line 230 of file sequential_composer_impl.cpp.

Referenced by foedus::storage::Composer::construct_root().

                                                                                         {
   debugging::StopWatch stop_watch;
 
   snapshot::SnapshotWriter* snapshot_writer = args.snapshot_writer_;
   std::vector<HeadPagePointer> all_head_pages;
   SequentialStorage storage(engine_, storage_id_);
   SnapshotPagePointer previous_root_page_pointer = storage.get_metadata()->root_snapshot_page_id_;
   for (SnapshotPagePointer page_id = previous_root_page_pointer; page_id != 0;) {
     // if there already is a root page, read them all.
     // we have to anyway re-write all of them, at least the next pointer.
     SequentialRootPage* root_page = reinterpret_cast<SequentialRootPage*>(
       args.gleaner_resource_->work_memory_.get_block());
     WRAP_ERROR_CODE(args.previous_snapshot_files_->read_page(page_id, root_page));
     ASSERT_ND(root_page->header().storage_id_ == storage_id_);
     ASSERT_ND(root_page->header().page_id_ == page_id);
     for (uint16_t i = 0; i < root_page->get_pointer_count(); ++i) {
       all_head_pages.push_back(root_page->get_pointers()[i]);
     }
     page_id = root_page->get_next_page();
   }
 
   // each root_info_page contains just one pointer to the head page.
   for (uint32_t i = 0; i < args.root_info_pages_count_; ++i) {
     const RootInfoPage* info_page = reinterpret_cast<const RootInfoPage*>(args.root_info_pages_[i]);
     ASSERT_ND(info_page->pointer_.page_id_ > 0);
     all_head_pages.push_back(info_page->pointer_);
   }
   VLOG(0) << to_string() << " construct_root() total head page pointers=" << all_head_pages.size();
 
   // now simply write out root pages that contain these pointers.
   SequentialRootPage* base = reinterpret_cast<SequentialRootPage*>(
     snapshot_writer->get_page_base());
   SequentialRootPage* cur_page = base;
   uint32_t allocated_pages = 1;
   SnapshotPagePointer root_of_root_page_id = snapshot_writer->get_next_page_id();
   cur_page->initialize_snapshot_page(storage_id_, root_of_root_page_id);
   for (uint32_t written_pointers = 0; written_pointers < all_head_pages.size();) {
     uint16_t count_in_this_page = std::min<uint64_t>(
       all_head_pages.size() - written_pointers,
       kRootPageMaxHeadPointers);
     cur_page->set_pointers(&all_head_pages[written_pointers], count_in_this_page);
     written_pointers += count_in_this_page;
     if (written_pointers < all_head_pages.size()) {
       // we need next page in root page.
       SequentialRootPage* new_page = cur_page + 1;
       new_page->initialize_snapshot_page(storage_id_, cur_page->header().page_id_ + 1);
       ASSERT_ND(extract_numa_node_from_snapshot_pointer(new_page->header().page_id_)
           == snapshot_writer->get_numa_node());
       ASSERT_ND(extract_snapshot_id_from_snapshot_pointer(new_page->header().page_id_)
           == snapshot_writer->get_snapshot_id());
       cur_page->set_next_page(new_page->header().page_id_);
       cur_page = new_page;
       ++allocated_pages;
     } else {
       ASSERT_ND(written_pointers == all_head_pages.size());
     }
   }
 
   // write out the new root pages
   WRAP_ERROR_CODE(snapshot_writer->dump_pages(0, allocated_pages));
   ASSERT_ND(snapshot_writer->get_next_page_id() == root_of_root_page_id + allocated_pages);
   *args.new_root_page_pointer_ = root_of_root_page_id;
 
   // In sequential, there is only one snapshot pointer to install, the root page.
   storage.get_control_block()->root_page_pointer_.snapshot_pointer_ = root_of_root_page_id;
   storage.get_control_block()->meta_.root_snapshot_page_id_ = root_of_root_page_id;
 
   stop_watch.stop();
   VLOG(0) << to_string() << " construct_root() done in " << stop_watch.elapsed_us() << "us."
     << " total head page pointers=" << all_head_pages.size()
     << ". new root head page=" << assorted::Hex(*args.new_root_page_pointer_)
     << ". root_page_count=" << allocated_pages;
   return kRetOk;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Composer::DropResult foedus::storage::sequential::SequentialComposer::drop_volatiles ( const Composer::DropVolatilesArguments & args )

Definition at line 309 of file sequential_composer_impl.cpp.

References ASSERT_ND, foedus::storage::combine_volatile_page_pointer(), foedus::thread::compose_thread_id(), foedus::storage::Composer::DropVolatilesArguments::drop(), foedus::Attachable< CONTROL_BLOCK >::get_control_block(), foedus::storage::sequential::SequentialPage::get_first_record_epoch(), foedus::Engine::get_memory_manager(), foedus::memory::EngineMemory::get_node_memory(), foedus::storage::VolatilePagePointer::get_numa_node(), foedus::storage::VolatilePagePointer::get_offset(), foedus::Engine::get_options(), foedus::storage::sequential::SequentialPage::get_record_count(), foedus::memory::PagePool::get_resolver(), foedus::memory::NumaNodeMemoryRef::get_volatile_pool(), foedus::thread::ThreadOptions::group_count_, foedus::storage::Composer::DropVolatilesArguments::my_partition_, foedus::storage::sequential::SequentialPage::next_page(), foedus::storage::Composer::DropVolatilesArguments::partitioned_drop_, foedus::memory::LocalPageResolver::resolve_offset(), foedus::storage::Composer::DropVolatilesArguments::snapshot_, foedus::EngineOptions::thread_, foedus::thread::ThreadOptions::thread_count_per_group_, foedus::snapshot::Snapshot::valid_until_epoch_, and foedus::storage::DualPagePointer::volatile_pointer_.

Referenced by foedus::storage::Composer::drop_volatiles().

                                               {
   // In sequential, no need to determine what volatile pages to keep.
   SequentialStorage storage(engine_, storage_id_);
   SequentialStoragePimpl pimpl(engine_, storage.get_control_block());
   uint16_t nodes = engine_->get_options().thread_.group_count_;
   uint16_t threads_per_node = engine_->get_options().thread_.thread_count_per_group_;
   for (uint16_t node = 0; node < nodes; ++node) {
     if (args.partitioned_drop_ && args.my_partition_ != node) {
       continue;
     }
     const memory::LocalPageResolver& resolver
       = engine_->get_memory_manager()->get_node_memory(node)->get_volatile_pool()->get_resolver();
     for (uint16_t local_ordinal = 0; local_ordinal < threads_per_node; ++local_ordinal) {
       thread::ThreadId thread_id = thread::compose_thread_id(node, local_ordinal);
       memory::PagePoolOffset* head_ptr = pimpl.get_head_pointer(thread_id);
       memory::PagePoolOffset* tail_ptr = pimpl.get_tail_pointer(thread_id);
       memory::PagePoolOffset tail_offset = *tail_ptr;
       if ((*head_ptr) == 0) {
         ASSERT_ND(tail_offset == 0);
         VLOG(0) << "No volatile pages for thread-" << thread_id << " in sequential-" << storage_id_;
         continue;
       }
 
       ASSERT_ND(tail_offset != 0);
       while (true) {
         memory::PagePoolOffset offset = *head_ptr;
         ASSERT_ND(offset != 0);
 
         // if the page is newer than the snapshot, keep them.
         // all volatile pages/records are appended in epoch order, so no need to check further.
         SequentialPage* head = reinterpret_cast<SequentialPage*>(resolver.resolve_offset(offset));
         ASSERT_ND(head->get_record_count() > 0);
         if (head->get_record_count() > 0
           && head->get_first_record_epoch() > args.snapshot_.valid_until_epoch_) {
           VLOG(0) << "Thread-" << thread_id << " in sequential-" << storage_id_ << " keeps volatile"
             << " pages at and after epoch-" << head->get_first_record_epoch();
           break;
         }
 
         // okay, drop this
         memory::PagePoolOffset next = head->next_page().volatile_pointer_.get_offset();
         ASSERT_ND(next != offset);
         ASSERT_ND(next == 0 || head->next_page().volatile_pointer_.get_numa_node() == node);
         args.drop(engine_, combine_volatile_page_pointer(node, offset));
         if (next == 0) {
           // it was the tail
           ASSERT_ND(tail_offset == offset);
           VLOG(0) << "Thread-" << thread_id << " in sequential-" << storage_id_ << " dropped all"
             << " volatile pages";
           *head_ptr = 0;
           *tail_ptr = 0;
           break;
         } else {
           // move head
           *head_ptr = next;
           DVLOG(2) << "Thread-" << thread_id << " in sequential-" << storage_id_ << " dropped a"
             << " page.";
         }
       }
     }
   }
   return Composer::DropResult(args);  // always everything dropped
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::string foedus::storage::sequential::SequentialComposer::to_string ( ) const

Definition at line 305 of file sequential_composer_impl.cpp.

Referenced by compose(), and construct_root().

                                               {
   return std::string("SequentialComposer-") + std::to_string(storage_id_);
 }

Here is the caller graph for this function:

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

/home/shino/foedus_code/foedus-core/include/foedus/storage/sequential/sequential_composer_impl.hpp
/home/shino/foedus_code/foedus-core/src/foedus/storage/sequential/sequential_composer_impl.cpp

Detailed Description

Classes

Public Member Functions

Constructor & Destructor Documentation

Member Function Documentation