xct_id.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_ID_HPP_
#define FOEDUS_XCT_XCT_ID_HPP_
#include <stdint.h>

#include <iosfwd>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/epoch.hpp"
#include "foedus/fwd.hpp"
#include "foedus/assorted/assorted_func.hpp"
#include "foedus/assorted/atomic_fences.hpp"
#include "foedus/assorted/endianness.hpp"
#include "foedus/assorted/raw_atomics.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/thread/thread_id.hpp"

namespace foedus {
namespace xct {

enum IsolationLevel {
  kDirtyRead,

  kSnapshot,

  kSerializable,
};

enum LockMode {
  kNoLock = 0,
  kReadLock,
  kWriteLock,
};

typedef uintptr_t UniversalLockId;

const UniversalLockId kNullUniversalLockId = 0;

typedef uint32_t LockListPosition;
const LockListPosition kLockListPositionInvalid = 0;


typedef uint32_t McsBlockIndex;
const uint64_t kMcsGuestId = -1;

struct AcquireAsyncRet {
  bool acquired_;
  McsBlockIndex block_index_;
};


struct McsWwBlockData {
  uint64_t word_;

  static uint64_t combine(uint32_t thread_id, McsBlockIndex block) ALWAYS_INLINE {
    uint64_t word = thread_id;
    word <<= 32;
    word |= block;
    return word;
  }
  static uint32_t decompose_thread_id(uint64_t word) ALWAYS_INLINE {
    return static_cast<uint32_t>((word >> 32) & 0xFFFFFFFFUL);
  }
  static McsBlockIndex decompose_block(uint64_t word) ALWAYS_INLINE {
    return static_cast<McsBlockIndex>(word & 0xFFFFFFFFUL);
  }

  McsWwBlockData() : word_(0) {}
  explicit McsWwBlockData(uint64_t word) : word_(word) {}
  McsWwBlockData(uint32_t thread_id, McsBlockIndex block) : word_(combine(thread_id, block)) {}

  bool operator==(const McsWwBlockData& other) const { return word_ == other.word_; }
  bool operator!=(const McsWwBlockData& other) const { return word_ != other.word_; }

  uint64_t get_word_acquire() const ALWAYS_INLINE {
    return assorted::atomic_load_acquire<uint64_t>(&word_);
  }
  uint64_t get_word_consume() const ALWAYS_INLINE {
    return assorted::atomic_load_consume<uint64_t>(&word_);
  }
  uint64_t get_word_atomic() const ALWAYS_INLINE {
    return assorted::atomic_load_seq_cst<uint64_t>(&word_);
  }
  uint64_t get_word_once() const ALWAYS_INLINE { return *(&word_); }
  McsWwBlockData copy_once() const ALWAYS_INLINE { return McsWwBlockData(get_word_once()); }
  McsWwBlockData copy_consume() const ALWAYS_INLINE { return McsWwBlockData(get_word_consume()); }
  McsWwBlockData copy_acquire() const ALWAYS_INLINE { return McsWwBlockData(get_word_acquire()); }
  McsWwBlockData copy_atomic() const ALWAYS_INLINE { return McsWwBlockData(get_word_atomic()); }

  bool is_valid_relaxed() const ALWAYS_INLINE { return word_ != 0; }
  bool is_valid_consume() const ALWAYS_INLINE { return get_word_consume() != 0; }
  bool is_valid_acquire() const ALWAYS_INLINE { return get_word_acquire() != 0; }
  bool is_valid_atomic()  const ALWAYS_INLINE { return get_word_atomic() != 0; }

  bool is_guest_relaxed() const ALWAYS_INLINE { return word_ == kMcsGuestId; }
  bool is_guest_acquire() const ALWAYS_INLINE { return get_word_acquire() == kMcsGuestId; }
  bool is_guest_consume() const ALWAYS_INLINE { return get_word_consume() == kMcsGuestId; }
  bool is_guest_atomic()  const ALWAYS_INLINE { return get_word_atomic() == kMcsGuestId; }
  inline uint32_t get_thread_id_relaxed() const ALWAYS_INLINE {
    return McsWwBlockData::decompose_thread_id(word_);
  }
  inline McsBlockIndex  get_block_relaxed() const ALWAYS_INLINE {
    return McsWwBlockData::decompose_block(word_);
  }
  void clear() ALWAYS_INLINE { word_ = 0; }
  void clear_atomic() ALWAYS_INLINE { assorted::atomic_store_seq_cst<uint64_t>(&word_, 0); }
  void clear_release() ALWAYS_INLINE { assorted::atomic_store_release<uint64_t>(&word_, 0); }
  void set_relaxed(uint32_t thread_id, McsBlockIndex block) ALWAYS_INLINE {
    word_ = McsWwBlockData::combine(thread_id, block);
  }
  void set_atomic(uint32_t thread_id, McsBlockIndex block) ALWAYS_INLINE {
    set_combined_atomic(McsWwBlockData::combine(thread_id, block));
  }
  void set_release(uint32_t thread_id, McsBlockIndex block) ALWAYS_INLINE {
    set_combined_release(McsWwBlockData::combine(thread_id, block));
  }
  void set_combined_atomic(uint64_t word) ALWAYS_INLINE {
    assorted::atomic_store_seq_cst<uint64_t>(&word_, word);
  }
  void set_combined_release(uint64_t word) ALWAYS_INLINE {
    assorted::atomic_store_release<uint64_t>(&word_, word);
  }
};

struct McsWwBlock {
  McsWwBlockData successor_;

