C++11 非阻塞生产者/消费者答案

【问题标题】：C++11 non-blocking producer/consumerC++11 非阻塞生产者/消费者
【发布时间】：2014-02-20 07:16:06
【问题描述】：

我有一个 C++11 应用程序，它有一个高优先级线程正在生成数据，还有一个低优先级线程正在使用它（在我的例子中，将它写入磁盘）。我想确保高优先级生产者线程永远不会被阻塞，即它只使用无锁算法。

使用无锁队列，我可以将数据从生产者线程推送到队列，并从消费者线程进行轮询，从而实现上述目标。我想修改我的程序，以便消费者线程在不活动时阻塞而不是轮询。

C++11 条件变量似乎对阻塞消费者线程很有用。谁能告诉我一个如何使用它的例子，同时避免消费者睡觉时数据仍在队列中的可能性？更具体地说，我想确保在生产者将最后一个项目推入队列后，消费者总是在有限的时间内被唤醒。生产者保持非阻塞状态也很重要。

【问题讨论】：

查看codereview.stackexchange.com/questions/39199/…中条件变量的使用示例
使用条件变量完全无锁是不可能的，因为要等待或发出信号，您必须获得锁。但实际上这并没有真正的区别（大多数锁都会有一个短暂的繁忙等待策略，这应该足以避免任何内核调用或其他原因，因为读者的关键部分非常短）
@Voo 不需要获取锁来表示条件变量。
@Voo Waiting 条件变量需要锁定，signaling 不需要。
@Casey 你有没有看到文档说明条件变量的信号是非阻塞的？我也这么认为，但我从来没有在任何地方找到明确的确认。

标签： multithreading c++11 producer-consumer

【解决方案1】：

C++11 条件变量似乎对阻塞消费者线程很有用。谁能告诉我一个如何使用它的例子，同时避免消费者睡觉时数据仍在队列中的可能性？

要使用条件变量，您需要一个互斥锁和一个条件。在您的情况下，条件将是“队列中有可用数据”。由于生产者将使用无锁更新来生产工作，消费者必须使用相同形式的同步来消费工作，因此互斥锁实际上不会用于同步，只有消费者线程需要，因为没有等待条件变量的其他方式。

// these variables are members or otherwise shared between threads
std::mutex m_mutex;
std::condition_variable m_cv;
lockfree_queue m_data;

// ...

// in producer thread:
while (true)
{
  // add work to queue
  m_data.push(x);
  m_cv.notify_one();
}

// in consumer thread:
while (true)
{
  std::unique_lock<std::mutex> lock(m_mutex);
  m_cv.wait(lock, []{ return !m_data.empty(); });
  // remove data from queue and process it
  auto x = m_data.pop();
}

如果队列在等待之前为空，则条件变量只会在wait 调用中阻塞。条件变量可能会被虚假唤醒，或者因为它是由生产者通知的，但在任何一种情况下，如果队列非空，则只会从 wait 调用返回（而不是再次休眠）。这可以通过使用带有谓词的condition_variable::wait 重载来保证，因为条件变量总是会为您重新检查谓词。

由于互斥锁仅由消费者线程使用，它实际上可能是该线程的本地（只要您只有一个消费者，有多个消费者都需要共享同一个互斥锁以等待同一个 condvar )。

【讨论】：

如果生产者在消费者调用 wait() 之前调用 notify_one()，消费者最终不会与队列中的数据一起休眠（因此如果该数据是最后一个数据，则永远休眠生产）？
不，因为对wait 的调用会在休眠前检查它的谓词，所以如果m_data_available 已经为真，它就不会休眠。但是在实际代码中，您可能不想使用简单的atomic<bool> 来指示数据可用，您需要检查m_queue.size() 或m_available_count，否则消费者不知道是否应该设置m_data_available=false消费单条数据后，可能会有更多可用数据。
我已经编辑了答案，将 atomic<bool> 替换为对无锁队列对象的操作，以解决上述评论中描述的问题，并澄清如果队列中已有数据
我相信wait()的形式不会原子地检查谓词并进入等待状态。因此，如果另一个线程在谓词检查和等待之间产生数据，它可能会永远等待队列中的数据。
啊，是的，这是真的 - 锁定互斥锁可以防止如果谓词的结果只会被持有互斥锁的线程更改，但在您的情况下，您不希望生产者使用互斥锁，因为它可能会阻塞。这个想法就这么多

【解决方案2】：

我过去发现的一个解决方案是使用 Windows 事件 (http://msdn.microsoft.com/en-us/library/windows/desktop/ms682396(v=vs.85).aspx)。在这种情况下，事件保持有信号状态，直到它唤醒一个等待的线程，如果没有线程在等待，它保持有信号状态。所以生产者只需要在将数据推送到队列后发出事件信号。然后我们保证消费者会在这之后的一段时间内醒来。

我无法找到使用标准库实现此功能的方法（至少在不阻塞生产者线程的情况下不会）。

【讨论】：

【解决方案3】：

我认为信号量可以用来安全地解决这个问题：

// in producer thread:
while (true)
{
  m_data.push();
  m_semaphore.release();
}

// in consumer thread:
while (true)
{
  m_semaphore.wait();
  m_data.pop();
}

不幸的是，我认为 C++11 不包含信号量？我也无法确认释放信号量是非阻塞操作。当然是互斥体的实现，例如C++0x has no semaphores? How to synchronize threads? 不允许非阻塞生产者线程。

【讨论】：