foedus::storage::sequential::SequentialStoragePimpl Class Reference

Lock-free list of records stored in the volatile part of sequential storage. More...

Detailed Description

Lock-free list of records stored in the volatile part of sequential storage.

Volatile parts of sequential storages are completely separated from stable snapshot pages and maintained as a lock-free list of SequentialPage.

The append of record in page involves one atomic CAS, and append of a new page involves one atomic CAS too. Because this list is an append/scan only list, this is way simpler than typical lock-free lists.

Concurrency Control

In this volatile list, we assume the following things to simplify concurrency control:

• Each SequentialPage contains records only in one epoch. When epoch switches, we insert new records to new page even if the page is almost vacant.
• The order in page does not necessarily reflect serialization order. The sequential storage provides a set semantics rather than list semantics in terms of serializability although it's loosely ordered.
• Each thread maintains its own head/tail pages so that they don't interfere each other at all. This, combined with the assumption above, makes it completely without blocking and atomic operations EXCEPT:
• For scanning cursors (which only rarely occur and are fundamentally slow anyways), we take an exclusive lock when they touch unsafe epochs. Further, the scanning thread must wait until all other threads did NOT start before the scanning thread take lock. (otherwise serializability is not guaranteed). We will have something like Xct::InCommitEpochGuard for this purpose. As scanning threads are rare, they can wait for a while, so it's okay for other threads to complete at least one transacion before they get aware of the lock.
• However, the above requirement is not mandatory if the scanning threads are running in snapshot mode or dirty-read mode. We just make sure that what is reads is within the epoch. Because other threads append record in serialization order to their own lists, this can be trivially achieved.

With these assumptions, sequential storages don't require any locking for serializability. Thus, we separate write-sets of sequential storages from other write-sets in transaction objects.

Page replacement policy after snapshot

This is an append-optimized storage, so there is no point to keep volatile pages for future use. We thus drop all volatile pages that were snapshot.

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.

Todo:

Implement the scanning functionality as above.

Definition at line 139 of file sequential_storage_pimpl.hpp.

#include <sequential_storage_pimpl.hpp>

Inheritance diagram for foedus::storage::sequential::SequentialStoragePimpl:

Collaboration diagram for foedus::storage::sequential::SequentialStoragePimpl:

Classes
struct	PointerPage

Public Member Functions
	SequentialStoragePimpl ()=delete
	SequentialStoragePimpl (SequentialStorage *storage)
	SequentialStoragePimpl (Engine *engine, SequentialStorageControlBlock *control_block)
bool	exists () const
StorageId	get_id () const
const StorageName &	get_name () const
ErrorStack	create (const SequentialMetadata &metadata)
ErrorStack	load (const StorageControlBlock &snapshot_block)
ErrorStack	initialize_head_tail_pages ()
ErrorStack	drop ()
ErrorStack	truncate (Epoch new_truncate_epoch, Epoch *commit_epoch)
void	apply_truncate (const SequentialTruncateLogType &the_log)
SequentialPage *	get_head (const memory::LocalPageResolver &resolver, thread::ThreadId thread_id) const
SequentialPage *	get_tail (const memory::LocalPageResolver &resolver, thread::ThreadId thread_id) const
memory::PagePoolOffset *	get_head_pointer (thread::ThreadId thread_id) const
memory::PagePoolOffset *	get_tail_pointer (thread::ThreadId thread_id) const
void	append_record (thread::Thread *context, xct::XctId owner_id, const void *payload, uint16_t payload_count)
	Appends an already-commited record to this volatile list. More...
template<typename HANDLER >
ErrorCode	for_every_page (HANDLER handler) const
	Traverse all pages and call back the handler for every page. More...
Public Member Functions inherited from foedus::Attachable< SequentialStorageControlBlock >
	Attachable ()
	Attachable (Engine *engine)
	Attachable (Engine *engine, SequentialStorageControlBlock *control_block)
	Attachable (SequentialStorageControlBlock *control_block)
	Attachable (const Attachable &other)
virtual	~Attachable ()
Attachable &	operator= (const Attachable &other)
virtual void	attach (SequentialStorageControlBlock *control_block)
	Attaches to the given shared memory. More...
bool	is_attached () const
	Returns whether the object has been already attached to some shared memory. More...