  inline bool has_successor_relaxed() const ALWAYS_INLINE { return successor_.is_valid_relaxed(); }
  inline bool has_successor_consume() const ALWAYS_INLINE { return successor_.is_valid_consume(); }
  inline bool has_successor_acquire() const ALWAYS_INLINE { return successor_.is_valid_acquire(); }
  inline bool has_successor_atomic()  const ALWAYS_INLINE { return successor_.is_valid_atomic(); }
  inline uint32_t         get_successor_thread_id_relaxed() const ALWAYS_INLINE {
    return successor_.get_thread_id_relaxed();
  }
  inline McsBlockIndex    get_successor_block_relaxed() const ALWAYS_INLINE {
    return successor_.get_block_relaxed();
  }
  inline void clear_successor_atomic() ALWAYS_INLINE { successor_.clear_atomic(); }
  inline void clear_successor_release() ALWAYS_INLINE { successor_.clear_release(); }
  inline void set_successor_atomic(thread::ThreadId thread_id, McsBlockIndex block) ALWAYS_INLINE {
    successor_.set_atomic(thread_id, block);
  }
  inline void set_successor_release(thread::ThreadId thread_id, McsBlockIndex block) ALWAYS_INLINE {
    successor_.set_release(thread_id, block);
  }
};

struct McsWwLock {
  McsWwLock() { tail_.clear(); }
  McsWwLock(thread::ThreadId tail_waiter, McsBlockIndex tail_waiter_block) {
    tail_.set_relaxed(tail_waiter, tail_waiter_block);
  }

  McsWwLock(const McsWwLock& other) CXX11_FUNC_DELETE;
  McsWwLock& operator=(const McsWwLock& other) CXX11_FUNC_DELETE;

  bool      is_locked() const { return tail_.is_valid_relaxed(); }

  void          ownerless_initial_lock();
  void          ownerless_acquire_lock();
  void          ownerless_release_lock();


  thread::ThreadId get_tail_waiter() const ALWAYS_INLINE { return tail_.get_thread_id_relaxed(); }
  McsBlockIndex get_tail_waiter_block() const ALWAYS_INLINE { return tail_.get_block_relaxed(); }

  McsWwBlockData get_tail_relaxed() const ALWAYS_INLINE { return tail_; }
  McsWwBlockData get_tail_once()    const ALWAYS_INLINE { return tail_.copy_once(); }
  McsWwBlockData get_tail_consume() const ALWAYS_INLINE { return tail_.copy_consume(); }
  McsWwBlockData get_tail_acquire() const ALWAYS_INLINE { return tail_.copy_acquire(); }
  McsWwBlockData get_tail_atomic()  const ALWAYS_INLINE { return tail_.copy_atomic(); }

  void  reset() ALWAYS_INLINE { tail_.clear(); }

  void  reset_guest_id_release() {
    tail_.set_combined_release(kMcsGuestId);
  }

  void  reset(thread::ThreadId tail_waiter, McsBlockIndex tail_waiter_block) ALWAYS_INLINE {
    tail_.set_relaxed(tail_waiter, tail_waiter_block);
  }

  void  reset_atomic() ALWAYS_INLINE { reset_atomic(0, 0); }
  void  reset_atomic(thread::ThreadId tail_waiter, McsBlockIndex tail_waiter_block) ALWAYS_INLINE {
    tail_.set_atomic(tail_waiter, tail_waiter_block);
  }
  void  reset_release() ALWAYS_INLINE { reset_release(0, 0); }
  void  reset_release(thread::ThreadId tail_waiter, McsBlockIndex tail_waiter_block) ALWAYS_INLINE {
    tail_.set_release(tail_waiter, tail_waiter_block);
  }

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

  McsWwBlockData tail_;
};


struct McsRwSimpleBlock {
  static const uint8_t kStateClassMask       = 3U;        // [LSB + 1, LSB + 2]
  static const uint8_t kStateClassReaderFlag = 1U;        // LSB binary = 01
  static const uint8_t kStateClassWriterFlag = 2U;        // LSB binary = 10

  static const uint8_t kStateBlockedFlag     = 1U << 7U;  // MSB binary = 1
  static const uint8_t kStateBlockedMask     = 1U << 7U;

  static const uint8_t kStateFinalizedMask   = 4U;

  static const uint8_t kSuccessorClassReader = 1U;
  static const uint8_t kSuccessorClassWriter = 2U;
  static const uint8_t kSuccessorClassNone   = 3U;        // LSB binary 11

  static const int32_t kTimeoutNever         = 0xFFFFFFFF;

