retrospective_lock_list.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_RETROSPECTIVE_LOCK_LIST_HPP_
#define FOEDUS_XCT_RETROSPECTIVE_LOCK_LIST_HPP_

#include <stdint.h>

#include <algorithm>
#include <iosfwd>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/error_code.hpp"
#include "foedus/memory/page_resolver.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/xct/fwd.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace xct {

struct LockEntry {
  UniversalLockId universal_lock_id_;

  RwLockableXctId* lock_;

  LockMode preferred_mode_;

  LockMode taken_mode_;

  McsBlockIndex mcs_block_;

  // want to have these.. but maitaining them is nasty after sort. let's revisit later
  // ReadXctAccess* read_set_;
  // WriteXctAccess* write_set_;

  void set(
    UniversalLockId id,
    RwLockableXctId* lock,
    LockMode preferred_mode,
    LockMode taken_mode) {
    universal_lock_id_ = id;
    lock_ = lock;
    preferred_mode_ = preferred_mode;
    taken_mode_ = taken_mode;
    mcs_block_ = 0;
  }

  bool is_locked() const { return taken_mode_ != kNoLock; }
  bool is_enough() const { return taken_mode_ >= preferred_mode_; }

  bool operator<(const LockEntry& rhs) const {
    return universal_lock_id_ < rhs.universal_lock_id_;
  }

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

  struct LessThan {
    bool operator()(UniversalLockId lhs, const LockEntry& rhs) const {
      return lhs < rhs.universal_lock_id_;
    }
    bool operator()(const LockEntry& lhs, UniversalLockId rhs) const {
      return lhs.universal_lock_id_ < rhs;
    }
  };
};

class RetrospectiveLockList {
 public:
  RetrospectiveLockList();
  ~RetrospectiveLockList();

  void init(
    LockEntry* array,
    uint32_t capacity,
    const memory::GlobalVolatilePageResolver& resolver);

  void uninit();
  void clear_entries();

  LockListPosition binary_search(UniversalLockId lock) const;

  LockListPosition lower_bound(UniversalLockId lock) const;

  void construct(thread::Thread* context, uint32_t read_lock_threshold);

  const LockEntry* get_array() const { return array_; }
  LockEntry* get_entry(LockListPosition pos) {
    ASSERT_ND(is_valid_entry(pos));
    return array_ + pos;
  }
  const LockEntry* get_entry(LockListPosition pos) const {
    ASSERT_ND(is_valid_entry(pos));
    return array_ + pos;
  }
  uint32_t get_capacity() const { return capacity_; }
  LockListPosition get_last_active_entry() const { return last_active_entry_; }
  bool is_valid_entry(LockListPosition pos) const {
    return pos != kLockListPositionInvalid && pos <= last_active_entry_;
  }
  bool is_empty() const { return last_active_entry_ == kLockListPositionInvalid; }

  friend std::ostream& operator<<(std::ostream& o, const RetrospectiveLockList& v);
  void assert_sorted() const ALWAYS_INLINE;
  void assert_sorted_impl() const;

  LockEntry* begin() { return array_ + 1U; }
  LockEntry* end() { return array_ + 1U + last_active_entry_; }
  const LockEntry* cbegin() const { return array_ + 1U; }
  const LockEntry* cend() const { return array_ + 1U + last_active_entry_; }
  const memory::GlobalVolatilePageResolver& get_volatile_page_resolver() const {
    return volatile_page_resolver_;
  }

 private:
  LockEntry* array_;
  memory::GlobalVolatilePageResolver volatile_page_resolver_;
  uint32_t capacity_;
  LockListPosition last_active_entry_;

  LockListPosition issue_new_position() {
    ++last_active_entry_;
    ASSERT_ND(last_active_entry_ < capacity_);
    return last_active_entry_;
  }
};

class CurrentLockList {
 public:
  CurrentLockList();
  ~CurrentLockList();

