sysxct_impl.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_SYSXCT_IMPL_HPP_
#define FOEDUS_XCT_SYSXCT_IMPL_HPP_

#include <algorithm>
#include <iosfwd>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/error_code.hpp"
#include "foedus/fwd.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/record.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/xct/fwd.hpp"
#include "foedus/xct/sysxct_functor.hpp"
#include "foedus/xct/xct_id.hpp"
#include "foedus/xct/xct_mcs_impl.hpp"

namespace foedus {
namespace xct {

// TASK(Hideaki) this comparator probably has a common use. move to somewhere.
inline UniversalLockId to_universal_lock_id(storage::VolatilePagePointer page_id, uintptr_t addr) {
  return to_universal_lock_id(page_id.get_numa_node(), page_id.get_offset(), addr);
}
struct PageComparator {
  bool operator()(storage::Page* p1, storage::Page* p2) const {
    auto id1 = to_universal_lock_id(p1->get_volatile_page_id(), reinterpret_cast<uintptr_t>(p1));
    auto id2 = to_universal_lock_id(p2->get_volatile_page_id(), reinterpret_cast<uintptr_t>(p2));
    return id1 < id2;
  }
};

struct SysxctLockEntry {
  UniversalLockId universal_lock_id_;

  uintptr_t       lock_;

  // No "preferred"/"taken" mode because system transactions always take W-locks, record or page.

  McsBlockIndex   mcs_block_;

  bool            page_lock_;

  bool            used_in_this_run_;

  void clear() {
    universal_lock_id_ = 0;
    lock_ = kNullUniversalLockId;
    mcs_block_ = 0;
    page_lock_ = false;
    used_in_this_run_ = false;
  }

  void set(UniversalLockId lock_id, uintptr_t lock, bool page_lock) {
    universal_lock_id_ = lock_id;
    lock_ = lock;
    mcs_block_ = 0;
    page_lock_ = page_lock;
    used_in_this_run_ = true;
  }

  RwLockableXctId* get_as_record_lock() const {
    ASSERT_ND(!page_lock_);
    return reinterpret_cast<RwLockableXctId*>(lock_);
  }
  storage::Page* get_as_page_lock() const {
    ASSERT_ND(page_lock_);
    return reinterpret_cast<storage::Page*>(lock_);
  }

  bool is_locked() const { return mcs_block_ != 0; }

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

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

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

class SysxctLockList {
 public:
  friend struct TestIsTryRequired;  // For unit test. A bit of whitebox test involved.

  enum Constants {
    kMaxSysxctLocks = 1 << 10,
  };

  SysxctLockList() { init(); }
  ~SysxctLockList() {}

  // No copy
  SysxctLockList(const SysxctLockList& other) = delete;
  SysxctLockList& operator=(const SysxctLockList& other) = delete;

  void init() {
    clear_no_assert(kNullUniversalLockId);
  }

  void clear_entries(UniversalLockId enclosing_max_lock_id) {
    ASSERT_ND(last_locked_entry_ == kLockListPositionInvalid);
    assert_sorted();
    clear_no_assert(enclosing_max_lock_id);
  }

  void compress_entries(UniversalLockId enclosing_max_lock_id);

  LockListPosition lower_bound(UniversalLockId lock) const;

  template<typename MCS_ADAPTOR>
  ErrorCode request_record_lock(
    MCS_ADAPTOR mcs_adaptor,
    storage::VolatilePagePointer page_id,
    RwLockableXctId* lock) {
    uintptr_t lock_addr = reinterpret_cast<uintptr_t>(lock);
    return request_lock_general(mcs_adaptor, page_id, lock_addr, false);
  }

  template<typename MCS_ADAPTOR>
  ErrorCode batch_request_record_locks(
    MCS_ADAPTOR mcs_adaptor,
    storage::VolatilePagePointer page_id,
    uint32_t lock_count,
    RwLockableXctId** locks) {
    uintptr_t* lock_addr = reinterpret_cast<uintptr_t*>(locks);
    return batch_request_locks_general(mcs_adaptor, page_id, lock_count, lock_addr, false);
  }