  union Self {
    uint16_t data_;                       // +2 => 2
    struct Components {
      uint8_t successor_class_;
      // state_ covers:
      // Bit 0-1: my **own** class (am I a reader or writer?)
      // Bit 2: whether we have checked the successor ("finalized", for readers only)
      // Bit 7: blocked (am I waiting for the lock or acquired?)
      uint8_t state_;
    } components_;
  } self_;
  // TODO(tzwang): make these two fields 8 bytes by themselves. Now we need
  // to worry about sub-word writes (ie have to use atomic ops even when
  // changing only these two fields because they are in the same byte as data_).
  thread::ThreadId successor_thread_id_;  // +2 => 4
  McsBlockIndex successor_block_index_;   // +4 => 8

  inline void init_reader() {
    self_.components_.state_ = kStateClassReaderFlag | kStateBlockedFlag;
    init_common();
  }
  inline void init_writer() {
    self_.components_.state_ = kStateClassWriterFlag | kStateBlockedFlag;
    init_common();
  }
  inline void init_common() ALWAYS_INLINE {
    self_.components_.successor_class_ = kSuccessorClassNone;
    successor_thread_id_ = 0;
    successor_block_index_ = 0;
    assorted::memory_fence_release();
  }

  inline bool is_reader() ALWAYS_INLINE {
    return (self_.components_.state_ & kStateClassMask) == kStateClassReaderFlag;
  }
  inline uint8_t read_state() {
    return assorted::atomic_load_acquire<uint8_t>(&self_.components_.state_);
  }
  inline void unblock() ALWAYS_INLINE {
    ASSERT_ND(read_state() & kStateBlockedFlag);
    assorted::raw_atomic_fetch_and_bitwise_and<uint8_t>(
      &self_.components_.state_,
      static_cast<uint8_t>(~kStateBlockedMask));
  }
  inline bool is_blocked() ALWAYS_INLINE {
    return read_state() & kStateBlockedMask;
  }
  inline bool is_granted() {
    return !is_blocked();
  }
  inline void set_finalized() {
    ASSERT_ND(is_reader());
    ASSERT_ND(!is_finalized());
    assorted::raw_atomic_fetch_and_bitwise_or<uint8_t>(
      &self_.components_.state_, kStateFinalizedMask);
    ASSERT_ND(is_finalized());
  }
  inline bool is_finalized() {
    ASSERT_ND(is_reader());
    return read_state() & kStateFinalizedMask;
  }
  bool timeout_granted(int32_t timeout);
  inline void set_successor_class_writer() {
    // In case the caller is a reader appending after a writer or waiting reader,
    // the requester should have already set the successor class to "reader" through by CASing
    // self_.data_ from [no-successor, blocked] to [reader successor, blocked].
    ASSERT_ND(self_.components_.successor_class_ == kSuccessorClassNone);
    assorted::raw_atomic_fetch_and_bitwise_and<uint8_t>(
      &self_.components_.successor_class_, kSuccessorClassWriter);
  }
  inline void set_successor_next_only(thread::ThreadId thread_id, McsBlockIndex block_index) {
    McsRwSimpleBlock tmp;
    tmp.self_.data_ = 0;
    tmp.successor_thread_id_ = thread_id;
    tmp.successor_block_index_ = block_index;
    ASSERT_ND(successor_thread_id_ == 0);
    ASSERT_ND(successor_block_index_ == 0);
    uint64_t *address = reinterpret_cast<uint64_t*>(this);
    uint64_t mask = *reinterpret_cast<uint64_t*>(&tmp);
    assorted::raw_atomic_fetch_and_bitwise_or<uint64_t>(address, mask);
  }
  inline bool has_successor() {
    return assorted::atomic_load_acquire<uint8_t>(
      &self_.components_.successor_class_) != kSuccessorClassNone;
  }
  inline bool successor_is_ready() {
    // Check block index only - thread ID could be 0
    return assorted::atomic_load_acquire<McsBlockIndex>(&successor_block_index_) != 0;
  }
  inline bool has_reader_successor() {
    uint8_t s = assorted::atomic_load_acquire<uint8_t>(&self_.components_.successor_class_);
    return s == kSuccessorClassReader;
  }
  inline bool has_writer_successor() {
    uint8_t s = assorted::atomic_load_acquire<uint8_t>(&self_.components_.successor_class_);
    return s == kSuccessorClassWriter;
  }

  uint16_t make_blocked_with_reader_successor_state() {
    // Only using the class bit, which doesn't change, so no need to use atomic ops.
    uint8_t state = self_.components_.state_ | kStateBlockedFlag;
    return (uint16_t)state << 8 | kSuccessorClassReader;
  }
  uint16_t make_blocked_with_no_successor_state() {
    uint8_t state = self_.components_.state_ | kStateBlockedFlag;
    return (uint16_t)state << 8 | kSuccessorClassNone;
  }
};

struct McsRwExtendedBlock {
  static const uint32_t kPredFlagWaiting         = 0U;
  static const uint32_t kPredFlagGranted         = 1U;
  static const uint32_t kPredFlagReader          = 0U;
  static const uint32_t kPredFlagWriter          = 1U << 31;
  static const uint32_t kPredFlagClassMask       = 1U << 31;

  static const uint32_t kSuccFlagWaiting         = 0U;
  static const uint32_t kSuccFlagLeaving         = 1U;
  static const uint32_t kSuccFlagDirectGranted   = 2U;
  static const uint32_t kSuccFlagLeavingGranted  = 3U;
  static const uint32_t kSuccFlagMask            = 3U;

