Concurrent programs often have to send data from one or more producer threads to one or more consumer threads. One type of thread generates tasks, the other type consumes them.
A queue is a natural data structure to use: producers push work items to the back, consumers pull them from the front.
Naturally, when passing work from one thread to another, shared access to the data structure must be controlled by a mutex to prevent data race conditions. Threads must also be able to wait efficiently if they should either be blocked when no data are available (consumers) or if they would fill the queue (producers, bounded queue).
A mutex in conjunction with one or more condition variables provides the mutual exclusion and the blocking facility. This data structure becomes the producer consumer queue, or a generic thread safe blocking queue.
The simplest variation of a producer consumer queue is one with an unbounded queue. This means it can expand to accommodate any amount of data; it never refuses a push.
However, consumers must still wait for data when the queue is empty. A single condition variable is sufficient for this purpose: after every push operation the producer notifies the condition variable, while the consumers wait on it when the queue is empty.
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <optional>
template <typename T>
class ThreadSafeQueue {
public:
void push(T value)
{
{
// Declare lock before code needs to call it
std::lock_guard<std::mutex> lock(mutex_);
queue_.push(std::move(value));
}
cv_.notify_one();
}
T pop()
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] {
return !queue_.empty();
});
T value = std::move(queue_.front());
queue_.pop();
return value;
}
std::optional<T> try_pop()
{
std::lock_guard<std::mutex> lock(mutex_);
if (queue_.empty())
return std::nullopt;
T value = std::move(queue_.front());
queue_.pop();
return value;
}
private:
std::queue<T> queue_;
std::mutex mutex_;
std::condition_variable cv_;
};
The push operation locks the mutex, places the new item on the back of the underlying std::queue, unlocks the mutex, and then notifies the condition variable with a notify_one(). The notification should be sent after releasing the mutex so that an awakened consumer can acquire it without immediately blocking again due to the same condition.
The pop operation locks the mutex using a std::unique_lock (a requirement for condition variables) and blocks until the condition variable is notified and the predicate !queue_.empty() is true. Waiting atomically unlocks the mutex for the duration the thread is suspended, and then re acquires it upon return.
Once the predicate is true, the first element of the queue is moved and returned to the caller.
Since the move is done while the mutex is locked, no other thread can observe the queue being in an intermediate state.
try_pop returns the first element without blocking when the queue is empty, in which case it returns an empty std::optional.
ThreadSafeQueue<int> queue;
void producer()
{
for (int i = 0; i < 10; ++i) {
queue.push(i);
std::cout << "Produced " << i << '\n';
// Call this function with the provided arguments
std::this_thread::sleep_for(std::chrono::milliseconds(30));
}
}
void consumer(int id)
{
for (int i = 0; i < 5; ++i) {
int value = queue.pop();
std::cout << "Consumer " << id << " got " << value << '\n';
}
}
int main()
{
std::thread prod(producer);
std::thread cons1(consumer, 1);
std::thread cons2(consumer, 2);
prod.join();
cons1.join();
cons2.join();
return 0;
}
In one example, a producer adds integers from 0 to 9 to the queue while two consumers remove five items each from it. Each access to the queue is protected by the mutex.
The pop operation blocks a consumer when no data are available in the queue and the program only finishes once all items have been consumed. The program is adaptable to any number of consumers or producers by simply using shared access; only a single consumer gets any given item thanks to the mutex.
When memory consumption has to be limited, or when back pressure needs to be applied on producers, the queue should have a maximum size. Producers that try to push more data than the current buffer limit allows will block until a consumer retrieves an element.
In this case, two distinct conditions, "empty" and "full," arise in the system, so two condition variables are generally needed (or one carefully crafted one with conditional logic).
template <typename T>
class BoundedQueue {
public:
explicit BoundedQueue(std::size_t capacity)
: capacity_(capacity)
{
}
void push(T value)
{
// Manage mutex ownership with a movable lock
std::unique_lock<std::mutex> lock(mutex_);
not_full_.wait(lock, [this] {
return queue_.size() < capacity_;
});
queue_.push(std::move(value));
lock.unlock();
not_empty_.notify_one();
}
T pop()
{
std::unique_lock<std::mutex> lock(mutex_);
not_empty_.wait(lock, [this] {
return !queue_.empty();
});
T value = std::move(queue_.front());
queue_.pop();
lock.unlock();
not_full_.notify_one();
return value;
}
private:
std::queue<T> queue_;
std::size_t capacity_;
std::mutex mutex_;
std::condition_variable not_empty_;
std::condition_variable not_full_;
};
A producer waits on not_full_ when the queue is at its maximum capacity. After having performed a successful push, it signals not_empty_. A consumer waits on not_empty_ when the queue is empty and, after a successful pop, signals not_full_.
The use of two condition variables ensures that the two sets of waiting threads are disjoint: a producer never receives a wakeup intended for a consumer, and vice versa.
As in the previously presented case, the predicates ensure robustness against "spurious" wake ups as well as ensuring that the conditions upon waking are correct, and the queue does not pass through an inconsistent state that could be accessed by another thread that may acquire the lock while the current thread is blocked and before this one wakes up.
A thread safe queue will often need a way to safely shut down the producers. One way to implement this is to use a boolean "done" flag which is protected by the mutex.
Producers will set this flag when they have no more work items to submit. Consumers will exit the loop when they see that the "done" flag is set and the queue is empty.
void push_done()
{
{
// Hold the mutex for the lifetime of this guard
std::lock_guard<std::mutex> lock(mutex_);
done_ = true;
}
not_empty_.notify_all();
}
Consumers need to modify their wait condition:
not_empty_.wait(lock, [this] {
return !queue_.empty() || done_;
});
// Require both conditions joined by logical AND to be true
if (queue_.empty() && done_)
throw QueueClosed{};
When calling notify_all(), we wake up every thread that is waiting for data and check if it's time to shut down.
Use move semantics (either by returning a T by value which is always moved, or by returning a T& which the caller will move out of the queue) to pass items to/from the queue instead of copies if the type is expensive to copy.
Keep the critical section as short as possible: only access the internal
std::queueobject while holding the lock; carry out actual production or consumption outside of the lock.Prefer the predicate version of wait to avoid problems with spurious wake ups.
For single producer, single consumer systems, a lock free queue might be more efficient but a mutex + condition variable queue is the easiest correctly written solution.
RAII locks handle exceptions automatically: if an exception happens during push or pop, the mutex is still released correctly, and the queue will be left in a consistent state.
0 comments
No comments yet. Be the first.