使用 std::atomic 的无锁队列

Question

我希望使用 std::atomic 创建一个无锁队列。
这是我第一次尝试这样做可能不太好：

template <typename T>
class atomic_queue
{
public:
    using value_type = T;
private:
    struct node
    {
        value_type m_value;
        node* m_next;
        node* m_prev;

        node(const value_type& value) :
            m_value(value),
            m_next(nullptr),
            m_prev(nullptr) {}
    };
private:
    std::atomic<node*> m_head = nullptr;
    std::atomic<node*> m_tail = nullptr;
public:
    void push(const value_type& value)
    {
        auto new_node = new node(value);

        node* tmp = nullptr;
        if (m_tail.compare_exchange_strong(tmp, new_node))
        {
            m_head.store(new_node, std::memory_order_relaxed);
            return;
        }

        node* old_tail;
        do {
            old_tail = m_tail;
            new_node->m_prev = old_tail;
        } while (!m_tail.compare_exchange_strong(old_tail, new_node));
        new_node->m_prev->m_next = new_node;
    }

    void pop()
    {
        if (m_head.load(std::memory_order_relaxed) == nullptr)
        {
            return;
        }

        node* tmp = nullptr;
        node* head = m_head;
        if (m_tail.compare_exchange_strong(head, tmp))
        {
            m_head.store(tmp, std::memory_order_relaxed);
            return;
        }

        node* old_head;
        do {
            old_head = m_head;
        } while (m_head && !m_head.compare_exchange_strong(old_head, old_head->m_next));
        if (old_head)
        {
            delete old_head;
        }
    }

    bool empty()
    {
        return m_head.load(std::memory_order_relaxed) == nullptr;
    }

    value_type& front()
    {
        node* head = m_head.load(std::memory_order_acquire);
        return head->m_value;
    }
};

这里要注意的是，我将 m_prev 存储在 node 上，这样我就可以在成功 push 之后更新 m_tail 的 m_next 而实际上通过 m_tail 这样做，以防它已被另一个线程更改。因此，即使另一个线程已经 push 一个新值，当前线程仍将 link 它所看到的 m_tail 的 m_next 到新节点。

据我所知，现在有一些事情并不是真正的线程安全的，而且我真的想不出解决这些问题的好方法：

让我们假设 thread1 pop 是队列中唯一的一项，然后我们进入以下 if 语句：

        node* tmp = nullptr;
        node* head = m_head;
        if (m_tail.compare_exchange_strong(head, tmp))
        {
            // Now thread2 kicks in
            m_head.store(tmp, std::memory_order_relaxed);
            return;
        }

让我们假设 thread2 在标记的位置踢到 push 队列的新值，将执行以下语句：

        node* tmp = nullptr;
        if (m_tail.compare_exchange_strong(tmp, new_node))
        {
            m_head.store(new_node, std::memory_order_relaxed);
            return;
        }

让我们假设它完成了 pushing 而没有 thread1 继续，然后 thread1 继续，然后 thread1 将执行：

        m_head.store(tmp, std::memory_order_relaxed);
        return;

并且基本上会通过将 m_head 设置为 nullptr 来撤消 thread2 的 push。据我所知，内存顺序在这种情况下帮不了我，所以我不确定我的选择是什么？

另一个有问题的场景是，假设我们有 2 个 reader 线程 thread3 和 thread4 做同样的工作：

    while (true)
    {
        if (!q.empty())
        {
            int v = q.front();
            q.pop();
            std::stringstream stream;
            stream << "thread_3/4: " << v << '\n';
            std::cout << stream.str();
        }
    }

并且让我们假设队列的大小为 1，因此他们都可以看到队列不为空并获取对前面数据的引用，然后弹出元素并打印相同的结果。
在我看来，锁定在这种情况下会有所帮助，但我不想使用锁定，而且我也不希望读取线程关心同步问题，因为接口本身应该负责，但由于 front 和 pop 是独立的我没有看到处理这个的好方法。
还有一个问题是 front 可能访问 nullptr，所以即使在这里我也不确定如何处理这个问题。我可以使接口 return 成为原始指针，或 std::optional 但在我看来这两种解决方案似乎都不正确所以很想听听关于应该在这里做什么的意见。

此外，我不确定我是否可以使用 CAS 以外的更便宜的方法，我知道我可以使用独特的槽方法，其中每个线程通过使用 fetch_add 获得固定数组的索引在 std::atomic<int> slot 类型的原子上，因此每个线程都会将队列推送到唯一索引，但我不喜欢这种方法，因为它限制了固定大小的队列。另一方面，使用 new 和 delete 可能也不是最快的，我可以使用排序的池分配器，但是我必须确保它也同步，这是一个新的疼痛程度。

我什至不确定这些都是问题，这些是我在实施过程中可以发现的问题，我确定我没有考虑到所有事情（或者我考虑过？），无论如何都会很想听听您对所描述问题的想法以及克服这些问题的方法。

Answer 1

您的实施存在几个问题，您已经正确识别了其中一些问题。

在 m_tail

m_head.store

此循环可能受到 ABA problem:

  do {
        old_head = m_head;
  } while (m_head && !m_head.compare_exchange_strong(old_head, old_head->m_next));

删除 pop 中的节点后，您立即 delete 它，但那时另一个线程可能仍然引用它并且访问它（例如 pop 中的另一个线程），导致 use-after-free。（这也称为内存回收问题。）
解释： 假设两个线程当前在 pop 中并且已将相同的值读入 old_head。第一个线程继续，在 m_head 上执行 CAS，并在下一步中立即删除 old_head。只是现在第二个线程继续尝试更新 m_head，使用 old_head->m_next 作为新值。这意味着线程二 取消引用指向刚刚删除的节点的指针。
您的设计需要两个单独的函数调用来从队列中弹出一个项目并获取它的值。

设计 lock-free 甚至 lock-less 算法本身就很困难。问题 2. 和 3. 都可以通过使用内存回收方案来解决。问题 4 通常通过不使用 front 操作来避免，而是使用 pop return 项目（直接通过 std::optional，或通过 try_pop通过引用获取 out-parameter 和 returns 指示操作是否成功的 bool 的版本。

无论哪种方式，我都建议使用一种已建立的 lock-free 算法，例如 Michael-Scott-Queue。不幸的是，如果您决定实施该算法，您仍然需要处理内存回收问题。

我可以推荐你参考我的 Xenium，它不仅提供了 Michael-Scott-Queue 的实现，而且还提供了几种内存回收方案，以防你仍然想自己试验一下，但又想避免安全内存回收的麻烦。

内存回收方案是一种解决内存回收问题的算法。有很多建议的解决方案来解决安全内存回收问题，例如危险指针或基于纪元的回收，但每种方案都有其缺点。这就是为什么内存回收问题仍然被视为共享内存并发中当前最难开放的问题。有关详细信息，我可以向您推荐我的硕士论文 Effective Memory Reclamation forLock-Free Data Structures in C++。它不仅解释了内存回收问题和大量提出的回收方案，还讨论了我基于通用接口对其中一些方案的实现。 Xenium 以该工作为基础。

使用 std::atomic 的无锁队列

Lockless queue using std::atomic

c++

queue

multithreading

c++11

stdatomic