  void init(
    LockEntry* array,
    uint32_t capacity,
    const memory::GlobalVolatilePageResolver& resolver);
  void uninit();
  void clear_entries();

  LockListPosition binary_search(UniversalLockId lock) const;

  LockListPosition get_or_add_entry(
    UniversalLockId lock_id,
    RwLockableXctId* lock,
    LockMode preferred_mode);

  LockListPosition lower_bound(UniversalLockId lock) const;

  const LockEntry* get_array() const { return array_; }
  LockEntry* get_array() { return array_; }
  LockEntry* get_entry(LockListPosition pos) {
    ASSERT_ND(is_valid_entry(pos));
    return array_ + pos;
  }
  const LockEntry* get_entry(LockListPosition pos) const {
    ASSERT_ND(is_valid_entry(pos));
    return array_ + pos;
  }
  uint32_t get_capacity() const { return capacity_; }
  LockListPosition get_last_active_entry() const { return last_active_entry_; }
  bool is_valid_entry(LockListPosition pos) const {
    return pos != kLockListPositionInvalid && pos <= last_active_entry_;
  }
  bool is_empty() const { return last_active_entry_ == kLockListPositionInvalid; }

  friend std::ostream& operator<<(std::ostream& o, const CurrentLockList& v);
  void assert_sorted() const ALWAYS_INLINE;
  void assert_sorted_impl() const;

  LockEntry* begin() { return array_ + 1U; }
  LockEntry* end() { return array_ + 1U + last_active_entry_; }
  const LockEntry* cbegin() const { return array_ + 1U; }
  const LockEntry* cend() const { return array_ + 1U + last_active_entry_; }
  const memory::GlobalVolatilePageResolver& get_volatile_page_resolver() const {
    return volatile_page_resolver_;
  }

  LockListPosition get_last_locked_entry() const { return last_locked_entry_; }
  UniversalLockId get_max_locked_id() const {
    if (last_locked_entry_ == kLockListPositionInvalid) {
      return kNullUniversalLockId;
    } else {
      return get_entry(last_locked_entry_)->universal_lock_id_;
    }
  }
  LockListPosition calculate_last_locked_entry() const {
    return calculate_last_locked_entry_from(last_active_entry_);
  }
  LockListPosition calculate_last_locked_entry_from(LockListPosition from) const {
    for (LockListPosition pos = from; pos > kLockListPositionInvalid; --pos) {
      if (array_[pos].is_locked()) {
        return pos;
      }
    }
    return kLockListPositionInvalid;
  }
  void assert_last_locked_entry() const {
#ifndef NDEBUG
    LockListPosition correct = calculate_last_locked_entry();
    ASSERT_ND(correct == last_locked_entry_);
#endif  // NDEBUG
  }

  void batch_insert_write_placeholders(const WriteXctAccess* write_set, uint32_t write_set_size);

  void prepopulate_for_retrospective_lock_list(const RetrospectiveLockList& rll);


  template<typename MCS_RW_IMPL>
  ErrorCode try_or_acquire_single_lock(LockListPosition pos, MCS_RW_IMPL* mcs_rw_impl);

  template<typename MCS_RW_IMPL>
  ErrorCode try_or_acquire_multiple_locks(LockListPosition upto_pos, MCS_RW_IMPL* mcs_rw_impl);

  template<typename MCS_RW_IMPL>
  void try_async_single_lock(LockListPosition pos, MCS_RW_IMPL* mcs_rw_impl);
  template<typename MCS_RW_IMPL>
  bool retry_async_single_lock(LockListPosition pos, MCS_RW_IMPL* mcs_rw_impl);
  template<typename MCS_RW_IMPL>
  void cancel_async_single_lock(LockListPosition pos, MCS_RW_IMPL* mcs_rw_impl);

  template<typename MCS_RW_IMPL>
  void try_async_multiple_locks(LockListPosition upto_pos, MCS_RW_IMPL* mcs_rw_impl);

