Producer-consumer with a shared buffer

In one minute

• Producers create work and consumers process it; a buffer between them lets each side run at its own pace, so a burst from producers does not have to be handled the instant it arrives.
• The buffer must be bounded. When it is full, producers block, slow down or drop; when it is empty, consumers block. A full buffer that pushes back on producers is backpressure, and it is the whole point of the pattern.
• The classic implementation is a lock plus two condition variables ("not full" and "not empty"), always waited on in a while loop. Most languages ship it ready-made: Go channels, Java's BlockingQueue, Python's queue.Queue.
• Shutdown is part of the design: closing the buffer (or sending a poison pill) tells consumers to drain what is left and stop, and producers must not send after close.
• The same shape appears at every scale: a Unix pipe, a TCP receive window, a Kafka topic and a thread pool's work queue are all producer-consumer buffers with different bounds and different full-buffer policies.

How it works

The problem

A log shipper reads lines from application files and sends them in batches to a central store. Reading is fast and bursty: a deploy can produce 50,000 lines in a second. Sending is slower and steady: the store accepts about 5,000 lines per second from one shipper. If the reader calls the sender directly, every read waits for a network round trip, and a slow store stalls the reader. If the reader appends to an unbounded list that the sender drains, a ten-minute store outage queues millions of lines and the shipper runs out of memory.

The producer-consumer pattern puts a bounded buffer between them. The reader (producer) puts lines in; the sender (consumer) takes them out. Short bursts are absorbed by the buffer. A long slowdown fills it, and then the producer waits, which in a log shipper means it simply reads the files later. Nothing is lost, and memory stays flat.

The three rules of a bounded buffer

1. A producer may add only when there is room; otherwise it waits (or follows an overflow policy).
2. A consumer may take only when there is an item; otherwise it waits.
3. Adding and taking must not interfere: two producers must not write the same slot, and a consumer must never see a half-written item.

The classic implementation

A ring buffer guarded by one mutex and two condition variables:

def put(self, item):
    with self.lock:
        while len(self.items) == self.capacity:
            self.not_full.wait()       # releases the lock while waiting
        self.items.append(item)
        self.not_empty.notify()

def take(self):
    with self.lock:
        while not self.items:
            self.not_empty.wait()
        item = self.items.popleft()
        self.not_full.notify()
        return item

The while is not optional. A waiting thread can wake up spuriously, and even a genuine notify does not guarantee the condition still holds: another consumer may have grabbed the item between the notify and this thread reacquiring the lock. Checking with if instead of while is the most common bug in hand-written versions. Using one condition variable for both conditions also works, but then every change must notify_all, waking producers that cannot proceed.

Dijkstra's original formulation used two counting semaphores, empty_slots (starting at capacity) and full_slots (starting at 0), plus a mutex for the buffer itself. A producer waits on empty_slots, adds, signals full_slots; a consumer does the reverse. The counts carry the state, so no while loop is needed.

Overflow policies

Blocking is only one answer to "the buffer is full". The others are the same menu a thread pool offers:

• Block the producer until there is room. Right when the producer can wait (reading files, a batch job).
• Fail the put immediately and let the producer decide (offer in Java, put_nowait in Python, a select with default in Go). Right for request handlers that should return 503 rather than hang.
• Drop the newest or the oldest item. Right for telemetry and video frames, where fresh data beats complete data.
• Block with a timeout, then fail. Usually the best default for services.

Shutdown

Consumers loop forever, so they need a way to know the stream has ended. Go's close(ch) makes receivers drain the remaining items and then get the zero value with ok == false; sending on a closed channel panics. In Java and Python, the convention is a poison pill: a special sentinel object put once per consumer. Either way, the rules are the same: only the producer side closes, everything buffered before close is still delivered, and anything blocked waiting to receive must wake up.

Many producers, many consumers

The pattern scales to M producers and N consumers on one buffer without code changes, which is why it is the basis of worker pools. What you lose is ordering: with several consumers, item 2 may finish before item 1. If order matters per key (all events for one account in order), partition: hash the key to one of N buffers, each with exactly one consumer. Kafka partitions apply exactly this rule across machines.

Batching

A consumer that takes one item at a time pays the per-item overhead (a lock, a syscall, a network round trip) every time. Draining everything available, up to a batch size, amortises that overhead and gets more efficient under load, which is the opposite of what most systems do. Java's drainTo and Go's "receive one, then non-blocking receive the rest" idiom both do this.

Reviewing agent-written code

Agents produce two typical shapes. One is an unbounded queue (queue.Queue() with no maxsize, an unbuffered list, new LinkedBlockingQueue<>()) that works in tests and fails under load. The other is a hand-rolled buffer with if instead of while around wait, or a busy loop that polls len(queue) with a sleep. Ask for the platform's blocking queue with an explicit capacity, a stated overflow policy, and a shutdown path, and for tests that cover a full buffer and close with waiters.

The same shape elsewhere

This is the overload problem shape solved with backpressure. A Unix pipeline zcat logs.gz | grep ERROR | sort runs three processes connected by kernel pipe buffers; when sort is slow, grep blocks on write and zcat blocks behind it. TCP's receive window is a producer-consumer buffer across a network: when the receiving application stops reading, the window shrinks to zero and the sender stops. A restaurant kitchen's ticket rail, with a limited number of clips, is the same buffer with humans on both sides.

