TL;DR: A bounded queue makes producers wait when consumers fall behind. Define how workers stop, how item failures are reported and what happens if producing the input fails. For the sentinel pattern, enqueue one stop marker per consumer and acknowledge every dequeued item, including markers. Keep expensive work outside the lock used to collect results.
How to approach it
State whether the pipeline drains accepted work after an error or cancels it. Choose a positive queue bound and worker count, implement that policy, then test multiple workers and both producer and consumer failures. Explain which memory is bounded and which is still retained elsewhere.
A strong answer
A preprocessing producer can read items faster than workers transform them. An unbounded queue stores the difference until memory is exhausted. A bounded queue can stop that backlog, but it does not automatically bound worker allocations, the input collection or the results retained by the caller.
Producer-Consumer Pipelines explains the pattern. Batching Queues and Backpressure connects the same waiting behavior to downstream services.
Implement one explicit shutdown policy
This finite-batch example continues after ordinary item exceptions. If the input iterator fails, it reports that error, drains already queued work and stops every worker. work must terminate and use ordinary Exception failures; there is no hard cancellation of a hung thread.
import queue
import threading
def run_pipeline(items, work, n_consumers=4, maxsize=8):
if type(n_consumers) is not int or n_consumers <= 0:
raise ValueError("n_consumers must be positive")
if type(maxsize) is not int or maxsize <= 0:
raise ValueError("maxsize must be positive")
q, stop = queue.Queue(maxsize=maxsize), object()
results, errors = [], []
lock = threading.Lock()
def consumer():
while True:
entry = q.get()
try:
if entry is stop:
return
index, item = entry
try:
value = work(item) # outside the result-collection lock
except Exception as exc:
with lock:
errors.append({"phase": "item", "index": index,
"error": repr(exc)})
else:
with lock:
results.append((index, value))
finally:
q.task_done()
workers = [threading.Thread(target=consumer) for _ in range(n_consumers)]
for worker in workers:
worker.start()
try:
for index, item in enumerate(items):
q.put((index, item)) # caller is the producer; blocks when full
except Exception as exc:
with lock:
errors.append({"phase": "producer", "index": None,
"error": repr(exc)})
finally:
for _ in workers:
q.put(stop)
q.join() # all items and stop markers have been acknowledged
for worker in workers:
worker.join()
return sorted(results), errors
Each worker exits on its first marker, so one marker cannot stop all the others. A FIFO queue puts the markers after accepted input. task_done() runs even when an item fails or a marker causes return, keeping the completion counter balanced. The Python queue documentation describes this acknowledgment contract.
Only appending shared results and errors takes the lock. Putting work(item) inside that lock would serialize the expensive part even with several threads. Results are sorted by input index before return; errors retain the order in which they were recorded.
This example uses one producer and four consumers, with ordinary item errors recorded.
- Producer queues items until done or failed
A full queue blocks new puts. Any input exception is recorded.
- Producer adds four stop markers
Markers follow accepted items. Each worker exits after consuming one marker.
- Acknowledge every dequeued entry
Items and markers call task_done. Join waits for acknowledgment and thread exit.
Test failure and concurrency paths
def transform(x):
if x == 13:
raise ValueError("bad item 13")
return x * 2
results, errors = run_pipeline(range(50), transform, n_consumers=4, maxsize=5)
assert len(results) == 49 and len(errors) == 1
assert errors[0]["phase"] == "item" and errors[0]["index"] == 13
assert results == [(i, i * 2) for i in range(50) if i != 13]
def broken_input():
yield 4
yield 5
raise ValueError("input failed")
results, errors = run_pipeline(broken_input(), lambda x: x + 1, maxsize=1)
assert results == [(0, 5), (1, 6)]
assert len(errors) == 1 and errors[0]["phase"] == "producer"
# Both workers must enter work concurrently to pass the barrier.
barrier = threading.Barrier(2, timeout=5)
def overlap(x):
barrier.wait()
return x
results, errors = run_pipeline([1, 2], overlap, n_consumers=2, maxsize=1)
assert results == [(0, 1), (1, 2)] and not errors
The barrier check detects accidental serialization of work. It tests coordination, not CPU speedup. Also test empty input, more consumers than items, several item failures and invalid bounds. A high-water mark alone does not prove backpressure blocked the producer; measure time waiting in put as well as service throughput.
Size the queue for memory and waiting time
Suppose production averages 3,000 items/s while consumption handles 1,000 items/s, and each queued payload is 10,000 bytes:
Unbounded backlog growth = 3,000 - 1,000 = 2,000 items/s
Payload growth = 2,000 × 10,000 = 20,000,000 bytes/s = 20 MB/s
16 GB of payload headroom / 20 MB/s = 800 seconds ≈ 13.3 minutes
A 100,000-item bound fills in 100,000 / 2,000 = 50 seconds
Its payload alone can occupy 100,000 × 10,000 = 1 GB
sanity: a large finite bound still applies backpressure; it may apply it too late.
These are fluid estimates excluding object overhead and other allocations. A bound of eight limits queued entries to eight, not total process memory. Up to four workers can each hold an item, the producer may hold the next item, and this example retains all results and errors until return. For an unbounded stream, use a suitably bounded output path or durable sink instead of growing lists.
Choose the bound from item sizes, acceptable delay and burst tolerance. “A few times the consumer count” can be a starting experiment, not a universal rule. A queue that stayed small in one run still protects against a future consumer slowdown.
Concurrency in Python, Go and C++ discusses execution choices. Threads suit blocking I/O and extensions that release the GIL in a conventional GIL-enabled CPython build. Pure Python CPU work may need processes or an appropriate free-threaded runtime; changing the runtime does not remove shared-state requirements. Async I/O needs an async queue and cooperative work rather than blocking calls in the event loop.
What interviewers probe next
- “How would you stop on the first error?” Add a coordinated cancellation policy with interruptible puts/gets and a way to drain or account for accepted work; a flag alone cannot wake a blocked call.
- “What if there are several producers?” Send stop markers only after all producers finish, using a coordinator or a checked completion count.
- “What if a worker never returns?” This implementation can hang; use operation timeouts and a suitable supervisor or process boundary for stronger isolation.
Common mistakes
- Enqueuing one marker for several consumers.
- Holding the result lock while doing the expensive work.
- Losing a producer exception while reporting partial results as complete.
- Claiming bounded input backlog makes an accumulating-results pipeline memory-bounded.
Key takeaways
- The example drains accepted work and reports both item and producer failures.
- A 100,000-item queue still fills after about 50 seconds under the stated rates.
- Bound all retained stages when the input is a stream, not only the input queue.