foedus::snapshot::SnapshotManagerPimpl Class Reference

Pimpl object of SnapshotManager. More...

Detailed Description

Pimpl object of SnapshotManager.

A private pimpl object for SnapshotManager. Do not include this header from a client program unless you know what you are doing.

Definition at line 204 of file snapshot_manager_pimpl.hpp.

#include <snapshot_manager_pimpl.hpp>

Inheritance diagram for foedus::snapshot::SnapshotManagerPimpl:

Collaboration diagram for foedus::snapshot::SnapshotManagerPimpl:

Public Member Functions
	SnapshotManagerPimpl ()=delete
	SnapshotManagerPimpl (Engine *engine)
ErrorStack	initialize_once () override
ErrorStack	uninitialize_once () override
const SnapshotOptions &	get_option () const
	shorthand for engine_->get_options().snapshot_. More...
Epoch	get_snapshot_epoch () const
Epoch	get_snapshot_epoch_weak () const
SnapshotId	get_previous_snapshot_id () const
SnapshotId	get_previous_snapshot_id_weak () const
ErrorStack	read_snapshot_metadata (SnapshotId snapshot_id, SnapshotMetadata *out)
void	trigger_snapshot_immediate (bool wait_completion, Epoch suggested_snapshot_epoch)
void	stop_snapshot_thread ()
	This is a hidden API called at the beginning of engine shutdown (namely restart manager). More...
SnapshotId	issue_next_snapshot_id ()
void	wakeup ()
void	sleep_a_while ()
bool	is_stop_requested () const
bool	is_gleaning () const
void	handle_snapshot ()
	Main routine for snapshot_thread_ in master engine. More...
ErrorStack	handle_snapshot_triggered (Snapshot *new_snapshot)
	handle_snapshot() calls this when it should start snapshotting. More...
void	handle_snapshot_child ()
	Main routine for snapshot_thread_ in child engines. More...
ErrorStack	glean_logs (const Snapshot &new_snapshot, std::map< storage::StorageId, storage::SnapshotPagePointer > *new_root_page_pointers)
	Sub-routine of handle_snapshot_triggered(). More...
ErrorStack	snapshot_metadata (const Snapshot &new_snapshot, const std::map< storage::StorageId, storage::SnapshotPagePointer > &new_root_page_pointers)
	Sub-routine of handle_snapshot_triggered(). More...
ErrorStack	snapshot_savepoint (const Snapshot &new_snapshot)
	Sub-routine of handle_snapshot_triggered(). More...
ErrorStack	drop_volatile_pages (const Snapshot &new_snapshot, const std::map< storage::StorageId, storage::SnapshotPagePointer > &new_root_page_pointers)
	Sub-routine of handle_snapshot_triggered(). More...
void	drop_volatile_pages_parallel (const Snapshot &new_snapshot, const std::map< storage::StorageId, storage::SnapshotPagePointer > &new_root_page_pointers, void *result_memory, uint16_t parallel_id)
	subroutine invoked by one thread for one node. More...
fs::Path	get_snapshot_metadata_file_path (SnapshotId snapshot_id) const
	each snapshot has a snapshot-metadata file "snapshot_metadata_<SNAPSHOT_ID>.xml" in first node's first partition folder. More...
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 ()

Public Attributes
Engine *const	engine_
SnapshotManagerControlBlock *	control_block_
std::vector< Snapshot >	snapshots_
	All previously taken snapshots. More...
std::atomic< bool >	stop_requested_
	To locally shutdown snapshot_thread_. More...
std::thread	snapshot_thread_
	The daemon thread of snapshot manager. More...
std::chrono::system_clock::time_point	previous_snapshot_time_
	When snapshot_thread_ took snapshot last time. More...
std::vector< LogMapper * >	local_mappers_
	Mappers in this node. More...
LogReducer *	local_reducer_
	Reducer in this node. More...
LogGleanerResource	gleaner_resource_
	Local resources for gleaner, which runs only in the master node. More...

Constructor & Destructor Documentation

foedus::snapshot::SnapshotManagerPimpl::SnapshotManagerPimpl ( )

delete

foedus::snapshot::SnapshotManagerPimpl::SnapshotManagerPimpl ( Engine *  engine )

inlineexplicit

Definition at line 207 of file snapshot_manager_pimpl.hpp.

    : engine_(engine), local_reducer_(nullptr) {}

Member Function Documentation

ErrorStack foedus::snapshot::SnapshotManagerPimpl::drop_volatile_pages	(	const Snapshot &	new_snapshot,
		const std::map< storage::StorageId, storage::SnapshotPagePointer > &	new_root_page_pointers
	)

Sub-routine of handle_snapshot_triggered().

Drop pointers to volatile pages based on the already-installed snapshot pointers.

Definition at line 488 of file snapshot_manager_pimpl.cpp.

References foedus::memory::AlignedMemory::alloc(), ASSERT_ND, foedus::memory::PagePoolOffsetChunk::clear(), foedus::storage::Composer::drop_root_volatile(), drop_volatile_pages_parallel(), foedus::debugging::StopWatch::elapsed_ms(), engine_, foedus::memory::AlignedMemory::get_block(), foedus::Engine::get_soc_count(), foedus::Engine::get_xct_manager(), foedus::Epoch::is_valid(), foedus::memory::AlignedMemory::kNumaAllocOnnode, foedus::kRetOk, foedus::storage::Composer::DropResult::max_observed_, foedus::snapshot::Snapshot::max_storage_id_, foedus::xct::XctManager::pause_accepting_xct(), foedus::memory::AlignedMemory::release_block(), foedus::xct::XctManager::resume_accepting_xct(), foedus::debugging::StopWatch::stop(), and foedus::snapshot::Snapshot::valid_until_epoch_.

