Producer-Consumer With Condition Variables

Implement BoundedBuffer, a fixed-capacity FIFO of long long for concurrent producers and consumers. Producers block when it is full, consumers block when it is empty, and close() shuts it down.

The correct solution is short. The usual mistakes are the lost wakeup, an if where a while is needed, a notify aimed at the wrong side, and a shutdown that strands a sleeping consumer. The interface:

• 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.

Requirements. The judge cannot see some of them, but follow all of 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.