  static const uint32_t kSuccFlagBusy            = 4U;

  static const uint32_t kSuccFlagSuccessorClassMask = 3U << 30;
  static const uint32_t kSuccFlagSuccessorReader    = 3U << 30;
  static const uint32_t kSuccFlagSuccessorNone      = 0U;
  static const uint32_t kSuccFlagSuccessorWriter    = 1U << 30;

  static const uint32_t kSuccIdSuccessorLeaving  = 0xFFFFFFFFU;
  static const uint32_t kSuccIdNoSuccessor       = 0xFFFFFFFEU;

  static const uint32_t kPredIdAcquired          = 0xFFFFFFFFU;

#ifndef NDEBUG
  static const uint64_t kSuccReleased            = 0xFFFFFFFFFFFFFFFFU;
#endif

  /* Special timeouts for instant return and unconditional acquire */
  static const int32_t kTimeoutNever = 0xFFFFFFFFU;
  static const int32_t kTimeoutZero  = 0U;

  union Field {
    uint64_t data_;
    struct Components {
      uint32_t flags_;
      uint32_t id_;
    } components_;
  };

  Field pred_;
  Field next_;

#ifndef NDEBUG
  inline void mark_released() {
    ASSERT_ND(pred_flag_is_granted());
    ASSERT_ND(next_flag_is_granted());
    set_next(kSuccReleased);
  }

  inline bool is_released() {
    return get_next() == kSuccReleased;
  }
#endif