  template<typename MCS_RW_IMPL>
  void        release_all_locks(MCS_RW_IMPL* mcs_rw_impl) {
    release_all_after(kNullUniversalLockId, mcs_rw_impl);
  }
  template<typename MCS_RW_IMPL>
  void        release_all_after(UniversalLockId address, MCS_RW_IMPL* mcs_rw_impl);
  template<typename MCS_RW_IMPL>
  void        release_all_at_and_after(UniversalLockId address, MCS_RW_IMPL* mcs_rw_impl);
  template<typename MCS_RW_IMPL>
  void        giveup_all_after(UniversalLockId address, MCS_RW_IMPL* mcs_rw_impl);
  template<typename MCS_RW_IMPL>
  void        giveup_all_at_and_after(UniversalLockId address, MCS_RW_IMPL* mcs_rw_impl);

 private:
  LockEntry* array_;
  memory::GlobalVolatilePageResolver volatile_page_resolver_;
  uint32_t capacity_;
  LockListPosition last_active_entry_;
  LockListPosition last_locked_entry_;

  LockListPosition issue_new_position() {
    assert_last_locked_entry();
    ++last_active_entry_;
    ASSERT_ND(last_active_entry_ < capacity_);
    return last_active_entry_;
  }

  void release_all_after_debuglog(
    uint32_t released_read_locks,
    uint32_t released_write_locks,
    uint32_t already_released_locks,
    uint32_t canceled_async_read_locks,
    uint32_t canceled_async_write_locks) const;

  void giveup_all_after_debuglog(
    uint32_t givenup_read_locks,
    uint32_t givenup_write_locks,
    uint32_t givenup_upgrades,
    uint32_t already_enough_locks,
    uint32_t canceled_async_read_locks,
    uint32_t canceled_async_write_locks) const;
};

inline void RetrospectiveLockList::assert_sorted() const {
  // In release mode, this code must be completely erased by compiler
#ifndef  NDEBUG
  assert_sorted_impl();
#endif  // NDEBUG
}

inline void CurrentLockList::assert_sorted() const {
#ifndef  NDEBUG
  assert_sorted_impl();
#endif  // NDEBUG
}

template<typename MCS_RW_IMPL>
inline ErrorCode CurrentLockList::try_or_acquire_single_lock(
  LockListPosition pos,
  MCS_RW_IMPL* mcs_rw_impl) {
  LockEntry* lock_entry = get_entry(pos);
  if (lock_entry->is_enough()) {
    return kErrorCodeOk;
  }
  ASSERT_ND(lock_entry->taken_mode_ != kWriteLock);

  McsRwLock* lock_addr = lock_entry->lock_->get_key_lock();
  if (lock_entry->taken_mode_ != kNoLock) {
    ASSERT_ND(lock_entry->preferred_mode_ == kWriteLock);
    ASSERT_ND(lock_entry->taken_mode_ == kReadLock);
    ASSERT_ND(lock_entry->mcs_block_);
    // This is reader->writer upgrade.
    // We simply release read-lock first then take write-lock in this case.
    // In traditional 2PL, such an unlock-then-lock violates serializability,
    // but we guarantee serializability by read-verification anyways.
    // We can release any lock anytime.. great flexibility!
    mcs_rw_impl->release_rw_reader(lock_addr, lock_entry->mcs_block_);
    lock_entry->taken_mode_ = kNoLock;
    last_locked_entry_ = calculate_last_locked_entry_from(pos - 1U);
    assert_last_locked_entry();
  } else {
    // This method is for unconditional acquire and try, not aync/retry.
    // If we have a queue node already, something was misused.
    ASSERT_ND(lock_entry->mcs_block_ == 0);
  }

  // Now we need to take the lock. Are we in canonical mode?
  if (last_locked_entry_ == kLockListPositionInvalid || last_locked_entry_ < pos) {
    // yay, we are in canonical mode. we can unconditionally get the lock
    ASSERT_ND(lock_entry->taken_mode_ == kNoLock);  // not a lock upgrade, either
    if (lock_entry->preferred_mode_ == kWriteLock) {
      lock_entry->mcs_block_ = mcs_rw_impl->acquire_unconditional_rw_writer(lock_addr);
    } else {
      ASSERT_ND(lock_entry->preferred_mode_ == kReadLock);
      lock_entry->mcs_block_ = mcs_rw_impl->acquire_unconditional_rw_reader(lock_addr);
    }
    lock_entry->taken_mode_ = lock_entry->preferred_mode_;
    last_locked_entry_ = pos;
  } else {
    // hmm, we violated canonical mode. has a risk of deadlock.
    // Let's just try acquire the lock and immediately give up if it fails.
    // The RLL will take care of the next run.
    // TODO(Hideaki) release some of the lock we have taken to restore canonical mode.
    // We haven't imlpemented this optimization yet.
    ASSERT_ND(lock_entry->mcs_block_ == 0);
    ASSERT_ND(lock_entry->taken_mode_ == kNoLock);
    if (lock_entry->preferred_mode_ == kWriteLock) {
      lock_entry->mcs_block_ = mcs_rw_impl->acquire_try_rw_writer(lock_addr);
    } else {
      ASSERT_ND(lock_entry->preferred_mode_ == kReadLock);
      lock_entry->mcs_block_ = mcs_rw_impl->acquire_try_rw_reader(lock_addr);
    }
    if (lock_entry->mcs_block_ == 0) {
      return kErrorCodeXctLockAbort;
    }
    lock_entry->taken_mode_ = lock_entry->preferred_mode_;
  }
  ASSERT_ND(lock_entry->mcs_block_);
  ASSERT_ND(lock_entry->is_locked());
  assert_last_locked_entry();
  return kErrorCodeOk;
}

