Concurrent bounded blocking queue

Expected question

"Implement a bounded queue with put(item) (blocks if full) and take() (blocks if empty), safe for multiple producers and consumers."

Variant forms

• "Implement a blocking queue like ArrayBlockingQueue."
• "Add offer/poll with timeouts."
• "How do you avoid lost wakeups with condition variables?"
• "Support fair ordering of waiting producers/consumers?"
• "Implement using locks vs lock-free ring buffer — trade-offs?"
• "Add drainTo(list, max) for batch consumers."
• "What invariants must hold after every put/take?"
• "How do you test for deadlock and race conditions in an interview?"

The question, as it might actually be asked

Implement a bounded queue with put(item) (blocks if full) and take() (blocks if empty), safe for multiple producers and consumers.

How you'd talk while coding

I'd clarify fixed capacity, whether we need timed offer/poll, fair locking, and null policy. Brute is an unbounded list with busy-spin — wrong under load. Correct is a mutex + condition: while full/empty, wait; after mutate, notify. I'd say out loud: "while, not if — spurious wakeups and competing waiters." Tests: capacity-1 second put blocks until take; empty take blocks until put. Staff+ stop line: "Bounded queue is backpressure. Shutdown is a poison pill or closed flag that wakes waiters — I won't sketch Michael-Scott lock-free unless contention is the measured problem."

Where this actually gets asked

Concurrency classic (producer/consumer). Staff+ cares about correct waiting, spurious wakeups, and fairness — not clever lock-free unless prompted.

Problem

Implement a bounded queue with put(item) (blocks if full) and take() (blocks if empty), safe for multiple producers and consumers.

Clarifying questions you should ask first

1. Capacity fixed?
2. Timeout variants needed (offer / poll)?
3. Fair locking required?
4. Null items allowed?

Approach ladder

Reference solution (Python)

from __future__ import annotations
from collections import deque
from threading import Condition
from typing import Deque, Generic, 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._q: Deque[T] = deque()
        self._cv = Condition()

    def put(self, item: T) -> None:
        with self._cv:
            while len(self._q) >= self._capacity:
                self._cv.wait()
            self._q.append(item)
            self._cv.notify_all()

    def take(self) -> T:
        with self._cv:
            while not self._q:
                self._cv.wait()
            item = self._q.popleft()
            self._cv.notify_all()
            return item

    def size(self) -> int:
        with self._cv:
            return len(self._q)

Complexity: O(1) put/take amortized; blocking as required.

Verbal tests to narrate

1. capacity=1: put blocks second producer until take
2. take on empty blocks until put
3. Multi producer/consumer smoke (narrate race freedom via monitor)

Staff+ deep dive

What not to discuss

• Lock-free Michael-Scott queue unless interviewer asks
• Equating this with Kafka (different problem)

What's expected at each level

• Mid-level: Mutex + queue; may miss wait conditions.
• Senior: Correct blocking with conditions.
• Staff+: Explains while-loops, shutdown, backpressure.
• Principal: Ties to real pipeline backpressure and overload control.

Follow-up questions to expect

• "Why a while-loop around wait?" — Spurious wakeups, and another waiter may have taken the slot before you run.
• "How does shutdown work?" — Closed flag or poison contract that wakes everyone safely — don't leave producers hung forever.
• "Lock-free instead?" — Only after measurements show the monitor is the bottleneck.
• "What metrics in production?" — Depth, producer wait time, dropped/rejected offers under overload.

Related

• ../general-system-design/04-distributed-job-scheduler-task-queue.md
• 00-staff-plus-coding-bar.md