  inline uint32_t read_pred_flags() {
    return assorted::atomic_load_acquire<uint32_t>(&pred_.components_.flags_);
  }
  inline uint32_t read_next_flags() {
    return assorted::atomic_load_acquire<uint32_t>(&next_.components_.flags_);
  }
  inline bool is_writer() { return read_pred_flags() & kPredFlagWriter; }
  inline bool is_reader() { return !is_writer(); }
  inline bool pred_flag_is_waiting() {
    return !pred_flag_is_granted();
  }
  inline bool pred_flag_is_granted() {
    return read_pred_flags() & kPredFlagGranted;
  }
  inline bool next_flag_is_direct_granted() {
    uint32_t f = (read_next_flags() & kSuccFlagMask);
    return f == kSuccFlagDirectGranted;
  }
  inline bool next_flag_is_leaving_granted() {
    uint32_t f = (read_next_flags() & kSuccFlagMask);
    return f == kSuccFlagLeavingGranted;
  }
  inline bool next_flag_is_granted() {
    uint32_t f = (read_next_flags() & kSuccFlagMask);
    return f == kSuccFlagLeavingGranted || f == kSuccFlagDirectGranted;
  }
  inline bool next_flag_is_leaving() {
    return (read_next_flags() & kSuccFlagMask) == kSuccFlagLeaving;
  }
  inline bool next_flag_is_waiting() {
    return (read_next_flags() & kSuccFlagMask) == kSuccFlagWaiting;
  }
  inline void set_next_flag_writer_successor() {
    ASSERT_ND(!next_flag_has_successor());
    assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
      &next_.components_.flags_, kSuccFlagSuccessorWriter);
  }
  inline void set_next_flag_reader_successor() {
    ASSERT_ND(!next_flag_has_successor());
    assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
      &next_.components_.flags_, kSuccFlagSuccessorReader);
  }
  inline void set_pred_flag_granted() {
    ASSERT_ND(pred_flag_is_waiting());
    assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
      &pred_.components_.flags_, kPredFlagGranted);
  }
  inline void set_next_flag_granted() {
    ASSERT_ND(pred_flag_is_granted());
    ASSERT_ND(next_flag_is_waiting() | next_flag_is_leaving());
    if (next_flag_is_waiting()) {
      assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
        &next_.components_.flags_, kSuccFlagDirectGranted);
    } else {
      ASSERT_ND(next_flag_is_leaving());
      assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
        &next_.components_.flags_, kSuccFlagLeavingGranted);
    }
  }
  inline void set_next_flag_busy_granted() {
    ASSERT_ND(pred_flag_is_granted());
    ASSERT_ND(next_flag_is_waiting() | next_flag_is_leaving());
    if (next_flag_is_waiting()) {
      assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
        &next_.components_.flags_, kSuccFlagDirectGranted | kSuccFlagBusy);
    } else {
      ASSERT_ND(next_flag_is_leaving());
      assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
        &next_.components_.flags_, kSuccFlagLeavingGranted | kSuccFlagBusy);
    }
  }
  inline void set_next_flag_leaving() {
    assorted::raw_atomic_exchange<uint16_t>(
      reinterpret_cast<uint16_t*>(&next_.components_.flags_),
      static_cast<uint16_t>(kSuccFlagLeaving));
  }
  inline void set_next_flag_no_successor() {
    assorted::raw_atomic_fetch_and_bitwise_and<uint32_t>(
      &next_.components_.flags_, ~kSuccFlagSuccessorClassMask);
  }
  inline void set_next(uint64_t next) {
    assorted::atomic_store_release<uint64_t>(&next_.data_, next);
  }
  inline bool cas_next_weak(uint64_t expected, uint64_t desired) {
    return assorted::raw_atomic_compare_exchange_weak<uint64_t>(
     &next_.data_, &expected, desired);
  }
  inline bool cas_next_strong(uint64_t expected, uint64_t desired) {
    return assorted::raw_atomic_compare_exchange_strong<uint64_t>(
     &next_.data_, &expected, desired);
  }
  inline void set_flags_granted() {
    set_pred_flag_granted();
    set_next_flag_granted();
  }
  inline bool next_flag_has_successor() {
    return read_next_flags() & kSuccFlagSuccessorClassMask;
  }
  inline bool next_flag_has_reader_successor() {
    return (read_next_flags() & kSuccFlagSuccessorClassMask) == kSuccFlagSuccessorReader;
  }
  inline bool next_flag_has_writer_successor() {
    return (read_next_flags() & kSuccFlagSuccessorClassMask) == kSuccFlagSuccessorWriter;
  }
  inline bool next_flag_is_busy() {
    return (read_next_flags() & kSuccFlagBusy) == kSuccFlagBusy;
  }
  inline void set_next_flag_busy() {
    ASSERT_ND(!next_flag_is_busy());
    assorted::raw_atomic_fetch_and_bitwise_or<uint32_t>(
      &next_.components_.flags_, kSuccFlagBusy);
  }
  inline void unset_next_flag_busy() {
    ASSERT_ND(next_flag_is_busy());
    assorted::raw_atomic_fetch_and_bitwise_and<uint32_t>(
      &next_.components_.flags_, ~kSuccFlagBusy);
  }
  inline uint32_t cas_val_next_flag_strong(uint32_t expected, uint32_t desired) {
    assorted::raw_atomic_compare_exchange_strong<uint32_t>(
      &next_.components_.flags_, &expected, desired);
    return expected;
  }
  inline uint32_t cas_val_next_flag_weak(uint32_t expected, uint32_t desired) {
    assorted::raw_atomic_compare_exchange_weak<uint32_t>(
      &next_.components_.flags_, &expected, desired);
    return expected;
  }
  inline uint64_t cas_val_next_strong(uint64_t expected, uint64_t desired) {
    assorted::raw_atomic_compare_exchange_strong<uint64_t>(
      &next_.data_, &expected, desired);
    return expected;
  }
  inline uint64_t cas_val_next_weak(uint64_t expected, uint64_t desired) {
    assorted::raw_atomic_compare_exchange_weak<uint64_t>(
      &next_.data_, &expected, desired);
    return expected;
  }
  inline uint32_t xchg_next_id(uint32_t id) {
    return assorted::raw_atomic_exchange<uint32_t>(&next_.components_.id_, id);
  }
  inline bool cas_next_id_strong(uint32_t expected, uint32_t desired) {
    return assorted::raw_atomic_compare_exchange_strong<uint32_t>(
      &next_.components_.id_, &expected, desired);
  }
  inline bool cas_next_id_weak(uint32_t expected, uint32_t desired) {
    return assorted::raw_atomic_compare_exchange_weak<uint32_t>(
      &next_.components_.id_, &expected, desired);
  }
  inline bool cas_pred_id_weak(uint32_t expected, uint32_t desired) {
    return assorted::raw_atomic_compare_exchange_weak<uint32_t>(
      &pred_.components_.id_, &expected, desired);
  }
  inline bool cas_pred_id_strong(uint32_t expected, uint32_t desired) {
    return assorted::raw_atomic_compare_exchange_strong<uint32_t>(
      &pred_.components_.id_, &expected, desired);
  }
  inline uint32_t cas_val_pred_id_weak(uint32_t expected, uint32_t desired) {
    assorted::raw_atomic_compare_exchange_weak<uint32_t>(
      &pred_.components_.id_, &expected, desired);
    return expected;
  }
  inline uint32_t make_next_flag_waiting_with_no_successor() { return kSuccFlagWaiting; }
  inline uint32_t make_next_flag_waiting_with_reader_successor() {
    return kSuccFlagWaiting | kSuccFlagSuccessorReader;
  }
  inline uint32_t get_pred_id() {
    return assorted::atomic_load_acquire<uint32_t>(&pred_.components_.id_);
  }
  inline uint32_t get_next_id() {
    return assorted::atomic_load_acquire<uint32_t>(&next_.components_.id_);
  }
  inline uint64_t get_next() {
    return assorted::atomic_load_acquire<uint64_t>(&next_.data_);
  }
  inline void set_pred_id(uint32_t id) {
    assorted::atomic_store_release<uint32_t>(&pred_.components_.id_, id);
  }
  inline void set_next_id(uint32_t id) {
    assorted::atomic_store_release<uint32_t>(&next_.components_.id_, id);
  }
  inline uint32_t xchg_pred_id(uint32_t id) {
    return assorted::raw_atomic_exchange<uint32_t>(&pred_.components_.id_, id);
  }
  inline void init_reader() {
    pred_.components_.flags_ = kPredFlagReader;
    next_.components_.flags_ = 0;
    pred_.components_.id_ = next_.components_.id_ = 0;
    ASSERT_ND(pred_flag_is_waiting());
    ASSERT_ND(next_flag_is_waiting());
    ASSERT_ND(is_reader());
  }
  inline void init_writer() {
    pred_.components_.flags_ = kPredFlagWriter;
    next_.components_.flags_ = 0;
    pred_.components_.id_ = next_.components_.id_ = 0;
    ASSERT_ND(pred_flag_is_waiting());
    ASSERT_ND(next_flag_is_waiting());
    ASSERT_ND(is_writer());
  }
  bool timeout_granted(int32_t timeout);
};