template<typename MCS_RW_IMPL>
inline ErrorCode CurrentLockList::try_or_acquire_multiple_locks(
  LockListPosition upto_pos,
  MCS_RW_IMPL* mcs_rw_impl) {
  ASSERT_ND(upto_pos != kLockListPositionInvalid);
  ASSERT_ND(upto_pos <= last_active_entry_);
  // Especially in this case, we probably should release locks after upto_pos first.
  for (LockListPosition pos = 1U; pos <= upto_pos; ++pos) {
    CHECK_ERROR_CODE(try_or_acquire_single_lock(pos, mcs_rw_impl));
  }
  return kErrorCodeOk;
}

template<typename MCS_RW_IMPL>
inline void CurrentLockList::try_async_single_lock(
  LockListPosition pos,
  MCS_RW_IMPL* mcs_rw_impl) {
  LockEntry* lock_entry = get_entry(pos);
  if (lock_entry->is_enough()) {
    return;
  }
  ASSERT_ND(lock_entry->taken_mode_ != kWriteLock);

  McsRwLock* lock_addr = lock_entry->lock_->get_key_lock();
  if (lock_entry->taken_mode_ != kNoLock) {
    ASSERT_ND(lock_entry->preferred_mode_ == kWriteLock);
    ASSERT_ND(lock_entry->taken_mode_ == kReadLock);
    ASSERT_ND(lock_entry->mcs_block_);
    // This is reader->writer upgrade.
    // We simply release read-lock first then take write-lock in this case.
    // In traditional 2PL, such an unlock-then-lock violates serializability,
    // but we guarantee serializability by read-verification anyways.
    // We can release any lock anytime.. great flexibility!
    mcs_rw_impl->release_rw_reader(lock_addr, lock_entry->mcs_block_);
    lock_entry->taken_mode_ = kNoLock;
    last_locked_entry_ = calculate_last_locked_entry_from(pos - 1U);
    assert_last_locked_entry();
  } else {
    // This function is for pushing the queue node in the extended rwlock.
    // Doomed if we already have a queue node.
    ASSERT_ND(lock_entry->mcs_block_ == 0);
  }

  // Don't really care canonical order here, just send out the request.
  AcquireAsyncRet async_ret;
  if (lock_entry->preferred_mode_ == kWriteLock) {
    async_ret = mcs_rw_impl->acquire_async_rw_writer(lock_addr);
  } else {
    ASSERT_ND(lock_entry->preferred_mode_ == kReadLock);
    async_ret = mcs_rw_impl->acquire_async_rw_reader(lock_addr);
  }
  ASSERT_ND(async_ret.block_index_);
  lock_entry->mcs_block_ = async_ret.block_index_;
  if (async_ret.acquired_) {
    lock_entry->taken_mode_ = lock_entry->preferred_mode_;
    ASSERT_ND(lock_entry->is_enough());
  }
  ASSERT_ND(lock_entry->mcs_block_);
}

