10.3 Bounded Producer-Consumer Queue

Worker threads communicate by passing work items through a shared queue. An unbounded queue can grow without limit under a fast producer; a bounded queue applies backpressure, slow consumers naturally throttle producers. This pattern appears in thread pools, audio pipelines, and capstone 10.4.

* Producers call push when data is ready.

* Consumers call pop when they can process another item.

* The bounded capacity forces producers to block when the buffer is full instead of allocating without end.

Lesson 9.2: protect std::queue mutations with a std::mutex. Lesson 9.3: use two condition variables:

* notEmpty: consumers wait here when the queue is empty.

* notFull: producers wait here when the queue is at capacity.

std::condition_variable::wait requires std::unique_lock (not lock_guard) because it must unlock the mutex while sleeping and re-lock before returning.

Always call wait(lock, predicate) so the thread goes back to sleep if it woke without a state change. Lesson 9.3 explained spurious wakeups, your predicate checks !queue.empty() or queue.size() < maxCapacity.

After push, notify notEmpty so a waiting consumer can proceed. After pop, notify notFull so a blocked producer can proceed. notify_one() is enough when any single waiter may proceed; use notify_all() only when you must wake every waiter (e.g. shutdown).

The template spawns a producer and consumer with deliberate timing. Run under ThreadSanitizer (Lesson 8.9) when your toolchain supports it to catch missing locks or data races.

Implement push and pop so that:

1. All queue access is mutex-protected.

2. push blocks when the queue is full.

3. pop blocks when the queue is empty.

4. Waiters use predicates and wake the opposite side after state changes.

Complete the // TODO sections in the playground template.

#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <cassert>

template <typename T>
class ThreadSafeQueue {
private:
    std::queue<T> dataQueue;
    size_t maxCapacity;
    std::mutex mtx;
    std::condition_variable notFull;
    std::condition_variable notEmpty;
public:
    ThreadSafeQueue(size_t cap) : maxCapacity{cap} {}

    // TODO: Block when full, push, notify consumers
    void push(const T& val) {
        // Your code here
    }

    // TODO: Block when empty, pop, notify producers
    T pop() {
        // Your code here
        return T{};
    }

    size_t size() {
        std::lock_guard<std::mutex> lock(mtx);
        return dataQueue.size();
    }
};

int main() {
    ThreadSafeQueue<int> q{2};

    std::thread producer([&]() {
        q.push(10);
        q.push(20);
        q.push(30);
    });

    std::thread consumer([&]() {
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
        int v1 = q.pop();
        int v2 = q.pop();
        int v3 = q.pop();
        assert(v1 == 10);
        assert(v2 == 20);
        assert(v3 == 30);
    });

    producer.join();
    consumer.join();

    std::cout << "All ThreadSafeQueue thread checks completed!\n";
    return 0;
}

If you get stuck, expand the reference solution below.

// - unique_lock is required because condition_variable.wait() needs to unlock/lock the mutex
// - wait() accepts a lambda predicate to protect against spurious wakeups
// - notify_one() wakes up one blocked thread from sleep

void push(const T& val) {
    std::unique_lock<std::mutex> lock(mtx);
    notFull.wait(lock, [this]() { return dataQueue.size() < maxCapacity; });
    dataQueue.push(val);
    notEmpty.notify_one();
}

T pop() {
    std::unique_lock<std::mutex> lock(mtx);
    notEmpty.wait(lock, [this]() { return !dataQueue.empty(); });
    T val = dataQueue.front();
    dataQueue.pop();
    notFull.notify_one();
    return val;
}