foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock > Class Template Reference

The Simple MCS-RW lock. More...

Detailed Description

template<typename ADAPTOR>
class foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >

The Simple MCS-RW lock.

These need partial specialization on RW_BLOCK=McsRwSimpleBlock. However, C++ standard doesn't allow function partial specialization. We thus make it a class partial specialization as below. Stupid? Totally agree!

Definition at line 348 of file xct_mcs_impl.cpp.

Public Member Functions
McsBlockIndex	acquire_try_rw_writer (McsRwLock *lock)
McsBlockIndex	acquire_try_rw_reader (McsRwLock *lock)
McsBlockIndex	acquire_unconditional_rw_reader (McsRwLock *mcs_rw_lock)
void	release_rw_reader (McsRwLock *mcs_rw_lock, McsBlockIndex block_index)
McsBlockIndex	acquire_unconditional_rw_writer (McsRwLock *mcs_rw_lock)
void	release_rw_writer (McsRwLock *mcs_rw_lock, McsBlockIndex block_index)
AcquireAsyncRet	acquire_async_rw_reader (McsRwLock *lock)
AcquireAsyncRet	acquire_async_rw_writer (McsRwLock *lock)
bool	retry_async_rw_reader (McsRwLock *lock, McsBlockIndex block_index)
bool	retry_async_rw_writer (McsRwLock *lock, McsBlockIndex block_index)
void	cancel_async_rw_reader (McsRwLock *, McsBlockIndex)
void	cancel_async_rw_writer (McsRwLock *, McsBlockIndex)

Static Public Member Functions
static bool	does_support_try_rw_reader ()

Member Function Documentation

template<typename ADAPTOR >

AcquireAsyncRet foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::acquire_async_rw_reader ( McsRwLock *  lock )

inline

Definition at line 544 of file xct_mcs_impl.cpp.

References foedus::xct::McsImpl< ADAPTOR, RW_BLOCK >::retry_async_rw_reader().

                                                           {
    // In simple version, no distinction between try/async/retry. Same logic.
    McsBlockIndex block_index = adaptor_.issue_new_block();
    bool success = retry_async_rw_reader(lock, block_index);
    return {success, block_index};
  }

Here is the call graph for this function:

template<typename ADAPTOR >

AcquireAsyncRet foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::acquire_async_rw_writer ( McsRwLock *  lock )

inline

Definition at line 550 of file xct_mcs_impl.cpp.

References foedus::xct::McsImpl< ADAPTOR, RW_BLOCK >::retry_async_rw_writer().

                                                           {
    McsBlockIndex block_index = adaptor_.issue_new_block();
    bool success = retry_async_rw_writer(lock, block_index);
    return {success, block_index};
  }

Here is the call graph for this function:

template<typename ADAPTOR >

McsBlockIndex foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::acquire_try_rw_reader ( McsRwLock *  lock )

inline

Definition at line 363 of file xct_mcs_impl.cpp.

References ASSERT_ND, and foedus::xct::McsImpl< ADAPTOR, RW_BLOCK >::retry_async_rw_reader().

                                                       {
    McsBlockIndex block_index = adaptor_.issue_new_block();
    bool success = retry_async_rw_reader(lock, block_index);
    if (success) {
#ifndef NDEBUG
      auto* my_block = adaptor_.get_rw_my_block(block_index);
      ASSERT_ND(my_block->is_finalized());
      ASSERT_ND(my_block->is_granted());
#endif  // NDEBUG
      return block_index;
    } else {
      // The block is never observed. reuse
      adaptor_.cancel_new_block(block_index);
      return 0;
    }
  }

Here is the call graph for this function:

template<typename ADAPTOR >

McsBlockIndex foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::acquire_try_rw_writer ( McsRwLock *  lock )

inline

Definition at line 350 of file xct_mcs_impl.cpp.