SequentialStorageControlBlock *	get_control_block () const
Engine *	get_engine () const
void	set_engine (Engine *engine)

Additional Inherited Members
Protected Attributes inherited from foedus::Attachable< SequentialStorageControlBlock >
Engine *	engine_
	Most attachable object stores an engine pointer (local engine), so we define it here. More...
SequentialStorageControlBlock *	control_block_
	The shared data on shared memory that has been initialized in some SOC or master engine. More...

Constructor & Destructor Documentation

foedus::storage::sequential::SequentialStoragePimpl::SequentialStoragePimpl ( )

delete

foedus::storage::sequential::SequentialStoragePimpl::SequentialStoragePimpl ( SequentialStorage *  storage )

inlineexplicit

Definition at line 145 of file sequential_storage_pimpl.hpp.

    : Attachable<SequentialStorageControlBlock>(
      storage->get_engine(),
      storage->get_control_block()) {
  }

foedus::storage::sequential::SequentialStoragePimpl::SequentialStoragePimpl ( Engine *  engine, SequentialStorageControlBlock *  control_block  )

inline

Definition at line 150 of file sequential_storage_pimpl.hpp.

    : Attachable<SequentialStorageControlBlock>(engine, control_block) {
  }

Member Function Documentation

void foedus::storage::sequential::SequentialStoragePimpl::append_record	(	thread::Thread *	context,
		xct::XctId	owner_id,
		const void *	payload,
		uint16_t	payload_count
	)

Appends an already-commited record to this volatile list.

This method is guaranteed to succeed, so it does not return error code, which is essential to be used after commit. Actually, there is one very rare case this method might fail: we need a new page and the page pool has zero free page. However, we can trivially avoid this case by checking if we have at least one free page in thread-local cache during pre-commit.

Definition at line 323 of file sequential_storage_pimpl.cpp.

References foedus::storage::sequential::SequentialPage::append_record_nosync(), ASSERT_ND, foedus::storage::sequential::SequentialPage::can_insert_record(), foedus::xct::XctId::get_epoch(), foedus::storage::sequential::SequentialPage::get_first_record_epoch(), get_head_pointer(), get_id(), foedus::thread::Thread::get_local_volatile_page_resolver(), foedus::thread::Thread::get_numa_node(), foedus::storage::sequential::SequentialPage::get_record_count(), get_tail_pointer(), foedus::thread::Thread::get_thread_id(), foedus::thread::Thread::get_thread_memory(), foedus::memory::NumaCoreMemory::grab_free_volatile_page(), foedus::storage::sequential::SequentialPage::initialize_volatile_page(), foedus::storage::sequential::SequentialPage::next_page(), foedus::memory::LocalPageResolver::resolve_offset(), foedus::memory::LocalPageResolver::resolve_offset_newpage(), foedus::storage::VolatilePagePointer::set(), UNLIKELY, and foedus::storage::DualPagePointer::volatile_pointer_.

