无锁队列——单生产者，多消费者答案

【问题标题】：Lock Free Queue -- Single Producer, Multiple Consumers无锁队列——单生产者，多消费者
【发布时间】：2011-02-11 17:44:41
【问题描述】：

我正在寻找一种方法来实现支持单个生产者和多个消费者的无锁队列数据结构。我看过 Maged Michael 和 Michael Scott (1996) 的经典方法，但他们的版本使用链表。我想要一个使用有界循环缓冲区的实现。使用原子变量的东西？

顺便说一句，我不确定为什么这些经典方法是为需要大量动态内存管理的链表设计的。在多线程程序中，所有的内存管理例程都是序列化的。将无锁方法与动态数据结构结合使用，难道我们没有打败它们的好处吗？

我正在尝试在英特尔 64 位架构上使用 pthread 库在 C/C++ 中对此进行编码。

谢谢你，希里什

【问题讨论】：

有限大小的缓冲区意味着如果其中没有空白空间，生产者可能会失败。你可以接受吗？
另请注意，在 C++ 中，您可以将自己的分配器提供给 std::list。因为你只有一个生产者，这个分配器不需要同步。例如，它可以从预先分配的缓冲区中“分配”列表节点，并且当空间不足时，使用全局同步的malloc() 分配一个新的缓冲区，类似于“真实”分配器。这意味着它只会在 1% 的调用中使用同步。
如果您希望优化线程的内存使用，tcmalloc 是一个很棒的库。由于它为每个线程维护内存池，它可能避免了内存例程序列化问题。
根据我的研究，英特尔 TBBMalloc 的可扩展性优于 Google TCMalloc。您还可以提供自己的分配器或使用英特尔 TBB cache_aligned_allocator 传递给 std::list 或 std::queue
有一些 malloc 实现是无锁的并且大部分是无锁的，还有任意数量的线程感知 malloc 可以避免在许多情况下锁定。

标签： c++ queue atomic lock-free

【解决方案1】：

循环缓冲区的使用使得锁成为必要，因为需要阻塞来防止头部越过尾部。但除此之外，头和尾指针可以很容易地以原子方式更新。或者在某些情况下，缓冲区可能太大以至于覆盖不是问题。（在现实生活中，你会在自动交易系统中看到这一点，循环缓冲区的大小可以保存 X 分钟的市场数据。如果你落后 X 分钟，你会遇到比覆盖缓冲区更糟糕的问题）。

当我在 C++ 中实现 MS 队列时，我使用堆栈构建了一个无锁分配器，这很容易实现。如果我有 MSQueue，那么在编译时我知道 sizeof(MSQueue::node)。然后我制作了一堆所需大小的 N 个缓冲区。 N 可以增长，即如果 pop() 返回 null，很容易向堆请求更多块，并将这些块推入堆栈。除了对更多内存的可能阻塞调用之外，这是一个无锁操作。

请注意，T 不能有非平凡的 dtor。我研究了一个确实允许非平凡的 dtors 的版本，它确实有效。但我发现将 T 设为我想要的 T 的指针更容易，生产者释放所有权，消费者获得所有权。这当然需要使用无锁方法分配 T 本身，但我使用堆栈创建的相同分配器也可以在这里工作。

无论如何，无锁编程的重点不在于数据结构本身更慢。要点是：

无锁使我独立于调度程序。基于锁的编程依赖于调度程序来确保锁的持有者正在运行，以便他们可以释放锁。这就是导致“优先级反转”的原因在 Linux 上，有一些锁定属性可以确保发生这种情况
如果我独立于调度程序，操作系统管理时间片的时间会容易得多，而且上下文切换会少得多
使用无锁方法更容易编写正确的多线程程序，因为我不必担心死锁、活锁、调度、同步等问题这对于共享内存实现尤其如此，在共享内存实现中进程可能会在持有共享锁时死亡内存，没有办法释放锁
无锁方法更容易扩展。事实上，我已经实现了使用网络消息传递的无锁方法。像这样的分布式锁是一场噩梦

也就是说，在很多情况下，基于锁的方法更可取和/或需要

更新昂贵或无法复制的内容时。大多数无锁方法使用某种版本控制，即制作对象的副本，更新它，并检查共享版本是否仍然与复制时相同，然后使当前版本更新版本。否则再次复制它，应用更新，然后再次检查。继续这样做，直到它起作用。当对象很小但它们很大或包含文件句柄等时，这很好，不推荐
大多数类型无法以无锁方式访问，例如任何 STL 容器。这些具有需要非原子访问的不变量，例如 assert(vector.size()==vector.end()-vector.begin())。因此，如果您要更新/读取共享的向量，则必须将其锁定。

【讨论】：

【解决方案2】：

这是一个老问题，但没有人提供公认的解决方案。因此，我将此信息提供给可能正在搜索的其他人。

本站：http://www.1024cores.net

提供了一些非常有用的无锁/无等待数据结构，并有详尽的解释。

您正在寻找一种针对读写器问题的无锁解决方案。

见：http://www.1024cores.net/home/lock-free-algorithms/reader-writer-problem

【讨论】：

【解决方案3】：

对于传统的单块循环缓冲区，我认为这根本无法通过原子操作安全地完成。你需要一口气读完这么多。假设你有一个这样的结构：

uint8_t* buf;
unsigned int size; // Actual max. buffer size
unsigned int length; // Actual stored data length (suppose in write prohibited from being > size)
unsigned int offset; // Start of current stored data

在阅读时，您需要执行以下操作（这就是我实现它的方式，您可以交换一些步骤，就像我稍后会讨论的那样）：

检查读取长度是否不超过存储长度
检查偏移量+读取长度是否不超过缓冲区边界
读出数据
增加偏移，减少长度

你当然应该做同步（如此原子）来完成这项工作？实际上将步骤 1 和 4 合并到一个原子步骤中，或者澄清一下：同步执行：

检查read_length，这可以像read_length=min(read_length,length);
使用 read_length 减少长度：length-=read_length
从偏移量unsigned int local_offset = offset获取本地副本
用 read_length 增加偏移量：offset+=read_length

之后，您可以从 local_offset 开始执行 memcpy（或其他），检查您的读取是否超过循环缓冲区大小（分成 2 个 memcpy），...。这是“相当”线程安全的，您的 write 方法仍然可以覆盖您正在读取的内存，因此请确保您的缓冲区确实足够大以尽量减少这种可能性。

现在，虽然我可以想象您可以将 3 和 4（我猜这就是他们在链表案例中所做的）甚至 1 和 2 组合在原子操作中，但我看不到您在一个原子操作中完成整个交易:)。

但是，如果您的消费者非常聪明并且总是知道要阅读什么，您可以尝试放弃“长度”检查。你还需要一个新的 woffset 变量，因为 (offset+length)%size 的旧方法来确定写入偏移量将不再有效。请注意，这与链表的情况很接近，实际上您总是从链表中读取一个元素（= 固定的、已知大小）。同样在这里，如果你把它做成一个循环链表，你可以读到很多，也可以写到你当时正在阅读的位置！

最后：我的建议是，只需使用锁，我使用 CircularBuffer 类，对于读取和写入完全安全）用于实时 720p60 视频流媒体，并且我完全没有遇到锁的速度问题。

【讨论】：

【解决方案4】：

这是一个古老的问题，但没有人提供准确回答它的答案。鉴于在（几乎）同一个问题的搜索结果中仍然很高，应该有一个答案，因为存在一个。

可能有不止一种解决方案，但这里有一个实现： https://github.com/tudinfse/FFQ 自述文件中引用的会议论文详细介绍了该算法。

【讨论】：