C++中的并发队列答案

【问题标题】：concurrent queue in C++C++中的并发队列
【发布时间】：2011-04-14 09:16:33
【问题描述】：

我正在尝试设计一个可以由多个读/写线程同时访问的队列。我更喜欢使用 2 个互斥锁，一个用于读写。写很简单，锁定写互斥体，追加数据，解锁就完成了。

问题在于阅读。如果队列中没有数据，我希望我的线程等到数据可用。一种明显的方法是获取读取互斥锁并每隔 N 个时钟周期轮询队列，但这显然不是最好的方法。这让我想到了条件变量。有没有人有任何好的资源来讨论在 C++ 中使用条件变量（最好是基于 pthreads）实现阻塞队列？

具体来说，我看到以下问题：

一旦写入完成，写入线程将执行 pthread_cond_signal 表明数据存在，但它如何知道某个读取线程正在等待？除非存在 pthread_cond_wait，否则调用 pthread_cond_signal 是非法的。
可以调用 pthread_cond_broadcast 而不是 pthread_cond_signal 吗？也许这可能会绕过 pthread_cond_wait 的问题。此外，这似乎更合乎逻辑，因为多个阅读器线程绝对是一个真正的可能性。
读取器和写入器线程似乎也必须使用相同的互斥锁锁定才能使用条件变量。如果这是真的，那么我们就有一个严重的问题。

【问题讨论】：

为什么你认为#3 是“一个严重的问题？”
@pilcrow 问题很严重，因为我希望读/写线程能够并行访问队列。如果只有一个互斥体，则不会发生并行读/写。
正如 Jim Lewis 所指出的，不清楚您是否有任何 read 线程，只有两种 writer（消费者和生产者）。在这种情况下，pthread_cond_wait 将在等待期间自动释放其互斥体，以便生产者（以及到达的消费者，就此而言）不会阻塞互斥体......

标签： c++ multithreading pthreads

【解决方案1】：

我有一个使用来自http://danborn.net/code/ 的相同互斥锁的实现，但就像我之前提到的，因为它使用条件变量，它也使用 1 个互斥锁。

这是 boost 版本，再次使用单个互斥锁：http://www.justsoftwaresolutions.co.uk/threading/implementing-a-thread-safe-queue-using-condition-variables.html

【讨论】：

【解决方案2】：

this C++ threading blog post 的示例 12-3 应该为您提供一个参考实现，但我认为您自己已经非常接近成功。解决您的具体问题：

在没有服务员的情况下发出信号不是非法的。这是完全合法的，您可以利用这一事实。（顺便说一句，你在哪里读到的？）
pthread_cond_broadcast 导致了一个主要问题：大象群 冲锋，其中所有 n 线程都唤醒并开始在缓存之间拉取内存，即使其中只有一个线程可能向前推进.它会起作用，但效率不高。
您可以从任何地方发出信号，甚至可以从不需要处理相关互斥体的线程发出信号。您的pthread_cond_wait 调用必须释放读取互斥锁；这是唯一的要求，也是唯一需要与条件变量交互的互斥体。

从 #3 开始，您的想法存在一个主要问题：编写器线程必须锁定读写互斥锁。如果没有，会发生什么当队列大小为 1 并且您有一个读取器和一个写入器同时执行时？

【讨论】：

这听起来像是一个非最佳参考，因为它没有显示 OP 想要的使用条件变量。
真的！我无法找到同时使用读/写互斥锁和条件变量的引用。 Arpan 的参考资料很好地展示了条件变量。
这里有一个例子：asgaard.homelinux.org/svn/cpp/threadqueue。对这个问题使用读和写互斥锁没有意义，你需要在读或写时使用相同的锁（尽管你可能需要两组条件，一组用于“队列空”，一组用于“队列满”如果你想在队列上设置一个上限）。
@Andreas 查看computing.llnl.gov/tutorials/pthreads/#ConditionVariables

【解决方案3】：

我认为你误解了一些东西——调用pthread_cond_signal 是完全合法的，即使在这种情况下没有线程在等待。

此外，在多线程编程的上下文中，将某事称为“读取”操作通常意味着它不会改变共享资源的状态。如果通过“阅读” 你的意思是“从队列的头部删除一个项目”，这改变了队列（换句话说，它也是一个“写”。）根据您的情况考虑“消费者和生产者”而不是“读者和作者”可能会更好。

一个互斥体（保证对队列的独占访问）和两个条件变量（“数据可用”、“可用空间”）应该足以满足您的情况。（如果队列可以动态增长，您不需要“可用空间”条件变量；我只是为了完整性而提到这一点。）

如果您的阅读线程是严格阅读器（也就是说，它们不会以任何方式修改共享队列数据结构，例如从queue)，pthread_rwlock 系列调用也可能是一个合适的解决方案。在这个范例中，有读锁（多个读取器可以同时持有，但强制写入器阻塞，直到读取器完成）和写入锁（确保持有写入锁的线程独占访问，阻塞任何其他写入器和读者）。

【讨论】：

有趣的想法。几点：1）在 pthread_cond_wait 从 computing.llnl.gov/tutorials/pthreads/#ConditionVariables 之前调用 pthread_cond_signal 是错误的 2）我希望我的读写操作并行发生，因此单个互斥锁对我不起作用。
@Arpan: 1) 教程在这一点上是完全错误的。 pthreads 不需要 pthread_cond_wait 在调用 pthread_cond_signal 时处于挂起状态。 2）如果允许同一数据结构的读取和写入并行运行，读取器可能会收到损坏的数据。这是一个极其常见的场景，使用单个互斥锁来保证独占访问是一种直接且常见的解决方案。
感谢您指出 1)。 2）我看不到阅读器如何接收损坏的数据——两个阅读器之间的争用由读取互斥锁保护。如果有人锁定了互斥体并且现在轮询队列是否为非空，这应该可以工作。
@Arpan：多个读者（严格意义上，如我的回答中所定义）不需要独占访问。确实需要防止读取器和写入器或多个写入器并行运行。考虑这种情况：写入线程在非原子写入操作中途中断，使数据暂时处于不一致状态。在那一刻，读取线程被调度，并且（因为没有互斥锁）读取部分写入的数据，得到垃圾。
我已经编辑了我的答案，提到了 pthread_rwlock 例程，这可能（也可能不是）是解决您问题的好方法，具体取决于您的读取线程在共享队列上执行的操作类型.

【解决方案4】：

您是否考虑过使用信号量而不是条件变量？信号量是可等待的，即使没有线程在等待，也可以表示“队列非空”状态。

【讨论】：

有效点，但信号量和读/写锁不太一样：后者是一种优化，在写不频繁的情况下。

【解决方案5】：

一个简单的 boost::condition 示例位于here。

【讨论】：