Detailed Description

Pimpl object of LogManager.

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

Definition at line 79 of file log_manager_pimpl.hpp.

#include <log_manager_pimpl.hpp>

Inheritance diagram for foedus::log::LogManagerPimpl:

[legend]

Collaboration diagram for foedus::log::LogManagerPimpl:

[legend]

Public Member Functions
	LogManagerPimpl ()=delete

	LogManagerPimpl (Engine *engine)

ErrorStack	initialize_once () override

ErrorStack	uninitialize_once () override

void	wakeup_loggers ()

ErrorCode	wait_until_durable (Epoch commit_epoch, int64_t wait_microseconds)

ErrorStack	refresh_global_durable_epoch ()

void	copy_logger_states (savepoint::Savepoint *new_savepoint)

Epoch	get_durable_global_epoch () const

Epoch	get_durable_global_epoch_weak () const

void	announce_new_durable_global_epoch (Epoch new_epoch)

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_

thread::ThreadGroupId	groups_

uint16_t	loggers_per_node_

std::vector< Logger * >	loggers_
	Local log writers. More...

std::vector< LoggerRef >	logger_refs_
	All log writers. More...

MetaLogger *	meta_logger_
	Metadata log writer. More...

MetaLogBuffer	meta_buffer_
	Metadata log buffer. More...

LogManagerControlBlock *	control_block_

Constructor & Destructor Documentation

foedus::log::LogManagerPimpl::LogManagerPimpl ( )

delete

foedus::log::LogManagerPimpl::LogManagerPimpl ( Engine * engine )

explicit

Definition at line 46 of file log_manager_pimpl.cpp.

47 : engine_(engine), meta_logger_(nullptr), control_block_(nullptr) {}

foedus::log::LogManagerPimpl::engine_

Engine *const engine_

Definition: log_manager_pimpl.hpp:100

foedus::log::LogManagerPimpl::control_block_

LogManagerControlBlock * control_block_

Definition: log_manager_pimpl.hpp:119

foedus::log::LogManagerPimpl::meta_logger_

MetaLogger * meta_logger_

Metadata log writer.

Definition: log_manager_pimpl.hpp:115

Member Function Documentation

void foedus::log::LogManagerPimpl::announce_new_durable_global_epoch ( Epoch new_epoch )

Definition at line 268 of file log_manager_pimpl.cpp.

References ASSERT_ND, control_block_, foedus::log::LogManagerControlBlock::durable_global_epoch_, foedus::log::LogManagerControlBlock::durable_global_epoch_advanced_, foedus::assorted::memory_fence_release(), foedus::soc::SharedPolling::signal(), and foedus::Epoch::value().