struct McsRwLock {
  static const thread::ThreadId kNextWriterNone = 0xFFFFU;

  McsRwLock() { reset(); }

  McsRwLock(const McsRwLock& other) CXX11_FUNC_DELETE;
  McsRwLock& operator=(const McsRwLock& other) CXX11_FUNC_DELETE;

  inline void reset() {
    tail_ = nreaders_ = 0;
    set_next_writer(kNextWriterNone);
    assorted::memory_fence_release();
  }
  inline void increment_nreaders() {
    assorted::raw_atomic_fetch_add<uint16_t>(&nreaders_, 1);
  }
  inline uint16_t decrement_nreaders() {
    return assorted::raw_atomic_fetch_add<uint16_t>(&nreaders_, -1);
  }
  inline uint16_t nreaders() {
    return assorted::atomic_load_acquire<uint16_t>(&nreaders_);
  }
  inline McsBlockIndex get_tail_waiter_block() const { return tail_ & 0xFFFFU; }
  inline thread::ThreadId get_tail_waiter() const { return tail_ >> 16U; }
  inline bool has_next_writer() const {
    return assorted::atomic_load_acquire<thread::ThreadId>(&next_writer_) != kNextWriterNone;
  }
  inline void set_next_writer(thread::ThreadId thread_id) {
    xchg_next_writer(thread_id);  // sub-word access...
  }
  inline thread::ThreadId get_next_writer() {
    return assorted::atomic_load_acquire<thread::ThreadId>(&next_writer_);
  }
  inline thread::ThreadId xchg_next_writer(thread::ThreadId id) {
    return assorted::raw_atomic_exchange<thread::ThreadId>(&next_writer_, id);
  }
  bool cas_next_writer_weak(thread::ThreadId expected, thread::ThreadId desired) {
    return assorted::raw_atomic_compare_exchange_weak<thread::ThreadId>(
      &next_writer_, &expected, desired);
  }
  bool cas_next_writer_strong(thread::ThreadId expected, thread::ThreadId desired) {
    return assorted::raw_atomic_compare_exchange_strong<thread::ThreadId>(
      &next_writer_, &expected, desired);
  }
  inline uint32_t xchg_tail(uint32_t new_tail) {
    return assorted::raw_atomic_exchange<uint32_t>(&tail_, new_tail);
  }
  inline bool cas_tail_strong(uint32_t expected, uint32_t desired) {
    return assorted::raw_atomic_compare_exchange_strong<uint32_t>(&tail_, &expected, desired);
  }
  inline bool cas_tail_weak(uint32_t expected, uint32_t desired) {
    return assorted::raw_atomic_compare_exchange_weak<uint32_t>(&tail_, &expected, desired);
  }
  static inline uint32_t to_tail_int(
    thread::ThreadId tail_waiter,
    McsBlockIndex tail_waiter_block) {
    ASSERT_ND(tail_waiter_block <= 0xFFFFU);
    return static_cast<uint32_t>(tail_waiter) << 16 | (tail_waiter_block & 0xFFFFU);
  }
  inline uint32_t get_tail_int() {
    return assorted::atomic_load_acquire<uint32_t>(&tail_);
  }
  bool is_locked() const {
    return (tail_ & 0xFFFFU) != 0 || nreaders_ > 0;
  }

  uint32_t tail_;                 // +4 => 4
  /* Note that threadId starts from 0, so we use 0xFFFF as the "invalid"
   * marker, unless we make the lock even larger than 8 bytes. This essentially
   * limits the largest allowed number of cores we support to 256 sockets x 256
   * cores per socket - 1.
   */
  thread::ThreadId next_writer_;  // +2 => 6
  uint16_t nreaders_;             // +2 => 8

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

struct McsRwAsyncMapping {
  UniversalLockId lock_id_;
  McsBlockIndex block_index_;
  char padding_[16 - sizeof(McsBlockIndex) - sizeof(UniversalLockId)];

  McsRwAsyncMapping(UniversalLockId lock_id, McsBlockIndex block) :
    lock_id_(lock_id), block_index_(block) {}
  McsRwAsyncMapping() : lock_id_(kNullUniversalLockId), block_index_(0) {}
};

const uint64_t kXctIdDeletedBit     = 1ULL << 63;
const uint64_t kXctIdMovedBit       = 1ULL << 62;
const uint64_t kXctIdBeingWrittenBit = 1ULL << 61;
const uint64_t kXctIdNextLayerBit    = 1ULL << 60;
const uint64_t kXctIdMaskSerializer = 0x0FFFFFFFFFFFFFFFULL;
const uint64_t kXctIdMaskEpoch      = 0x0FFFFFFF00000000ULL;
const uint64_t kXctIdMaskOrdinal    = 0x00000000FFFFFFFFULL;

const uint64_t kMaxXctOrdinal       = (1ULL << 24) - 1U;

struct XctId {
  XctId() : data_(0) {}