Down the stack

Diagram

sequenceDiagram
  participant P as Producer
  participant B as Buffer (cap 2)
  participant C as Consumer
  C->>B: recv()
  Note over C,B: empty: consumer parks
  P->>B: send(a)
  B-->>C: hand a directly
  P->>B: send(b)
  P->>B: send(c)
  Note over P,B: buffered b, c (full)
  P->>B: send(d)
  Note over P,B: full: producer parks with d
  C->>B: recv() gets b
  B-->>P: d moved into buffer, producer wakes
  P->>B: close()
  C->>B: recv() gets c, then d, then closed

Worked example

The log shipper's numbers: the store accepts batches of up to 1,000 lines with a 40 ms round trip, so one consumer sends at most 25,000 lines per second; the production steady state is 3,000 lines per second with deploy bursts of 50,000 lines over about 2 seconds.

With a buffer of 100,000 lines (about 20 MB at 200 bytes per line), a burst of 100,000 lines arrives while the consumer drains 25,000 per second, so the buffer peaks around 50,000 and empties within 4 seconds. No producer blocks. During a 10-minute store outage, the consumer stops draining, the buffer fills in about 30 seconds at the steady rate, and the reader then blocks on put. Because the reader tracks its file offset, it resumes where it stopped when the store recovers; with the buffer full, it catches up at 25,000 minus 3,000, about 22,000 lines per second, clearing a 10-minute backlog of 1.8 million lines in under 90 seconds.

An earlier version used an unbounded queue. During the same outage it grew to 1.8 million lines, about 360 MB, and the container, limited to 256 MB, was killed by the OOM killer, losing the in-memory lines. The bound changed the failure from "crash and lose data" to "pause and catch up".

Batching mattered as much: sending one line per request at 40 ms each caps the consumer at 25 lines per second. Draining up to 1,000 available lines per request gives the 1,000-times improvement above.

Trade-offs

Pitfalls in production

• Unbounded queues turning a slow consumer into an out-of-memory crash. Fix: bound every in-process queue and choose an overflow policy explicitly.
• if instead of while around a condition wait, so a consumer takes from an empty buffer after a race or spurious wakeup. Fix: always recheck in a loop, or use a library queue.
• Consumers that never exit because nobody closed the queue or sent enough poison pills. Fix: close from the producer side once all producers are done (in Go, a sync.WaitGroup and one closer goroutine), or one pill per consumer.
• Lost ordering after adding consumers. Fix: partition by key so each key has one consumer.
• Deadlock when a producer holds a lock that the consumer needs while blocking on a full buffer. Fix: never block on a queue while holding other locks.
• Queue depth not monitored, so a slow consumer goes unnoticed until latency explodes. Fix: export depth, time-in-queue and blocked-producer time, and alert on sustained high water.

Leadership lens

• Ask of every queue in a design: what is its bound, what happens when it is full, and who notices. "It's async" is not an answer to overload.
• Size buffers in time, not items: capacity divided by drain rate is the latency you are willing to add. A buffer of 10 minutes of work is a 10-minute delay you have signed up for.
• Choose between in-process buffers and a broker deliberately; a broker adds durability and fan-out, and also an operational dependency. Point teams at Message Queues when work must survive a crash.
• Put producer-consumer correctness in the review checklist for agent-written concurrency code: bound, overflow policy, close path, and tests with full and empty buffers.

Check yourself

1. Why must a condition wait be inside a while loop rather than an if?

   Show answer

   Wakeups can be spurious, and even after a real notify another thread may have changed the state before this one reacquired the lock, so the condition must be checked again.

2. A Go channel has a parked receiver and an empty buffer when a send happens. Where does the value go?

   Show answer

   Directly to the waiting receiver, which is made runnable; the buffer is bypassed, saving a copy and keeping FIFO fairness.

3. You add a second consumer to speed up event processing and customers see account events applied out of order. What do you change?

   Show answer

   Partition by account: hash the account ID to one of N queues, each with a single consumer, so each account's events stay in order while different accounts run in parallel.

4. How do you choose a buffer's capacity?

   Show answer

   From the burst you need to absorb and the latency you can accept: capacity is roughly burst size minus what consumers drain during the burst, and capacity divided by drain rate must stay under the latency budget.

Watch

• GopherCon 2017: Understanding Channels (Kavya Joshi): the hchan structure, its wait queues and the direct handoff to parked goroutines.
• Google I/O 2012: Go Concurrency Patterns (Rob Pike): producer-consumer pipelines, fan-in and fan-out built from channels.

Free resources

• Go channel source (chan.go): chansend, chanrecv and closechan, the reference implementation of a bounded buffer with wait queues.
• Go Concurrency Patterns: Pipelines and cancellation (Go blog): stages, fan-out, and how to close and cancel pipelines correctly.
• ArrayBlockingQueue (Java SE 21 docs): the classic bounded buffer with fairness options and the full blocking, timed and failing API.
• queue module (Python docs): bounded queues, task_done and join for waiting on consumers.
• LMAX Disruptor technical paper: how a lock-free ring buffer with padded sequences reaches very high throughput.
• pipe(7) Linux manual page: the kernel's own bounded buffer, its capacity and blocking rules.