Step 6 of 6

Condition variables and producer/consumer

A very common shape for concurrent programs is producer/consumer: some threads produce work (requests arriving, files to process, frames to encode), and others consume it. A queue sits between them. Consumers need to wait when the queue is empty, without burning a core in a loop that keeps checking.

A std::condition_variable (from <condition_variable>) lets a thread sleep until another thread says something changed. It always works together with a mutex and a condition, like "the queue isn't empty":

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

class BlockingQueue {
public:
    void push(int v) {
        {
            std::lock_guard<std::mutex> lock(m_);
            q_.push(v);
        }
        cv_.notify_one();                  // wake one sleeping consumer
    }
    void close() {
        {
            std::lock_guard<std::mutex> lock(m_);
            closed_ = true;
        }
        cv_.notify_all();                  // wake everyone so they can finish
    }
    // Waits for an item. Returns nullopt once the queue is closed and empty.
    std::optional<int> pop() {
        std::unique_lock<std::mutex> lock(m_);
        cv_.wait(lock, [&] { return !q_.empty() || closed_; });
        if (q_.empty()) return std::nullopt;
        int v = q_.front();
        q_.pop();
        return v;
    }

private:
    std::mutex m_;
    std::condition_variable cv_;
    std::queue<int> q_;
    bool closed_ = false;
};

int main() {
    BlockingQueue jobs;
    long long total = 0;
    int handled = 0;
    std::thread consumer([&] {
        while (auto job = jobs.pop()) {    // sleeps whenever the queue is empty
            total += *job;
            handled++;
        }
    });
    for (int i = 1; i <= 100; i++) jobs.push(i);
    jobs.close();
    consumer.join();
    std::cout << handled << " jobs, total " << total << "\n";
}

100 jobs, total 5050

How it works

cv_.wait(lock, condition) checks the condition while holding the lock. If it's false, it releases the mutex and sleeps in one step, so a producer can get in and push. When woken, it re-locks the mutex and checks the condition again, returning only once it's true.
Always wait with a condition (the lambda). Threads can wake up without any notification (a "spurious wakeup"), and a notification sent before the consumer started waiting would otherwise be missed. The condition covers both.
wait needs a std::unique_lock, not a lock_guard, because it has to unlock and re-lock the mutex.
The producer changes the state under the lock, then notifies. Notifying after releasing the lock (the extra { } block) avoids waking a consumer that would immediately block on the still-held mutex.
close() is how everything shuts down cleanly: consumers drain what's left, then pop() returns nullopt and their loops end. Forgetting it leaves the consumer asleep forever and join() never returns.

Bounded queues

This queue can grow without limit. If producers are faster than consumers, memory fills up. Real systems use a bounded queue: when it's full, push waits too (on a second condition variable, "not full"), which slows the producers down to the consumers' speed. That's called backpressure.

The storage inside a bounded queue is usually a ring buffer: a fixed array with a head index where the oldest item is and a count of items. Pushing writes at (head + count) % capacity; popping reads at head and advances it with % capacity, wrapping around the end of the array. Nothing is ever shifted or reallocated, so every operation is O(1). The blocking queue is then this ring buffer, plus a mutex around every operation, plus the two condition variables.

Your turn: write the ring buffer. push stores a value and returns true, or returns false if the buffer is full. pop removes and returns the oldest value, or std::nullopt if it's empty. Add size, empty and full.

Previous: Deadlock and lock ordering