xct.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_XCT_XCT_HPP_
#define FOEDUS_XCT_XCT_HPP_

#include <iosfwd>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/epoch.hpp"
#include "foedus/error_code.hpp"
#include "foedus/fwd.hpp"
#include "foedus/log/common_log_types.hpp"

// For log verification. Only in debug mode
#ifndef NDEBUG
#include "foedus/log/log_type_invoke.hpp"
#endif  // NDEBUG

#include "foedus/memory/fwd.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/record.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/thread/thread_id.hpp"
#include "foedus/xct/fwd.hpp"
#include "foedus/xct/retrospective_lock_list.hpp"
#include "foedus/xct/xct_access.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace xct {

class Xct {
 public:
  enum Constants {
    kMaxPointerSets = 1024,
    kMaxPageVersionSets = 1024,
  };

  Xct(Engine* engine, thread::Thread* context, thread::ThreadId thread_id);

  // No copy
  Xct(const Xct& other) CXX11_FUNC_DELETE;
  Xct& operator=(const Xct& other) CXX11_FUNC_DELETE;

  void initialize(
    memory::NumaCoreMemory* core_memory,
    uint32_t* mcs_block_current,
    uint32_t* mcs_rw_async_mapping_current);

  void                activate(IsolationLevel isolation_level) {
    ASSERT_ND(!active_);
    active_ = true;
    enable_rll_for_this_xct_ = default_rll_for_this_xct_;
    hot_threshold_for_this_xct_ = default_hot_threshold_for_this_xct_;
    rll_threshold_for_this_xct_ = default_rll_threshold_for_this_xct_;
    isolation_level_ = isolation_level;
    pointer_set_size_ = 0;
    page_version_set_size_ = 0;
    read_set_size_ = 0;
    write_set_size_ = 0;
    lock_free_read_set_size_ = 0;
    lock_free_write_set_size_ = 0;
    *mcs_block_current_ = 0;
    *mcs_rw_async_mapping_current_ = 0;
    local_work_memory_cur_ = 0;
    current_lock_list_.clear_entries();
    if (!retrospective_lock_list_.is_empty()) {
      // If we have RLL, we will highly likely lock all of them.
      // So, let's make CLL entries for all of them at the beginning.
      // This is both for simplicity and performance.
      current_lock_list_.prepopulate_for_retrospective_lock_list(retrospective_lock_list_);
    }
  }

  void                deactivate() {
    ASSERT_ND(active_);
    ASSERT_ND(current_lock_list_.is_empty());
    active_ = false;
    *mcs_block_current_ = 0;
    *mcs_rw_async_mapping_current_ = 0;
  }

  uint32_t            get_mcs_block_current() const { return *mcs_block_current_; }
  uint32_t            increment_mcs_block_current() { return ++(*mcs_block_current_); }
  void                decrement_mcs_block_current() { --(*mcs_block_current_); }

  bool                is_active() const { return active_; }

  bool  is_enable_rll_for_this_xct() const { return enable_rll_for_this_xct_; }
  void  set_enable_rll_for_this_xct(bool value) { enable_rll_for_this_xct_ = value; }
  bool  is_default_rll_for_this_xct() const { return default_rll_for_this_xct_ ; }
  void  set_default_rll_for_this_xct(bool value) { default_rll_for_this_xct_ = value; }

  uint16_t  get_hot_threshold_for_this_xct() const { return hot_threshold_for_this_xct_; }
  void  set_hot_threshold_for_this_xct(uint16_t value) { hot_threshold_for_this_xct_ = value; }
  uint16_t  get_default_hot_threshold_for_this_xct() const {
    return default_hot_threshold_for_this_xct_ ; }
  void  set_default_hot_threshold_for_this_xct(uint16_t value) {
    default_hot_threshold_for_this_xct_ = value; }

  uint16_t  get_rll_threshold_for_this_xct() const { return rll_threshold_for_this_xct_; }
  void  set_rll_threshold_for_this_xct(uint16_t value) { rll_threshold_for_this_xct_ = value; }
  uint16_t  get_default_rll_threshold_for_this_xct() const {
    return default_rll_threshold_for_this_xct_ ; }
  void  set_default_rll_threshold_for_this_xct(uint16_t value) {
    default_rll_threshold_for_this_xct_ = value; }

  SysxctWorkspace* get_sysxct_workspace() const { return sysxct_workspace_; }