Referenced by handle_snapshot_triggered().

                                                                                      {
  // To speed up, we parallelize this process per node, and use the same partitioning scheme.
  LOG(INFO) << "Dropping volatile pointers...";

  // initializations done.
  // below, we should release the resources before exiting. So, let's not just use CHECK_ERROR.
  const uint16_t soc_count = engine_->get_soc_count();

  // collect results of pointer dropping for all storages for all nodes.
  // this is just to drop the root page. DropResult[storage_id][node].
  memory::AlignedMemory result_memory;
  result_memory.alloc(
    sizeof(storage::Composer::DropResult) * soc_count * (new_snapshot.max_storage_id_ + 2U),
    1U << 12,
    memory::AlignedMemory::kNumaAllocOnnode,
    0);
  memory::AlignedMemory chunks_memory;  // only for drop_root
  chunks_memory.alloc(
    sizeof(memory::PagePoolOffsetChunk) * soc_count,
    1U << 12,
    memory::AlignedMemory::kNumaAllocOnnode,
    0);

  // So far, we pause transaction executions during this step to simplify the algorithm.
  // Without this simplification, not only this thread but also normal transaction executions
  // have to do several complex and expensive checks.
  engine_->get_xct_manager()->pause_accepting_xct();
  // It will take a while for individual worker threads to complete the currently running xcts.
  // Just wait for a while to let that happen.
  std::this_thread::sleep_for(std::chrono::milliseconds(100));  // almost forever in OLTP xcts.
  LOG(INFO) << "Paused transaction executions to safely drop volatile pages and waited enough"
    << " to let currently running xcts end. Now start replace pointers.";
  debugging::StopWatch stop_watch;

  std::vector< std::thread > threads;
  for (uint16_t node = 0; node < soc_count; ++node) {
    threads.emplace_back(
      &SnapshotManagerPimpl::drop_volatile_pages_parallel,
      this,
      new_snapshot,
      new_root_page_pointers,
      result_memory.get_block(),
      node);
  }

  for (std::thread& thr : threads) {
    thr.join();
  }

  LOG(INFO) << "Joined child threads. Now consider dropping root pages";

  // At last, we consider dropping root volatile pages.
  // usually, this happens only when the root page doesn't have any volatile pointer.
  // As an exceptional case, we might drop ALL volatile pages of a storage whose max_observed
  // is not updated but for some reason dropped_all is false, eg non-matching boundaries.
  // even in that case, we can drop all volatile pages safely because this is within pause.
  memory::PagePoolOffsetChunk* dropped_chunks = reinterpret_cast<memory::PagePoolOffsetChunk*>(
    chunks_memory.get_block());
  for (uint16_t node = 0; node < soc_count; ++node) {
    dropped_chunks[node].clear();
  }
  storage::Composer::DropResult* results
    = reinterpret_cast<storage::Composer::DropResult*>(result_memory.get_block());
  for (storage::StorageId id = 1; id <= new_snapshot.max_storage_id_; ++id) {
    VLOG(1) << "Considering to drop root page of storage-" << id << " ...";
    bool cannot_drop = false;
    for (uint16_t node = 0; node < soc_count; ++node) {
      storage::Composer::DropResult result = results[soc_count * id + node];
      ASSERT_ND(result.max_observed_.is_valid());
      ASSERT_ND(result.max_observed_ >= new_snapshot.valid_until_epoch_);
      if (result.max_observed_ > new_snapshot.valid_until_epoch_) {
        cannot_drop = true;
        break;
      }
    }
    if (cannot_drop) {
      continue;
    }
    LOG(INFO) << "Looks like we can drop ALL volatile pages of storage-" << id << "!!!";
    // Still, the specific implementation of the storage might not choose to do so.
    // We call a method in composer.
    uint64_t dropped_count = 0;
    storage::Composer::DropVolatilesArguments args = {
      new_snapshot,
      0,
      false,
      dropped_chunks,
      &dropped_count};
    storage::Composer composer(engine_, id);
    composer.drop_root_volatile(args);
    LOG(INFO) << "As a result, we dropped " << dropped_count << " pages from storage-" << id;
  }


  engine_->get_xct_manager()->resume_accepting_xct();

  stop_watch.stop();
  LOG(INFO) << "Total: Dropped volatile pages in " << stop_watch.elapsed_ms() << "ms.";

  chunks_memory.release_block();
  result_memory.release_block();

  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::snapshot::SnapshotManagerPimpl::drop_volatile_pages_parallel	(	const Snapshot &	new_snapshot,
		const std::map< storage::StorageId, storage::SnapshotPagePointer > &	new_root_page_pointers,
		void *	result_memory,
		uint16_t	parallel_id
	)

subroutine invoked by one thread for one node.

Definition at line 595 of file snapshot_manager_pimpl.cpp.

References foedus::memory::AlignedMemory::alloc(), ASSERT_ND, foedus::memory::PagePoolOffsetChunk::clear(), foedus::storage::Composer::drop_volatiles(), foedus::storage::Composer::DropResult::dropped_all_, foedus::debugging::StopWatch::elapsed_ms(), foedus::debugging::StopWatch::elapsed_sec(), foedus::memory::PagePoolOffsetChunk::empty(), engine_, foedus::memory::AlignedMemory::get_block(), foedus::Engine::get_memory_manager(), foedus::memory::EngineMemory::get_node_memory(), foedus::Engine::get_soc_count(), foedus::storage::StorageManager::get_storage(), foedus::Engine::get_storage_manager(), foedus::memory::NumaNodeMemoryRef::get_volatile_pool(), foedus::memory::AlignedMemory::kNumaAllocOnnode, foedus::storage::Composer::DropResult::max_observed_, foedus::snapshot::Snapshot::max_storage_id_, foedus::storage::StorageControlBlock::meta_, foedus::storage::Metadata::name_, foedus::Epoch::one_more(), foedus::memory::PagePool::release(), foedus::memory::AlignedMemory::release_block(), foedus::storage::StorageControlBlock::root_page_pointer_, foedus::storage::Metadata::root_snapshot_page_id_, foedus::memory::PagePoolOffsetChunk::size(), foedus::debugging::StopWatch::stop(), and foedus::snapshot::Snapshot::valid_until_epoch_.

Referenced by drop_volatile_pages().

                        {
  // this thread is pinned on its own socket. We use the same partitioning scheme as reducer
  // so that this method mostly hits local pages
  thread::NumaThreadScope numa_scope(parallel_id);

  const uint16_t soc_count = engine_->get_soc_count();
  storage::Composer::DropResult* results  // DropResult[storage_id][node]
    = reinterpret_cast<storage::Composer::DropResult*>(result_memory);

  // To avoid invoking volatile pool for every dropped page, we cache them in chunks
  memory::AlignedMemory chunks_memory;
  chunks_memory.alloc(
    sizeof(memory::PagePoolOffsetChunk) * soc_count,
    1U << 12,
    memory::AlignedMemory::kNumaAllocOnnode,
    parallel_id);
  memory::PagePoolOffsetChunk* dropped_chunks = reinterpret_cast<memory::PagePoolOffsetChunk*>(
    chunks_memory.get_block());
  for (uint16_t node = 0; node < soc_count; ++node) {
    dropped_chunks[node].clear();
  }

  LOG(INFO) << "Thread-" << parallel_id << " started dropping volatile pages.";

  uint64_t dropped_count_total = 0;
  debugging::StopWatch stop_watch;
  for (storage::StorageId id = 1; id <= new_snapshot.max_storage_id_; ++id) {
    const auto& it = new_root_page_pointers.find(id);
    if (it != new_root_page_pointers.end()) {
      VLOG(0) << "Dropping pointers for storage-" << id << " ...";
      storage::SnapshotPagePointer new_root_page_pointer = it->second;
      ASSERT_ND(new_root_page_pointer != 0);
      storage::Composer composer(engine_, id);
      uint64_t dropped_count = 0;
      storage::Composer::DropVolatilesArguments args = {
        new_snapshot,
        parallel_id,
        true,
        dropped_chunks,
        &dropped_count};
      debugging::StopWatch watch;
      storage::Composer::DropResult result = composer.drop_volatiles(args);
      ASSERT_ND(engine_->get_storage_manager()->get_storage(id)->root_page_pointer_.
        snapshot_pointer_ == new_root_page_pointer);
      ASSERT_ND(engine_->get_storage_manager()->get_storage(id)->meta_.root_snapshot_page_id_
        == new_root_page_pointer);
      dropped_count_total += dropped_count;
      watch.stop();
      LOG(INFO) << "Thread-" << parallel_id << " drop_volatiles for storage-" << id
        << " (" << engine_->get_storage_manager()->get_storage(id)->meta_.name_ << ")"
        << " took " << watch.elapsed_sec() << "s. dropped_count=" << dropped_count
        << ". result =" << result;
      results[soc_count * id + parallel_id] = result;
    } else {
      VLOG(0) << "Thread-" << parallel_id << " storage-"
        << id << " wasn't changed no drop pointers";
      results[soc_count * id + parallel_id].max_observed_
        = new_snapshot.valid_until_epoch_.one_more();  // do NOT drop root page in this case.
      results[soc_count * id + parallel_id].dropped_all_ = false;
    }
  }

  stop_watch.stop();
  LOG(INFO) << "Thread-" << parallel_id << " dropped " << dropped_count_total
    << " volatile pointers in " << stop_watch.elapsed_ms() << "ms.";

  for (uint16_t node = 0; node < soc_count; ++node) {
    memory::PagePoolOffsetChunk* chunk = dropped_chunks + node;
    memory::PagePool* volatile_pool
      = engine_->get_memory_manager()->get_node_memory(node)->get_volatile_pool();
    if (!chunk->empty()) {
      volatile_pool->release(chunk->size(), chunk);
    }
    ASSERT_ND(chunk->empty());
  }
  chunks_memory.release_block();
}

Here is the call graph for this function:

Here is the caller graph for this function:

const SnapshotOptions & foedus::snapshot::SnapshotManagerPimpl::get_option

(

)

const

shorthand for engine_->get_options().snapshot_.

Definition at line 57 of file snapshot_manager_pimpl.cpp.

References engine_, foedus::Engine::get_options(), and foedus::EngineOptions::snapshot_.

Referenced by get_snapshot_metadata_file_path(), handle_snapshot(), and snapshot_metadata().

                                                              {
  return engine_->get_options().snapshot_;
}

Here is the call graph for this function:

Here is the caller graph for this function:

SnapshotId foedus::snapshot::SnapshotManagerPimpl::get_previous_snapshot_id ( ) const

inline

Definition at line 218 of file snapshot_manager_pimpl.hpp.

References control_block_, and foedus::snapshot::SnapshotManagerControlBlock::get_previous_snapshot_id().

Referenced by foedus::snapshot::SnapshotManager::get_previous_snapshot_id().

{ return control_block_->get_previous_snapshot_id(); }

Here is the call graph for this function:

Here is the caller graph for this function:

SnapshotId foedus::snapshot::SnapshotManagerPimpl::get_previous_snapshot_id_weak ( ) const

inline

Definition at line 219 of file snapshot_manager_pimpl.hpp.

References control_block_, and foedus::snapshot::SnapshotManagerControlBlock::get_previous_snapshot_id_weak().

Referenced by foedus::snapshot::SnapshotManager::get_previous_snapshot_id_weak().

                                                    {
    return control_block_->get_previous_snapshot_id_weak();
  }

Here is the call graph for this function:

Here is the caller graph for this function:

Epoch foedus::snapshot::SnapshotManagerPimpl::get_snapshot_epoch ( ) const

inline

Definition at line 215 of file snapshot_manager_pimpl.hpp.

References control_block_, and foedus::snapshot::SnapshotManagerControlBlock::get_snapshot_epoch().

Referenced by foedus::snapshot::SnapshotManager::get_snapshot_epoch(), handle_snapshot(), handle_snapshot_triggered(), and trigger_snapshot_immediate().

{ return control_block_->get_snapshot_epoch(); }

Here is the call graph for this function:

Here is the caller graph for this function:

Epoch foedus::snapshot::SnapshotManagerPimpl::get_snapshot_epoch_weak ( ) const

inline

Definition at line 216 of file snapshot_manager_pimpl.hpp.

References control_block_, and foedus::snapshot::SnapshotManagerControlBlock::get_snapshot_epoch_weak().

Referenced by foedus::snapshot::SnapshotManager::get_snapshot_epoch_weak().

{ return control_block_->get_snapshot_epoch_weak(); }

Here is the call graph for this function:

Here is the caller graph for this function:

fs::Path foedus::snapshot::SnapshotManagerPimpl::get_snapshot_metadata_file_path

(

SnapshotId

snapshot_id

)

const

each snapshot has a snapshot-metadata file "snapshot_metadata_<SNAPSHOT_ID>.xml" in first node's first partition folder.

Definition at line 480 of file snapshot_manager_pimpl.cpp.

References get_option().

Referenced by read_snapshot_metadata(), and snapshot_metadata().

                                                                                         {
  fs::Path folder(get_option().get_primary_folder_path());
  fs::Path file(folder);
  file /= std::string("snapshot_metadata_")
    + std::to_string(snapshot_id) + std::string(".xml");
  return file;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::SnapshotManagerPimpl::glean_logs	(	const Snapshot &	new_snapshot,
		std::map< storage::StorageId, storage::SnapshotPagePointer > *	new_root_page_pointers
	)

Sub-routine of handle_snapshot_triggered().

Read log files, distribute them to each partition, and construct snapshot files at each partition. After successful completion, all snapshot files become also durable (LogGleaner's uninitialize() makes it sure). Thus, now we can start installing pointers to the new snapshot file pages.

Definition at line 386 of file snapshot_manager_pimpl.cpp.

References engine_, foedus::snapshot::LogGleaner::execute(), foedus::snapshot::LogGleaner::get_new_root_page_pointers(), gleaner_resource_, and foedus::ErrorStack::is_error().

Referenced by handle_snapshot_triggered().

                                                                                {
  // Log gleaner is an object allocated/deallocated per snapshotting.
  // Gleaner runs on this thread (snapshot_thread_)
  LogGleaner gleaner(engine_, &gleaner_resource_, new_snapshot);
  ErrorStack result = gleaner.execute();
  if (result.is_error()) {
    LOG(ERROR) << "Log Gleaner encountered either an error or early termination request";
  }
  // the output is list of pointers to new root pages
  *new_root_page_pointers = gleaner.get_new_root_page_pointers();
  return result;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::snapshot::SnapshotManagerPimpl::handle_snapshot

(

)

Main routine for snapshot_thread_ in master engine.

This method keeps taking snapshot periodically. When there are no logs in all the private buffers for a while, it goes into sleep. This method exits when this object's uninitialize() is called.

Definition at line 191 of file snapshot_manager_pimpl.cpp.

References control_block_, engine_, foedus::log::LogManager::get_durable_global_epoch(), foedus::Engine::get_log_manager(), get_option(), foedus::snapshot::SnapshotManagerControlBlock::get_requested_snapshot_epoch(), get_snapshot_epoch(), handle_snapshot_triggered(), foedus::ErrorStack::is_error(), foedus::Engine::is_initialized(), is_stop_requested(), foedus::Epoch::is_valid(), foedus::assorted::memory_fence_acquire(), previous_snapshot_time_, sleep_a_while(), and SPINLOCK_WHILE.

Referenced by initialize_once().

                                           {
  LOG(INFO) << "Snapshot daemon started";
  // The actual snapshotting can't start until all other modules are initialized.
  SPINLOCK_WHILE(!is_stop_requested() && !engine_->is_initialized()) {
    assorted::memory_fence_acquire();
  }

  LOG(INFO) << "Snapshot daemon now starts taking snapshot";
  while (!is_stop_requested()) {
    sleep_a_while();
    if (is_stop_requested()) {
      break;
    }
    // should we start snapshotting? or keep sleeping?
    bool triggered = false;
    std::chrono::system_clock::time_point until = previous_snapshot_time_ +
      std::chrono::milliseconds(get_option().snapshot_interval_milliseconds_);
    Epoch durable_epoch = engine_->get_log_manager()->get_durable_global_epoch();
    Epoch previous_epoch = get_snapshot_epoch();
    if (previous_epoch.is_valid() && previous_epoch == durable_epoch) {
      LOG(INFO) << "Current snapshot is already latest. durable_epoch=" << durable_epoch;
    } else if (control_block_->get_requested_snapshot_epoch().is_valid()
        && (!previous_epoch.is_valid()
            || control_block_->get_requested_snapshot_epoch() > previous_epoch)) {
      // if someone requested immediate snapshot, do it.
      triggered = true;
      LOG(INFO) << "Immediate snapshot request detected. snapshotting..";
    } else if (std::chrono::system_clock::now() >= until) {
      triggered = true;
      LOG(INFO) << "Snapshot interval has elapsed. snapshotting..";
    } else {
      // TASK(Hideaki): check free pages in page pool and compare with configuration.
    }

    if (triggered) {
      Snapshot new_snapshot;
      ErrorStack stack = handle_snapshot_triggered(&new_snapshot);
      if (stack.is_error()) {
        LOG(ERROR) << "Snapshot failed:" << stack;
      }
    } else {
      VLOG(1) << "Snapshotting not triggered. going to sleep again";
    }
  }

  LOG(INFO) << "Snapshot daemon ended. ";
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::snapshot::SnapshotManagerPimpl::handle_snapshot_child

(

)

Main routine for snapshot_thread_ in child engines.

All this does is to launch mappers/reducers threads when asked by master engine.

Definition at line 239 of file snapshot_manager_pimpl.cpp.

References foedus::soc::SharedPolling::acquire_ticket(), control_block_, foedus::snapshot::LogGleanerControlBlock::cur_snapshot_, engine_, foedus::Engine::get_soc_id(), foedus::snapshot::SnapshotManagerControlBlock::gleaner_, foedus::snapshot::Snapshot::id_, is_gleaning(), is_stop_requested(), foedus::snapshot::MapReduceBase::join_thread(), foedus::snapshot::MapReduceBase::launch_thread(), local_mappers_, local_reducer_, foedus::snapshot::SnapshotManagerControlBlock::snapshot_children_wakeup_, and foedus::soc::SharedPolling::timedwait().

Referenced by initialize_once().

                                                 {
  LOG(INFO) << "Child snapshot daemon-" << engine_->get_soc_id() << " started";
  thread::NumaThreadScope scope(engine_->get_soc_id());
  SnapshotId previous_id = control_block_->gleaner_.cur_snapshot_.id_;
  while (!is_stop_requested()) {
    {
      uint64_t demand = control_block_->snapshot_children_wakeup_.acquire_ticket();
      if (!is_stop_requested() && !is_gleaning()) {
        control_block_->snapshot_children_wakeup_.timedwait(demand, 100000ULL);
      }
    }
    if (is_stop_requested()) {
      break;
    } else if (!is_gleaning() || previous_id == control_block_->gleaner_.cur_snapshot_.id_) {
      continue;
    }
    SnapshotId current_id = control_block_->gleaner_.cur_snapshot_.id_;
    LOG(INFO) << "Child snapshot daemon-" << engine_->get_soc_id() << " received a request"
      << " for snapshot-" << current_id;
    local_reducer_->launch_thread();
    for (LogMapper* mapper : local_mappers_) {
      mapper->launch_thread();
    }
    LOG(INFO) << "Child snapshot daemon-" << engine_->get_soc_id() << " launched mappers/reducer"
      " for snapshot-" << current_id;
    for (LogMapper* mapper : local_mappers_) {
      mapper->join_thread();
    }
    local_reducer_->join_thread();
    LOG(INFO) << "Child snapshot daemon-" << engine_->get_soc_id() << " joined mappers/reducer"
      " for snapshot-" << current_id;
    previous_id = current_id;
  }

  LOG(INFO) << "Child snapshot daemon-" << engine_->get_soc_id() << " ended";
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::SnapshotManagerPimpl::handle_snapshot_triggered

(

Snapshot *

new_snapshot

)

handle_snapshot() calls this when it should start snapshotting.

In other words, this function is the main routine of snapshotting.

Definition at line 316 of file snapshot_manager_pimpl.cpp.

References ASSERT_ND, foedus::snapshot::Snapshot::base_epoch_, CHECK_ERROR, control_block_, foedus::snapshot::LogGleanerControlBlock::cur_snapshot_, drop_volatile_pages(), engine_, foedus::log::LogManager::get_durable_global_epoch(), foedus::storage::StorageManager::get_largest_storage_id(), foedus::Engine::get_log_manager(), foedus::snapshot::SnapshotManagerControlBlock::get_requested_snapshot_epoch(), get_snapshot_epoch(), foedus::Engine::get_storage_manager(), glean_logs(), foedus::snapshot::SnapshotManagerControlBlock::gleaner_, foedus::snapshot::Snapshot::id_, foedus::snapshot::increment(), foedus::storage::StorageManager::is_initialized(), foedus::Engine::is_master(), foedus::Epoch::is_valid(), foedus::snapshot::kNullSnapshotId, foedus::kRetOk, foedus::snapshot::Snapshot::max_storage_id_, foedus::assorted::memory_fence_release(), foedus::snapshot::SnapshotManagerControlBlock::previous_snapshot_id_, previous_snapshot_time_, foedus::soc::SharedPolling::signal(), foedus::snapshot::SnapshotManagerControlBlock::snapshot_epoch_, snapshot_metadata(), snapshot_savepoint(), foedus::snapshot::SnapshotManagerControlBlock::snapshot_taken_, foedus::snapshot::Snapshot::valid_until_epoch_, and foedus::Epoch::value().

Referenced by handle_snapshot(), and foedus::restart::RestartManagerPimpl::recover().

                                                                                 {
  ASSERT_ND(engine_->is_master());
  ASSERT_ND(engine_->get_storage_manager()->is_initialized());  // snapshot relied on storage module
  Epoch durable_epoch = engine_->get_log_manager()->get_durable_global_epoch();
  Epoch previous_epoch = get_snapshot_epoch();
  LOG(INFO) << "Taking a new snapshot. durable_epoch=" << durable_epoch
    << ", requested_snapshot_epoch=" << control_block_->get_requested_snapshot_epoch()
    << ". previous_snapshot=" << previous_epoch;
  ASSERT_ND(durable_epoch.is_valid() &&
    (!previous_epoch.is_valid() || durable_epoch > previous_epoch));
  new_snapshot->base_epoch_ = previous_epoch;
  Epoch requested_epoch = control_block_->get_requested_snapshot_epoch();
  if (requested_epoch.is_valid()) {
    ASSERT_ND(requested_epoch <= durable_epoch);
    ASSERT_ND(!previous_epoch.is_valid() || requested_epoch > previous_epoch);
    new_snapshot->valid_until_epoch_ = requested_epoch;
  } else {
    new_snapshot->valid_until_epoch_ = durable_epoch;
  }
  new_snapshot->max_storage_id_ = engine_->get_storage_manager()->get_largest_storage_id();
  ASSERT_ND(new_snapshot->max_storage_id_
    >= control_block_->gleaner_.cur_snapshot_.max_storage_id_);

  // determine the snapshot ID
  SnapshotId snapshot_id;
  if (control_block_->previous_snapshot_id_ == kNullSnapshotId) {
    snapshot_id = 1;
  } else {
    snapshot_id = increment(control_block_->previous_snapshot_id_);
  }
  LOG(INFO) << "Issued ID for this snapshot:" << snapshot_id;
  new_snapshot->id_ = snapshot_id;

  // okay, let's start the snapshotting.
  // The procedures below will take long time, so we keep checking our "is_stop_requested"
  // and stops our child threads when it happens.

  // For each storage that was modified in this snapshotting,
  // this holds the pointer to new root page.
  std::map<storage::StorageId, storage::SnapshotPagePointer> new_root_page_pointers;

  // Log gleaners design partitioning and do scatter-gather to consume the logs.
  // This will create snapshot files at each partition and tell us the new root pages of
  // each storage.
  CHECK_ERROR(glean_logs(*new_snapshot, &new_root_page_pointers));

  // Write out the metadata file.
  CHECK_ERROR(snapshot_metadata(*new_snapshot, new_root_page_pointers));

  // Invokes savepoint module to make sure this snapshot has "happened".
  CHECK_ERROR(snapshot_savepoint(*new_snapshot));

  // install pointers to snapshot pages and drop volatile pages.
  CHECK_ERROR(drop_volatile_pages(*new_snapshot, new_root_page_pointers));

  Epoch new_snapshot_epoch = new_snapshot->valid_until_epoch_;
  ASSERT_ND(new_snapshot_epoch.is_valid() &&
    (!get_snapshot_epoch().is_valid() || new_snapshot_epoch > get_snapshot_epoch()));

  // done. notify waiters if exist
  Epoch::EpochInteger epoch_after = new_snapshot_epoch.value();
  control_block_->previous_snapshot_id_ = snapshot_id;
  previous_snapshot_time_ = std::chrono::system_clock::now();

  control_block_->snapshot_epoch_ = epoch_after;
  assorted::memory_fence_release();
  control_block_->snapshot_taken_.signal();
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::SnapshotManagerPimpl::initialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 61 of file snapshot_manager_pimpl.cpp.

References foedus::snapshot::LogGleanerControlBlock::all_count_, foedus::snapshot::LogGleanerResource::allocate(), ASSERT_ND, CHECK_ERROR, control_block_, foedus::storage::PartitionerMetadata::data_size_, engine_, ERROR_STACK, foedus::soc::SharedMemoryRepo::get_global_memory_anchors(), foedus::savepoint::SavepointManager::get_latest_snapshot_epoch(), foedus::savepoint::SavepointManager::get_latest_snapshot_id(), foedus::Engine::get_log_manager(), foedus::Engine::get_options(), foedus::Engine::get_savepoint_manager(), foedus::soc::SocManager::get_shared_memory_repo(), foedus::Engine::get_soc_count(), foedus::Engine::get_soc_manager(), foedus::snapshot::SnapshotManagerControlBlock::gleaner_, gleaner_resource_, foedus::thread::ThreadOptions::group_count_, handle_snapshot(), handle_snapshot_child(), foedus::snapshot::SnapshotManagerControlBlock::initialize(), foedus::DefaultInitializable::initialize(), foedus::storage::PartitionerMetadata::initialize(), foedus::log::LogManager::is_initialized(), foedus::soc::SharedMutex::is_initialized(), foedus::Engine::is_master(), foedus::Epoch::kEpochInvalid, foedus::kErrorCodeDepedentModuleUnavailableInit, foedus::kRetOk, local_mappers_, local_reducer_, foedus::EngineOptions::log_, foedus::log::LogOptions::loggers_per_node_, foedus::snapshot::LogGleanerControlBlock::mappers_count_, foedus::storage::StorageOptions::max_storages_, foedus::storage::PartitionerMetadata::mutex_, foedus::storage::StorageOptions::partitioner_data_memory_mb_, foedus::soc::GlobalMemoryAnchors::partitioner_metadata_, foedus::snapshot::SnapshotManagerControlBlock::previous_snapshot_id_, previous_snapshot_time_, foedus::snapshot::LogGleanerControlBlock::reducers_count_, foedus::snapshot::SnapshotManagerControlBlock::requested_snapshot_epoch_, foedus::snapshot::SnapshotManagerControlBlock::snapshot_epoch_, foedus::soc::GlobalMemoryAnchors::snapshot_manager_memory_, snapshot_thread_, stop_requested_, foedus::EngineOptions::storage_, foedus::EngineOptions::thread_, and foedus::Epoch::value().

                                                 {
  LOG(INFO) << "Initializing SnapshotManager..";
  if (!engine_->get_log_manager()->is_initialized()) {
    return ERROR_STACK(kErrorCodeDepedentModuleUnavailableInit);
  }
  soc::SharedMemoryRepo* repo = engine_->get_soc_manager()->get_shared_memory_repo();
  control_block_ = repo->get_global_memory_anchors()->snapshot_manager_memory_;
  if (engine_->is_master()) {
    control_block_->initialize();
    // get snapshot status from savepoint
    control_block_->snapshot_epoch_
      = engine_->get_savepoint_manager()->get_latest_snapshot_epoch().value();
    control_block_->previous_snapshot_id_
      = engine_->get_savepoint_manager()->get_latest_snapshot_id();
    LOG(INFO) << "Latest snapshot: id=" << control_block_->previous_snapshot_id_ << ", epoch="
      << control_block_->snapshot_epoch_;
    control_block_->requested_snapshot_epoch_.store(Epoch::kEpochInvalid);

    const EngineOptions& options = engine_->get_options();
    uint32_t reducer_count = options.thread_.group_count_;
    uint32_t mapper_count = reducer_count * options.log_.loggers_per_node_;
    control_block_->gleaner_.reducers_count_ = reducer_count;
    control_block_->gleaner_.mappers_count_ = mapper_count;
    control_block_->gleaner_.all_count_ = reducer_count + mapper_count;

    // also initialize the shared memory for partitioner
    uint32_t max_storages = engine_->get_options().storage_.max_storages_;
    storage::PartitionerMetadata* partitioner_metadata
      = repo->get_global_memory_anchors()->partitioner_metadata_;
    for (storage::StorageId i = 0; i < max_storages; ++i) {
      storage::PartitionerMetadata* meta = partitioner_metadata + i;
      ASSERT_ND(!meta->mutex_.is_initialized());
      meta->initialize();
      ASSERT_ND(meta->mutex_.is_initialized());
    }
    // set the size of partitioner data
    repo->get_global_memory_anchors()->partitioner_metadata_[0].data_size_
      = engine_->get_options().storage_.partitioner_data_memory_mb_ * (1ULL << 20);

    // And master-only resources for running gleaner. These are NOT shared memory. local to master.
    gleaner_resource_.allocate(engine_->get_soc_count());
  }

  // in child engines, we instantiate local mappers/reducer objects (but not the threads yet)
  previous_snapshot_time_ = std::chrono::system_clock::now();
  stop_requested_ = false;
  if (!engine_->is_master()) {
    local_reducer_ = new LogReducer(engine_);
    CHECK_ERROR(local_reducer_->initialize());
    for (uint16_t i = 0; i < engine_->get_options().log_.loggers_per_node_; ++i) {
      local_mappers_.push_back(new LogMapper(engine_, i));
      CHECK_ERROR(local_mappers_[i]->initialize());
    }
  }

  // Launch the daemon thread at last
  if (engine_->is_master()) {
    snapshot_thread_ = std::move(std::thread(&SnapshotManagerPimpl::handle_snapshot, this));
  } else {
    snapshot_thread_ = std::move(std::thread(&SnapshotManagerPimpl::handle_snapshot_child, this));
  }
  return kRetOk;
}

Here is the call graph for this function:

bool foedus::snapshot::SnapshotManagerPimpl::is_gleaning ( ) const

inline

Definition at line 251 of file snapshot_manager_pimpl.hpp.

References control_block_, foedus::snapshot::SnapshotManagerControlBlock::gleaner_, and foedus::snapshot::LogGleanerControlBlock::gleaning_.

Referenced by handle_snapshot_child().

{ return control_block_->gleaner_.gleaning_; }

Here is the caller graph for this function:

bool foedus::snapshot::SnapshotManagerPimpl::is_stop_requested ( ) const

inline

Definition at line 250 of file snapshot_manager_pimpl.hpp.

References stop_requested_.

Referenced by handle_snapshot(), handle_snapshot_child(), sleep_a_while(), and trigger_snapshot_immediate().

{ return stop_requested_; }

Here is the caller graph for this function:

SnapshotId foedus::snapshot::SnapshotManagerPimpl::issue_next_snapshot_id ( )

inline

Definition at line 239 of file snapshot_manager_pimpl.hpp.

References control_block_, foedus::snapshot::increment(), foedus::snapshot::kNullSnapshotId, and foedus::snapshot::SnapshotManagerControlBlock::previous_snapshot_id_.

                                      {
    if (control_block_->previous_snapshot_id_ == kNullSnapshotId) {
      control_block_->previous_snapshot_id_ = 1;
    } else {
      control_block_->previous_snapshot_id_ = increment(control_block_->previous_snapshot_id_);
    }
    return control_block_->previous_snapshot_id_;
  }

Here is the call graph for this function:

ErrorStack foedus::snapshot::SnapshotManagerPimpl::read_snapshot_metadata	(	SnapshotId	snapshot_id,
		SnapshotMetadata *	out
	)

Definition at line 453 of file snapshot_manager_pimpl.cpp.

References ASSERT_ND, CHECK_ERROR, foedus::debugging::StopWatch::elapsed_ms(), foedus::fs::file_size(), get_snapshot_metadata_file_path(), foedus::snapshot::SnapshotMetadata::id_, foedus::kRetOk, foedus::externalize::Externalizable::load_from_file(), and foedus::debugging::StopWatch::stop().

Referenced by foedus::snapshot::SnapshotManager::read_snapshot_metadata().

                         {
  fs::Path file = get_snapshot_metadata_file_path(snapshot_id);
  LOG(INFO) << "Reading snapshot metadata file fullpath=" << file;

  debugging::StopWatch stop_watch;
  CHECK_ERROR(out->load_from_file(file));
  stop_watch.stop();
  LOG(INFO) << "Read a snapshot metadata file. size=" << fs::file_size(file) << " bytes"
    << ", elapsed time to read+parse=" << stop_watch.elapsed_ms() << "ms.";

  ASSERT_ND(out->id_ == snapshot_id);
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::snapshot::SnapshotManagerPimpl::sleep_a_while

(

)

Definition at line 181 of file snapshot_manager_pimpl.cpp.

References foedus::soc::SharedPolling::acquire_ticket(), control_block_, is_stop_requested(), foedus::snapshot::SnapshotManagerControlBlock::snapshot_wakeup_, and foedus::soc::SharedPolling::timedwait().

Referenced by handle_snapshot().

                                         {
  uint64_t demand = control_block_->snapshot_wakeup_.acquire_ticket();
  if (!is_stop_requested()) {
    control_block_->snapshot_wakeup_.timedwait(demand, 20000ULL);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::SnapshotManagerPimpl::snapshot_metadata	(	const Snapshot &	new_snapshot,
		const std::map< storage::StorageId, storage::SnapshotPagePointer > &	new_root_page_pointers
	)

Sub-routine of handle_snapshot_triggered().

Write out a snapshot metadata file that contains metadata of all storages and a few other global metadata.

Definition at line 401 of file snapshot_manager_pimpl.cpp.

References ASSERT_ND, foedus::snapshot::Snapshot::base_epoch_, foedus::snapshot::SnapshotMetadata::base_epoch_, CHECK_ERROR, foedus::storage::StorageManager::clone_all_storage_metadata(), foedus::fs::create_directories(), foedus::debugging::StopWatch::elapsed_ms(), engine_, ERROR_STACK, foedus::fs::exists(), foedus::fs::file_size(), foedus::fs::fsync(), foedus::snapshot::SnapshotMetadata::get_metadata(), get_option(), get_snapshot_metadata_file_path(), foedus::Engine::get_storage_manager(), foedus::snapshot::Snapshot::id_, foedus::snapshot::SnapshotMetadata::id_, foedus::storage::StorageControlBlock::is_valid_status(), foedus::kErrorCodeFsMkdirFailed, foedus::kRetOk, foedus::snapshot::SnapshotMetadata::largest_storage_id_, foedus::snapshot::Snapshot::max_storage_id_, foedus::externalize::Externalizable::save_to_file(), foedus::debugging::StopWatch::start(), foedus::debugging::StopWatch::stop(), foedus::snapshot::SnapshotMetadata::storage_control_blocks_, foedus::snapshot::Snapshot::valid_until_epoch_, foedus::snapshot::SnapshotMetadata::valid_until_epoch_, and foedus::Epoch::value().

Referenced by handle_snapshot_triggered().

                                                                                      {
  // construct metadata object
  SnapshotMetadata metadata;
  metadata.id_ = new_snapshot.id_;
  metadata.base_epoch_ = new_snapshot.base_epoch_.value();
  metadata.valid_until_epoch_ = new_snapshot.valid_until_epoch_.value();
  metadata.largest_storage_id_ = new_snapshot.max_storage_id_;
  CHECK_ERROR(engine_->get_storage_manager()->clone_all_storage_metadata(&metadata));

  // we modified the root page.
  uint32_t installed_root_pages_count = 0;
  for (storage::StorageId id = 1; id <= metadata.largest_storage_id_; ++id) {
    const auto& it = new_root_page_pointers.find(id);
    ASSERT_ND(metadata.storage_control_blocks_[id].is_valid_status());
    storage::Metadata* meta = metadata.get_metadata(id);
    if (it != new_root_page_pointers.end()) {
      storage::SnapshotPagePointer new_pointer = it->second;
      // composer's construct_root should have been already set the new root pointer
      ASSERT_ND(new_pointer == meta->root_snapshot_page_id_);
      ++installed_root_pages_count;
    }
  }
  LOG(INFO) << "Out of " << metadata.largest_storage_id_ << " storages, "
    << installed_root_pages_count << " changed their root pages.";
  ASSERT_ND(installed_root_pages_count == new_root_page_pointers.size());

  // save it to a file
  fs::Path folder(get_option().get_primary_folder_path());
  if (!fs::exists(folder)) {
    if (!fs::create_directories(folder, true)) {
      LOG(ERROR) << "Failed to create directory:" << folder << ". check permission.";
      return ERROR_STACK(kErrorCodeFsMkdirFailed);
    }
  }

  fs::Path file = get_snapshot_metadata_file_path(new_snapshot.id_);
  LOG(INFO) << "New snapshot metadata file fullpath=" << file;

  debugging::StopWatch stop_watch;
  CHECK_ERROR(metadata.save_to_file(file));
  stop_watch.stop();
  LOG(INFO) << "Wrote a snapshot metadata file. size=" << fs::file_size(file) << " bytes"
    << ", elapsed time to write=" << stop_watch.elapsed_ms() << "ms. now fsyncing...";
  stop_watch.start();
  fs::fsync(file, true);
  stop_watch.stop();
  LOG(INFO) << "fsynced the file and the folder! elapsed=" << stop_watch.elapsed_ms() << "ms.";
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::SnapshotManagerPimpl::snapshot_savepoint

(

const Snapshot &

new_snapshot

)

Sub-routine of handle_snapshot_triggered().

Invokes the savepoint module to take savepoint pointing to this snapshot. Until that, the snapshot is not yet deemed as "happened".

Definition at line 469 of file snapshot_manager_pimpl.cpp.

References ASSERT_ND, CHECK_ERROR, engine_, foedus::savepoint::SavepointManager::get_latest_snapshot_epoch(), foedus::savepoint::SavepointManager::get_latest_snapshot_id(), foedus::Engine::get_savepoint_manager(), foedus::snapshot::Snapshot::id_, foedus::kRetOk, foedus::savepoint::SavepointManager::take_savepoint_after_snapshot(), and foedus::snapshot::Snapshot::valid_until_epoch_.

Referenced by handle_snapshot_triggered().

                                                                                {
  LOG(INFO) << "Taking savepoint to include this new snapshot....";
  CHECK_ERROR(engine_->get_savepoint_manager()->take_savepoint_after_snapshot(
    new_snapshot.id_,
    new_snapshot.valid_until_epoch_));
  ASSERT_ND(engine_->get_savepoint_manager()->get_latest_snapshot_id() == new_snapshot.id_);
  ASSERT_ND(engine_->get_savepoint_manager()->get_latest_snapshot_epoch()
    == new_snapshot.valid_until_epoch_);
  return kRetOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::snapshot::SnapshotManagerPimpl::stop_snapshot_thread

(

)

This is a hidden API called at the beginning of engine shutdown (namely restart manager).

Snapshot Manager initializes before Storage because it must read previous snapshot, but it must stop snapshot thread before Storage because the snapshot thread relies on storage module. To solve the issue, we call this method from restart manager's uninit to stop snapshot thread. This method is also called by snapshot manager's own uninit() in case restart manager didn't initialize. Thus, this method must be idempotent.

Definition at line 164 of file snapshot_manager_pimpl.cpp.

References foedus::snapshot::LogGleanerControlBlock::cancelled_, control_block_, engine_, foedus::snapshot::SnapshotManagerControlBlock::gleaner_, foedus::Engine::is_master(), snapshot_thread_, stop_requested_, foedus::snapshot::LogGleanerControlBlock::terminating_, wakeup(), and foedus::snapshot::SnapshotManagerControlBlock::wakeup_snapshot_children().

Referenced by foedus::restart::RestartManagerPimpl::uninitialize_once(), and uninitialize_once().

                                                {
  LOG(INFO) << "Stopping the snapshot thread...";
  if (snapshot_thread_.joinable()) {
    // whether from master or not, just make sure all reducers/mappers notice that it's closing
    stop_requested_ = true;
    control_block_->gleaner_.cancelled_ = true;
    control_block_->gleaner_.terminating_ = true;
    if (engine_->is_master()) {
      wakeup();
    } else {
      control_block_->wakeup_snapshot_children();
    }
    snapshot_thread_.join();
  }
  LOG(INFO) << "Stopped the snapshot thread.";
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::snapshot::SnapshotManagerPimpl::trigger_snapshot_immediate	(	bool	wait_completion,
		Epoch	suggested_snapshot_epoch
	)

Definition at line 277 of file snapshot_manager_pimpl.cpp.

References foedus::soc::SharedPolling::acquire_ticket(), ASSERT_ND, control_block_, engine_, foedus::log::LogManager::get_durable_global_epoch(), foedus::Engine::get_log_manager(), get_snapshot_epoch(), is_stop_requested(), foedus::Epoch::is_valid(), foedus::snapshot::SnapshotManagerControlBlock::requested_snapshot_epoch_, foedus::snapshot::SnapshotManagerControlBlock::snapshot_taken_, foedus::soc::SharedPolling::timedwait(), and wakeup().

Referenced by foedus::snapshot::SnapshotManager::trigger_snapshot_immediate().

                                  {
  LOG(INFO) << "Requesting to immediately take a snapshot. suggested_snapshot_epoch="
    << suggested_snapshot_epoch;
  Epoch before = get_snapshot_epoch();
  Epoch durable_epoch = engine_->get_log_manager()->get_durable_global_epoch();
  ASSERT_ND(durable_epoch.is_valid());

  Epoch new_snapshot_epoch = durable_epoch;
  if (suggested_snapshot_epoch.is_valid()) {
    if (suggested_snapshot_epoch > durable_epoch) {
      LOG(WARNING) << "You can't specify non-durable epoch for snapshot.";
    } else {
      new_snapshot_epoch = suggested_snapshot_epoch;
    }
  }

  if (before.is_valid() && before >= durable_epoch) {
    LOG(INFO) << "Current snapshot already satisfies. new_snapshot_epoch=" << new_snapshot_epoch;
    return;
  }

  control_block_->requested_snapshot_epoch_.store(new_snapshot_epoch.value());
  wakeup();
  if (wait_completion) {
    VLOG(0) << "Waiting for the completion of snapshot... before=" << before;
    while (!is_stop_requested() &&
        (!get_snapshot_epoch().is_valid() || new_snapshot_epoch > get_snapshot_epoch())) {
      uint64_t demand = control_block_->snapshot_taken_.acquire_ticket();
      if (!is_stop_requested() &&
        (!get_snapshot_epoch().is_valid() || new_snapshot_epoch > get_snapshot_epoch())) {
        control_block_->snapshot_taken_.timedwait(demand, 100000ULL);
      }
    }
  }
  LOG(INFO) << "Observed the completion of snapshot! after=" << get_snapshot_epoch();
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::snapshot::SnapshotManagerPimpl::uninitialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 125 of file snapshot_manager_pimpl.cpp.

References ASSERT_ND, control_block_, foedus::snapshot::LogGleanerResource::deallocate(), foedus::ErrorStackBatch::emprace_back(), engine_, ERROR_STACK, foedus::soc::SharedMemoryRepo::get_global_memory_anchors(), foedus::Engine::get_log_manager(), foedus::Engine::get_options(), foedus::soc::SocManager::get_shared_memory_repo(), foedus::Engine::get_soc_manager(), gleaner_resource_, foedus::log::LogManager::is_initialized(), foedus::soc::SharedMutex::is_initialized(), foedus::Engine::is_master(), foedus::kErrorCodeDepedentModuleUnavailableUninit, local_mappers_, local_reducer_, foedus::storage::StorageOptions::max_storages_, foedus::storage::PartitionerMetadata::mutex_, foedus::soc::GlobalMemoryAnchors::partitioner_metadata_, foedus::snapshot::LogGleanerResource::per_node_resources_, stop_snapshot_thread(), foedus::EngineOptions::storage_, SUMMARIZE_ERROR_BATCH, foedus::snapshot::SnapshotManagerControlBlock::uninitialize(), foedus::DefaultInitializable::uninitialize(), and foedus::storage::PartitionerMetadata::uninitialize().

                                                   {
  LOG(INFO) << "Uninitializing SnapshotManager..";
  ErrorStackBatch batch;
  if (!engine_->get_log_manager()->is_initialized()) {
    batch.emprace_back(ERROR_STACK(kErrorCodeDepedentModuleUnavailableUninit));
  }
  stop_snapshot_thread();
  if (engine_->is_master()) {
    // also uninitialize the shared memory for partitioner
    soc::GlobalMemoryAnchors* anchors
      = engine_->get_soc_manager()->get_shared_memory_repo()->get_global_memory_anchors();
    uint32_t max_storages = engine_->get_options().storage_.max_storages_;
    for (storage::StorageId i = 0; i < max_storages; ++i) {
      ASSERT_ND(anchors->partitioner_metadata_[i].mutex_.is_initialized());
      anchors->partitioner_metadata_[i].uninitialize();
      ASSERT_ND(!anchors->partitioner_metadata_[i].mutex_.is_initialized());
    }

    control_block_->uninitialize();
    ASSERT_ND(local_reducer_ == nullptr);
    ASSERT_ND(local_mappers_.size() == 0);

    gleaner_resource_.deallocate();
  } else {
    if (local_reducer_) {
      batch.emprace_back(local_reducer_->uninitialize());
      delete local_reducer_;
      local_reducer_ = nullptr;
    }
    for (LogMapper* mapper : local_mappers_) {
      batch.emprace_back(mapper->uninitialize());
      delete mapper;
    }
    local_mappers_.clear();
    ASSERT_ND(gleaner_resource_.per_node_resources_.empty());
  }

  return SUMMARIZE_ERROR_BATCH(batch);
}

Here is the call graph for this function:

void foedus::snapshot::SnapshotManagerPimpl::wakeup

(

)

Definition at line 187 of file snapshot_manager_pimpl.cpp.

References control_block_, foedus::soc::SharedPolling::signal(), and foedus::snapshot::SnapshotManagerControlBlock::snapshot_wakeup_.

Referenced by stop_snapshot_thread(), and trigger_snapshot_immediate().

                                  {
  control_block_->snapshot_wakeup_.signal();
}

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

SnapshotManagerControlBlock* foedus::snapshot::SnapshotManagerPimpl::control_block_

Definition at line 323 of file snapshot_manager_pimpl.hpp.

Referenced by get_previous_snapshot_id(), get_previous_snapshot_id_weak(), get_snapshot_epoch(), get_snapshot_epoch_weak(), handle_snapshot(), handle_snapshot_child(), handle_snapshot_triggered(), initialize_once(), is_gleaning(), issue_next_snapshot_id(), sleep_a_while(), stop_snapshot_thread(), trigger_snapshot_immediate(), uninitialize_once(), and wakeup().

Engine* const foedus::snapshot::SnapshotManagerPimpl::engine_

Definition at line 321 of file snapshot_manager_pimpl.hpp.

Referenced by drop_volatile_pages(), drop_volatile_pages_parallel(), get_option(), glean_logs(), handle_snapshot(), handle_snapshot_child(), handle_snapshot_triggered(), initialize_once(), snapshot_metadata(), snapshot_savepoint(), stop_snapshot_thread(), trigger_snapshot_immediate(), and uninitialize_once().

LogGleanerResource foedus::snapshot::SnapshotManagerPimpl::gleaner_resource_

Local resources for gleaner, which runs only in the master node.

Empty in child nodes.

Definition at line 356 of file snapshot_manager_pimpl.hpp.

Referenced by glean_logs(), initialize_once(), and uninitialize_once().

std::vector<LogMapper*> foedus::snapshot::SnapshotManagerPimpl::local_mappers_

Mappers in this node.

Index is logger ordinal. Empty in master engine.

Definition at line 351 of file snapshot_manager_pimpl.hpp.

Referenced by handle_snapshot_child(), initialize_once(), and uninitialize_once().

LogReducer* foedus::snapshot::SnapshotManagerPimpl::local_reducer_

Reducer in this node.

Null in master engine.

Definition at line 353 of file snapshot_manager_pimpl.hpp.

Referenced by handle_snapshot_child(), initialize_once(), and uninitialize_once().

std::chrono::system_clock::time_point foedus::snapshot::SnapshotManagerPimpl::previous_snapshot_time_

When snapshot_thread_ took snapshot last time.

Read and written only by snapshot_thread_.

Definition at line 348 of file snapshot_manager_pimpl.hpp.

Referenced by handle_snapshot(), handle_snapshot_triggered(), and initialize_once().

std::thread foedus::snapshot::SnapshotManagerPimpl::snapshot_thread_

The daemon thread of snapshot manager.

In master engine, this occasionally wakes up and serves as the main managing thread for snapshotting, which consists of several child threads and multiple phases. In child engine, this receives requests from master engine's snapshot_thread_ and launch mappers/reducers in this node.

Definition at line 342 of file snapshot_manager_pimpl.hpp.

Referenced by initialize_once(), and stop_snapshot_thread().

std::vector< Snapshot > foedus::snapshot::SnapshotManagerPimpl::snapshots_

All previously taken snapshots.

Access to this data must be protected with mutex. This is populated only in master engine.

Definition at line 330 of file snapshot_manager_pimpl.hpp.

std::atomic<bool> foedus::snapshot::SnapshotManagerPimpl::stop_requested_

To locally shutdown snapshot_thread_.

This is not a shared memory.

Definition at line 333 of file snapshot_manager_pimpl.hpp.

Referenced by initialize_once(), is_stop_requested(), and stop_snapshot_thread().

---

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

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