Referenced by foedus::storage::sequential::SequentialStorage::apply_append_record().

                          {
  thread::ThreadId thread_id = context->get_thread_id();
  thread::ThreadGroupId node = context->get_numa_node();

  // the list is local to this core, so no race possible EXCEPT scanning thread
  // and snapshot thread, but they are read-only or only dropping pages.
  memory::PagePoolOffset* tail_pointer = get_tail_pointer(thread_id);
  ASSERT_ND(tail_pointer);
  SequentialPage* tail = nullptr;
  if (*tail_pointer != 0) {
    tail = reinterpret_cast<SequentialPage*>(
      context->get_local_volatile_page_resolver().resolve_offset(*tail_pointer));
  }
  if (tail == nullptr ||
      !tail->can_insert_record(payload_count) ||
      // note: we make sure no volatile page has records from two epochs.
      // this makes us easy to drop volatile pages after snapshotting.
      (tail->get_record_count() > 0 && tail->get_first_record_epoch() != owner_id.get_epoch())) {
    memory::PagePoolOffset new_page_offset
      = context->get_thread_memory()->grab_free_volatile_page();
    if (UNLIKELY(new_page_offset == 0)) {
      LOG(FATAL) << " Unexpected error. we ran out of free page while inserting to sequential"
        " storage after commit.";
    }
    VolatilePagePointer new_page_pointer;
    new_page_pointer.set(node, new_page_offset);
    SequentialPage* new_page = reinterpret_cast<SequentialPage*>(
      context->get_local_volatile_page_resolver().resolve_offset_newpage(new_page_offset));
    new_page->initialize_volatile_page(get_id(), new_page_pointer);

    if (tail == nullptr) {
      // this is the first access to this head pointer. Let's install the first page.
      ASSERT_ND(*tail_pointer == 0);
      memory::PagePoolOffset* head_pointer = get_head_pointer(thread_id);
      ASSERT_ND(*head_pointer == 0);
      *head_pointer = new_page_offset;
      *tail_pointer = new_page_offset;
    } else {
      ASSERT_ND(*get_head_pointer(thread_id) != 0);
      *tail_pointer = new_page_offset;
      tail->next_page().volatile_pointer_ = new_page_pointer;
    }
    tail = new_page;
  }

  ASSERT_ND(tail &&
    tail->can_insert_record(payload_count) &&
    (tail->get_record_count() == 0 || tail->get_first_record_epoch() == owner_id.get_epoch()));
  tail->append_record_nosync(owner_id, payload_count, payload);
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::storage::sequential::SequentialStoragePimpl::apply_truncate

(

const SequentialTruncateLogType &

the_log

)

Definition at line 313 of file sequential_storage_pimpl.cpp.

References ASSERT_ND, foedus::Attachable< SequentialStorageControlBlock >::control_block_, get_name(), foedus::log::BaseLogType::header_, foedus::storage::sequential::SequentialTruncateLogType::new_truncate_epoch_, foedus::Epoch::value(), and foedus::log::LogHeader::xct_id_.

Referenced by foedus::storage::sequential::SequentialStorage::apply_truncate().

                                                                                    {
  // this method is called only during restart, so no race.
  ASSERT_ND(control_block_->exists());
  control_block_->cur_truncate_epoch_tid_.xct_id_ = the_log.header_.xct_id_;
  control_block_->cur_truncate_epoch_.store(the_log.new_truncate_epoch_.value());
  control_block_->meta_.truncate_epoch_ = the_log.new_truncate_epoch_.value();
  LOG(INFO) << "Applied redo-log of truncation on sequential storage- " << get_name()
    << " epoch=" << control_block_->cur_truncate_epoch_;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::storage::sequential::SequentialStoragePimpl::create

(

const SequentialMetadata &

metadata

)

Definition at line 147 of file sequential_storage_pimpl.cpp.

References CHECK_ERROR, foedus::Attachable< SequentialStorageControlBlock >::control_block_, foedus::Attachable< SequentialStorageControlBlock >::engine_, ERROR_STACK, exists(), foedus::Engine::get_earliest_epoch(), get_name(), initialize_head_tail_pages(), foedus::kErrorCodeStrAlreadyExists, foedus::storage::kExists, foedus::kRetOk, and foedus::Epoch::value().

Referenced by foedus::storage::sequential::SequentialStorage::create().

                                                                            {
  if (exists()) {
    LOG(ERROR) << "This sequential-storage already exists: " << get_name();
    return ERROR_STACK(kErrorCodeStrAlreadyExists);
  }

  control_block_->meta_ = metadata;
  const Epoch initial_truncate_epoch = engine_->get_earliest_epoch();
  control_block_->meta_.truncate_epoch_ = initial_truncate_epoch.value();
  control_block_->cur_truncate_epoch_.store(initial_truncate_epoch.value());
  control_block_->cur_truncate_epoch_tid_.reset();
  control_block_->cur_truncate_epoch_tid_.xct_id_.set_epoch(initial_truncate_epoch);
  control_block_->cur_truncate_epoch_tid_.xct_id_.set_ordinal(1);

  CHECK_ERROR(initialize_head_tail_pages());
  control_block_->status_ = kExists;
  LOG(INFO) << "Newly created a sequential-storage " << get_name();
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::storage::sequential::SequentialStoragePimpl::drop

(

)

Definition at line 86 of file sequential_storage_pimpl.cpp.

References ASSERT_ND, foedus::thread::compose_thread_id(), foedus::Attachable< SequentialStorageControlBlock >::control_block_, foedus::Attachable< SequentialStorageControlBlock >::engine_, foedus::memory::PagePoolOffsetChunk::full(), get_head(), foedus::Engine::get_memory_manager(), get_name(), foedus::memory::EngineMemory::get_node_memory(), foedus::Engine::get_options(), foedus::memory::PagePool::get_resolver(), foedus::memory::NumaNodeMemoryRef::get_volatile_pool(), foedus::thread::ThreadOptions::group_count_, foedus::kRetOk, foedus::memory::PagePoolOffsetChunk::push_back(), foedus::memory::PagePool::release(), foedus::memory::PagePool::release_one(), foedus::memory::LocalPageResolver::resolve_offset(), foedus::memory::PagePoolOffsetChunk::size(), foedus::EngineOptions::thread_, foedus::thread::ThreadOptions::thread_count_per_group_, and foedus::storage::VolatilePagePointer::word.

Referenced by foedus::storage::sequential::SequentialStorage::drop().

                                        {
  LOG(INFO) << "Uninitializing an sequential-storage " << get_name();
  // release all pages in this list.
  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) {
    // we are sure these pages are from only one NUMA node, so we can easily batch-return.
    memory::NumaNodeMemoryRef* memory = engine_->get_memory_manager()->get_node_memory(node);
    memory::PagePool* pool = memory->get_volatile_pool();
    const memory::LocalPageResolver& resolver = pool->get_resolver();
    memory::PagePoolOffsetChunk chunk;
    for (uint16_t local_ordinal = 0; local_ordinal < threads_per_node; ++local_ordinal) {
      thread::ThreadId thread_id = thread::compose_thread_id(node, local_ordinal);
      for (SequentialPage* page = get_head(resolver, thread_id); page;) {
        ASSERT_ND(page->header().page_id_);
        VolatilePagePointer cur_pointer;
        cur_pointer.word = page->header().page_id_;
        ASSERT_ND(page == reinterpret_cast<SequentialPage*>(resolver.resolve_offset(
          cur_pointer.get_offset())));
        ASSERT_ND(node == cur_pointer.get_numa_node());
        VolatilePagePointer next_pointer = page->next_page().volatile_pointer_;
        if (chunk.full()) {
          pool->release(chunk.size(), &chunk);
        }
        ASSERT_ND(!chunk.full());
        chunk.push_back(cur_pointer.get_offset());

        if (!next_pointer.is_null()) {
          ASSERT_ND(node == next_pointer.get_numa_node());
          page = reinterpret_cast<SequentialPage*>(resolver.resolve_offset(
            next_pointer.get_offset()));
        } else {
          page = nullptr;
        }
      }
    }
    if (chunk.size() > 0) {
      pool->release(chunk.size(), &chunk);
    }
    ASSERT_ND(chunk.size() == 0);
  }

  // release pointer pages
  for (uint16_t p = 0; p * 4 < nodes; ++p) {
    uint16_t node = p * 4;
    memory::NumaNodeMemoryRef* memory = engine_->get_memory_manager()->get_node_memory(node);
    memory::PagePool* pool = memory->get_volatile_pool();
    if (!control_block_->head_pointer_pages_[p].is_null()) {
      ASSERT_ND(control_block_->head_pointer_pages_[p].get_numa_node() == node);
      pool->release_one(control_block_->head_pointer_pages_[p].get_offset());
    }
    if (!control_block_->tail_pointer_pages_[p].is_null()) {
      ASSERT_ND(control_block_->tail_pointer_pages_[p].get_numa_node() == node);
      pool->release_one(control_block_->tail_pointer_pages_[p].get_offset());
    }
  }
  std::memset(control_block_->head_pointer_pages_, 0, sizeof(control_block_->head_pointer_pages_));
  std::memset(control_block_->tail_pointer_pages_, 0, sizeof(control_block_->tail_pointer_pages_));
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool foedus::storage::sequential::SequentialStoragePimpl::exists ( ) const

inline

Definition at line 154 of file sequential_storage_pimpl.hpp.

References foedus::Attachable< SequentialStorageControlBlock >::control_block_.

Referenced by create().

{ return control_block_->exists(); }

Here is the caller graph for this function:

template<typename HANDLER >

ErrorCode foedus::storage::sequential::SequentialStoragePimpl::for_every_page ( HANDLER  handler ) const

inline

Traverse all pages and call back the handler for every page.

Handler must look like "ErrorCode func(SequentialPage* page) { ... } ".

Definition at line 196 of file sequential_storage_pimpl.hpp.

References ASSERT_ND, CHECK_ERROR_CODE, foedus::thread::compose_thread_id(), foedus::Attachable< SequentialStorageControlBlock >::engine_, get_head(), foedus::Engine::get_memory_manager(), foedus::memory::EngineMemory::get_node_memory(), foedus::storage::VolatilePagePointer::get_offset(), foedus::Engine::get_options(), foedus::memory::PagePool::get_resolver(), foedus::memory::NumaNodeMemoryRef::get_volatile_pool(), foedus::thread::ThreadOptions::group_count_, foedus::kErrorCodeOk, foedus::memory::LocalPageResolver::resolve_offset(), foedus::EngineOptions::thread_, and foedus::thread::ThreadOptions::thread_count_per_group_.

Referenced by foedus::storage::sequential::operator<<().

                                                  {
    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) {
      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);
        for (SequentialPage* page = get_head(resolver, thread_id); page;) {
          ASSERT_ND(page->header().page_id_);
          CHECK_ERROR_CODE(handler(page));

          VolatilePagePointer next_pointer = page->next_page().volatile_pointer_;
          memory::PagePoolOffset offset = next_pointer.get_offset();
          if (offset != 0) {
            page = reinterpret_cast<SequentialPage*>(resolver.resolve_offset(offset));
          } else {
            page = nullptr;
          }
        }
      }
    }
    return kErrorCodeOk;
  }