  template<typename MCS_ADAPTOR>
  ErrorCode request_page_lock(MCS_ADAPTOR mcs_adaptor, storage::Page* page) {
    uintptr_t lock_addr = reinterpret_cast<uintptr_t>(page);
    return request_lock_general(mcs_adaptor, page->get_volatile_page_id(), lock_addr, true);
  }

  template<typename MCS_ADAPTOR>
  ErrorCode batch_request_page_locks(
    MCS_ADAPTOR mcs_adaptor,
    uint32_t lock_count,
    storage::Page** pages) {
    std::sort(pages, pages + lock_count, PageComparator());
    for (uint32_t i = 0; i < lock_count; ++i) {
      CHECK_ERROR_CODE(request_page_lock(mcs_adaptor, pages[i]));
    }
    return kErrorCodeOk;
  }

  template<typename MCS_ADAPTOR>
  void release_all_locks(MCS_ADAPTOR mcs_adaptor);

  LockListPosition get_or_add_entry(
    storage::VolatilePagePointer page_id,
    uintptr_t lock_addr,
    bool page_lock);

  LockListPosition batch_get_or_add_entries(
    storage::VolatilePagePointer page_id,
    uint32_t lock_count,
    uintptr_t* lock_addr,
    bool page_lock);

  const SysxctLockEntry* get_array() const { return array_; }
  SysxctLockEntry* get_array() { return array_; }
  SysxctLockEntry* get_entry(LockListPosition pos) {
    ASSERT_ND(is_valid_entry(pos));
    return array_ + pos;
  }
  const SysxctLockEntry* get_entry(LockListPosition pos) const {
    ASSERT_ND(is_valid_entry(pos));
    return array_ + pos;
  }
  uint32_t  get_capacity() const { return kMaxSysxctLocks; }
  bool      is_full() const { return last_active_entry_ == kMaxSysxctLocks; }
  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 SysxctLockList& v);
  void assert_sorted() const ALWAYS_INLINE;
  void assert_sorted_impl() const;

  SysxctLockEntry* begin() { return array_ + 1U; }
  SysxctLockEntry* end() { return array_ + 1U + last_active_entry_; }
  const SysxctLockEntry* cbegin() const { return array_ + 1U; }
  const SysxctLockEntry* cend() const { return array_ + 1U + last_active_entry_; }
  LockListPosition to_pos(const SysxctLockEntry* entry) const {
    ASSERT_ND(entry > array_);
    LockListPosition pos = entry - array_;
    ASSERT_ND(pos <= get_last_active_entry());
    return pos;
  }

  LockListPosition get_last_locked_entry() const { return last_locked_entry_; }
  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
  }

  UniversalLockId get_enclosing_max_lock_id() const { return enclosing_max_lock_id_; }
  bool is_try_mode_required(LockListPosition pos) const {
    const SysxctLockEntry* entry = get_entry(pos);
    // If the enclosing thread took a lock after this, we must be careful.
    if (entry->universal_lock_id_ <= enclosing_max_lock_id_) {
      return true;
    }

    // If this list contains an already-taken lock after this, we must be careful.
    if (pos <= last_locked_entry_) {
      return true;
    }

    return false;
  }

 private:
  LockListPosition last_active_entry_;
  LockListPosition last_locked_entry_;

  UniversalLockId enclosing_max_lock_id_;

  SysxctLockEntry array_[kMaxSysxctLocks];

  void clear_no_assert(UniversalLockId enclosing_max_lock_id) {
    last_locked_entry_ = kLockListPositionInvalid;
    last_active_entry_ = kLockListPositionInvalid;
    enclosing_max_lock_id_ = enclosing_max_lock_id;
    array_[kLockListPositionInvalid].clear();
  }

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

  template<typename MCS_ADAPTOR>
  ErrorCode try_or_acquire_multiple_locks(
    MCS_ADAPTOR mcs_adaptor,
    LockListPosition from_pos,
    UniversalLockId last_lock_id,
    bool mandatory_locks);

