c++ - 什么时候可以安全地从compare_exchange中删除memory_order_acquire或memory_order_release？

Question

我引用了Lewiss Baker的协程教程中的代码。

https://lewissbaker.github.io/2017/11/17/understanding-operator-co-await

bool async_manual_reset_event::awaiter::await_suspend(
  std::experimental::coroutine_handle<> awaitingCoroutine) noexcept
{
  // Special m_state value that indicates the event is in the 'set' state.
  const void* const setState = &m_event;

  // Remember the handle of the awaiting coroutine.
  m_awaitingCoroutine = awaitingCoroutine;

  // Try to atomically push this awaiter onto the front of the list.
  void* oldValue = m_event.m_state.load(std::memory_order_acquire);
  do
  {
    // Resume immediately if already in 'set' state.
    if (oldValue == setState) return false; 

    // Update linked list to point at current head.
    m_next = static_cast<awaiter*>(oldValue);

    // Finally, try to swap the old list head, inserting this awaiter
    // as the new list head.
  } while (!m_event.m_state.compare_exchange_weak(
             oldValue,
             this,
             std::memory_order_release,
             std::memory_order_acquire));

  // Successfully enqueued. Remain suspended.
  return true;
}

其中m_state只是一个 std::atomic<void *>。

bool async_manual_reset_event::is_set() const noexcept
{
  return m_state.load(std::memory_order_acquire) == this;
}
void async_manual_reset_event::reset() noexcept
{
  void* oldValue = this;
  m_state.compare_exchange_strong(oldValue, nullptr, std::memory_order_acquire);
}
void async_manual_reset_event::set() noexcept
{
  // Needs to be 'release' so that subsequent 'co_await' has
  // visibility of our prior writes.
  // Needs to be 'acquire' so that we have visibility of prior
  // writes by awaiting coroutines.
  void* oldValue = m_state.exchange(this, std::memory_order_acq_rel);
  if (oldValue != this)
  {
    // Wasn't already in 'set' state.
    // Treat old value as head of a linked-list of waiters
    // which we have now acquired and need to resume.
    auto* waiters = static_cast<awaiter*>(oldValue);
    while (waiters != nullptr)
    {
      // Read m_next before resuming the coroutine as resuming
      // the coroutine will likely destroy the awaiter object.
      auto* next = waiters->m_next;
      waiters->m_awaitingCoroutine.resume();
      waiters = next;
    }
  }
}

请注意，在 m_state.exchange方法的 set()中，上面的注释清楚地说明了为什么调用交换需要获取和释放。

我想知道为什么在 m_state.compare_exchange_weak方法的 await_suspend()中，第三个参数是std::memory_order_release但不是memory_order_acq_rel(获取已删除)。

作者(Lewis)确实解释了我们需要在compare_exchange_weak中释放，因为我们需要稍后让set()看到compare_exchange_weak中的写入。但是，为什么不要求其他线程中的其他compare_exchange_weak来查看当前compare_exchange_weak中的写操作呢？

是因为发布顺序？即，在发布链中(首先写发布，所有中间操作都是“读取获取然后写发布”操作，最后一个操作是读获取)，那么您无需告诉他们在中间进行获取？

在以下代码中，我尝试实现一个共享锁，

    struct lock {
        uint64_t exclusive : 1;
        uint64_t id : 48;
        uint64_t shared_count : 15;
    };
    std::atomic<lock> lock_ { {0, 0, 0} };
    bool try_lock_shared() noexcept {
        lock currentlock = lock_.load(std::memory_order_acquire);
        if (currentlock.exclusive == 1) {
            return false;
        }
        lock newlock;
        do {
            newlock = currentlock;
            newlock.shared_count++;
        }
        while(!lock_.compare_exchange_weak(currentlock, newlock, std::memory_order_acq_rel) && currentlock.exclusive == 0);

        return currentlock.exclusive == 0;
    }
    bool try_lock() noexcept {
        uint64_t id = utils::get_thread_id();
        lock currentlock = lock_.load(std::memory_order_acquire);
        if (currentlock.exclusive == 1) {
            assert(currentlock.id != id);
            return false;
        }

        bool result = false;
        lock newlock { 1, id, 0 };
        do {
            newlock.shared_count = currentlock.shared_count;
        }
        while(!(result = lock_.compare_exchange_weak(currentlock, newlock, std::memory_order_acq_rel)) && currentlock.exclusive == 0);

        return result;
    }

