Concurrent queue – C++11

This is an implementation of the classic multiple producer, multiple consumer thread-safe queue concept. Much has been said about this before, but I need a simple, pure C++ version of this for future posts, so I thought I might as well say a few words about it. I had set a variant on this theme as an exercise in a C++11 workshop, and I also encountered a simplified variant of it in a job interview assessment question. It asked to implement a thread safe queue a single producer and a single consumer thread. The interface was minimal and clean:

template <typename T>
struct Queue
{
  // pop an element and return a copy. Block if queue empty.
  T pop();
  //  push an element to the back of the queue.
  void push(const T& item);
};

Let’s spend a moment looking at the advantages and disadvantages of this interface. For reference, we look at a part of the std::queue container adapter. A simplified version, with the methods needed for enqueueing and popping elements, would look something like this:

template <typename T>
class queue
{
  T& front();
  void push( const T& value );
  void pop();
  bool empty() const; // we need this to avoid UB when calling front()
};

We see immediately that we have twice as many methods to achieve the same two operations. There good reasons for this:

• Exception safety: the standard library call to front() does not return return a copy of the element and pop it off the queue because if the resulting copy construction throws an exception, the element is lost: it has been removed from the queue, and the caller has failed to obtain a copy. This would break the strong exception guarantee. We are required to copy the front() element first, and only then pop() it from the queue.
• Non-blocking access: a call to front() is synchronous and must return a valid reference to an element of the queue. If the queue is empty, the call cannot block, and results in undefined behaviour. The empty() method allows us to check before calling front().

These requirements are at odds with concurrent access: we have to carefully synchronise four operations (std::queue has more even methods that would require synchronization, amongst them size() and back()). In fact, the standard library containers and container adapters have not been designed considering thread safety in the presence of race conditions (that is, more than one thread accessing them with at least one of them performing write operations). The Queue interface in the first code sample simplifies the problem by condensing the de-queueing operations into a single returning, blocking pop() member function, at the cost of some exception safety.

There is no clean way to solve this last problem. One option is to modify Queue::pop() to take a reference to an element:

template <typename T>
struct Queue
{
 ...
  void pop(T& firstInQueue);
};

This makes it ugly on the caller side:

Queue<SomeType> q;
...
SomeType elem;
q.pop(elem);

We have a choice between elegance and the strong exception safety guarantee. We can opt for the former, and only provide the basic guarantee, meaning that the invariants of the Queue component are maintained and no resources are leaked. Or we can provide the safe and unsafe methods and let the caller decide. In any case, lets move on from exception safety and onto thread safety.

Since I want to use this queue for examples requiring more than one writer and one reader thread, let’s up the requirements to deal with multi-writer, multi-reader threads. It is a good moment to go over the requirements:

• Concurrent access from multiple writer and reader threads
• pop() blocks if queue is empty
• At least one means to pop an element from queue satisfying the strong exception guarantee
• No user actions should result in deadlock
• Performance: minimize lock contention, polling.

In terms of the implementation, there are three key issues to consider:

1. Only one thread can be pushing an element at any given time
2. Only one thread can be popping an element at any given time
3. Threads must wait for elements to be in the queue before they can pop

We can implement this functionality using a single std::mutex, together with an std::condition_variable, following this excellent example.

The mutex prevents concurrent reads and writes, and the condition variable allows consuming threads to wait for elements to be available in the queue without excessive mutex contention and without using expensive and inefficient polling.

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

template <typename T>
class Queue
{
 ...
 std::queue<T> queue_;
 std::mutex mutex_;
 std::condition_variable cond_;
};

The core of our pop() methods looks like this:

  std::unique_lock<std::mutex> mlock(mutex_); // mutex scope lock
  while (queue_.empty()) // check condition to be safe against spurious wakes
  {
    cond_.wait(mlock); // release lock and go join the waiting thread queue
  }
  // copy front(), pop() and return
}

The method is protected by a mutex, which is locked using scope guard std::unique_lock. If the queue is empty, we wait on the condition variable cond_. This releases the lock to other threads and blocks until we are notified that the condition has been met. In this case, the call to wait() returns and we check the while condition again. This extra check is because conditions may experience spurious wakes: we could be wrongly notified while the queue is still empty.

The condition notification will be provided when an element is pushed into the queue, as shown here in the push() method:

  void push(const T& item)
  {
    std::unique_lock<std::mutex> mlock(mutex_);
    queue_.push(item);
    mlock.unlock();     // unlock before notificiation to minimize mutex contention
    cond_.notify_one(); // notify one waiting thread
  }

We notify with every call to push() to wake up threads that could be waiting eventhough there is data in the queue.

Our final queue looks like this:

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

template <typename T>
class Queue
{
 public:

  T pop()
  {
    std::unique_lock<std::mutex> mlock(mutex_);
    while (queue_.empty())
    {
      cond_.wait(mlock);
    }
    auto item = queue_.front();
    queue_.pop();
    return item;
  }

  void pop(T& item)
  {
    std::unique_lock<std::mutex> mlock(mutex_);
    while (queue_.empty())
    {
      cond_.wait(mlock);
    }
    item = queue_.front();
    queue_.pop();
  }

  void push(const T& item)
  {
    std::unique_lock<std::mutex> mlock(mutex_);
    queue_.push(item);
    mlock.unlock();
    cond_.notify_one();
  }

  void push(T&& item)
  {
    std::unique_lock<std::mutex> mlock(mutex_);
    queue_.push(std::move(item));
    mlock.unlock();
    cond_.notify_one();
  }

 private:
  std::queue<T> queue_;
  std::mutex mutex_;
  std::condition_variable cond_;
};

where we have provided the two versions of the pop() method discussed earlier, plus an of push() taking an rvalue reference T&& to benefit from move semantics. We can think of a few useful extensions to this queue class. For example, it would be nice to have a timeout pop() facility, and it could be wise to restrict the queue to a maximum length, beyond which calls to push() would block. These improvements may form the basis of a future post.

Sample code can be found on github.