  bool                is_read_only() const {
    return write_set_size_ == 0 && lock_free_write_set_size_ == 0;
  }
  IsolationLevel      get_isolation_level() const { return isolation_level_; }
  const XctId&        get_id() const { return id_; }
  thread::Thread*     get_thread_context() { return context_; }
  thread::ThreadId    get_thread_id() const { return thread_id_; }
  uint32_t            get_pointer_set_size() const { return pointer_set_size_; }
  uint32_t            get_page_version_set_size() const { return page_version_set_size_; }
  uint32_t            get_read_set_size() const { return read_set_size_; }
  uint32_t            get_write_set_size() const { return write_set_size_; }
  uint32_t            get_lock_free_read_set_size() const { return lock_free_read_set_size_; }
  uint32_t            get_lock_free_write_set_size() const { return lock_free_write_set_size_; }
  const PointerAccess*   get_pointer_set() const { return pointer_set_; }
  const PageVersionAccess*  get_page_version_set() const { return page_version_set_; }
  ReadXctAccess*      get_read_set()  { return read_set_; }
  WriteXctAccess*     get_write_set() { return write_set_; }
  LockFreeReadXctAccess* get_lock_free_read_set() { return lock_free_read_set_; }
  LockFreeWriteXctAccess* get_lock_free_write_set() { return lock_free_write_set_; }


  void                issue_next_id(XctId max_xct_id, Epoch *epoch);

  ErrorCode           add_to_pointer_set(
    const storage::VolatilePagePointer* pointer_address,
    storage::VolatilePagePointer observed);

  void                overwrite_to_pointer_set(
    const storage::VolatilePagePointer* pointer_address,
    storage::VolatilePagePointer observed);

  ErrorCode           add_to_page_version_set(
    const storage::PageVersion* version_address,
    storage::PageVersionStatus observed);

  ErrorCode           on_record_read(
    bool intended_for_write,
    RwLockableXctId* tid_address,
    XctId* observed_xid,
    ReadXctAccess** read_set_address,
    bool no_readset_if_moved = false,
    bool no_readset_if_next_layer = false);
  ErrorCode           on_record_read(
    bool intended_for_write,
    RwLockableXctId* tid_address,
    bool no_readset_if_moved = false,
    bool no_readset_if_next_layer = false) {
    XctId dummy_xctid;
    ReadXctAccess* dummy_read_set;
    return on_record_read(
      intended_for_write,
      tid_address,
      &dummy_xctid,
      &dummy_read_set,
      no_readset_if_moved ,
      no_readset_if_next_layer);
  }
  void on_record_read_take_locks_if_needed(
    bool intended_for_write,
    const storage::Page* page_address,
    UniversalLockId lock_id,
    RwLockableXctId* tid_address);

  ErrorCode           add_related_write_set(
    ReadXctAccess* related_read_set,
    RwLockableXctId* tid_address,
    char* payload_address,
    log::RecordLogType* log_entry);

  ErrorCode           add_to_read_set(
    storage::StorageId storage_id,
    XctId observed_owner_id,
    RwLockableXctId* owner_id_address,
    ReadXctAccess** read_set_address);
  ErrorCode           add_to_read_set(
    storage::StorageId storage_id,
    XctId observed_owner_id,
    UniversalLockId owner_lock_id,
    RwLockableXctId* owner_id_address,
    ReadXctAccess** read_set_address);

  ErrorCode           add_to_write_set(
    storage::StorageId storage_id,
    RwLockableXctId* owner_id_address,
    char* payload_address,
    log::RecordLogType* log_entry);

  ErrorCode           add_to_write_set(
    storage::StorageId storage_id,
    storage::Record* record,
    log::RecordLogType* log_entry) {
    return add_to_write_set(storage_id, &record->owner_id_, record->payload_, log_entry);
  }

  ErrorCode           add_to_read_and_write_set(
    storage::StorageId storage_id,
    XctId observed_owner_id,
    RwLockableXctId* owner_id_address,
    char* payload_address,
    log::RecordLogType* log_entry);

  ErrorCode           add_to_lock_free_read_set(
    storage::StorageId storage_id,
    XctId observed_owner_id,
    RwLockableXctId* owner_id_address);

  ErrorCode           add_to_lock_free_write_set(
    storage::StorageId storage_id,
    log::RecordLogType* log_entry);