template<typename MCS_RW_IMPL>
inline bool CurrentLockList::retry_async_single_lock(
  LockListPosition pos,
  MCS_RW_IMPL* mcs_rw_impl) {
  LockEntry* lock_entry = get_entry(pos);
  // Must be not taken yet, and must have pushed a qnode to the lock queue
  ASSERT_ND(!lock_entry->is_enough());
  ASSERT_ND(lock_entry->taken_mode_ == kNoLock);
  ASSERT_ND(lock_entry->mcs_block_);
  ASSERT_ND(!lock_entry->is_locked());

  McsRwLock* lock_addr = lock_entry->lock_->get_key_lock();
  bool acquired = false;
  if (lock_entry->preferred_mode_ == kWriteLock) {
    acquired = mcs_rw_impl->retry_async_rw_writer(lock_addr, lock_entry->mcs_block_);
  } else {
    ASSERT_ND(lock_entry->preferred_mode_ == kReadLock);
    acquired = mcs_rw_impl->retry_async_rw_reader(lock_addr, lock_entry->mcs_block_);
  }
  if (acquired) {
    lock_entry->taken_mode_ = lock_entry->preferred_mode_;
    ASSERT_ND(lock_entry->is_locked());
    last_locked_entry_ = std::max(last_locked_entry_, pos);
    assert_last_locked_entry();
  }
  return acquired;
}

template<typename MCS_RW_IMPL>
inline void CurrentLockList::cancel_async_single_lock(
  LockListPosition pos,
  MCS_RW_IMPL* mcs_rw_impl) {
  LockEntry* lock_entry = get_entry(pos);
  ASSERT_ND(!lock_entry->is_enough());
  ASSERT_ND(lock_entry->taken_mode_ == kNoLock);
  ASSERT_ND(lock_entry->mcs_block_);
  McsRwLock* lock_addr = lock_entry->lock_->get_key_lock();
  if (lock_entry->preferred_mode_ == kReadLock) {
    mcs_rw_impl->cancel_async_rw_reader(lock_addr, lock_entry->mcs_block_);
  } else {
    ASSERT_ND(lock_entry->preferred_mode_ == kReadLock);
    mcs_rw_impl->cancel_async_rw_writer(lock_addr, lock_entry->mcs_block_);
  }
  lock_entry->mcs_block_ = 0;
}

template<typename MCS_RW_IMPL>
inline void CurrentLockList::try_async_multiple_locks(
  LockListPosition upto_pos,
  MCS_RW_IMPL* mcs_rw_impl) {
  ASSERT_ND(upto_pos != kLockListPositionInvalid);
  ASSERT_ND(upto_pos <= last_active_entry_);
  for (LockListPosition pos = 1U; pos <= upto_pos; ++pos) {
    try_async_single_lock(pos, mcs_rw_impl);
  }
}

