79

我正在处理的一个项目使用多个线程来处理一组文件。每个线程都可以将文件添加到要处理的文件列表中，因此我将（我认为是）一个线程安全队列放在一起。相关部分如下：

// qMutex is a std::mutex intended to guard the queue
// populatedNotifier is a std::condition_variable intended to
//                   notify waiting threads of a new item in the queue

void FileQueue::enqueue(std::string&& filename)
{
    std::lock_guard<std::mutex> lock(qMutex);
    q.push(std::move(filename));

    // Notify anyone waiting for additional files that more have arrived
    populatedNotifier.notify_one();
}

std::string FileQueue::dequeue(const std::chrono::milliseconds& timeout)
{
    std::unique_lock<std::mutex> lock(qMutex);
    if (q.empty()) {
        if (populatedNotifier.wait_for(lock, timeout) == std::cv_status::no_timeout) {
            std::string ret = q.front();
            q.pop();
            return ret;
        }
        else {
            return std::string();
        }
    }
    else {
        std::string ret = q.front();
        q.pop();
        return ret;
    }
}

但是，我偶尔会在if (...wait_for(lock, timeout) == std::cv_status::no_timeout) { }块内发生段错误，并且 gdb 中的检查表明由于队列为空而正在发生段错误。这怎么可能？据我了解，wait_for只有在收到通知时才会返回cv_status::no_timeout，并且这应该只在FileQueue::enqueue刚刚将新项目推送到队列后才会发生。

4

8 回答 8

74

最好使条件（由您的条件变量监控）成为 while-loop: 的逆条件 while(!some_condition)。在这个循环中，如果条件失败，您将进入睡眠状态，从而触发循环体。

这样，如果您的线程被唤醒（可能是虚假的），您的循环仍将在继续之前检查条件。将条件视为感兴趣的状态，并将条件变量更多地视为来自系统的信号，表明该状态可能已准备好。循环将完成实际确认它是真实的繁重工作，如果不是，则进入睡眠状态。

我刚刚为异步队列编写了一个模板，希望对您有所帮助。这q.empty()是我们想要的相反条件：队列中有东西。所以它作为while循环的检查。

#ifndef SAFE_QUEUE
#define SAFE_QUEUE

#include <queue>
#include <mutex>
#include <condition_variable>

// A threadsafe-queue.
template <class T>
class SafeQueue
{
public:
  SafeQueue(void)
    : q()
    , m()
    , c()
  {}

  ~SafeQueue(void)
  {}

  // Add an element to the queue.
  void enqueue(T t)
  {
    std::lock_guard<std::mutex> lock(m);
    q.push(t);
    c.notify_one();
  }

  // Get the "front"-element.
  // If the queue is empty, wait till a element is avaiable.
  T dequeue(void)
  {
    std::unique_lock<std::mutex> lock(m);
    while(q.empty())
    {
      // release lock as long as the wait and reaquire it afterwards.
      c.wait(lock);
    }
    T val = q.front();
    q.pop();
    return val;
  }

private:
  std::queue<T> q;
  mutable std::mutex m;
  std::condition_variable c;
};
#endif
于 2013-04-18T06:00:48.270 回答
35

根据标准condition_variables允许虚假唤醒，即使事件没有发生。在虚假唤醒的情况下，它会返回cv_status::no_timeout（因为它是唤醒而不是超时），即使它没有被通知。正确的解决方案当然是在继续之前检查唤醒是否真的合法。

详细信息在标准§30.5.1 [thread.condition.condvar] 中指定：

— 当调用 notify_one()、调用 notify_all()、abs_time 指定的绝对超时 (30.2.4) 到期或虚假发出信号时，该函数将解除阻塞。

...

返回：如果 abs_time 指定的绝对超时 (30.2.4) 已过期，则返回 cv_status::timeout，否则为 cv_status::no_timeout。

于 2013-03-07T18:01:33.463 回答
18

这可能是您应该这样做的方式：

void push(std::string&& filename)
{
    {
        std::lock_guard<std::mutex> lock(qMutex);

        q.push(std::move(filename));
    }

    populatedNotifier.notify_one();
}

bool try_pop(std::string& filename, std::chrono::milliseconds timeout)
{
    std::unique_lock<std::mutex> lock(qMutex);

    if(!populatedNotifier.wait_for(lock, timeout, [this] { return !q.empty(); }))
        return false;

    filename = std::move(q.front());
    q.pop();

    return true;    
}
于 2013-03-07T18:08:37.297 回答
13

除了接受的答案之外，我想说实现正确的多生产者/多消费者队列很困难（不过，自 C++11 以来更容易）

我建议您尝试（非常好的）无锁 boost 库，“队列”结构将做您想做的事，具有无等待/无锁保证并且不需要 C++11 编译器。

我现在添加这个答案是因为无锁库对于提升来说是相当新的（我相信从 1.53 开始）

于 2014-05-20T21:09:36.017 回答
5

我会将您的出列函数重写为：

std::string FileQueue::dequeue(const std::chrono::milliseconds& timeout)
{
    std::unique_lock<std::mutex> lock(qMutex);
    while(q.empty()) {
        if (populatedNotifier.wait_for(lock, timeout) == std::cv_status::timeout ) 
           return std::string();
    }
    std::string ret = q.front();
    q.pop();
    return ret;
}

它更短，并且没有像您那样重复的代码。仅发出它可能会等待更长的超时时间。为防止您需要记住循环之前的开始时间，请检查超时并相应地调整等待时间。或指定等待条件的绝对时间。

于 2013-03-07T18:10:11.570 回答
1

这个案例也有 GLib 解决方案，我还没试过，但我相信这是一个很好的解决方案。 https://developer.gnome.org/glib/2.36/glib-Asynchronous-Queues.html#g-async-queue-new

于 2015-02-17T19:40:57.303 回答
1

BlockingCollection是一个 C++11 线程安全的集合类，它提供对队列、堆栈和优先级容器的支持。它处理您描述的“空”队列场景。以及“完整”队列。

于 2018-10-15T17:01:45.283 回答
0

你可能喜欢 lfqueue，https://github.com/Taymindis/lfqueue。它是无锁并发队列。我目前正在使用它来消耗来自多个来电的队列，并且像一个魅力一样工作。

于 2018-07-17T23:48:58.093 回答