shared_memory_repo.hpp

Go to the documentation of this file.

/*
 * Copyright (c) 2014-2015, Hewlett-Packard Development Company, LP.
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the Free
 * Software Foundation; either version 2 of the License, or (at your option)
 * any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details. You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 * HP designates this particular file as subject to the "Classpath" exception
 * as provided by HP in the LICENSE.txt file that accompanied this code.
 */
#ifndef FOEDUS_SOC_SHARED_MEMORY_REPO_HPP_
#define FOEDUS_SOC_SHARED_MEMORY_REPO_HPP_

#include <stdint.h>

#include <cstring>
#include <string>

#include "foedus/assert_nd.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/engine.hpp"
#include "foedus/engine_options.hpp"
#include "foedus/error_stack.hpp"
#include "foedus/module_type.hpp"
#include "foedus/assorted/atomic_fences.hpp"
#include "foedus/assorted/protected_boundary.hpp"
#include "foedus/log/fwd.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/memory/shared_memory.hpp"
#include "foedus/proc/fwd.hpp"
#include "foedus/proc/proc_id.hpp"
#include "foedus/restart/fwd.hpp"
#include "foedus/savepoint/fwd.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/soc/fwd.hpp"
#include "foedus/soc/soc_id.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/thread/thread_id.hpp"
#include "foedus/xct/fwd.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace soc {

struct MasterEngineStatus CXX11_FINAL {
  enum StatusCode {
    kInitial = 0,
    kSharedMemoryAllocated,
    kSharedMemoryReservedReclamation,

    kRunning,

    kWaitingForChildTerminate,
    kTerminated,
    kFatalError,
  };

  // only for reinterpret_cast
  MasterEngineStatus() CXX11_FUNC_DELETE;
  ~MasterEngineStatus() CXX11_FUNC_DELETE;
  MasterEngineStatus(const MasterEngineStatus &other) CXX11_FUNC_DELETE;
  MasterEngineStatus& operator=(const MasterEngineStatus &other) CXX11_FUNC_DELETE;

  void change_status_atomic(StatusCode new_status) {
    ASSERT_ND(new_status == kFatalError || new_status == kWaitingForChildTerminate ||
      static_cast<int>(new_status) == static_cast<int>(status_code_) + 1);
    assorted::memory_fence_acq_rel();
    status_code_ = new_status;
    assorted::memory_fence_acq_rel();
  }
  StatusCode read_status_atomic() const {
    assorted::memory_fence_acq_rel();
    return status_code_;
  }

  void change_init_atomic(ModuleType value) {
    ASSERT_ND(static_cast<int>(initialized_modules_) == static_cast<int>(value) - 1);
    assorted::memory_fence_acq_rel();
    initialized_modules_ = value;
    assorted::memory_fence_acq_rel();
  }
  void change_uninit_atomic(ModuleType value) {
    ASSERT_ND(static_cast<int>(uninitialized_modules_) == static_cast<int>(value) + 1);
    assorted::memory_fence_acq_rel();
    uninitialized_modules_ = value;
    assorted::memory_fence_acq_rel();
  }

  StatusCode status_code_;
  ModuleType initialized_modules_;
  ModuleType uninitialized_modules_;
  // so far this is the only information.
};

struct ChildEngineStatus CXX11_FINAL {
  enum StatusCode {
    kInitial = 0,
    kSharedMemoryAttached,
    kWaitingForMasterInitialization,

    kRunning,

    kTerminated,
    kFatalError,
  };

  // only for reinterpret_cast
  ChildEngineStatus() CXX11_FUNC_DELETE;
  ~ChildEngineStatus() CXX11_FUNC_DELETE;
  ChildEngineStatus(const ChildEngineStatus &other) CXX11_FUNC_DELETE;
  ChildEngineStatus& operator=(const ChildEngineStatus &other) CXX11_FUNC_DELETE;

  void change_status_atomic(StatusCode new_status) {
    ASSERT_ND(new_status == kFatalError ||
      static_cast<int>(new_status) == static_cast<int>(status_code_) + 1);
    assorted::memory_fence_acq_rel();
    status_code_ = new_status;
    assorted::memory_fence_acq_rel();
  }
  StatusCode read_status_atomic() const {
    assorted::memory_fence_acq_rel();
    return status_code_;
  }

  void change_init_atomic(ModuleType value) {
    ASSERT_ND(static_cast<int>(initialized_modules_) == static_cast<int>(value) - 1);
    assorted::memory_fence_acq_rel();
    initialized_modules_ = value;
    assorted::memory_fence_acq_rel();
  }
  void change_uninit_atomic(ModuleType value) {
    ASSERT_ND(static_cast<int>(uninitialized_modules_) == static_cast<int>(value) + 1);
    assorted::memory_fence_acq_rel();
    uninitialized_modules_ = value;
    assorted::memory_fence_acq_rel();
  }

  StatusCode status_code_;
  ModuleType initialized_modules_;
  ModuleType uninitialized_modules_;
  // so far this is the only information.
};