Here is the call graph for this function:

Here is the caller graph for this function:

SequentialPage * foedus::storage::sequential::SequentialStoragePimpl::get_head	(	const memory::LocalPageResolver &	resolver,
		thread::ThreadId	thread_id
	)		const

Definition at line 407 of file sequential_storage_pimpl.cpp.

References get_head_pointer(), and foedus::memory::LocalPageResolver::resolve_offset().

Referenced by drop(), and for_every_page().

                                  {
  memory::PagePoolOffset offset = *get_head_pointer(thread_id);
  if (offset == 0) {
    return nullptr;
  }
  return reinterpret_cast<SequentialPage*>(resolver.resolve_offset(offset));
}

Here is the call graph for this function:

Here is the caller graph for this function:

memory::PagePoolOffset * foedus::storage::sequential::SequentialStoragePimpl::get_head_pointer

(

thread::ThreadId

thread_id

)

const

Definition at line 378 of file sequential_storage_pimpl.cpp.

References ASSERT_ND, foedus::Attachable< SequentialStorageControlBlock >::control_block_, foedus::thread::decompose_numa_local_ordinal(), foedus::thread::decompose_numa_node(), foedus::Attachable< SequentialStorageControlBlock >::engine_, foedus::memory::EngineMemory::get_global_volatile_page_resolver(), foedus::Engine::get_memory_manager(), foedus::Engine::get_options(), foedus::storage::sequential::get_pointer_page_and_index(), foedus::thread::ThreadOptions::group_count_, foedus::storage::sequential::SequentialStoragePimpl::PointerPage::pointers_, foedus::memory::GlobalVolatilePageResolver::resolve_offset_newpage(), foedus::EngineOptions::thread_, and foedus::thread::ThreadOptions::thread_count_per_group_.

Referenced by append_record(), and get_head().

                                                                                             {
  ASSERT_ND(thread::decompose_numa_node(thread_id) < engine_->get_options().thread_.group_count_);
  ASSERT_ND(thread::decompose_numa_local_ordinal(thread_id)
    < engine_->get_options().thread_.thread_count_per_group_);
  uint16_t page;
  uint16_t index;
  get_pointer_page_and_index(thread_id, &page, &index);
  const memory::GlobalVolatilePageResolver& resolver
    = engine_->get_memory_manager()->get_global_volatile_page_resolver();
  PointerPage* head_page
    = reinterpret_cast<PointerPage*>(resolver.resolve_offset_newpage(
      control_block_->head_pointer_pages_[page]));
  return head_page->pointers_ + index;
}

Here is the call graph for this function:

Here is the caller graph for this function:

StorageId foedus::storage::sequential::SequentialStoragePimpl::get_id ( ) const

inline

Definition at line 155 of file sequential_storage_pimpl.hpp.

References foedus::Attachable< SequentialStorageControlBlock >::control_block_.

Referenced by append_record(), and truncate().

{ return control_block_->meta_.id_; }

Here is the caller graph for this function:

const StorageName& foedus::storage::sequential::SequentialStoragePimpl::get_name ( ) const