References foedus::xct::McsImpl< ADAPTOR, RW_BLOCK >::retry_async_rw_writer().

                                                       {
    McsBlockIndex block_index = adaptor_.issue_new_block();
    bool success = retry_async_rw_writer(lock, block_index);
    if (success) {
      return block_index;
    } else {
      // The block is never observed. reuse
      adaptor_.cancel_new_block(block_index);
      return 0;
    }
  }

Here is the call graph for this function:

template<typename ADAPTOR >

McsBlockIndex foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::acquire_unconditional_rw_reader ( McsRwLock *  mcs_rw_lock )

inline

Definition at line 380 of file xct_mcs_impl.cpp.

References ASSERT_ND, foedus::xct::McsRwLock::increment_nreaders(), foedus::xct::spin_until(), foedus::xct::McsRwLock::tail_, and foedus::xct::McsRwLock::to_tail_int().

                                                                        {
    ASSERT_ND(adaptor_.get_cur_block() < 0xFFFFU);
    const thread::ThreadId id = adaptor_.get_my_id();
    const McsBlockIndex block_index = adaptor_.issue_new_block();
    ASSERT_ND(block_index > 0);
    ASSERT_ND(sizeof(McsRwSimpleBlock) == sizeof(McsWwBlock));
    auto* my_block = adaptor_.get_rw_my_block(block_index);

    // So I'm a reader
    my_block->init_reader();
    ASSERT_ND(my_block->is_blocked() && my_block->is_reader());
    ASSERT_ND(!my_block->has_successor());
    ASSERT_ND(my_block->successor_block_index_ == 0);

    // Now ready to XCHG
    uint32_t tail_desired = McsRwLock::to_tail_int(id, block_index);
    uint32_t* tail_address = &(mcs_rw_lock->tail_);
    uint32_t pred_tail_int = assorted::raw_atomic_exchange<uint32_t>(tail_address, tail_desired);

    if (pred_tail_int == 0) {
      mcs_rw_lock->increment_nreaders();
      my_block->unblock();  // reader successors will know they don't need to wait
    } else {
      // See if the predecessor is a reader; if so, if it already acquired the lock.
      auto* pred_block = adaptor_.dereference_rw_tail_block(pred_tail_int);
      uint16_t* pred_state_address = &pred_block->self_.data_;
      uint16_t pred_state_expected = pred_block->make_blocked_with_no_successor_state();
      uint16_t pred_state_desired = pred_block->make_blocked_with_reader_successor_state();
      if (!pred_block->is_reader() || assorted::raw_atomic_compare_exchange_strong<uint16_t>(
        pred_state_address,
        &pred_state_expected,
        pred_state_desired)) {
        // Predecessor is a writer or a waiting reader. The successor class field and the
        // blocked state in pred_block are separated, so we can blindly set_successor().
        pred_block->set_successor_next_only(id, block_index);
        spin_until([my_block]{ return my_block->is_granted(); });
      } else {
        // Join the active, reader predecessor
        ASSERT_ND(!pred_block->is_blocked());
        mcs_rw_lock->increment_nreaders();
        pred_block->set_successor_next_only(id, block_index);
        my_block->unblock();
      }
    }
    finalize_acquire_reader_simple(mcs_rw_lock, my_block);
    ASSERT_ND(my_block->is_finalized());
    return block_index;
  }

Here is the call graph for this function:

template<typename ADAPTOR >

McsBlockIndex foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::acquire_unconditional_rw_writer ( McsRwLock *  mcs_rw_lock )

inline

Definition at line 476 of file xct_mcs_impl.cpp.

