Producer-Consumer With Condition Variables

The bounded buffer is the interview classic of thread coordination for a reason: the correct solution is fifteen lines, and there are at least four famous ways to write fifteen subtly wrong ones (the lost wakeup, the if instead of while, the notify aimed at the wrong side, the shutdown that strands a sleeping consumer). Desks use it as the fastest possible probe of whether you have internalised the condition-variable discipline or just seen it once.

Implement BoundedBuffer, a fixed-capacity FIFO of long long for concurrent producers and consumers:

• explicit BoundedBuffer(std::size_t capacity): at most capacity buffered elements. capacity >= 1.
• bool put(long long value): blocks while the buffer is full and not closed. Returns true once enqueued; returns false (without enqueuing) if the buffer is closed, including when close() arrives while blocked.
• std::optional<long long> take(): blocks while the buffer is empty and not closed. Returns the oldest element; after close(), keeps returning buffered elements until the buffer is drained, then returns std::nullopt to every caller, forever.
• void close(): idempotent. Wakes all blocked producers and consumers so they can observe the shutdown.

Discipline requirements, all graded in interviews even when a judge cannot see them:

• One mutex and two condition variables (space-available and data-available), each waited on inside a predicate loop. A bare wait() guarded by an if is wrong on every platform: spurious wakeups are permitted by the standard, and a woken thread may find another thread took its element first.
• No busy-waiting anywhere: no spinning on a flag, no sleep_for polling, no yield loops. Blocked threads must consume no CPU.
• FIFO exactly: no element lost, duplicated, or reordered.

Examples

[]{
    BoundedBuffer buffer(16);
    std::thread producer([&buffer] {
        for (long long i = 0; i < 100000; ++i) buffer.put(i);
        buffer.close();
    });
    long long count = 0, violations = 0, next = 0;
    while (auto v = buffer.take()) {
        if (*v != next) ++violations;
        ++next;
        ++count;
    }
    producer.join();
    return std::pair<long long, long long>(count, violations);
}()
// (100000, 0)  (100k values squeezed through 16 slots, in exact order)

[]{ BoundedBuffer buffer(8); std::vector<long long> out; out.push_back(buffer.put(5)); out.push_back(buffer.put(6)); out.push_back(buffer.put(7)); buffer.close(); out.push_back(buffer.put(8)); out.push_back(buffer.take().value_or(-1)); out.push_back(buffer.take().value_or(-1)); out.push_back(buffer.take().value_or(-1)); out.push_back(buffer.take().has_value()); return out; }()
// [1, 1, 1, 0, 5, 6, 7, 0]  (close rejects new puts but drains what was accepted)

Constraints

• Up to 3 threads (tests use one producer with one or two consumers) and \(10^5\) items per test; one test forces every element through a 2-slot buffer.
• A take() blocked on an empty buffer when close() fires must wake and return std::nullopt, not sleep forever: fold the closed flag into the wait predicate.
• The judge measures results after all threads join (counts, sums, per-consumer ordering), which are invariant under interleavings. It cannot detect a busy-wait or an if-instead-of-while that this scheduler happened to forgive; the editorial covers the bugs the judge cannot see, and ThreadSanitizer plus code review is how they are caught in practice.