  void set(Epoch::EpochInteger epoch_int, uint32_t ordinal) {
    ASSERT_ND(epoch_int < Epoch::kEpochIntOverflow);
    ASSERT_ND(ordinal <= kMaxXctOrdinal);
    data_ = static_cast<uint64_t>(epoch_int) << 32 | ordinal;
  }

  Epoch   get_epoch() const ALWAYS_INLINE { return Epoch(get_epoch_int()); }
  void    set_epoch(Epoch epoch) ALWAYS_INLINE { set_epoch_int(epoch.value()); }
  Epoch::EpochInteger get_epoch_int() const ALWAYS_INLINE {
    return (data_ & kXctIdMaskEpoch) >> 32;
  }
  void    set_epoch_int(Epoch::EpochInteger epoch_int) ALWAYS_INLINE {
    ASSERT_ND(epoch_int < Epoch::kEpochIntOverflow);
    data_ = (data_ & ~kXctIdMaskEpoch) | (static_cast<uint64_t>(epoch_int) << 32);
  }
  bool    is_valid() const ALWAYS_INLINE { return get_epoch_int() != Epoch::kEpochInvalid; }


  uint32_t  get_ordinal() const ALWAYS_INLINE {
    ASSERT_ND(static_cast<uint32_t>(data_) <= kMaxXctOrdinal);
    return static_cast<uint32_t>(data_);
  }
  void      set_ordinal(uint32_t ordinal) ALWAYS_INLINE {
    ASSERT_ND(ordinal <= kMaxXctOrdinal);
    data_ = (data_ & (~kXctIdMaskOrdinal)) | ordinal;
  }
  void      increment_ordinal() ALWAYS_INLINE {
    uint32_t ordinal = get_ordinal();
    set_ordinal(ordinal + 1U);
  }
  int       compare_epoch_and_orginal(const XctId& other) const ALWAYS_INLINE {
    // compare epoch
    if (get_epoch_int() != other.get_epoch_int()) {
      Epoch this_epoch = get_epoch();
      Epoch other_epoch = other.get_epoch();
      ASSERT_ND(this_epoch.is_valid());
      ASSERT_ND(other_epoch.is_valid());
      if (this_epoch < other_epoch) {
        return -1;
      } else {
        ASSERT_ND(this_epoch > other_epoch);
        return 1;
      }
    }

    // if the epoch is the same, compare in_epoch_ordinal_.
    ASSERT_ND(get_epoch() == other.get_epoch());
    if (get_ordinal() < other.get_ordinal()) {
      return -1;
    } else if (get_ordinal() > other.get_ordinal()) {
      return 1;
    } else {
      return 0;
    }
  }

  void    set_being_written() ALWAYS_INLINE { data_ |= kXctIdBeingWrittenBit; }
  void    set_write_complete() ALWAYS_INLINE { data_ &= (~kXctIdBeingWrittenBit); }
  XctId   spin_while_being_written() const ALWAYS_INLINE;
  void    set_deleted() ALWAYS_INLINE { data_ |= kXctIdDeletedBit; }
  void    set_notdeleted() ALWAYS_INLINE { data_ &= (~kXctIdDeletedBit); }
  void    set_moved() ALWAYS_INLINE { data_ |= kXctIdMovedBit; }
  void    set_next_layer() ALWAYS_INLINE {
    // Delete-bit has no meaning for a next-layer record. To avoid confusion, turn it off.
    data_ = (data_ & (~kXctIdDeletedBit)) | kXctIdNextLayerBit;
  }
  // note, we should not need this method because becoming a next-layer-pointer is permanent.
  // we never revert it, which simplifies a concurrency control.
  // void    set_not_next_layer() ALWAYS_INLINE { data_ &= (~kXctIdNextLayerBit); }

  bool    is_being_written() const ALWAYS_INLINE {
    return (assorted::atomic_load_acquire<uint64_t>(&data_) & kXctIdBeingWrittenBit) != 0; }
  bool    is_deleted() const ALWAYS_INLINE { return (data_ & kXctIdDeletedBit) != 0; }
  bool    is_moved() const ALWAYS_INLINE { return (data_ & kXctIdMovedBit) != 0; }
  bool    is_next_layer() const ALWAYS_INLINE { return (data_ & kXctIdNextLayerBit) != 0; }
  bool    needs_track_moved() const ALWAYS_INLINE {
    return (data_ & (kXctIdMovedBit | kXctIdNextLayerBit)) != 0;
  }


  bool operator==(const XctId &other) const ALWAYS_INLINE { return data_ == other.data_; }
  bool operator!=(const XctId &other) const ALWAYS_INLINE { return data_ != other.data_; }

  void store_max(const XctId& other) ALWAYS_INLINE {
    if (!other.is_valid()) {
      return;
    }

    if (before(other)) {
      operator=(other);
    }
  }

  bool before(const XctId &other) const ALWAYS_INLINE {
    ASSERT_ND(other.is_valid());
    // compare epoch, then ordinal
    if (get_epoch_int() != other.get_epoch_int()) {
      return get_epoch().before(other.get_epoch());
    }
    return get_ordinal() < other.get_ordinal();
  }