References ASSERT_ND, foedus::xct::McsRwLock::get_next_writer(), foedus::xct::McsRwLock::kNextWriterNone, foedus::xct::McsRwLock::next_writer_, foedus::xct::McsRwLock::nreaders(), foedus::xct::spin_until(), foedus::xct::McsRwLock::tail_, and foedus::xct::McsRwLock::to_tail_int().

                            {
    const thread::ThreadId id = adaptor_.get_my_id();
    const McsBlockIndex block_index = adaptor_.issue_new_block();
    ASSERT_ND(adaptor_.get_cur_block() < 0xFFFFU);
    ASSERT_ND(block_index > 0);
    ASSERT_ND(sizeof(McsRwSimpleBlock) == sizeof(McsWwBlock));
    auto* my_block = adaptor_.get_rw_my_block(block_index);

    my_block->init_writer();
    ASSERT_ND(my_block->is_blocked() && !my_block->is_reader());
    ASSERT_ND(!my_block->has_successor());
    ASSERT_ND(my_block->successor_block_index_ == 0);

    // Now ready to XCHG
    uint32_t tail_desired = McsRwLock::to_tail_int(id, block_index);
    uint32_t* tail_address = &(mcs_rw_lock->tail_);
    uint32_t pred_tail_int = assorted::raw_atomic_exchange<uint32_t>(tail_address, tail_desired);
    ASSERT_ND(pred_tail_int != tail_desired);
    thread::ThreadId old_next_writer = 0xFFFFU;
    if (pred_tail_int == 0) {
      ASSERT_ND(mcs_rw_lock->get_next_writer() == McsRwLock::kNextWriterNone);
      assorted::raw_atomic_exchange<thread::ThreadId>(&mcs_rw_lock->next_writer_, id);
      if (mcs_rw_lock->nreaders() == 0) {
        old_next_writer = assorted::raw_atomic_exchange<thread::ThreadId>(
          &mcs_rw_lock->next_writer_,
          McsRwLock::kNextWriterNone);
        if (old_next_writer == id) {
          my_block->unblock();
          return block_index;
        }
      }
    } else {
      auto* pred_block = adaptor_.dereference_rw_tail_block(pred_tail_int);
      pred_block->set_successor_class_writer();
      pred_block->set_successor_next_only(id, block_index);
    }
    spin_until([my_block]{ return my_block->is_granted(); });
    return block_index;
  }

Here is the call graph for this function:

template<typename ADAPTOR >

void foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::cancel_async_rw_reader ( McsRwLock *  , McsBlockIndex    )

inline

Definition at line 643 of file xct_mcs_impl.cpp.

                                                             {
    // In simple version, we don't actually have any mechanism to retry.
    // so, we don't have to do any cancel, either. No-op.
  }

template<typename ADAPTOR >

void foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::cancel_async_rw_writer ( McsRwLock *  , McsBlockIndex    )

inline

Definition at line 647 of file xct_mcs_impl.cpp.

                                                             {
  }

template<typename ADAPTOR >

static bool foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::does_support_try_rw_reader ( )

inlinestatic

Definition at line 362 of file xct_mcs_impl.cpp.

{ return false; }

template<typename ADAPTOR >

void foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::release_rw_reader ( McsRwLock *  mcs_rw_lock, McsBlockIndex  block_index  )

inline

Definition at line 429 of file xct_mcs_impl.cpp.