inline

Definition at line 156 of file sequential_storage_pimpl.hpp.

References foedus::Attachable< SequentialStorageControlBlock >::control_block_.

Referenced by apply_truncate(), create(), drop(), load(), and truncate().

{ return control_block_->meta_.name_; }

Here is the caller graph for this function:

SequentialPage * foedus::storage::sequential::SequentialStoragePimpl::get_tail	(	const memory::LocalPageResolver &	resolver,
		thread::ThreadId	thread_id
	)		const

Definition at line 417 of file sequential_storage_pimpl.cpp.

References get_tail_pointer(), and foedus::memory::LocalPageResolver::resolve_offset().

                                  {
  memory::PagePoolOffset offset = *get_tail_pointer(thread_id);
  if (offset == 0) {
    return nullptr;
  }
  return reinterpret_cast<SequentialPage*>(resolver.resolve_offset(offset));
}

Here is the call graph for this function:

memory::PagePoolOffset * foedus::storage::sequential::SequentialStoragePimpl::get_tail_pointer

(

thread::ThreadId

thread_id

)

const

Definition at line 392 of file sequential_storage_pimpl.cpp.

References ASSERT_ND, foedus::Attachable< SequentialStorageControlBlock >::control_block_, foedus::thread::decompose_numa_local_ordinal(), foedus::thread::decompose_numa_node(), foedus::Attachable< SequentialStorageControlBlock >::engine_, foedus::memory::EngineMemory::get_global_volatile_page_resolver(), foedus::Engine::get_memory_manager(), foedus::Engine::get_options(), foedus::storage::sequential::get_pointer_page_and_index(), foedus::thread::ThreadOptions::group_count_, foedus::storage::sequential::SequentialStoragePimpl::PointerPage::pointers_, foedus::memory::GlobalVolatilePageResolver::resolve_offset_newpage(), foedus::EngineOptions::thread_, and foedus::thread::ThreadOptions::thread_count_per_group_.