  template<typename MCS_ADAPTOR>
  ErrorCode request_lock_general(
    MCS_ADAPTOR mcs_adaptor,
    storage::VolatilePagePointer page_id,
    uintptr_t lock,
    bool page_lock) {
    // Just reuse the batched version. Shouldn't have any significant overhead.
    // Using the same codepath is better also for testing/robustness. Not just I'm lazy!
    return batch_request_locks_general(mcs_adaptor, page_id, 1U, &lock, page_lock);
  }

  template<typename MCS_ADAPTOR>
  ErrorCode batch_request_locks_general(
    MCS_ADAPTOR mcs_adaptor,
    storage::VolatilePagePointer page_id,
    uint32_t lock_count,
    uintptr_t* locks,
    bool page_lock);
};

struct SysxctWorkspace {
  bool running_sysxct_;       // +1 -> 1
  char pad1_[7];              // +7 -> 8

  thread::Thread*  thread_;   // +8 -> 16
  char pad2_[48];             // +48 -> 64

  SysxctLockList lock_list_;

  void init(thread::Thread* enclosing_thread) {
    ASSERT_ND(enclosing_thread);
    running_sysxct_ = false;
    thread_ = enclosing_thread;
    lock_list_.init();
  }

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

template<typename MCS_ADAPTOR, typename ENCLOSURE_RELEASE_ALL_LOCKS_FUNCTOR>
inline ErrorCode run_nested_sysxct_impl(
  SysxctFunctor* functor,
  MCS_ADAPTOR mcs_adaptor,
  uint32_t max_retries,
  SysxctWorkspace* workspace,
  UniversalLockId enclosing_max_lock_id,
  ENCLOSURE_RELEASE_ALL_LOCKS_FUNCTOR enclosure_release_all_locks_functor) {
  if (UNLIKELY(workspace->running_sysxct_)) {
    // so far each thread can convey only one system transaction.
    // this is 1) to reuse SysxctWorkspace, and 2) to avoid deadlocks.
    return kErrorCodeXctTwoSysXcts;
  }

  workspace->running_sysxct_ = true;
  uint32_t cur_retries = 0;
  auto* lock_list = &(workspace->lock_list_);
  lock_list->clear_entries(enclosing_max_lock_id);
  while (true) {
    ErrorCode ret = functor->run(workspace);
    // In any case, release all locks.
    lock_list->release_all_locks(mcs_adaptor);

    // The first run of the system transaction takes locks only in "try" mode
    // to avoid deadlocks. They thus have higher chance to get aborted.
    if (UNLIKELY(ret == kErrorCodeXctRaceAbort || ret == kErrorCodeXctLockAbort)) {
      // consider retry
      if (cur_retries < max_retries) {
        // When we retry, we release all locks in _outer_ transaction to eliminate the
        // chance of deadlocks. Then the next run of the system transaction can
        // take locks unconditionally, waiting for an arbitrary period.
        if (enclosing_max_lock_id != kNullUniversalLockId) {
          enclosure_release_all_locks_functor();
          lock_list->compress_entries(kNullUniversalLockId);
          enclosing_max_lock_id = kNullUniversalLockId;
          // Now the next run, helped by inherited entries, should be in unconditional mode
        }
        ++cur_retries;
        continue;
      }
    }

    workspace->running_sysxct_ = false;
    return ret;
  }
}


// TASK(Hideaki) standalone sysxct... Needed in a few places.


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

template<typename MCS_ADAPTOR>
inline ErrorCode SysxctLockList::try_or_acquire_multiple_locks(
  MCS_ADAPTOR mcs_adaptor,
  LockListPosition from_pos,
  UniversalLockId last_lock_id,
  bool mandatory_locks) {
  ASSERT_ND(from_pos != kLockListPositionInvalid);
  assert_sorted();
  for (LockListPosition pos = from_pos;; ++pos) {
    if (pos > last_active_entry_) {
      break;
    } else if (array_[pos].universal_lock_id_ > last_lock_id) {
      // The only use of this func is batch_request_locks_general, where last_lock_id surely exists
      ASSERT_ND(array_[pos - 1].universal_lock_id_ == last_lock_id);
      break;
    }

    SysxctLockEntry* lock_entry = get_entry(pos);
    if (mandatory_locks) {
      lock_entry->used_in_this_run_ = true;
    }
    if (lock_entry->is_locked()) {
      continue;
    }

    const bool needs_try = is_try_mode_required(pos);
    if (lock_entry->page_lock_) {
      McsWwImpl< MCS_ADAPTOR > impl(mcs_adaptor);
      storage::Page* page = lock_entry->get_as_page_lock();
      McsWwLock* lock_addr = &page->get_header().page_version_.lock_;
      if (needs_try) {
        lock_entry->mcs_block_ = impl.acquire_try(lock_addr);
      } else {
        lock_entry->mcs_block_ = impl.acquire_unconditional(lock_addr);
        ASSERT_ND(lock_entry->is_locked());
      }
    } else {
      McsImpl< MCS_ADAPTOR, typename MCS_ADAPTOR::ThisRwBlock > impl(mcs_adaptor);
      McsRwLock* lock_addr = &lock_entry->get_as_record_lock()->lock_;
      if (needs_try) {
        lock_entry->mcs_block_ = impl.acquire_try_rw_writer(lock_addr);
      } else {
        lock_entry->mcs_block_ = impl.acquire_unconditional_rw_writer(lock_addr);
        ASSERT_ND(lock_entry->is_locked());
      }
    }
    if (lock_entry->mcs_block_) {
      if (last_locked_entry_ < pos) {
        last_locked_entry_ = pos;
      }
      assert_last_locked_entry();
    } else {
      return kErrorCodeXctRaceAbort;
    }
  }

  return kErrorCodeOk;
}

template<typename MCS_ADAPTOR>
inline ErrorCode SysxctLockList::batch_request_locks_general(
  MCS_ADAPTOR mcs_adaptor,
  storage::VolatilePagePointer page_id,
  uint32_t lock_count,
  uintptr_t* locks,
  bool page_lock) {
  if (lock_count == 0) {
    return kErrorCodeOk;
  }

  // First, put corresponding entries to the list. either existing or just confirming.
  LockListPosition from_pos = batch_get_or_add_entries(page_id, lock_count, locks, page_lock);
  if (from_pos > 1U) {
    // To reduce the chance of deadlocks, we take locks before these locks.
    // But, if we have already taken any lock after that, it won't help.
    UniversalLockId first_lock_id = to_universal_lock_id(page_id, locks[0]);
    if (last_locked_entry_ < from_pos && enclosing_max_lock_id_ < first_lock_id) {
      // These are non-mandatory locks, so we don't turn on the "used_in_this_run_" flags
      CHECK_ERROR_CODE(try_or_acquire_multiple_locks(mcs_adaptor, 1U, first_lock_id, false));
    }
  }


  // Upto which lock ID are we locking? Note, this param is lock-id, not pos.
  UniversalLockId last_lock_id = to_universal_lock_id(page_id, locks[lock_count - 1]);

  // Then, lock them all! These are "mandatory" locks.
  return try_or_acquire_multiple_locks(mcs_adaptor, from_pos, last_lock_id, true);
}

template<typename MCS_ADAPTOR>
inline void SysxctLockList::release_all_locks(MCS_ADAPTOR mcs_adaptor) {
  for (auto* entry = begin(); entry != end(); ++entry) {
    if (!entry->is_locked()) {
      continue;
    }
    if (entry->page_lock_) {
      McsWwImpl< MCS_ADAPTOR > impl(mcs_adaptor);
      storage::Page* page = entry->get_as_page_lock();
      McsWwLock* lock_addr = &page->get_header().page_version_.lock_;
      impl.release(lock_addr, entry->mcs_block_);
      entry->mcs_block_ = 0;
    } else {
      McsImpl< MCS_ADAPTOR, typename MCS_ADAPTOR::ThisRwBlock > impl(mcs_adaptor);
      McsRwLock* lock_addr = &entry->get_as_record_lock()->lock_;
      impl.release_rw_writer(lock_addr, entry->mcs_block_);
      entry->mcs_block_ = 0;
    }
  }
  last_locked_entry_ = kLockListPositionInvalid;
}

}  // namespace xct
}  // namespace foedus
#endif  // FOEDUS_XCT_SYSXCT_IMPL_HPP_