struct GlobalMemoryAnchors {
  enum Constants {
    kMasterStatusMemorySize = 1 << 12,
    kLogManagerMemorySize = 1 << 12,
    kMetaLoggerSize = 1 << 13,
    kRestartManagerMemorySize = 1 << 12,
    kSavepointManagerMemorySize = (3 << 19) + (1 << 12),
    kSnapshotManagerMemorySize = 1 << 12,
    kStorageManagerMemorySize = 1 << 12,
    kXctManagerMemorySize = 1 << 12,
    kStorageMemorySize = 1 << 12,
    kMaxBoundaries = 1 << 7,
  };

  GlobalMemoryAnchors() { clear(); }
  ~GlobalMemoryAnchors() {}
  void clear() { std::memset(this, 0, sizeof(*this)); }

  // No copying
  GlobalMemoryAnchors(const GlobalMemoryAnchors &other) CXX11_FUNC_DELETE;
  GlobalMemoryAnchors& operator=(const GlobalMemoryAnchors &other) CXX11_FUNC_DELETE;

  uint64_t        options_xml_length_;
  char*           options_xml_;

  MasterEngineStatus* master_status_memory_;

  log::LogManagerControlBlock*              log_manager_memory_;
  log::MetaLogControlBlock*                 meta_logger_memory_;
  restart::RestartManagerControlBlock*      restart_manager_memory_;
  savepoint::SavepointManagerControlBlock*  savepoint_manager_memory_;
  snapshot::SnapshotManagerControlBlock*    snapshot_manager_memory_;
  storage::StorageManagerControlBlock*      storage_manager_memory_;
  xct::XctManagerControlBlock*              xct_manager_memory_;

  storage::PartitionerMetadata*             partitioner_metadata_;
  void*                                     partitioner_data_;

  storage::StorageId*                       storage_name_sort_memory_;

  storage::StorageControlBlock*             storage_memories_;

  void*                                     user_memory_;

  assorted::ProtectedBoundary*              protected_boundaries_[kMaxBoundaries];
  uint32_t                                  protected_boundaries_count_;
  bool                                      protected_boundaries_needs_release_;
};

struct NodeMemoryAnchors {
  enum Constants {
    kChildStatusMemorySize = 1 << 12,
    kPagePoolMemorySize = 1 << 12,
    kLogReducerMemorySize = 1 << 12,
    kLoggerMemorySize = 1 << 21,
    kProcManagerMemorySize = 1 << 12,
    kMaxBoundaries = 1 << 12,
  };

  NodeMemoryAnchors() { clear(); }
  ~NodeMemoryAnchors() { deallocate_arrays(); }
  void clear() { std::memset(this, 0, sizeof(*this)); }

  // No copying
  NodeMemoryAnchors(const NodeMemoryAnchors &other) CXX11_FUNC_DELETE;
  NodeMemoryAnchors& operator=(const NodeMemoryAnchors &other) CXX11_FUNC_DELETE;

  void allocate_arrays(const EngineOptions& options);
  void deallocate_arrays();

  ChildEngineStatus*  child_status_memory_;

  memory::PagePoolControlBlock*   volatile_pool_status_;

  proc::ProcManagerControlBlock*  proc_manager_memory_;
  proc::ProcAndName*  proc_memory_;
  proc::LocalProcId*  proc_name_sort_memory_;

  snapshot::LogReducerControlBlock* log_reducer_memory_;

  void*               log_reducer_buffers_[2];

  storage::Page*      log_reducer_root_info_pages_;

  log::LoggerControlBlock** logger_memories_;

  ThreadMemoryAnchors*  thread_anchors_;

  void*               volatile_page_pool_;

  assorted::ProtectedBoundary*              protected_boundaries_[kMaxBoundaries];
  uint32_t                                  protected_boundaries_count_;
  bool                                      protected_boundaries_needs_release_;
};

struct ThreadMemoryAnchors {
  enum Constants {
    kThreadMemorySize = 1 << 15,
    kTaskInputMemorySize = 1 << 19,
    kTaskOutputMemorySize = 1 << 19,
    kMcsWwLockMemorySize = 1 << 19,
    kMcsRwLockMemorySize = 1 << 19,
    kMcsRwAsyncMappingMemorySize  = 1 << 19,
  };
  ThreadMemoryAnchors() { std::memset(this, 0, sizeof(*this)); }
  ~ThreadMemoryAnchors() {}

  // No copying
  ThreadMemoryAnchors(const ThreadMemoryAnchors &other) CXX11_FUNC_DELETE;
  ThreadMemoryAnchors& operator=(const ThreadMemoryAnchors &other) CXX11_FUNC_DELETE;

  thread::ThreadControlBlock* thread_memory_;

  void*           task_input_memory_;

  void*           task_output_memory_;

  xct::McsWwBlock*          mcs_ww_lock_memories_;
  // These two might be integrated, but for now we allocate both.
  xct::McsRwSimpleBlock*    mcs_rw_simple_lock_memories_;
  xct::McsRwExtendedBlock*  mcs_rw_extended_lock_memories_;
  xct::McsRwAsyncMapping*   mcs_rw_async_mappings_memories_;
};

