C ++ 11非阻塞生产者/消费者 [英] C++11 non-blocking producer/consumer

问题描述

我有一个C ++ 11应用程序，其中的一个高优先级线程正在生成数据，而一个低优先级线程正在使用它(在我的情况下是将其写入磁盘).我想确保高优先级生产者线程不会被阻塞，即它仅使用无锁算法.

I have a C++11 application with a high-priority thread that's producing data, and a low-priority thread that's consuming it (in my case, writing it to disk). I'd like to make sure the high-priority producer thread is never blocked, i.e. it uses only lock-free algorithms.

使用无锁队列，我可以将数据从生产者线程推送到队列，并从消费者线程轮询数据，从而达到我的目标.我想修改程序，以使使用者线程在非活动状态而不是轮询状态时阻塞.

With a lock-free queue, I can push data to the queue from the producer thread, and poll it from the consumer thread, thus meeting my goals above. I'd like to modify my program so that the consumer thread blocks when inactive instead of polling.

似乎C ++ 11条件变量可能对阻止使用者线程有用.任何人都可以向我展示如何使用它的示例，同时避免消费者在队列中仍存有数据的情况下入睡吗?更具体地说，我想确保在生产者将最后一个项目推入队列后，始终在一定的时间内唤醒消费者.生产者保持不阻塞也很重要.

It seems like the C++11 condition variable might be useful to block the consumer thread. Can anyone show me an example of how to use it, while avoiding the possibility that the consumer sleeps with data still in the queue? More specifically, I want to make sure that the consumer is always woken up some finite time after the producer pushes the last item into the queue. It's also important that the producer remains non-blocking.

推荐答案

似乎C ++ 11条件变量可能对阻止使用者线程有用.任何人都可以向我展示如何使用它的示例，同时避免消费者在队列中仍存有数据的情况下入睡吗?

It seems like the C++11 condition variable might be useful to block the consumer thread. Can anyone show me an example of how to use it, while avoiding the possibility that the consumer sleeps with data still in the queue?

要使用条件变量，您需要互斥体和条件.在您的情况下，条件将是队列中有可用数据".由于生产者将使用无锁更新来生产工作，因此消费者必须使用相同形式的同步来消耗工作，因此互斥锁实际上将不会用于同步，并且仅由消费者线程使用，因为没有等待条件变量的另一种方法.

To use a condition variable you need a mutex and a condition. In your case the condition will be "there is data available in the queue". Since the producer will be using lock-free updates to produce work, the consumer has to use the same form of synchronisation to consume the work, so the mutex will not actually be used for synchronisation and is only needed by the consumer thread because there's no other way to wait on a condition variable.

// 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重载可以保证这一点，因为条件变量始终会为您重新检查谓词.

The condition variable will only block in the wait call if the queue is empty before the wait. The condition variable might wake up spuriously, or because it was notified by the producer, but in either case will only return from the wait call (rather than sleeping again) if the queue is non-empty. That's guaranteed by using the condition_variable::wait overload that takes a predicate, because the condition variable always re-checks the predicate for you.

由于互斥锁仅由使用者线程使用，因此实际上可能是该线程的局部变量(只要您只有一个使用者，并且有多个使用者，则它们都需要共享同一个互斥锁来等待同一个condvar ).

Since the mutex is only used by the consumer thread it could in fact be local to that thread (as long as you only have one consumer, with more than one they all need to share the same mutex to wait on the same condvar).

这篇关于C ++ 11非阻塞生产者/消费者的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！