template<typename MCS_RW_IMPL>
inline void CurrentLockList::release_all_after(UniversalLockId address, MCS_RW_IMPL* mcs_rw_impl) {
  // Only this and below logics are implemented here because this needs to know about CLL.
  // This is not quite about the lock algorithm itself. It's something on higher level.
  assert_sorted();
  uint32_t released_read_locks = 0;
  uint32_t released_write_locks = 0;
  uint32_t already_released_locks = 0;
  uint32_t canceled_async_read_locks = 0;
  uint32_t canceled_async_write_locks = 0;

  LockListPosition new_last_locked_entry = kLockListPositionInvalid;
  for (LockEntry* entry = begin(); entry != end(); ++entry) {
    if (entry->universal_lock_id_ <= address) {
      if (entry->is_locked()) {
        new_last_locked_entry = entry - array_;
      }
      continue;
    }
    if (entry->is_locked()) {
      if (entry->taken_mode_ == kReadLock) {
        mcs_rw_impl->release_rw_reader(entry->lock_->get_key_lock(), entry->mcs_block_);
        ++released_read_locks;
      } else {
        ASSERT_ND(entry->taken_mode_ == kWriteLock);
        mcs_rw_impl->release_rw_writer(entry->lock_->get_key_lock(), entry->mcs_block_);
        ++released_write_locks;
      }
      entry->mcs_block_ = 0;
      entry->taken_mode_ = kNoLock;
    } else if (entry->mcs_block_) {
      // Not locked yet, but we have mcs_block_ set, this means we tried it in
      // async mode, then still waiting or at least haven't confirmed that we acquired it.
      // Cancel these "retrieable" locks to which we already pushed our qnode.
      ASSERT_ND(entry->taken_mode_ == kNoLock);
      if (entry->preferred_mode_ == kReadLock) {
        mcs_rw_impl->cancel_async_rw_reader(entry->lock_->get_key_lock(), entry->mcs_block_);
        ++canceled_async_read_locks;
      } else {
        ASSERT_ND(entry->preferred_mode_ == kWriteLock);
        mcs_rw_impl->cancel_async_rw_writer(entry->lock_->get_key_lock(), entry->mcs_block_);
        ++canceled_async_write_locks;
      }
      entry->mcs_block_ = 0;
    } else {
      ASSERT_ND(entry->taken_mode_ == kNoLock);
      ++already_released_locks;
    }
  }

  last_locked_entry_ = new_last_locked_entry;
  assert_last_locked_entry();

#ifndef NDEBUG
  release_all_after_debuglog(
    released_read_locks,
    released_write_locks,
    already_released_locks,
    canceled_async_read_locks,
    canceled_async_write_locks);
#endif  // NDEBUG
}
template<typename MCS_RW_IMPL>
inline void CurrentLockList::release_all_at_and_after(
  UniversalLockId address,
  MCS_RW_IMPL* mcs_rw_impl) {
  if (address == kNullUniversalLockId) {
    release_all_after<MCS_RW_IMPL>(kNullUniversalLockId, mcs_rw_impl);
  } else {
    release_all_after<MCS_RW_IMPL>(address - 1U, mcs_rw_impl);
  }
}

