A thread-safe queue lets one thread add work while other threads retrieve it without coordinating around the queue themselves. Encapsulating the mutex and condition variable keeps synchronization at the point where data changes hands. A waiting consumer can sleep until an item is available rather than repeatedly checking whether the queue is empty. section 4.1.1 section 6.2.2
Its interface differs from std::queue because retrieval must account for an empty queue and concurrent access:
- push() adds an item under the queue’s lock and notifies a waiting thread. section 4.1.2
- try_pop() does not wait. One overload writes into a reference and returns success or failure; another returns a shared pointer that is null if no item was retrieved. section 4.1.2
- wait_and_pop() waits for a nonempty queue, then retrieves and removes an item while it remains protected by the lock. It also has reference and shared-pointer overloads. section 4.1.2
Listing 4.4. Extracting push() and wait_and_pop() from listing 4.1
#include <queue>
#include <mutex>
#include <condition_variable>
template<typename T>
class threadsafe_queue
{
private:
std::mutex mut;
std::queue<T> data_queue;
std::condition_variable data_cond;
public:
void push(T new_value)
{
std::lock_guard<std::mutex> lk(mut);
data_queue.push(new_value);
data_cond.notify_one();
}
void wait_and_pop(T& value)
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk,[this]{return !data_queue.empty();});
value=data_queue.front();
data_queue.pop();
}
};
threadsafe_queue<data_chunk> data_queue; #1
void data_preparation_thread()
{
while(more_data_to_prepare())
{
data_chunk const data=prepare_data();
data_queue.push(data); #2
}
}
void data_processing_thread()
{
while(true)
{
data_chunk data;
data_queue.wait_and_pop(data); #3
process(data);
if(is_last_chunk(data))
break;
}
}
!@%STYLE%@!
{"css":"{\"css\": \"font-weight: bold;\"}","target":"[[{\"line\":25,\"ch\":46},{\"line\":25,\"ch\":48}],[{\"line\":31,\"ch\":46},{\"line\":31,\"ch\":48}],[{\"line\":39,\"ch\":46},{\"line\":39,\"ch\":48}]]"}
!@%STYLE%@!
The compact implementation shows the central pairing: push() changes the underlying std::queue and calls notify_one(); wait_and_pop() uses a predicate-based condition-variable wait before accessing front() and pop(). The mutex and condition variable belong to the queue object. section 4.1.2
The predicate is important: waking does not by itself mean that this thread can take an item. Waiting for !data_queue.empty() makes the condition explicit, and the lock still protects retrieval after the wait. A const empty() member must also lock: another thread may hold a non-const reference and modify the same queue. Its mutex is therefore mutable. section 4.1.2 section 6.2.2
With multiple consumers, notify_one() wakes one waiter, not a particular waiter. There is an exception-safety wrinkle: if the awakened consumer throws while constructing a returned shared pointer, other consumers may remain asleep despite an available item. Where that matters, notify_all() is an alternative that wakes all waiters. section 6.2.2
A single mutex is straightforward but serializes pushes and pops. A finer-grained design uses a linked queue with a dummy node and separate head and tail mutexes, allowing much of a pop to run alongside a push. Its empty test compares head with tail; waiting on that condition requires care about which mutex protects each access. In this design, push() finishes its locked update before notifying, so the awakened consumer need not immediately contend with the producer for that lock. section 6.2.3
Listing 6.9. A thread-safe queue with locking and waiting: wait_and_pop()
template<typename T>
class threadsafe_queue
{
private:
node* get_tail()
{
std::lock_guard<std::mutex> tail_lock(tail_mutex);
return tail;
}
std::unique_ptr<node> pop_head() #1
{
std::unique_ptr<node> old_head=std::move(head);
head=std::move(old_head->next);
return old_head;
}
std::unique_lock<std::mutex> wait_for_data() #2
{
std::unique_lock<std::mutex> head_lock(head_mutex);
data_cond.wait(head_lock,[&]{return head.get()!=get_tail();});
return std::move(head_lock); #3
}
std::unique_ptr<node> wait_pop_head()
{
std::unique_lock<std::mutex> head_lock(wait_for_data()); #4
return pop_head();
}
std::unique_ptr<node> wait_pop_head(T& value)
{
std::unique_lock<std::mutex> head_lock(wait_for_data()); #5
value=std::move(*head->data);
return pop_head();
}
public:
std::shared_ptr<T> wait_and_pop()
{
std::unique_ptr<node> const old_head=wait_pop_head();
return old_head->data;
}
void wait_and_pop(T& value)
{
std::unique_ptr<node> const old_head=wait_pop_head(value);
}
};
!@%STYLE%@!
{"css":"{\"css\": \"font-weight: bold;\"}","target":"[[{\"line\":9,\"ch\":65},{\"line\":9,\"ch\":67}],[{\"line\":15,\"ch\":65},{\"line\":15,\"ch\":67}],[{\"line\":19,\"ch\":65},{\"line\":19,\"ch\":67}],[{\"line\":23,\"ch\":65},{\"line\":23,\"ch\":67}],[{\"line\":28,\"ch\":65},{\"line\":28,\"ch\":67}]]"}
!@%STYLE%@!
The fine-grained wait acquires the head lock and tests head against the tail obtained under the tail lock. Returning the held head lock lets the subsequent pop proceed under the same protection. section 6.2.3
