请求审查 mpsc 队列的内存顺序

问题描述

我将 http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue 的 C 代码转换为 C++。

在撰写本文时，它看起来如下：

struct MpscNode
{
  std::atomic<MpscNode*> m_next;
};

class MpscQueue
{
 private:
  std::atomic<MpscNode*> m_head;
  MpscNode*              m_tail;
  MpscNode               m_stub;

 public:
  MpscQueue() : m_head(&m_stub),m_tail(&m_stub),m_stub{{nullptr}} { }

  void push(MpscNode* node)
  {
    node->m_next.store(nullptr,std::memory_order_relaxed);
    MpscNode* prev = m_head.exchange(node,std::memory_order_relaxed);
    prev->m_next.store(node,std::memory_order_release);
  }

  MpscNode* pop()
  {
    MpscNode* tail = m_tail;
    MpscNode* next = tail->m_next.load(std::memory_order_acquire);
    if (tail == &m_stub)
    {
      if (nullptr == next)
        return nullptr;
      m_tail = next;
      tail = next;
      next = next->m_next.load(std::memory_order_acquire);
    }
    if (next)
    {
      m_tail = next;
      return tail;
    }
    MpscNode* head = m_head.load(std::memory_order_relaxed);
    if (tail != head)
      return nullptr;
    push(&m_stub);
    next = tail->m_next.load(std::memory_order_acquire);
    if (next)
    {
      // Remove node and return it.
      m_tail = next;
      return tail;
    }
    return nullptr;
  }
};

您可以在此处查看最新版本，包括我的评论：https://github.com/CarloWood/ai-utils/blob/master/threading/MpscQueue.h

我的问题是关于 push 中的第二行：

MpscNode* prev = m_head.exchange(node,std::memory_order_relaxed);

我不完全确定这里使用的内存顺序。看起来是对的，但我想回顾一下我使用的内存顺序。

我无法测试，因为我只有一个 Intel cpu - 而不是一个弱的原子的（比如 -say- ARM）。

解决方法

是的，您的代码是正确的！ m_head 上的操作可以放宽，因为它们不用于任何同步。 pop 仅加载 m_head 以检查它是否等于 tail，但没有依赖于 m_head 上的同步关系的操作。

重要的操作在 m_next 上，用于建立必要的同步，并在那里正确使用获取/释放订单。

所以从我的角度来看，这段代码看起来不错！

FWIW：您不一定需要具有弱内存模型的 CPU 来检测数据竞争； thread-sanitizer 在 x86 CPU 上也做得很好。

编辑：

让我尝试详细说明...首先，让我们注意一些一般性观察。

队列总是至少包含一个节点，即 m_tail 和 m_head 总是指向某个节点并且永远不会为空。
当且仅当 m_tail 和 m_head 指向 m_stub 时队列为空。如果 m_tail 和 m_head 都指向某个不是 m_stub 的节点，则队列包含一个元素（即该节点），因此我们必须推送 m_stub 以便能够使该节点出列并满足我们的要求，即队列始终至少包含一个节点。
队列不可线性化！

在推理记忆顺序时，我们首先需要确定我们的要求，即，什么是必要的发生之前的关系。让我们使用以下符号：

happens-before    -hb->
synchronize-with  -sw->
sequenced-before  -sb->
read-from         -rf->

在这种情况下很简单 - 我们需要 push 和 pop 操作之间的happens-before 关系。特别是我们需要显示以下内容（node1 和 node2 是单独的变量，但都指向同一个节点；init 和 consume 是一些任意函数，它们对节点）：

init(node1) -sb-> push(node1) -hb-> node2 = pop() -sb-> consume(node2)

由于happens-before 关系的传递性，它遵循init(node1) -hb-> consume(node2)。那么我们如何确保呢？让我们看看 push 操作：

node->m_next.store() -sb-> m_head.exchange() -sb-> prev-m_next.store()

到 prev->m_next 的存储是我们的同步点。只有当此存储对消费者可见时，才能使用该节点。

现在让我们看看 pop 操作。首先，我们可以注意到有两条路径不返回nullptr，并且这两条路径都返回变量tail。 tail 有两个赋值，一个来自 m_tail，另一个来自 next 如果 tail == &m_stub。我们将在这里忽略后一种情况，因为很容易形成类似的论点。 m_tail 不能用于与 pop 建立线程间发生的关系，因此我们必须考虑 m_tail 如何接收其值。

m_tail 有两个赋值（我们一直忽略 tail == &m_stub 的情况），都赋值给 next，而 tail->m_next.load() 本身被赋值为 pop_1 pop_2 v-------------------------^---------------------v v-----------------^----------------v next=tail->m_next.load() -sb-> m_tail=next -sb-> ... -sb-> tail=m_tail -sb-> return tail Note: pop_1 is the pop call preceeding pop_2. 的结果。所以我们最终得到以下（简化）：

tail->m_next

此代码路径仅包含单个原子操作 - pop 的负载 - 因此此操作必须建立必要的发生前关系。这是通过使用获取/释放订单来实现的。如果 push 操作中的负载观察到 pop_2 中存储写入的值，则这两个操作建立同步关系。值得注意的是，pop_1 所要求的线程间发生在之前的关系实际上是由 thread1 thread2 v----------------^-------------v v------------------------^---------------------v prev->m_next.store(mo_release) -rf-> next=tail->m_next.load(mo_acquire) -sb-> pop_2 => prev->m_next.store(mo_release) -sw-> next=tail->m_next.load(mo_acquire) -sb-> pop_2 => prev->m_next.store(mo_release) -hb-> next=tail->m_next.load(mo_acquire) -sb-> pop_2 => push -hb-> pop_2 建立的：

m_head

现在让我们看看执行 next 加载的代码路径。首先，此路径仅在 m_head 为空时才相关。其次，如果 tail 与 push 不同，我们会立即返回一个 nullptr（如果 m_next 已经执行了交换，但还没有存储到 m_head - 回想一下，这个队列是不可线性化的）。所以我们只在 tail 等于 m_stub 时才继续（即，如果我们在队列中剩下一个不是 tail 的节点），但我们只是讨论了如何我们已经为 m_head 建立了必要的发生前关系，所以我们不需要 m_head！如果在 memory_order_release 上没有获取操作，那么使用 {{1}} 进行交换操作是没有意义的，因为没有任何东西可以与该交换同步。

c++memory-barriers memory-model stdatomic thread-safety