我到处都使用了 lock_.compare_exchange_weak(currentlock, newlock, std::memory_order_acq_rel)，可以安全地将它们替换为 compare_exchange_weak(currentlock, newlock, std::memory_order_release, std::memory_order_acquire)吗？

我还可以看到从 memory_order_release中删除 compare_exchange_strong的示例(请参阅Lewis代码的 compare_exchange_strong函数中的 reset())，在这里您只需要std::memory_order_acquire作为compare_exchange_strong(但不发布)。我没有真正看到从弱函数中删除了memory_order_release也不从强函数中删除了memory_order_acquire。

这让我想知道是否存在更深层次的规则，我不理解或不理解。

谢谢。

Answer 1

memory_order_acquire仅对读取值的操作有意义，而memory_order_release仅对写入值的操作有意义。由于读-修改-写操作可以读写，因此可以合并这些存储顺序，但并非总是必要的。
m_event.m_state.compare_exchange_weak使用memory_order_release写入新值，因为它试图替换以前使用memory_order_acquire读取的值:

  // load initial value using memory_order_acquire
  void* oldValue = m_event.m_state.load(std::memory_order_acquire);
  do {
    ...
  } while (!m_event.m_state.compare_exchange_weak(oldValue, this,
             std::memory_order_release,
             // in case of failure, load new value using memory_order_acquire
             std::memory_order_acquire));

恕我直言，在这种情况下甚至根本不需要使用memory_order_acquire，因为从不取消引用oldValue，而是仅将其存储为下一个指针，即可以完美地发现将这两个memory_order_acquire替换为memory_order_relaxed。

在 async_manual_reset_event::set()中，情况不同:

  void* oldValue = m_state.exchange(this, std::memory_order_acq_rel);
  if (oldValue != this)
  {
    auto* waiters = static_cast<awaiter*>(oldValue);
    while (waiters != nullptr)
    {
      // we are de-referencing the pointer read from m_state!
      auto* next = waiters->m_next;
      waiters->m_awaitingCoroutine.resume();
      waiters = next;
    }

因为我们要取消引用从 m_state读取的指针，所以我们必须确保这些读取在写入这些服务对象之后发生。这是通过 m_state上的sync-with关系来确保的。通过 memory_order_release通过前面讨论的compare_exchange添加了writer。交换的获取部分与release-compare_exchange(实际上是释放序列的一部分，所有先前的release-compare_exchange)同步，从而提供了必要的事前发生关系。

老实说，我不确定为什么这种交换需要发行部分。我认为作者可能希望站在“安全的一面”，因为其他几项操作也比必要的要强(我已经提到 await_suspend不需要memory_order_acquire， is_set和 reset也是一样)。

对于您的锁实现，它非常简单-当您想要获取锁( try_lock_shared / try_lock)时，仅将 memory_order_acquire用于比较交换操作。释放锁必须使用 memory_order_release。

该参数也非常简单:您必须确保在获取锁后，当前所有者对可见的锁所保护的数据所做的任何更改对当前所有者都是可见的，也就是说，您必须确保这些更改发生在锁之前。获取锁定后将要执行的操作。这是通过在 try_lock(acquire-CAS)与先前的 unlock(发行商店)之间建立同步关系来实现的。

当试图基于C++内存模型的语义争论实现的正确性时，我通常按以下步骤进行操作:

标识必要的事前关系(例如您的锁)

确保在所有代码路径上正确建立了这些事前关系

而且我总是注释原子操作，以记录如何建立这些关系(即，涉及其他哪些操作)。例如:

  // (1) - this acquire-load synchronizes-with the release-CAS (11)
  auto n = head.load(std::memory_order_acquire);

  // (8) - this acquire-load synchronizes-with the release-CAS (11)
  h.acquire(head, std::memory_order_acquire);

  // (11) - this release-CAS synchronizes-with the acquire-load (1, 8)
  if (head.compare_exchange_weak(expected, next, std::memory_order_release, std::memory_order_relaxed))

(有关完整代码，请参见 https://github.com/mpoeter/xenium/blob/master/xenium/michael_scott_queue.hpp)

有关C++内存模型的更多详细信息，我可以推荐我与他人合着的这篇论文: Memory Models for C/C++ Programmers