  void                remember_previous_xct_id(XctId new_id) {
    ASSERT_ND(id_.before(new_id));
    id_ = new_id;
    ASSERT_ND(id_.get_ordinal() > 0);
    ASSERT_ND(id_.is_valid());
  }

  ErrorCode           acquire_local_work_memory(uint32_t size, void** out, uint32_t alignment = 8) {
    if (size % alignment != 0) {
      size = ((size / alignment) + 1U) * alignment;
    }
    uint64_t begin = local_work_memory_cur_;
    if (begin % alignment != 0) {
      begin = ((begin / alignment) + 1U) * alignment;
    }
    if (UNLIKELY(size + begin > local_work_memory_size_)) {
      return kErrorCodeXctNoMoreLocalWorkMemory;
    }
    local_work_memory_cur_ = size + begin;
    *out = reinterpret_cast<char*>(local_work_memory_) + begin;
    return kErrorCodeOk;
  }

  xct::CurrentLockList*       get_current_lock_list() { return &current_lock_list_; }
  const xct::CurrentLockList* get_current_lock_list() const { return &current_lock_list_; }
  xct::RetrospectiveLockList* get_retrospective_lock_list() {
    return &retrospective_lock_list_;
  }

  bool assert_related_read_write() const ALWAYS_INLINE;

  friend std::ostream& operator<<(std::ostream& o, const Xct& v);

 private:
  Engine* const engine_;
  thread::Thread* const context_;

  const thread::ThreadId thread_id_;

  XctId               id_;

  IsolationLevel      isolation_level_;

  bool                active_;

  bool                enable_rll_for_this_xct_;
  bool                default_rll_for_this_xct_;
  uint16_t            hot_threshold_for_this_xct_;
  uint16_t            default_hot_threshold_for_this_xct_;
  uint16_t            rll_threshold_for_this_xct_;
  uint16_t            default_rll_threshold_for_this_xct_;


  SysxctWorkspace*    sysxct_workspace_;

  uint32_t*           mcs_block_current_;

  uint32_t*           mcs_rw_async_mapping_current_;

  ReadXctAccess*      read_set_;
  uint32_t            read_set_size_;
  uint32_t            max_read_set_size_;

  WriteXctAccess*     write_set_;
  uint32_t            write_set_size_;
  uint32_t            max_write_set_size_;

  LockFreeReadXctAccess*  lock_free_read_set_;
  uint32_t                lock_free_read_set_size_;
  uint32_t                max_lock_free_read_set_size_;

  LockFreeWriteXctAccess* lock_free_write_set_;
  uint32_t                lock_free_write_set_size_;
  uint32_t                max_lock_free_write_set_size_;

  PointerAccess*      pointer_set_;
  uint32_t            pointer_set_size_;

  PageVersionAccess*  page_version_set_;
  uint32_t            page_version_set_size_;

  xct::CurrentLockList    current_lock_list_;

  xct::RetrospectiveLockList  retrospective_lock_list_;

  void*               local_work_memory_;
  uint64_t            local_work_memory_size_;
  uint64_t            local_work_memory_cur_;
};

inline bool Xct::assert_related_read_write() const {
#ifndef NDEBUG
  for (uint32_t i = 0; i < write_set_size_; ++i) {
    WriteXctAccess* write = write_set_ + i;
    if (write->related_read_) {
      ASSERT_ND(write->related_read_ >= read_set_);
      uint32_t index = write->related_read_ - read_set_;
      ASSERT_ND(index < read_set_size_);
      ASSERT_ND(write->owner_id_address_ == write->related_read_->owner_id_address_);
      ASSERT_ND(write == write->related_read_->related_write_);
    }
  }

  for (uint32_t i = 0; i < read_set_size_; ++i) {
    ReadXctAccess* read = read_set_ + i;
    if (read->related_write_) {
      ASSERT_ND(read->related_write_ >= write_set_);
      uint32_t index = read->related_write_ - write_set_;
      ASSERT_ND(index < write_set_size_);
      ASSERT_ND(read->owner_id_address_ == read->related_write_->owner_id_address_);
      ASSERT_ND(read == read->related_write_->related_read_);
    }
  }
#endif  // NDEBUG
  return true;
}


}  // namespace xct
}  // namespace foedus
#endif  // FOEDUS_XCT_XCT_HPP_