References ASSERT_ND, foedus::xct::McsRwLock::decrement_nreaders(), foedus::xct::McsRwSimpleBlock::is_blocked(), foedus::xct::McsRwSimpleBlock::is_reader(), foedus::xct::McsRwLock::kNextWriterNone, foedus::xct::McsRwLock::next_writer_, foedus::xct::McsRwLock::nreaders(), foedus::xct::spin_until(), foedus::xct::McsRwLock::tail_, foedus::xct::McsRwLock::to_tail_int(), and foedus::xct::McsRwSimpleBlock::unblock().

                               {
    const thread::ThreadId id = adaptor_.get_my_id();
    ASSERT_ND(block_index > 0);
    ASSERT_ND(adaptor_.get_cur_block() >= block_index);
    McsRwSimpleBlock* my_block = adaptor_.get_rw_my_block(block_index);
    ASSERT_ND(my_block->is_finalized());
    // Make sure there is really no successor or wait for it
    uint32_t* tail_address = &mcs_rw_lock->tail_;
    uint32_t expected = McsRwLock::to_tail_int(id, block_index);
    if (my_block->successor_is_ready() ||
      !assorted::raw_atomic_compare_exchange_strong<uint32_t>(tail_address, &expected, 0)) {
      // Have to wait for the successor to install itself after me
      // Don't check for curr_block->has_successor()! It only tells whether the state bit
      // is set, not whether successor_thread_id_ and successor_block_index_ are set.
      // But remember to skip trying readers who failed.
      spin_until([my_block]{ return my_block->successor_is_ready(); });
      if (my_block->has_writer_successor()) {
        assorted::raw_atomic_exchange<thread::ThreadId>(
          &mcs_rw_lock->next_writer_,
          my_block->successor_thread_id_);
      }
    }

    if (mcs_rw_lock->decrement_nreaders() == 1) {
      // I'm the last active reader
      thread::ThreadId next_writer
        = assorted::atomic_load_acquire<thread::ThreadId>(&mcs_rw_lock->next_writer_);
      if (next_writer != McsRwLock::kNextWriterNone &&
          mcs_rw_lock->nreaders() == 0 &&
          assorted::raw_atomic_compare_exchange_strong<thread::ThreadId>(
            &mcs_rw_lock->next_writer_,
            &next_writer,
            McsRwLock::kNextWriterNone)) {
        // I have a waiting writer, wake it up
        // Assuming a thread can wait for one and only one MCS lock at any instant
        // before starting to acquire the next.
        McsBlockIndex next_cur_block = adaptor_.get_other_cur_block(next_writer);
        McsRwSimpleBlock *writer_block = adaptor_.get_rw_other_block(next_writer, next_cur_block);
        ASSERT_ND(writer_block->is_blocked());
        ASSERT_ND(!writer_block->is_reader());
        writer_block->unblock();
      }
    }
  }

Here is the call graph for this function:

template<typename ADAPTOR >

void foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::release_rw_writer ( McsRwLock *  mcs_rw_lock, McsBlockIndex  block_index  )

inline

Definition at line 517 of file xct_mcs_impl.cpp.

References ASSERT_ND, foedus::xct::McsRwLock::increment_nreaders(), foedus::xct::spin_until(), foedus::xct::McsRwLock::tail_, foedus::xct::McsRwLock::to_tail_int(), and UNLIKELY.

                               {
    const thread::ThreadId id = adaptor_.get_my_id();
    ASSERT_ND(block_index > 0);
    ASSERT_ND(adaptor_.get_cur_block() >= block_index);
    auto* my_block = adaptor_.get_rw_my_block(block_index);
    uint32_t expected = McsRwLock::to_tail_int(id, block_index);
    uint32_t* tail_address = &mcs_rw_lock->tail_;
    if (my_block->successor_is_ready() ||
      !assorted::raw_atomic_compare_exchange_strong<uint32_t>(tail_address, &expected, 0)) {
      if (UNLIKELY(!my_block->successor_is_ready())) {
        spin_until([my_block]{ return my_block->successor_is_ready(); });
      }
      ASSERT_ND(my_block->successor_is_ready());
      auto* successor_block = adaptor_.get_rw_other_block(
        my_block->successor_thread_id_,
        my_block->successor_block_index_);
      ASSERT_ND(successor_block->is_blocked());
      if (successor_block->is_reader()) {
        mcs_rw_lock->increment_nreaders();
      }
      successor_block->unblock();
    }
  }

Here is the call graph for this function:

template<typename ADAPTOR >

bool foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::retry_async_rw_reader ( McsRwLock *  lock, McsBlockIndex  block_index  )

inline

Definition at line 556 of file xct_mcs_impl.cpp.