template<typename MCS_RW_IMPL>
inline void CurrentLockList::giveup_all_after(UniversalLockId address, MCS_RW_IMPL* mcs_rw_impl) {
  assert_sorted();
  uint32_t givenup_read_locks = 0;
  uint32_t givenup_write_locks = 0;
  uint32_t givenup_upgrades = 0;
  uint32_t already_enough_locks = 0;
  uint32_t canceled_async_read_locks = 0;
  uint32_t canceled_async_write_locks = 0;

  for (LockEntry* entry = begin(); entry != end(); ++entry) {
    if (entry->universal_lock_id_ <= address) {
      continue;
    }
    if (entry->preferred_mode_ == kNoLock) {
      continue;
    }
    if (entry->is_enough()) {
      ++already_enough_locks;
      continue;
    }

    if (entry->is_locked()) {
      ASSERT_ND(entry->taken_mode_ == kReadLock);
      ASSERT_ND(entry->preferred_mode_ == kWriteLock);
      ++givenup_upgrades;
      entry->preferred_mode_ = entry->taken_mode_;
    } else if (entry->mcs_block_) {
      if (entry->preferred_mode_ == kReadLock) {
        mcs_rw_impl->cancel_async_rw_reader(entry->lock_->get_key_lock(), entry->mcs_block_);
        ++canceled_async_read_locks;
      } else {
        ASSERT_ND(entry->preferred_mode_ == kWriteLock);
        mcs_rw_impl->cancel_async_rw_writer(entry->lock_->get_key_lock(), entry->mcs_block_);
        ++canceled_async_write_locks;
      }
      entry->mcs_block_ = 0;
      entry->preferred_mode_ = kNoLock;
    } else {
      ASSERT_ND(entry->taken_mode_ == kNoLock);
      if (entry->preferred_mode_ == kReadLock) {
        ++givenup_read_locks;
      } else {
        ASSERT_ND(entry->preferred_mode_ == kWriteLock);
        ++givenup_write_locks;
      }
      entry->preferred_mode_ = kNoLock;
    }
  }

#ifndef NDEBUG
  giveup_all_after_debuglog(
    givenup_read_locks,
    givenup_write_locks,
    givenup_upgrades,
    already_enough_locks,
    canceled_async_read_locks,
    canceled_async_write_locks);
#endif  // NDEBUG
}
template<typename MCS_RW_IMPL>
inline void CurrentLockList::giveup_all_at_and_after(
  UniversalLockId address,
  MCS_RW_IMPL* mcs_rw_impl) {
  if (address == kNullUniversalLockId) {
    giveup_all_after<MCS_RW_IMPL>(kNullUniversalLockId, mcs_rw_impl);
  } else {
    giveup_all_after<MCS_RW_IMPL>(address - 1U, mcs_rw_impl);
  }
}

template<typename LOCK_LIST, typename LOCK_ENTRY>
inline LockListPosition lock_lower_bound(
  const LOCK_LIST& list,
  UniversalLockId lock) {
  LockListPosition last_active_entry = list.get_last_active_entry();
  if (last_active_entry == kLockListPositionInvalid) {
    return kLockListPositionInvalid + 1U;
  }
  // Check the easy cases first. This will be an wasted cost if it's not, but still cheap.
  const LOCK_ENTRY* array = list.get_array();
  // For example, [dummy, 3, 5, 7] (last_active_entry=3).
  // id=7: 3, larger: 4, smaller: need to check more
  if (array[last_active_entry].universal_lock_id_ == lock) {
    return last_active_entry;
  } else if (array[last_active_entry].universal_lock_id_ < lock) {
    return last_active_entry + 1U;
  }

  LockListPosition pos
    = std::lower_bound(
        array + 1U,
        array + last_active_entry + 1U,
        lock,
        typename LOCK_ENTRY::LessThan())
      - array;
  // in the above example, id=6: 3, id=4,5: 2, smaller: 1
  ASSERT_ND(pos != kLockListPositionInvalid);
  ASSERT_ND(pos <= last_active_entry);  // otherwise we went into the branch above
  ASSERT_ND(array[pos].universal_lock_id_ >= lock);
  ASSERT_ND(pos == 1U || array[pos - 1U].universal_lock_id_ < lock);
  return pos;
}

template<typename LOCK_LIST, typename LOCK_ENTRY>
inline LockListPosition lock_binary_search(
  const LOCK_LIST& list,
  UniversalLockId lock) {
  LockListPosition last_active_entry = list.get_last_active_entry();
  LockListPosition pos = lock_lower_bound<LOCK_LIST, LOCK_ENTRY>(list, lock);
  if (pos != kLockListPositionInvalid && pos <= last_active_entry) {
    const LOCK_ENTRY* array = list.get_array();
    if (array[pos].universal_lock_id_ == lock) {
      return pos;
    }
  }
  return kLockListPositionInvalid;
}


}  // namespace xct
}  // namespace foedus
#endif  // FOEDUS_XCT_RETROSPECTIVE_LOCK_LIST_HPP_