DevOps NetworkDevOpsNetwork
Join Free
Practical infra coding

Producer/consumer log pipeline with a bounded queue

mediumConcurrency basics

Problem statement

A log shipper reads lines from a file much faster than it can parse them. If the reader simply loads everything into memory, a 50 GB log takes the host down. Build a pipeline with one producer (the main thread here) that reads lines and three consumer threads that parse them, connected by a queue.Queue(maxsize=5).

  • When the queue is full, the producer must wait (back-pressure), so memory stays bounded however big the input is.
  • Each line looks like 2026-09-24T10:03:00Z api INFO request handled. Consumers count lines per (service, level).
  • Malformed lines (fewer than three fields) are counted as skipped; they must not kill a consumer thread.
  • When the input ends, every consumer must stop cleanly, and the program must not hang. Use a sentinel value.
  • Print counts sorted by service and level, the parsed/skipped totals, and whether the queue ever grew beyond its limit.

The input comes from a generator standing in for the file: 60 lines, service ["api", "web", "worker"][i % 3], level ERROR when i % 10 == 0 else INFO, and every line with i % 25 == 24 replaced by garbage-line-without-fields.

Examples

Example 1

Input: python solution.py

Output: api ERROR 2 api INFO 17 web ERROR 2 web INFO 17 worker ERROR 2 worker INFO 18 parsed 58 lines, skipped 2 malformed queue never held more than 5: True

Explanation: Lines 24 and 49 are garbage. Line 24 would have been api and line 49 web, so worker, which lost nothing, has one more INFO line than the others. Lines 0, 30 (api), 10, 40 (web) and 20, 50 (worker) are errors.

Hints

Approach

Optimal

  1. Bounded queue. queue.Queue(maxsize=5) is thread-safe and blocks the producer in put() when it holds 5 items. The producer can never get more than 5 lines ahead of the consumers, so memory use is constant whatever the file size. The script records the largest qsize() it saw to show the bound held.
  2. Shutdown with sentinels. After the last line the producer puts one STOP object per consumer. A consumer that takes STOP returns. Using a unique object() rather than None or an empty string means no real input can be mistaken for the signal. Because each consumer takes exactly one sentinel, all of them stop.
  3. Failures stay local. Parsing a malformed line raises ValueError on the unpacking; it is recorded and the loop continues. The outer finally hands the consumer's partial Counter back even if something unexpected happens, and task_done() in the inner finally keeps the queue's bookkeeping right, so q.join() would still work.
  4. No shared mutable state during the work. Each consumer counts into its own Counter. The main thread adds them together after join, so the counting itself needs no lock.
  5. No silent hangs. join(timeout=10) plus an is_alive() check turns a stuck consumer into a clear error instead of a process that never exits.
  6. Deterministic output. Which consumer handled which line differs every run, but the merged totals do not, and they are printed sorted.

n is the number of lines and k the number of (service, level) pairs.

ComplexityTime O(n)Space O(queue size + k)
Python
import queue
import threading
import time
from collections import Counter
QUEUE_SIZE = 5
N_CONSUMERS = 3
STOP = object() # sentinel: "no more work"
def fake_log_lines():
# Stands in for tailing a big log file: 60 lines, a few of them malformed.
for i in range(60):
if i % 25 == 24:
yield "garbage-line-without-fields"
else:
svc = ["api", "web", "worker"][i % 3]
level = "ERROR" if i % 10 == 0 else "INFO"
yield f"2026-09-24T10:{i:02d}:00Z {svc} {level} request handled"
def producer(q, stats):
for line in fake_log_lines():
q.put(line) # blocks while the queue is full: back-pressure
stats["max_qsize"] = max(stats["max_qsize"], q.qsize())
for _ in range(N_CONSUMERS):
q.put(STOP) # one sentinel per consumer
def consumer(q, results, errors):
local = Counter() # no shared state while working
try:
while True:
item = q.get()
try:
if item is STOP:
return
time.sleep(0.001) # pretend parsing takes a moment
_, svc, level, *_ = item.split()
local[(svc, level)] += 1
except ValueError:
errors.append(item) # list.append is thread-safe in CPython
finally:
q.task_done()
finally:
results.append(local) # hand the partial result back even on failure
q = queue.Queue(maxsize=QUEUE_SIZE)
stats = {"max_qsize": 0}
results, errors = [], []
consumers = [threading.Thread(target=consumer, args=(q, results, errors)) for _ in range(N_CONSUMERS)]
for t in consumers:
t.start()
producer(q, stats)
for t in consumers:
t.join(timeout=10)
if t.is_alive():
raise SystemExit("a consumer is stuck")
total = sum(results, Counter())
for (svc, level), n in sorted(total.items()):
print(f"{svc:<7}{level:<6}{n}")
print(f"parsed {sum(total.values())} lines, skipped {len(errors)} malformed")
print(f"queue never held more than {QUEUE_SIZE}: {stats['max_qsize'] <= QUEUE_SIZE}")

Follow-up questions

  • Add a second stage: parsed records go through another bounded queue to a single writer thread.
  • Handle Ctrl+C: stop reading, let the consumers drain what is queued, then exit.
  • The consumer occasionally hangs on one record. Add a per-record timeout without killing the thread.

Frequently asked questions

Log shippers, metrics pipelines and job workers are all producer/consumer systems. Interviewers use this to check that you understand back-pressure, clean shutdown and error isolation, the three things that go wrong in real pipelines: memory blow-ups, hung processes and one bad record killing a worker.

Then the producer never waits. If it is faster than the consumers, the queue grows until the process runs out of memory, which is the exact failure the pipeline is meant to prevent. A bounded queue turns a speed mismatch into waiting instead of a crash.

Not much on standard CPython, because the GIL lets only one thread run Python bytecode at a time. For CPU-bound parsing, use multiprocessing (with multiprocessing.Queue) or concurrent.futures.ProcessPoolExecutor. The design (bounded queue, sentinels, merge at the end) stays the same.