References foedus::xct::McsRwLock::increment_nreaders(), foedus::xct::McsRwLock::tail_, and foedus::xct::McsRwLock::to_tail_int().

                               {
    /* The following turns out to be unsafe because there might be writers
     who are not yet detected by the pred-readers. We are still trying
     to find a way without resorting to extended version.. but for now disabled.
     does_support_try_rw_reader() returns false for this reason.
    const thread::ThreadId id = adaptor_.get_my_id();
    // take a look at the whole lock word, and cas if it's a reader or null
    uint64_t lock_word
      = assorted::atomic_load_acquire<uint64_t>(reinterpret_cast<uint64_t*>(lock));
    McsRwLock ll;
    std::memcpy(&ll, &lock_word, sizeof(ll));
    // Note: it's tempting to put this whole function under an infinite retry
    // loop and only break when this condition is true. That works fine with
    // a single lock, but might cause deadlocks and making this try version
    // not really a try, consider this example with two locks A and B.
    //
    // Lock: requester 1 -> requester 2
    //
    // A: T1 holding as writer -> T2 waiting unconditionally as a writer in canonical mode
    // B: T2 holding as writer -> T1 trying as a reader in non-canonical mode
    //
    // In this case, T1 always sees next_writer=none because T2 consumed it when it got the
    // lock, and the below CAS fails because now B.tail is T2, a writer. T1 would stay in
    // the loop forever...
    if (ll.next_writer_ != McsRwLock::kNextWriterNone) {
      return false;
    }
    McsRwSimpleBlock* block = nullptr;
    if (ll.tail_) {
      block = adaptor_.dereference_rw_tail_block(ll.tail_);
    }
    if (ll.tail_ == 0 || (block->is_granted() && block->is_reader())) {
      ll.increment_nreaders();
      ll.tail_ = McsRwLock::to_tail_int(id, block_index);
      uint64_t desired = *reinterpret_cast<uint64_t*>(&ll);
      auto* my_block = adaptor_.get_rw_my_block(block_index);
      my_block->init_reader();

      if (assorted::raw_atomic_compare_exchange_weak<uint64_t>(
        reinterpret_cast<uint64_t*>(lock), &lock_word, desired)) {
        if (block) {
          block->set_successor_next_only(id, block_index);
        }
        my_block->unblock();
        finalize_acquire_reader_simple(lock, my_block);
        return true;
      }
    }
    */

    // Instead, a best-effort impl here, which is basically same as the writer-case below.
    // This returns false even if there only are readers.
    const thread::ThreadId id = adaptor_.get_my_id();
    auto* my_block = adaptor_.get_rw_my_block(block_index);
    my_block->init_reader();

    McsRwLock tmp;
    uint64_t expected = *reinterpret_cast<uint64_t*>(&tmp);
    McsRwLock tmp2;
    tmp2.increment_nreaders();
    tmp2.tail_ = McsRwLock::to_tail_int(id, block_index);
    uint64_t desired = *reinterpret_cast<uint64_t*>(&tmp2);
    if (assorted::raw_atomic_compare_exchange_weak<uint64_t>(
      reinterpret_cast<uint64_t*>(lock), &expected, desired)) {
      my_block->unblock();
      finalize_acquire_reader_simple(lock, my_block);
      return true;
    }
    return false;
  }

Here is the call graph for this function:

template<typename ADAPTOR >

bool foedus::xct::McsImpl< ADAPTOR, McsRwSimpleBlock >::retry_async_rw_writer ( McsRwLock *  lock, McsBlockIndex  block_index  )

inline

Definition at line 628 of file xct_mcs_impl.cpp.

References foedus::xct::McsRwLock::tail_, and foedus::xct::McsRwLock::to_tail_int().

                                                                         {
    const thread::ThreadId id = adaptor_.get_my_id();
    auto* my_block = adaptor_.get_rw_my_block(block_index);
    my_block->init_writer();

    McsRwLock tmp;
    uint64_t expected = *reinterpret_cast<uint64_t*>(&tmp);
    McsRwLock tmp2;
    tmp2.tail_ = McsRwLock::to_tail_int(id, block_index);
    uint64_t desired = *reinterpret_cast<uint64_t*>(&tmp2);
    my_block->unblock();
    return assorted::raw_atomic_compare_exchange_weak<uint64_t>(
      reinterpret_cast<uint64_t*>(lock), &expected, desired);
  }

Here is the call graph for this function:

---

The documentation for this class was generated from the following file:

• /home/shino/foedus_code/foedus-core/src/foedus/xct/xct_mcs_impl.cpp