使用std :: atomic的无锁队列

如何解决使用std :: atomic的无锁队列

我希望使用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上，以便在成功完成m_next之后可以更新m_tail的{{1}}而不实际这样做通过push，以防其他线程已更改它。因此，即使另一个线程已经达到m_tail的新值，当前线程仍将其看到的push的{{1}}链接到新节点。

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

让我们假设队列中有m_tail m_next个唯一的项目，然后我们进入以下if语句：

thread1

让我们假设pop在标记的位置插入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,std::memory_order_relaxed); return; }，只有push继续，然后thread1才会执行：

thread1

，并且基本上可以通过将thread1设置为m_head.store(tmp,std::memory_order_relaxed); return;来撤消thread2的{{1}}。据我了解，在这种情况下，内存命令对我无济于事，所以我不确定我的选择是什么？

另一个有问题的情况是，假设我们有2个阅读器线程push和m_head 做同样的事情：

nullptr

让我们假设队列的大小为1，因此他们两个都可以看到队列不为空，并获得对前数据的引用，然后弹出该元素并打印相同的结果。
在我看来，锁定在这种情况下会有所帮助，但是我不希望使用锁定，而且我也不希望读取线程关心同步问题，因为接口本身应该是负责任的，但是自{{1} }和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(); } }，所以即使在这里，我也不确定如何处理此问题。我可以使接口返回原始指针或front，但在我看来这两种解决方案似乎都不正确，因此很乐意听取有关在此处应执行的操作的意见。

此外，我不确定是否可以使用除CAS之外的更便宜的方法，我知道我可以采用独特的插槽方法，其中每个线程通过在原子上使用pop来获得到固定数组的索引类型为front的对象，因此每个线程都将队列推送到唯一索引，但是我不喜欢这种方法，因为它限制了固定大小的队列。另一方面，使用nullptr和std::optional可能也不是最快的方法，我可以使用某种池分配器，但是我必须确保它也已同步，这是一个新的疼痛程度。

我什至不确定这些是所有问题，还是我可以在实现过程中发现的问题，我确定我没有考虑所有事情（或者也许我做了？），无论如何我都会听听您对上述问题的想法，以及解决这些问题的方法。

解决方法

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

在m_head.store上执行CAS后，两个m_tail操作之间的竞争
此循环可能会受到ABA problem的影响：

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

在pop中删除节点后，您立即delete，但是那时另一个线程可能仍然引用该节点并访问（例如另一个线程）弹出式广告），从而可以免费使用。（这也称为内存回收问题。）
说明：假设两个线程当前位于pop中，并且已经将相同的值读入old_head中。第一个线程继续执行，对m_head执行CAS，然后在下一步中立即删除old_head。直到现在，第二个线程才继续尝试使用m_head作为新值来更新old_head->m_next。这意味着第二个线程取消对刚删除的节点的指针。
您的设计需要两个单独的函数调用才能从队列中弹出项目并获取其值。

设计无锁甚至无锁算法本质上是困难的。问题2和3.都可以使用内存回收方案来解决。通常可以通过不使用front操作来避免问题4，而可以让pop返回该项（直接通过std::optional或通过使用try_pop版本的{按引用超出参数，并返回布尔值，指示操作是否成功。

无论哪种方式，我都建议使用已建立的无锁算法之一，例如Michael-Scott-Queue。不幸的是，如果您决定实施该算法，则仍然必须解决内存回收问题。

我可以向您介绍我的Xenium，它不仅提供了Michael-Scott-Queue的实现，而且还提供了几种内存回收方案，以防您仍然想尝试一下自己，但又想避免安全内存回收带来麻烦。

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

使用std :: atomic的无锁队列

如何解决使用std :: atomic的无锁队列

解决方法

相关推荐