BoundedBuffer: a blocking queue from one Condition

Build a fixed-capacity blocking FIFO on one threading.Condition — no queue.Queue, no polling loop.

class BoundedBuffer[T]:
    def __init__(self, capacity: int) -> None: ...
    def put(self, item: T, timeout: float | None = None) -> None: ...
    def get(self, timeout: float | None = None) -> T: ...

• put appends, waiting for room if the buffer is full. If timeout elapses first, raise TimeoutError.
• get removes and returns the oldest item, waiting for one to arrive. If timeout elapses first, raise TimeoutError.
• timeout=None means wait forever.
• Store the condition as self._cond. The report checks it is really a threading.Condition, which is how “you built this yourself” is enforced.

What the report proves

The driver runs one producer and consumers consumers, then probes the full and empty edges directly.

• count / checksum — every item arrives exactly once. Nothing lost to a missed wakeup, nothing duplicated by two consumers popping the same slot.
• in_order — each consumer’s own subsequence is increasing. That is what FIFO means when several threads drain one queue: the global interleaving is the OS’s business, but no consumer may ever see an older item after a newer one.
• put_timeout / get_timeout — a full put and an empty get raise TimeoutError rather than blocking forever.
• sleeps — the driver replaces time.sleep with a counter for the duration of the run. A correct implementation never calls it. Any polling loop fails here, which is the point of the exercise.

Two mistakes this problem is built to catch

if instead of while. A single check after wait() returns is wrong: wait() can return with the predicate still false, because another thread was woken first and took the slot. Condition.wait_for(predicate, timeout) is that loop written correctly, including the deadline arithmetic that a hand-rolled while cond.wait(timeout) gets wrong (each wait(timeout) restarts the clock, so a loop of them can wait arbitrarily long). It returns the predicate’s final value — False means timed out, and that return is what you convert into TimeoutError.

notify() instead of notify_all(). One condition is serving two predicates here: “not full” and “not empty”. notify() wakes one arbitrary waiter, which may be a producer that still has no room — it re-checks, sleeps again, and the consumer that could have made progress was never woken. The system stalls with no error at all. Test case 2 (capacity 1, four consumers) is built to hit this.

Where the type system earns its keep

BoundedBuffer[T] is generic with PEP 695 syntax, so BoundedBuffer[int]() gives you get() -> int and rejects put("x"). Under --strict the internal list needs its own annotation (self._items: list[T] = []) — an empty list has no inferable element type. And self._cond: Final[threading.Condition] states that the condition is created once and never rebound, which is the invariant the whole class rests on.