  void clear_status_bits() { data_ &= kXctIdMaskSerializer; }

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

  uint64_t            data_;
};

struct LockableXctId {
  McsWwLock       lock_;
  XctId         xct_id_;

  McsWwLock* get_key_lock() ALWAYS_INLINE { return &lock_; }
  bool is_keylocked() const ALWAYS_INLINE { return lock_.is_locked(); }
  bool is_deleted() const ALWAYS_INLINE { return xct_id_.is_deleted(); }
  bool is_moved() const ALWAYS_INLINE { return xct_id_.is_moved(); }
  bool is_next_layer() const ALWAYS_INLINE { return xct_id_.is_next_layer(); }
  bool needs_track_moved() const ALWAYS_INLINE { return xct_id_.needs_track_moved(); }
  bool is_being_written() const ALWAYS_INLINE { return xct_id_.is_being_written(); }

  void    reset() ALWAYS_INLINE {
    lock_.reset();
    xct_id_.data_ = 0;
  }
  friend std::ostream& operator<<(std::ostream& o, const LockableXctId& v);
};

struct RwLockableXctId {
  McsRwLock       lock_;

  XctId         xct_id_;

  McsRwLock* get_key_lock() ALWAYS_INLINE { return &lock_; }
  bool is_keylocked() const ALWAYS_INLINE { return lock_.is_locked(); }
  bool is_deleted() const ALWAYS_INLINE { return xct_id_.is_deleted(); }
  bool is_moved() const ALWAYS_INLINE { return xct_id_.is_moved(); }
  bool is_next_layer() const ALWAYS_INLINE { return xct_id_.is_next_layer(); }
  bool needs_track_moved() const ALWAYS_INLINE { return xct_id_.needs_track_moved(); }
  bool is_being_written() const ALWAYS_INLINE { return xct_id_.is_being_written(); }
  bool is_hot(thread::Thread* context) const;
  void hotter(thread::Thread* context) const;

  void    reset() ALWAYS_INLINE {
    lock_.reset();
    xct_id_.data_ = 0;
  }
  friend std::ostream& operator<<(std::ostream& o, const RwLockableXctId& v);
};

class McsOwnerlessLockScope {
 public:
  McsOwnerlessLockScope();
  McsOwnerlessLockScope(
    McsWwLock* lock,
    bool acquire_now = true,
    bool non_racy_acquire = false);
  ~McsOwnerlessLockScope();

  bool is_valid() const { return lock_; }
  bool is_locked_by_me() const { return locked_by_me_; }

  void acquire(bool non_racy_acquire);
  void release();

 private:
  McsWwLock*        lock_;
  bool            locked_by_me_;
};

struct TrackMovedRecordResult {
  TrackMovedRecordResult()
    : new_owner_address_(CXX11_NULLPTR), new_payload_address_(CXX11_NULLPTR) {}
  TrackMovedRecordResult(RwLockableXctId* new_owner_address, char* new_payload_address)
    : new_owner_address_(new_owner_address), new_payload_address_(new_payload_address) {}

  RwLockableXctId* new_owner_address_;
  char* new_payload_address_;
};

inline XctId XctId::spin_while_being_written() const {
  uint64_t copied_data = assorted::atomic_load_acquire<uint64_t>(&data_);
  if (UNLIKELY(copied_data & kXctIdBeingWrittenBit)) {
    while (copied_data & kXctIdBeingWrittenBit) {
      copied_data = assorted::atomic_load_acquire<uint64_t>(&data_);
    }
  }
  XctId ret;
  ret.data_ = copied_data;
  return ret;
}


UniversalLockId to_universal_lock_id(
  const memory::GlobalVolatilePageResolver& resolver,
  uintptr_t lock_ptr);

const uint64_t kLockPageSize = 1 << 12;
inline UniversalLockId to_universal_lock_id(
  uint64_t numa_node,
  uint64_t local_page_index,
  uintptr_t lock_ptr) {
  const uint64_t in_page_offset = lock_ptr % kLockPageSize;
  return (numa_node << 48) | (local_page_index * kLockPageSize + in_page_offset);
}

// just shorthands.
inline UniversalLockId xct_id_to_universal_lock_id(
  const memory::GlobalVolatilePageResolver& resolver,
  RwLockableXctId* lock) {
  return to_universal_lock_id(resolver, reinterpret_cast<uintptr_t>(lock));
}
inline UniversalLockId rw_lock_to_universal_lock_id(
  const memory::GlobalVolatilePageResolver& resolver,
  McsRwLock* lock) {
  return to_universal_lock_id(resolver, reinterpret_cast<uintptr_t>(lock));
}

RwLockableXctId* from_universal_lock_id(
  const memory::GlobalVolatilePageResolver& resolver,
  const UniversalLockId universal_lock_id);

// sizeof(XctId) must be 64 bits.
STATIC_SIZE_CHECK(sizeof(XctId), sizeof(uint64_t))
STATIC_SIZE_CHECK(sizeof(McsWwLock), 8)
STATIC_SIZE_CHECK(sizeof(LockableXctId), 16)

}  // namespace xct
}  // namespace foedus
#endif  // FOEDUS_XCT_XCT_ID_HPP_