Problem 83 · Concurrency & Python

Bounded Blocking Queue

MediumClassic DSA & systems

01Problem

Implement a thread-safe bounded queue where put blocks while full and get blocks while empty, with optional timeouts.

02Key insight

Use one lock with two Condition variables (not_full, not_empty) and always wait in a predicate loop (wait_for) to handle spurious wake-ups. Notify the opposite condition after each state change.

03Solution

solution.pypython36 lines

import threading
from collections import deque
from typing import Generic, Optional, TypeVar

T = TypeVar("T")


class BoundedBlockingQueue(Generic[T]):
    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            raise ValueError("capacity must be positive")
        self._capacity = capacity
        self._items: deque[T] = deque()
        lock = threading.Lock()
        self._not_full = threading.Condition(lock)
        self._not_empty = threading.Condition(lock)

    def put(self, item: T, timeout: Optional[float] = None) -> bool:
        with self._not_full:
            if not self._not_full.wait_for(lambda: len(self._items) < self._capacity, timeout):
                return False
            self._items.append(item)
            self._not_empty.notify()
            return True

    def get(self, timeout: Optional[float] = None) -> T:
        with self._not_empty:
            if not self._not_empty.wait_for(lambda: len(self._items) > 0, timeout):
                raise TimeoutError("queue empty")
            item = self._items.popleft()
            self._not_full.notify()
            return item

    def __len__(self) -> int:
        with self._not_full:
            return len(self._items)

Try it first.

TimeO(1) per op

SpaceO(capacity)

tests.py · 3 passing

tests.pypython19 lines

import threading
q = BoundedBlockingQueue(3)
out = []
def consumer():
    for _ in range(500):
        out.append(q.get(timeout=2))
t = threading.Thread(target=consumer)
t.start()
for i in range(500):
    assert q.put(i, timeout=2)
t.join()
assert out == list(range(500))
full = BoundedBlockingQueue(1)
full.put("a")
assert full.put("b", timeout=0.01) is False
try:
    BoundedBlockingQueue(1).get(timeout=0.01); assert False
except TimeoutError:
    pass

04Staff-level follow-up

Multiple producers and consumers - is notify() enough? Yes here, because each state change can satisfy at most one waiter of the opposite kind; use notify_all only if waiters have heterogeneous predicates (e.g. put_many).