class SharedMemoryRepo CXX11_FINAL {
 public:
  SharedMemoryRepo() :
    soc_count_(0),
    my_soc_id_(0),
    node_memories_(CXX11_NULLPTR),
    node_memory_anchors_(CXX11_NULLPTR) {}
  ~SharedMemoryRepo() { deallocate_shared_memories(); }

  // Disable copy constructor
  SharedMemoryRepo(const SharedMemoryRepo &other) CXX11_FUNC_DELETE;
  SharedMemoryRepo& operator=(const SharedMemoryRepo &other) CXX11_FUNC_DELETE;

  ErrorStack  allocate_shared_memories(uint64_t upid, Eid eid, const EngineOptions& options);

  ErrorStack  attach_shared_memories(
    uint64_t master_upid,
    Eid master_eid,
    SocId my_soc_id,
    EngineOptions* options);

  void        mark_for_release();

  void        deallocate_shared_memories();

  void*       get_global_memory() { return global_memory_.get_block(); }
  GlobalMemoryAnchors* get_global_memory_anchors() { return &global_memory_anchors_; }
  void*       get_global_user_memory() { return global_memory_anchors_.user_memory_; }

  void        change_master_status(MasterEngineStatus::StatusCode new_status);
  MasterEngineStatus::StatusCode get_master_status() const;

  void*       get_node_memory(SocId node) { return node_memories_[node].get_block(); }
  NodeMemoryAnchors* get_node_memory_anchors(SocId node) { return &node_memory_anchors_[node]; }

  void        change_child_status(SocId node, ChildEngineStatus::StatusCode new_status);
  ChildEngineStatus::StatusCode get_child_status(SocId node) const;

  ThreadMemoryAnchors* get_thread_memory_anchors(thread::ThreadId thread_id) {
    SocId node = thread::decompose_numa_node(thread_id);
    uint16_t thread_ordinal = thread::decompose_numa_local_ordinal(thread_id);
    return &node_memory_anchors_[node].thread_anchors_[thread_ordinal];
  }

  void* get_volatile_pool(SocId node) { return node_memory_anchors_[node].volatile_page_pool_; }

  static uint64_t calculate_global_memory_size(uint64_t xml_size, const EngineOptions& options);
  static uint64_t calculate_node_memory_size(const EngineOptions& options);

 private:
  SocId                 soc_count_;
  SocId                 my_soc_id_;

  memory::SharedMemory  global_memory_;
  GlobalMemoryAnchors   global_memory_anchors_;

  memory::SharedMemory* node_memories_;
  NodeMemoryAnchors*    node_memory_anchors_;

  void init_empty(const EngineOptions& options);

  void set_global_memory_anchors(
    uint64_t xml_size,
    const EngineOptions& options,
    bool reset_boundaries);
  void put_global_memory_boundary(
    uint64_t* position,
    const std::string& name,
    bool reset_boundaries) {
    char* base = global_memory_.get_block();
    assorted::ProtectedBoundary* boundary
      = reinterpret_cast<assorted::ProtectedBoundary*>(base + (*position));
    if (reset_boundaries) {
      boundary->reset(name);
    }
    uint32_t next_index = global_memory_anchors_.protected_boundaries_count_;
    ASSERT_ND(next_index < GlobalMemoryAnchors::kMaxBoundaries);
    global_memory_anchors_.protected_boundaries_[next_index] = boundary;
    ++global_memory_anchors_.protected_boundaries_count_;
    *position += sizeof(assorted::ProtectedBoundary);
  }

  void set_node_memory_anchors(SocId node, const EngineOptions& options, bool reset_boundaries);
  void put_node_memory_boundary(
    SocId node,
    uint64_t* position,
    const std::string& name,
    bool reset_boundaries) {
    char* base = node_memories_[node].get_block();
    assorted::ProtectedBoundary* boundary
      = reinterpret_cast<assorted::ProtectedBoundary*>(base + (*position));
    if (reset_boundaries) {
      boundary->reset(name);
    }
    uint32_t next_index = node_memory_anchors_[node].protected_boundaries_count_;
    ASSERT_ND(next_index < NodeMemoryAnchors::kMaxBoundaries);
    node_memory_anchors_[node].protected_boundaries_[next_index] = boundary;
    ++node_memory_anchors_[node].protected_boundaries_count_;
    *position += sizeof(assorted::ProtectedBoundary);
  }


  static void allocate_one_node(
    uint64_t upid,
    Eid eid,
    uint16_t node,
    uint64_t node_memory_size,
    bool rigorous_memory_boundary_check,
    bool rigorous_page_boundary_check,
    ErrorStack* alloc_result,
    SharedMemoryRepo* repo);
};

CXX11_STATIC_ASSERT(
  sizeof(MasterEngineStatus) <= (1 << 12),
  "Size of MasterEngineStatus exceeds 4kb");

CXX11_STATIC_ASSERT(
  sizeof(ChildEngineStatus) <= (1 << 12),
  "Size of ChildEngineStatus exceeds 4kb");

}  // namespace soc
}  // namespace foedus
#endif  // FOEDUS_SOC_SHARED_MEMORY_REPO_HPP_