Referenced by foedus::log::LogManager::announce_new_durable_global_epoch().

                                                                        {
   ASSERT_ND(new_epoch >= Epoch(control_block_->durable_global_epoch_));
   control_block_->durable_global_epoch_ = new_epoch.value();
   assorted::memory_fence_release();
   control_block_->durable_global_epoch_advanced_.signal();
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::log::LogManagerPimpl::copy_logger_states ( savepoint::Savepoint * new_savepoint )

Definition at line 276 of file log_manager_pimpl.cpp.

References foedus::savepoint::Savepoint::current_log_files_, foedus::savepoint::Savepoint::current_log_files_offset_durable_, logger_refs_, foedus::savepoint::Savepoint::oldest_log_files_, and foedus::savepoint::Savepoint::oldest_log_files_offset_begin_.

Referenced by foedus::log::LogManager::copy_logger_states().

                                                                           {
   new_savepoint->oldest_log_files_.clear();
   new_savepoint->oldest_log_files_offset_begin_.clear();
   new_savepoint->current_log_files_.clear();
   new_savepoint->current_log_files_offset_durable_.clear();
   for (const LoggerRef& logger : logger_refs_) {
     logger.copy_logger_state(new_savepoint);
   }
 }

Here is the caller graph for this function:

Epoch foedus::log::LogManagerPimpl::get_durable_global_epoch ( ) const

inline

Definition at line 91 of file log_manager_pimpl.hpp.

References control_block_, and foedus::log::LogManagerControlBlock::durable_global_epoch_.

Referenced by foedus::log::LogManager::get_durable_global_epoch(), initialize_once(), refresh_global_durable_epoch(), and wait_until_durable().

                                                {
     return Epoch(control_block_->durable_global_epoch_.load());
   }

Here is the caller graph for this function:

Epoch foedus::log::LogManagerPimpl::get_durable_global_epoch_weak ( ) const

inline

Definition at line 94 of file log_manager_pimpl.hpp.

References control_block_, and foedus::log::LogManagerControlBlock::durable_global_epoch_.

Referenced by foedus::log::LogManager::get_durable_global_epoch_weak().

                                                     {
     return Epoch(control_block_->durable_global_epoch_.load(std::memory_order_relaxed));
   }

Here is the caller graph for this function:

ErrorStack foedus::log::LogManagerPimpl::initialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 49 of file log_manager_pimpl.cpp.

References ASSERT_ND, foedus::Attachable< CONTROL_BLOCK >::attach(), CHECK_ERROR, CHECK_OUTOFMEMORY, COERCE_ERROR, foedus::thread::compose_thread_id(), control_block_, foedus::log::LogOptions::convert_folder_path_pattern(), foedus::fs::create_directories(), foedus::log::LogManagerControlBlock::durable_global_epoch_, engine_, ERROR_STACK, foedus::fs::exists(), get_durable_global_epoch(), foedus::soc::SharedMemoryRepo::get_global_memory_anchors(), foedus::savepoint::SavepointManager::get_initial_durable_epoch(), foedus::soc::SharedMemoryRepo::get_node_memory_anchors(), foedus::Engine::get_options(), foedus::Engine::get_savepoint_manager(), foedus::soc::SocManager::get_shared_memory_repo(), foedus::Engine::get_soc_id(), foedus::Engine::get_soc_manager(), foedus::Engine::get_thread_pool(), foedus::thread::ThreadOptions::get_total_thread_count(), foedus::thread::ThreadOptions::group_count_, groups_, foedus::log::LogManagerControlBlock::initialize(), foedus::DefaultInitializable::initialize(), foedus::savepoint::SavepointManager::is_initialized(), foedus::thread::ThreadPool::is_initialized(), foedus::Engine::is_master(), foedus::kErrorCodeDepedentModuleUnavailableInit, foedus::kErrorCodeLogInvalidLoggerCount, foedus::kRetOk, foedus::EngineOptions::log_, foedus::soc::GlobalMemoryAnchors::log_manager_memory_, foedus::soc::NodeMemoryAnchors::logger_memories_, logger_refs_, loggers_, foedus::log::LogOptions::loggers_per_node_, loggers_per_node_, meta_buffer_, meta_logger_, foedus::soc::GlobalMemoryAnchors::meta_logger_memory_, foedus::Attachable< CONTROL_BLOCK >::set_engine(), foedus::EngineOptions::thread_, foedus::thread::ThreadOptions::thread_count_per_group_, and foedus::Epoch::value().

                                             {
   groups_ = engine_->get_options().thread_.group_count_;
   loggers_per_node_ = engine_->get_options().log_.loggers_per_node_;
   const LoggerId total_loggers = loggers_per_node_ * groups_;
   const uint16_t total_threads = engine_->get_options().thread_.get_total_thread_count();
   LOG(INFO) << "Initializing LogManager. #loggers_per_node=" << loggers_per_node_
     << ", #NUMA-nodes=" << static_cast<int>(groups_) << ", #total_threads=" << total_threads;
   if (!engine_->get_thread_pool()->is_initialized()
     || !engine_->get_savepoint_manager()->is_initialized()) {
     return ERROR_STACK(kErrorCodeDepedentModuleUnavailableInit);
   }
   // see comments in LogOptions#log_paths_
   if (total_loggers == 0 || total_loggers % groups_ != 0 || total_threads % total_loggers != 0
     || total_loggers > total_threads) {
     return ERROR_STACK(kErrorCodeLogInvalidLoggerCount);
   }
 
   // attach control block
   soc::SharedMemoryRepo* memory_repo = engine_->get_soc_manager()->get_shared_memory_repo();
   control_block_ = memory_repo->get_global_memory_anchors()->log_manager_memory_;
 
   // attach logger_refs
   const uint16_t cores_per_logger = total_threads / total_loggers;
   for (thread::ThreadGroupId node = 0; node < groups_; ++node) {
     soc::NodeMemoryAnchors* node_anchors = memory_repo->get_node_memory_anchors(node);
     for (uint16_t j = 0; j < loggers_per_node_; ++j) {
       LoggerControlBlock* logger_block = node_anchors->logger_memories_[j];
       logger_refs_.emplace_back(LoggerRef(
         engine_,
         logger_block,
         node * loggers_per_node_ + j,
         node,
         j));
     }
   }
 
   // meta_buffer_ exists in all engines
   meta_buffer_.set_engine(engine_);
   meta_buffer_.attach(memory_repo->get_global_memory_anchors()->meta_logger_memory_);
 
   if (engine_->is_master()) {
     // In master, we initialize the control block. No local loggers.
     // Initialize durable_global_epoch_
     control_block_->initialize();
     control_block_->durable_global_epoch_
       = engine_->get_savepoint_manager()->get_initial_durable_epoch().value();
     LOG(INFO) << "durable_global_epoch_=" << get_durable_global_epoch();
     meta_logger_ = new MetaLogger(engine_);
     CHECK_ERROR(meta_logger_->initialize());
   } else {
     // In SOC, we don't have to initialize the control block, but have to launch local loggers.
     // evenly distribute loggers to NUMA nodes, then to cores.
     soc::SocId node = engine_->get_soc_id();
     thread::ThreadLocalOrdinal current_ordinal = 0;
     soc::NodeMemoryAnchors* node_anchors = memory_repo->get_node_memory_anchors(node);
     for (uint16_t j = 0; j < loggers_per_node_; ++j) {
       LoggerControlBlock* logger_block = node_anchors->logger_memories_[j];
       std::vector< thread::ThreadId > assigned_thread_ids;
       for (auto k = 0; k < cores_per_logger; ++k) {
         assigned_thread_ids.push_back(thread::compose_thread_id(node, current_ordinal));
         current_ordinal++;
       }
       std::string folder = engine_->get_options().log_.convert_folder_path_pattern(node, j);
       // to avoid race, create the root log folder now.
       fs::Path path(folder);
       if (!fs::exists(path)) {
         fs::create_directories(path);
       }
       Logger* logger = new Logger(
         engine_,
         logger_block,
         node * loggers_per_node_ + j,
         node,
         j,
         fs::Path(folder),
         assigned_thread_ids);
       CHECK_OUTOFMEMORY(logger);
       loggers_.push_back(logger);
     }
     ASSERT_ND(current_ordinal == engine_->get_options().thread_.thread_count_per_group_);
 
     // call initialize() of each logger.
     // this might take long, so do it in parallel.
     std::vector<std::thread> init_threads;
     for (auto j = 0; j < loggers_per_node_; ++j) {
       Logger* logger = loggers_[j];
       init_threads.push_back(std::thread([logger]() {
         COERCE_ERROR(logger->initialize());  // TASK(Hideaki) collect errors
       }));
     }
     LOG(INFO) << "Launched threads to initialize loggers in node-" << node << ". waiting..";
     for (auto& init_thread : init_threads) {
       init_thread.join();
     }
     LOG(INFO) << "All loggers in node-" << node << " were initialized!";
   }
 
   return kRetOk;
 }

Here is the call graph for this function:

ErrorStack foedus::log::LogManagerPimpl::refresh_global_durable_epoch ( )

Definition at line 179 of file log_manager_pimpl.cpp.

References ASSERT_ND, CHECK_ERROR, control_block_, foedus::log::LogManagerControlBlock::durable_global_epoch_, foedus::log::LogManagerControlBlock::durable_global_epoch_advanced_, foedus::log::LogManagerControlBlock::durable_global_epoch_savepoint_mutex_, engine_, get_durable_global_epoch(), foedus::Engine::get_savepoint_manager(), foedus::Epoch::is_valid(), foedus::kRetOk, logger_refs_, foedus::assorted::memory_fence_acquire(), foedus::assorted::memory_fence_release(), foedus::soc::SharedPolling::signal(), foedus::Epoch::store_min(), foedus::savepoint::SavepointManager::take_savepoint(), and foedus::Epoch::value().

Referenced by foedus::log::LogManager::refresh_global_durable_epoch().

                                                          {
   assorted::memory_fence_acquire();
   Epoch min_durable_epoch;
   ASSERT_ND(!min_durable_epoch.is_valid());
   for (const LoggerRef& logger : logger_refs_) {
     min_durable_epoch.store_min(logger.get_durable_epoch());
   }
   ASSERT_ND(min_durable_epoch.is_valid());
 
   if (min_durable_epoch <= get_durable_global_epoch()) {
     VLOG(0) << "durable_global_epoch_ not advanced";
     return kRetOk;
   }
 
   VLOG(0) << "Global durable epoch is about to advance from " << get_durable_global_epoch()
     << " to " << min_durable_epoch;
   {
     soc::SharedMutexScope guard(&control_block_->durable_global_epoch_savepoint_mutex_);
     if (min_durable_epoch <= get_durable_global_epoch()) {
       VLOG(0) << "oh, I lost the race.";
       return kRetOk;
     }
 
     CHECK_ERROR(engine_->get_savepoint_manager()->take_savepoint(min_durable_epoch));
 
     // set durable_global_epoch_ within the SharedCond's mutex scope, and then broadcast.
     // this is required to avoid lost signals.
     control_block_->durable_global_epoch_ = min_durable_epoch.value();
     assorted::memory_fence_release();
     control_block_->durable_global_epoch_advanced_.signal();
   }
   return kRetOk;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorStack foedus::log::LogManagerPimpl::uninitialize_once ( )

overridevirtual

Implements foedus::DefaultInitializable.

Definition at line 149 of file log_manager_pimpl.cpp.

References ASSERT_ND, control_block_, foedus::ErrorStackBatch::emprace_back(), engine_, ERROR_STACK, foedus::Engine::get_savepoint_manager(), foedus::Engine::get_thread_pool(), foedus::savepoint::SavepointManager::is_initialized(), foedus::thread::ThreadPool::is_initialized(), foedus::Engine::is_master(), foedus::kErrorCodeDepedentModuleUnavailableUninit, logger_refs_, loggers_, meta_logger_, SUMMARIZE_ERROR_BATCH, foedus::log::LogManagerControlBlock::uninitialize(), foedus::DefaultInitializable::uninitialize(), and foedus::ErrorStackBatch::uninitialize_and_delete_all().

                                               {
   LOG(INFO) << "Uninitializing LogManager..";
   ErrorStackBatch batch;
   if (!engine_->get_thread_pool()->is_initialized()
     || !engine_->get_savepoint_manager()->is_initialized()) {
     batch.emprace_back(ERROR_STACK(kErrorCodeDepedentModuleUnavailableUninit));
   }
   if (engine_->is_master()) {
     ASSERT_ND(loggers_.empty());
     if (meta_logger_) {
       batch.emprace_back(meta_logger_->uninitialize());
       delete meta_logger_;
       meta_logger_ = nullptr;
     }
   } else {
     ASSERT_ND(meta_logger_ == nullptr);
   }
   batch.uninitialize_and_delete_all(&loggers_);
   logger_refs_.clear();
   if (engine_->is_master()) {
     control_block_->uninitialize();
   }
   return SUMMARIZE_ERROR_BATCH(batch);
 }

Here is the call graph for this function:

ErrorCode foedus::log::LogManagerPimpl::wait_until_durable	(	Epoch	commit_epoch,
		int64_t	wait_microseconds
	)

Definition at line 214 of file log_manager_pimpl.cpp.

References foedus::soc::SharedPolling::acquire_ticket(), control_block_, foedus::log::LogManagerControlBlock::durable_global_epoch_advanced_, get_durable_global_epoch(), foedus::kErrorCodeOk, foedus::kErrorCodeTimeout, logger_refs_, foedus::assorted::memory_fence_acquire(), SPINLOCK_WHILE, foedus::soc::SharedPolling::timedwait(), and foedus::soc::SharedPolling::wait().

Referenced by foedus::log::LogManager::wait_until_durable().

                                                                                            {
   assorted::memory_fence_acquire();
   if (commit_epoch <= get_durable_global_epoch()) {
     DVLOG(1) << "Already durable. commit_epoch=" << commit_epoch << ", durable_global_epoch_="
       << get_durable_global_epoch();
     return kErrorCodeOk;
   }
 
   if (wait_microseconds == 0) {
     DVLOG(1) << "Conditional check: commit_epoch=" << commit_epoch << ", durable_global_epoch_="
       << get_durable_global_epoch();
     return kErrorCodeTimeout;
   }
 
   std::chrono::high_resolution_clock::time_point now = std::chrono::high_resolution_clock::now();
   std::chrono::high_resolution_clock::time_point until
     = now + std::chrono::microseconds(wait_microseconds);
   // @spinlock, but with sleep (not frequently called)
   SPINLOCK_WHILE(commit_epoch > get_durable_global_epoch()) {
     for (LoggerRef& logger : logger_refs_) {
       logger.wakeup_for_durable_epoch(commit_epoch);
     }
 
     VLOG(0) << "Synchronously waiting for commit_epoch " << commit_epoch;
     if (wait_microseconds <= 0)  {
       // set durable_global_epoch_ within the SharedCond's mutex scope, and then wait.
       // this is required to avoid lost signals.
       uint64_t demand = control_block_->durable_global_epoch_advanced_.acquire_ticket();
       if (commit_epoch <= get_durable_global_epoch()) {
         break;
       }
       control_block_->durable_global_epoch_advanced_.wait(demand);
       continue;
     }
 
     if (std::chrono::high_resolution_clock::now() >= until) {
       LOG(WARNING) << "Timeout occurs. wait_microseconds=" << wait_microseconds;
       return kErrorCodeTimeout;
     }
 
     {
       uint64_t demand = control_block_->durable_global_epoch_advanced_.acquire_ticket();
       if (commit_epoch <= get_durable_global_epoch()) {
         break;
       }
       control_block_->durable_global_epoch_advanced_.timedwait(
         demand,
         wait_microseconds);  // a bit lazy. we sleep longer in case of spurrious wakeup
     }
   }
 
   VLOG(0) << "durable epoch advanced. durable_global_epoch_=" << get_durable_global_epoch();
   return kErrorCodeOk;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::log::LogManagerPimpl::wakeup_loggers ( )

Definition at line 173 of file log_manager_pimpl.cpp.

References logger_refs_.

Referenced by foedus::log::LogManager::wakeup_loggers().

                                      {
   for (LoggerRef& logger : logger_refs_) {
     logger.wakeup();
   }
 }

Here is the caller graph for this function:

Member Data Documentation

LogManagerControlBlock* foedus::log::LogManagerPimpl::control_block_

Definition at line 119 of file log_manager_pimpl.hpp.

Referenced by announce_new_durable_global_epoch(), get_durable_global_epoch(), get_durable_global_epoch_weak(), initialize_once(), refresh_global_durable_epoch(), uninitialize_once(), and wait_until_durable().

Engine* const foedus::log::LogManagerPimpl::engine_

Definition at line 100 of file log_manager_pimpl.hpp.

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

thread::ThreadGroupId foedus::log::LogManagerPimpl::groups_

Definition at line 102 of file log_manager_pimpl.hpp.

Referenced by initialize_once().

std::vector< LoggerRef > foedus::log::LogManagerPimpl::logger_refs_

All log writers.

Index is global logger ordinal.

Definition at line 112 of file log_manager_pimpl.hpp.

Referenced by copy_logger_states(), foedus::log::LogManager::get_logger(), initialize_once(), refresh_global_durable_epoch(), uninitialize_once(), wait_until_durable(), and wakeup_loggers().

std::vector< Logger* > foedus::log::LogManagerPimpl::loggers_

Local log writers.

Index is local logger ordinal. Empty in master engine.

Definition at line 109 of file log_manager_pimpl.hpp.

Referenced by initialize_once(), and uninitialize_once().

uint16_t foedus::log::LogManagerPimpl::loggers_per_node_

Definition at line 103 of file log_manager_pimpl.hpp.

Referenced by initialize_once().

MetaLogBuffer foedus::log::LogManagerPimpl::meta_buffer_

Metadata log buffer.

Exists in all engines

Definition at line 117 of file log_manager_pimpl.hpp.

Referenced by foedus::log::LogManager::get_meta_buffer(), and initialize_once().

MetaLogger* foedus::log::LogManagerPimpl::meta_logger_

Metadata log writer.

Instantiated only in master

Definition at line 115 of file log_manager_pimpl.hpp.

Referenced by foedus::log::LogManager::get_meta_logger(), initialize_once(), and uninitialize_once().

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

/home/shino/foedus_code/foedus-core/include/foedus/log/log_manager_pimpl.hpp
/home/shino/foedus_code/foedus-core/src/foedus/log/log_manager_pimpl.cpp

Detailed Description

Public Member Functions

Public Attributes

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation