xct_manager_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_XCT_XCT_MANAGER_PIMPL_HPP_
#define FOEDUS_XCT_XCT_MANAGER_PIMPL_HPP_
#include <atomic>
#include <thread>

#include "foedus/epoch.hpp"
#include "foedus/fwd.hpp"
#include "foedus/initializable.hpp"
#include "foedus/soc/shared_memory_repo.hpp"
#include "foedus/soc/shared_polling.hpp"
#include "foedus/thread/condition_variable_impl.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/thread/stoppable_thread_impl.hpp"
#include "foedus/xct/fwd.hpp"
#include "foedus/xct/retrospective_lock_list.hpp"  // to inline CurrentLockListIteratorForWriteSet
#include "foedus/xct/xct_access.hpp"               // same above. iterator must be fast...
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace xct {
struct XctManagerControlBlock {
  // this is backed by shared memory. not instantiation. just reinterpret_cast.
  XctManagerControlBlock() = delete;
  ~XctManagerControlBlock() = delete;

  void initialize() {
    current_global_epoch_advanced_.initialize();
    epoch_chime_wakeup_.initialize();
    new_transaction_paused_ = false;
  }
  void uninitialize() {
  }

  std::atomic<Epoch::EpochInteger>  current_global_epoch_;
  std::atomic<Epoch::EpochInteger>  requested_global_epoch_;

  soc::SharedPolling                current_global_epoch_advanced_;

  soc::SharedPolling                epoch_chime_wakeup_;
  std::atomic<bool>                 epoch_chime_terminate_requested_;

  std::atomic<bool>                 new_transaction_paused_;
};

class XctManagerPimpl final : public DefaultInitializable {
 public:
  XctManagerPimpl() = delete;
  explicit XctManagerPimpl(Engine* engine) : engine_(engine) {}
  ErrorStack  initialize_once() override;
  ErrorStack  uninitialize_once() override;

  Epoch       get_current_global_epoch() const {
    return Epoch(control_block_->current_global_epoch_.load());
  }
  Epoch       get_requested_global_epoch() const {
    return Epoch(control_block_->requested_global_epoch_.load());
  }
  Epoch       get_current_global_epoch_weak() const {
    return Epoch(control_block_->current_global_epoch_.load(std::memory_order_relaxed));
  }

  ErrorCode   begin_xct(thread::Thread* context, IsolationLevel isolation_level);
  ErrorCode   precommit_xct(thread::Thread* context, Epoch *commit_epoch);
  ErrorCode   abort_xct(thread::Thread* context);

  ErrorCode   wait_for_commit(Epoch commit_epoch, int64_t wait_microseconds);
  void        set_requested_global_epoch(Epoch request);
  void        advance_current_global_epoch();
  void        wait_for_current_global_epoch(Epoch target_epoch, int64_t wait_microseconds);
  void        wakeup_epoch_chime_thread();

  ErrorCode   precommit_xct_readonly(thread::Thread* context, Epoch *commit_epoch);
  ErrorCode   precommit_xct_readwrite(thread::Thread* context, Epoch *commit_epoch);

  bool        precommit_xct_lock_track_write(thread::Thread* context, WriteXctAccess* entry);
  bool        precommit_xct_verify_track_read(thread::Thread* context, ReadXctAccess* entry);
  ErrorCode   precommit_xct_lock(thread::Thread* context, XctId* max_xct_id);
  ErrorCode   precommit_xct_lock_batch_track_moved(thread::Thread* context);
  bool        precommit_xct_verify_readonly(thread::Thread* context, Epoch *commit_epoch);
  bool        precommit_xct_verify_readwrite(thread::Thread* context, XctId* max_xct_id);
  bool        precommit_xct_verify_pointer_set(thread::Thread* context);
  bool        precommit_xct_verify_page_version_set(thread::Thread* context);
  void        precommit_xct_apply(thread::Thread* context, XctId max_xct_id, Epoch *commit_epoch);
  void        release_and_clear_all_current_locks(thread::Thread* context);
  bool        precommit_xct_acquire_writer_lock(thread::Thread* context, WriteXctAccess *write);
  void        precommit_xct_sort_access(thread::Thread* context);
  bool        precommit_xct_try_acquire_writer_locks(thread::Thread* context);
  bool        precommit_xct_request_writer_lock(thread::Thread* context, WriteXctAccess* write);

  void        handle_epoch_chime();
  void        handle_epoch_chime_wait_grace_period(Epoch grace_epoch);
  bool        is_stop_requested() const;

  void        pause_accepting_xct();
  void        resume_accepting_xct();
  void        wait_until_resume_accepting_xct(thread::Thread* context);

  Engine* const                 engine_;
  XctManagerControlBlock*       control_block_;

  std::thread epoch_chime_thread_;
};


struct CurrentLockListIteratorForWriteSet {
  CurrentLockListIteratorForWriteSet(
    const WriteXctAccess* write_set,
    const CurrentLockList* cll,
    uint32_t write_set_size);

  void next_writes();
  bool is_valid() const { return write_cur_pos_ < write_next_pos_; }

  const WriteXctAccess* const   write_set_;
  const CurrentLockList* const  cll_;
  const uint32_t    write_set_size_;

  uint32_t          write_cur_pos_;
  uint32_t          write_next_pos_;
  LockListPosition  cll_pos_;
};

inline CurrentLockListIteratorForWriteSet::CurrentLockListIteratorForWriteSet(
  const WriteXctAccess* write_set,
  const CurrentLockList* cll,
  uint32_t write_set_size)
  : write_set_(write_set),
  cll_(cll),
  write_set_size_(write_set_size)  {
  write_cur_pos_ = 0;
  write_next_pos_ = 0;
  cll_pos_ = kLockListPositionInvalid;
  cll_->assert_sorted();

  next_writes();  // set to initial record.
}

inline void CurrentLockListIteratorForWriteSet::next_writes() {
  write_cur_pos_ = write_next_pos_;
  ++cll_pos_;
  if (write_cur_pos_ >= write_set_size_) {
    return;
  }
  const WriteXctAccess* write = write_set_ + write_cur_pos_;
  const UniversalLockId write_id = write->owner_lock_id_;

  // CLL must contain all entries in write-set. We are reading in-order.
  // So, we must find a valid CLL entry that is == write_id
  const LockEntry* l = cll_->get_entry(cll_pos_);
  while (l->universal_lock_id_ < write_id) {
    ASSERT_ND(cll_pos_ < cll_->get_last_active_entry());
    ++cll_pos_;
    l = cll_->get_entry(cll_pos_);
  }

  ASSERT_ND(l->universal_lock_id_ == write_id);
  ASSERT_ND(write_next_pos_ < write_set_size_);
  while (true) {
    ++write_next_pos_;
    if (write_next_pos_ == write_set_size_) {
      break;
    }
    const WriteXctAccess* next_write = write_set_ + write_next_pos_;
    const UniversalLockId next_write_id = next_write->owner_lock_id_;
    ASSERT_ND(write_id <= next_write_id);
    if (write_id < next_write_id) {
      break;
    }
  }
}

static_assert(
  sizeof(XctManagerControlBlock) <= soc::GlobalMemoryAnchors::kXctManagerMemorySize,
  "XctManagerControlBlock is too large.");
}  // namespace xct
}  // namespace foedus
#endif  // FOEDUS_XCT_XCT_MANAGER_PIMPL_HPP_