C++11 thread-safe queue

Question

A project I'm working on uses multiple threads to do work on a collection of files. Each thread can add files to the list of files to be processed, so I put together (what I thought was) a thread-safe queue. Relevant portions follow:

// 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;
    }
}

However, I am occasionally segfaulting inside the if (...wait_for(lock, timeout) == std::cv_status::no_timeout) { } block, and inspection in gdb indicates that the segfaults are occurring because the queue is empty. How is this possible? It was my understanding that wait_for only returns cv_status::no_timeout when it has been notified, and this should only happen after FileQueue::enqueue has just pushed a new item to the queue.

Answer 1

According to the standard condition_variables are allowed to wakeup spuriously, even if the event hasn't occured. In case of a spurious wakeup it will return cv_status::no_timeout (since it woke up instead of timing out), even though it hasn't been notified. The correct solution for this is of course to check if the wakeup was actually legit before proceding.

The details are specified in the standard §30.5.1 [thread.condition.condvar]:

—The function will unblock when signaled by a call to notify_one(), a call to notify_all(), expiration of the absolute timeout (30.2.4) specified by abs_time, or spuriously.

...

Returns: cv_status::timeout if the absolute timeout (30.2.4) specifiedby abs_time expired, other-ise cv_status::no_timeout.

Answer 2

It is best to make the condition (monitored by your condition variable) the inverse condition of a while-loop: while(!some_condition). Inside this loop, you go to sleep if your condition fails, triggering the body of the loop.

This way, if your thread is awoken--possibly spuriously--your loop will still check the condition before proceeding. Think of the condition as the state of interest, and think of the condition variable as more of a signal from the system that this state might be ready. The loop will do the heavy lifting of actually confirming that it's true, and going to sleep if it's not.

I just wrote a template for an async queue, hope this helps. Here, q.empty() is the inverse condition of what we want: for the queue to have something in it. So it serves as the check for the while loop.

#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

Answer 3

This is probably how you should do it:

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;    
}

Answer 4

Adding to the accepted answer, I would say that implementing a correct multi producers / multi consumers queue is difficult (easier since C++11, though)

I would suggest you to try the (very good) lock free boost library, the "queue" structure will do what you want, with wait-free/lock-free guarantees and without the need for a C++11 compiler.

I am adding this answer now because the lock-free library is quite new to boost (since 1.53 I believe)

Answer 5

I would rewrite your dequeue function as:

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;
}

It is shorter and does not have duplicate code like your did. Only issue it may wait longer that timeout. To prevent that you would need to remember start time before loop, check for timeout and adjust wait time accordingly. Or specify absolute time on wait condition.

Answer 6

There is also GLib solution for this case, I did not try it yet, but I believe it is a good solution. https://developer.gnome.org/glib/2.36/glib-Asynchronous-Queues.html#g-async-queue-new

Answer 7

BlockingCollection is a C++11 thread safe collection class that provides support for queue, stack and priority containers. It handles the "empty" queue scenario you described. As well as a "full" queue.

Answer 8

You may like lfqueue, https://github.com/Taymindis/lfqueue. It’s lock free concurrent queue. I’m currently using it to consuming the queue from multiple incoming calls and works like a charm.