sequential_storage_pimpl.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_STORAGE_SEQUENTIAL_SEQUENTIAL_STORAGE_PIMPL_HPP_
#define FOEDUS_STORAGE_SEQUENTIAL_SEQUENTIAL_STORAGE_PIMPL_HPP_
#include <stdint.h>

#include <atomic>

#include "foedus/assert_nd.hpp"
#include "foedus/attachable.hpp"
#include "foedus/compiler.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/engine.hpp"
#include "foedus/engine_options.hpp"
#include "foedus/fwd.hpp"
#include "foedus/memory/engine_memory.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/memory/numa_node_memory.hpp"
#include "foedus/soc/shared_memory_repo.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/storage/storage.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/storage/sequential/fwd.hpp"
#include "foedus/storage/sequential/sequential_id.hpp"
#include "foedus/storage/sequential/sequential_metadata.hpp"
#include "foedus/storage/sequential/sequential_page_impl.hpp"
#include "foedus/storage/sequential/sequential_storage.hpp"
#include "foedus/thread/fwd.hpp"

namespace foedus {
namespace storage {
namespace sequential {
struct SequentialStorageControlBlock final {
  // this is backed by shared memory. not instantiation. just reinterpret_cast.
  SequentialStorageControlBlock() = delete;
  ~SequentialStorageControlBlock() = delete;

  bool exists() const { return status_ == kExists || status_ == kMarkedForDeath; }
  ErrorCode optimistic_read_truncate_epoch(thread::Thread* context, Epoch* out) const;

  soc::SharedMutex    status_mutex_;
  StorageStatus       status_;
  DualPagePointer     root_page_pointer_;
  SequentialMetadata  meta_;

  // Do NOT reorder members up to here. The layout must be compatible with StorageControlBlock
  // Type-specific shared members below.

  xct::RwLockableXctId  cur_truncate_epoch_tid_;

  std::atomic< Epoch::EpochInteger >  cur_truncate_epoch_;

  VolatilePagePointer   head_pointer_pages_[kPointerPageCount];
  VolatilePagePointer   tail_pointer_pages_[kPointerPageCount];
};

class SequentialStoragePimpl final : public Attachable<SequentialStorageControlBlock> {
 public:
  struct PointerPage {
    memory::PagePoolOffset pointers_[kPointersPerPage];
  };
  SequentialStoragePimpl() = delete;
  explicit SequentialStoragePimpl(SequentialStorage* storage)
    : Attachable<SequentialStorageControlBlock>(
      storage->get_engine(),
      storage->get_control_block()) {
  }
  SequentialStoragePimpl(Engine* engine, SequentialStorageControlBlock* control_block)
    : Attachable<SequentialStorageControlBlock>(engine, control_block) {
  }

  bool        exists() const { return control_block_->exists(); }
  StorageId   get_id() const { return control_block_->meta_.id_; }
  const StorageName& get_name() const { return control_block_->meta_.name_; }
  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);

  template <typename HANDLER>
  ErrorCode for_every_page(HANDLER handler) const {
    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;
  }
};

static_assert(sizeof(SequentialStoragePimpl) <= kPageSize, "SequentialStoragePimpl is too large");
static_assert(
  sizeof(SequentialStorageControlBlock) <= soc::GlobalMemoryAnchors::kStorageMemorySize,
  "SequentialStorageControlBlock is too large.");
}  // namespace sequential
}  // namespace storage
}  // namespace foedus
#endif  // FOEDUS_STORAGE_SEQUENTIAL_SEQUENTIAL_STORAGE_PIMPL_HPP_