Referenced by append_record(), and get_tail().

                                                                                             {
  ASSERT_ND(thread::decompose_numa_node(thread_id) < engine_->get_options().thread_.group_count_);
  ASSERT_ND(thread::decompose_numa_local_ordinal(thread_id)
    < engine_->get_options().thread_.thread_count_per_group_);
  uint16_t page;
  uint16_t index;
  get_pointer_page_and_index(thread_id, &page, &index);
  const memory::GlobalVolatilePageResolver& resolver
    = engine_->get_memory_manager()->get_global_volatile_page_resolver();
  PointerPage* tail_page
    = reinterpret_cast<PointerPage*>(resolver.resolve_offset_newpage(
      control_block_->tail_pointer_pages_[page]));
  return tail_page->pointers_ + index;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::storage::sequential::SequentialStoragePimpl::initialize_head_tail_pages

(

)

Definition at line 188 of file sequential_storage_pimpl.cpp.

References foedus::Attachable< SequentialStorageControlBlock >::control_block_, foedus::Attachable< SequentialStorageControlBlock >::engine_, foedus::Engine::get_memory_manager(), foedus::memory::EngineMemory::get_node_memory(), foedus::Engine::get_options(), foedus::memory::PagePool::get_resolver(), foedus::memory::NumaNodeMemoryRef::get_volatile_pool(), foedus::memory::PagePool::grab_one(), foedus::thread::ThreadOptions::group_count_, foedus::storage::kPageSize, foedus::kRetOk, foedus::memory::LocalPageResolver::resolve_offset_newpage(), foedus::EngineOptions::thread_, and WRAP_ERROR_CODE.

Referenced by create(), and load().

                                                              {
  std::memset(control_block_->head_pointer_pages_, 0, sizeof(control_block_->head_pointer_pages_));
  std::memset(control_block_->tail_pointer_pages_, 0, sizeof(control_block_->tail_pointer_pages_));
  // we pre-allocate pointer pages for all required nodes.
  // 2^10 pointers (threads) per page : 4 nodes per page

  uint32_t nodes = engine_->get_options().thread_.group_count_;
  for (uint16_t p = 0; p * 4 < nodes; ++p) {
    uint16_t node = p * 4;
    memory::NumaNodeMemoryRef* memory = engine_->get_memory_manager()->get_node_memory(node);
    memory::PagePool* pool = memory->get_volatile_pool();
    memory::PagePoolOffset head_offset, tail_offset;
    // minor todo: gracefully fail in case of out of memory
    WRAP_ERROR_CODE(pool->grab_one(&head_offset));
    WRAP_ERROR_CODE(pool->grab_one(&tail_offset));
    control_block_->head_pointer_pages_[p].set(node, head_offset);
    control_block_->tail_pointer_pages_[p].set(node, tail_offset);
    void* head_page =  pool->get_resolver().resolve_offset_newpage(head_offset);
    void* tail_page =  pool->get_resolver().resolve_offset_newpage(tail_offset);
    std::memset(head_page, 0, kPageSize);
    std::memset(tail_page, 0, kPageSize);
  }
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::storage::sequential::SequentialStoragePimpl::load

(

const StorageControlBlock &

snapshot_block

)

Definition at line 166 of file sequential_storage_pimpl.cpp.

References ASSERT_ND, CHECK_ERROR, foedus::Attachable< SequentialStorageControlBlock >::control_block_, foedus::Attachable< SequentialStorageControlBlock >::engine_, foedus::Engine::get_earliest_epoch(), get_name(), initialize_head_tail_pages(), foedus::Epoch::is_valid(), foedus::storage::kExists, foedus::kRetOk, foedus::storage::StorageControlBlock::meta_, and foedus::Epoch::value().

Referenced by foedus::storage::sequential::SequentialStorage::load().

                                                                                 {
  // for sequential storage, whether the snapshot root pointer is null or not doesn't matter.
  // essentially load==create, except that it just sets the snapshot root pointer.
  control_block_->meta_ = static_cast<const SequentialMetadata&>(snapshot_block.meta_);
  Epoch initial_truncate_epoch(control_block_->meta_.truncate_epoch_);
  if (!initial_truncate_epoch.is_valid()) {
    initial_truncate_epoch = engine_->get_earliest_epoch();
  }
  ASSERT_ND(initial_truncate_epoch.is_valid());
  control_block_->meta_.truncate_epoch_ = initial_truncate_epoch.value();
  control_block_->cur_truncate_epoch_.store(initial_truncate_epoch.value());
  control_block_->cur_truncate_epoch_tid_.reset();
  control_block_->cur_truncate_epoch_tid_.xct_id_.set_epoch(initial_truncate_epoch);
  control_block_->cur_truncate_epoch_tid_.xct_id_.set_ordinal(1);

  CHECK_ERROR(initialize_head_tail_pages());
  control_block_->root_page_pointer_.snapshot_pointer_
    = control_block_->meta_.root_snapshot_page_id_;
  control_block_->status_ = kExists;
  LOG(INFO) << "Loaded a sequential-storage " << get_name();
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::storage::sequential::SequentialStoragePimpl::truncate	(	Epoch	new_truncate_epoch,
		Epoch *	commit_epoch
	)

Definition at line 236 of file sequential_storage_pimpl.cpp.

References foedus::log::MetaLogBuffer::commit(), foedus::Attachable< SequentialStorageControlBlock >::control_block_, foedus::Attachable< SequentialStorageControlBlock >::engine_, ERROR_STACK, foedus::Engine::get_current_global_epoch(), get_id(), foedus::Engine::get_log_manager(), foedus::log::LogManager::get_meta_buffer(), get_name(), foedus::log::BaseLogType::header_, foedus::Epoch::is_valid(), foedus::kErrorCodeInvalidParameter, foedus::kRetOk, foedus::log::LogHeader::log_length_, foedus::log::LogHeader::log_type_code_, foedus::assorted::memory_fence_acq_rel(), foedus::assorted::memory_fence_release(), foedus::storage::sequential::SequentialTruncateLogType::new_truncate_epoch_, foedus::xct::XctId::set(), foedus::log::LogHeader::storage_id_, and foedus::Epoch::value().

Referenced by foedus::storage::sequential::SequentialStorage::truncate().

                                                                                         {
  LOG(INFO) << "Truncating " << get_name() << " upto Epoch " << new_truncate_epoch
    << ". old value=" << control_block_->cur_truncate_epoch_;
  if (!new_truncate_epoch.is_valid()) {
    LOG(ERROR) << "truncate() was called with an invalid epoch";
    return ERROR_STACK(kErrorCodeInvalidParameter);
  } else if (new_truncate_epoch < engine_->get_earliest_epoch()) {
    LOG(ERROR) << "too-old epoch for this system. " << new_truncate_epoch;
    return ERROR_STACK(kErrorCodeInvalidParameter);
  }

  // will check this again after locking.
  if (Epoch(control_block_->cur_truncate_epoch_) >= new_truncate_epoch) {
    LOG(INFO) << "Already truncated up to " << Epoch(control_block_->cur_truncate_epoch_)
      << ". Requested = " << new_truncate_epoch;
    return kRetOk;
  }

  if (new_truncate_epoch > engine_->get_current_global_epoch()) {
    LOG(WARNING) << "Ohh? we don't prohibit it, but are you sure? Truncating up to a future"
      << " epoch-" << new_truncate_epoch << ". cur_global=" << engine_->get_current_global_epoch();
  }


  // We lock it first so that there are no concurrent truncate.
  {
    // TODO(Hideaki) Ownerless-lock here
    // xct::McsOwnerlessLockScope scope(&control_block_->cur_truncate_epoch_tid_);

    if (Epoch(control_block_->cur_truncate_epoch_) >= new_truncate_epoch) {
      LOG(INFO) << "Already truncated up to " << Epoch(control_block_->cur_truncate_epoch_)
        << ". Requested = " << new_truncate_epoch;
      return kRetOk;
    }

    // First, let scanner know that something is happening.
    // 1. Scanners that didn't observe this and commit before us: fine.
    // 2. Scanners that didn't observe this and commit after us:
    //      will see this flag and abort in precommit, fine.
    // 3. Scanners that observe this:
    //      spin until we are done, fine (see optimistic_read_truncate_epoch()).
    // NOTE: Below, we must NOT have any error-return path. Otherwise being_written state is left.
    control_block_->cur_truncate_epoch_tid_.xct_id_.set_being_written();
    assorted::memory_fence_acq_rel();

    // Log this operation as a metadata operation. We get a commit_epoch here.
    {
      char log_buffer[sizeof(SequentialTruncateLogType)];
      std::memset(log_buffer, 0, sizeof(log_buffer));
      SequentialTruncateLogType* the_log = reinterpret_cast<SequentialTruncateLogType*>(log_buffer);
      the_log->header_.storage_id_ = get_id();
      the_log->header_.log_type_code_ = log::get_log_code<SequentialTruncateLogType>();
      the_log->header_.log_length_ = sizeof(SequentialTruncateLogType);
      the_log->new_truncate_epoch_ = new_truncate_epoch;
      engine_->get_log_manager()->get_meta_buffer()->commit(the_log, commit_epoch);
    }

    // Then, apply it. This also clears the being_written flag
    {
      xct::XctId xct_id;
      xct_id.set(commit_epoch->value(), 1);  // no dependency, so minimal ordinal is always correct
      control_block_->cur_truncate_epoch_.store(new_truncate_epoch.value());  // atomic!
      control_block_->cur_truncate_epoch_tid_.xct_id_ = xct_id;
      assorted::memory_fence_release();

      // Also set to the metadata to make this permanent.
      // The metadata will be written out in next snapshot.
      // Until that, REDO-operation below will re-apply that after crash.
      control_block_->meta_.truncate_epoch_ = new_truncate_epoch.value();
      assorted::memory_fence_release();
    }
  }

  LOG(INFO) << "Truncated";
  return kRetOk;
}

Here is the call graph for this function:

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_storage_pimpl.hpp
• /home/shino/foedus_code/foedus-core/src/foedus/storage/sequential/sequential_storage_pimpl.cpp