Knowledge Guide
Home › Hands-On Builds › Concurrency Katas

Build a Bounded Producer–Consumer

Build a Bounded Producer–Consumer

Every worker pool, message broker client, and log shipper you'll ever build is a producer and a consumer disagreeing about speed. Get the hand-off wrong and you either explode memory or burn a CPU core spinning on nothing — and it's one of the most common "write it live" concurrency questions at Meta/Google/Amazon-tier interviews because it's impossible to fake: you either understand condition variables or your code deadlocks in front of the interviewer. This lab makes you build a bounded blocking buffer from an empty file, in Java and Go, break it on purpose, and fix it for real. It also underpins the Bounded Producer-Consumer & the Web Crawler pattern — do this lab first, that page will read like a victory lap.

1. The trap

You're asked to wire a producer thread (say, a log tailer) to a consumer thread (say, a batch uploader) that's slower. The "obvious" first draft:

Queue<LogLine> queue = new LinkedList<>();   // unbounded

// producer thread
while (running) queue.add(nextLine());

// consumer thread
while (running) {
    if (!queue.isEmpty()) upload(queue.poll());
}

Run it against a bursty source for ten minutes. Two ways to die, and you'll hit both:

Both failures come from the same root cause: the queue has no opinion about capacity, and the consumer has no way to sleep until there's work. You need back-pressure (bound it) and a real wake-up mechanism (don't spin). That's this whole lab.

2. Scope it like a senior

Before writing a line, pin down the contract. A candidate who starts coding immediately is the one who builds the wrong thing correctly. Ask:

Answering these up front is what "senior" looks like: you're negotiating the spec, not guessing at it.

3. Reason to the design

Attempt 0 — lock + busy-wait. Put a mutex around the queue so at least the data structure isn't corrupted, but the consumer still polls in a loop when empty:

synchronized(lock) {
    while (queue.isEmpty()) { /* spin, holding nothing, wasting a core */ }
    item = queue.poll();
}

This is actually worse than the naive version — a real mutex held (or repeatedly re-acquired) while spinning either burns CPU or, if you spin outside the lock, races on isEmpty() vs poll(). Either way, you're checking the condition without any way to be notified when it changes. What you actually want is: "stop running entirely, and put me back on the run queue exactly when the condition I'm waiting for becomes true." That's a condition variable — not a queue feature, a scheduler feature.

Attempt 1 — one lock, one condition variable. A single Condition (or sync.Cond) that producers and consumers both wait on and both signal. It works, but every wake-up is a coin flip: a producer waiting for space might get woken by another producer finishing a put (irrelevant to it) instead of by a consumer freeing space. With signalAll/Broadcast this "just works" because everyone re-checks their own condition on waking — but every waiter is woken on every state change, which is wasted scheduling work under load (the thundering-herd problem, scoped small).

Attempt 2 — two condition variables, one lock (the standard design). Split the single condition into notFull (producers wait here) and notEmpty (consumers wait here), both guarded by the same lock. A producer that adds an item only needs to wake takers, so it signals notEmpty; a consumer that removes an item only needs to wake putters, so it signals notFull. Same core mechanism, but each wake-up call now only disturbs the threads that could possibly care. This is exactly what java.util.concurrent.ArrayBlockingQueue does internally, and it's what we hand-roll below.

The one rule that makes or breaks this: after waking from await()/Wait(), you MUST re-check the condition in a loop, never assume it still holds. Two independent reasons force this, and interviewers expect both:

  1. Spurious wakeups. POSIX condition variables (which both the JVM and Go's runtime sit on top of) are explicitly allowed to wake a waiter with no signal having happened at all. Rare, platform-dependent, but real.
  2. Lost/stale signals under multiple waiters (the one you'll actually hit). With multiple consumers waiting on notEmpty and a producer that adds exactly one item and calls signalAll/Broadcast, every waiting consumer wakes up — but only one item exists. If a consumer trusts the wake-up and proceeds unconditionally (if), a second consumer reads a slot nobody refilled. That's not a rare race — that's the deterministic outcome of one item and three waiters, reproduced below, every single run.

The fix in both languages is one word: use a loop (while in Java, for in Go) around the wait, not a one-shot if. This single design decision is the entire lesson of this lab.

4. Build it — milestones

Build a bounded buffer with this contract, growing it milestone by milestone. Attempt each milestone yourself before reading the reference implementation below — the reveal is positioned after the tests on purpose.

Reference implementation — Java (ReentrantLock + two Conditions, hand-rolled)

This is the actual mechanism inside ArrayBlockingQueue, written out so you own every line. Notice the while in both put and take — that's Movement 3's rule, and Movement 5 shows exactly what breaks if you change it to if.

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/** A hand-rolled bounded blocking buffer: one lock, two condition variables. */
public final class BoundedBuffer<T> {
    private final Object[] items;
    private int head = 0, tail = 0, count = 0;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull  = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private volatile boolean closed = false;

    public BoundedBuffer(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException("capacity must be > 0");
        items = new Object[capacity];
    }

    /** Blocks until there is room, or the buffer is closed. */
    public void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length && !closed) {   // while, not if -- see Movement 5
                notFull.await();
            }
            if (closed) throw new IllegalStateException("put() on a closed buffer");
            items[tail] = item;
            tail = (tail + 1) % items.length;
            count++;
            notEmpty.signalAll();   // wake ALL waiting takers -- while must recheck
        } finally {
            lock.unlock();
        }
    }

    /** Blocks until an item is available. Returns null once closed and drained (M4). */
    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0 && !closed) {               // while, not if -- see Movement 5
                notEmpty.await();
            }
            if (count == 0 && closed) return null;         // drained + shut down
            T item = (T) items[head];
            items[head] = null;
            head = (head + 1) % items.length;
            count--;
            notFull.signalAll();
            return item;
        } finally {
            lock.unlock();
        }
    }

    /** M3: bounded wait instead of blocking forever. */
    public boolean offer(T item, long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        lock.lock();
        try {
            while (count == items.length && !closed) {
                if (nanos <= 0L) return false;
                nanos = notFull.awaitNanos(nanos);
            }
            if (closed) return false;
            items[tail] = item;
            tail = (tail + 1) % items.length;
            count++;
            notEmpty.signalAll();
            return true;
        } finally {
            lock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    public T poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        lock.lock();
        try {
            while (count == 0 && !closed) {
                if (nanos <= 0L) return null;
                nanos = notEmpty.awaitNanos(nanos);
            }
            if (count == 0) return null;
            T item = (T) items[head];
            items[head] = null;
            head = (head + 1) % items.length;
            count--;
            notFull.signalAll();
            return item;
        } finally {
            lock.unlock();
        }
    }

    /** M4: graceful shutdown -- wakes every blocked thread so it can observe `closed`. */
    public void close() {
        lock.lock();
        try {
            closed = true;
            notEmpty.signalAll();
            notFull.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public int size() {
        lock.lock();
        try { return count; } finally { lock.unlock(); }
    }
}

Reference implementation — Go (channel idiom AND sync.Cond idiom)

Go gives you two genuinely different ways to build this, and you should be able to write both. First, the idiomatic one — a buffered channel's send/receive blocking is the bounded buffer, no lock required:

package pcbuffer

import (
    "context"
    "errors"
)

// ChanBuffer is a bounded producer/consumer buffer built directly on a
// buffered channel. The channel's own send/receive blocking IS the
// back-pressure -- no lock, no condition variable needed.
type ChanBuffer[T any] struct {
    ch chan T
}

func NewChanBuffer[T any](capacity int) *ChanBuffer[T] {
    return &ChanBuffer[T]{ch: make(chan T, capacity)}
}

// Put blocks until there is room in the buffer.
func (b *ChanBuffer[T]) Put(item T) { b.ch <- item }

// Take blocks until an item is available. ok=false means the channel was
// closed and fully drained -- Go's version of the poison pill (M4).
func (b *ChanBuffer[T]) Take() (item T, ok bool) {
    item, ok = <-b.ch
    return item, ok
}

var ErrTimeout = errors.New("pcbuffer: timed out")

// Offer is Put with a deadline (M3) -- select gives this for free.
func (b *ChanBuffer[T]) Offer(ctx context.Context, item T) error {
    select {
    case b.ch <- item:
        return nil
    case <-ctx.Done():
        return ErrTimeout
    }
}

// Poll is Take with a deadline (M3).
func (b *ChanBuffer[T]) Poll(ctx context.Context) (item T, err error) {
    select {
    case v, ok := <-b.ch:
        if !ok { return item, errClosed }
        return v, nil
    case <-ctx.Done():
        return item, ErrTimeout
    }
}

var errClosed = errors.New("pcbuffer: closed")

// Close is the graceful shutdown (M4): no more Puts after this; ranging
// consumers drain whatever is left, then see ok=false.
func (b *ChanBuffer[T]) Close() { close(b.ch) }

Second, the manual idiom — a ring buffer with a Mutex and two *sync.Cond, mirroring the Java design closely (same lock-plus-two-conditions structure, same while/for-recheck rule). One intentional divergence: Java's put() throws IllegalStateException on a closed buffer, while Go's Put() below silently drops the item on close (returns with no error) — don't assume the two are interchangeable on that edge. Go's sync.Cond only gives you Wait/Signal/Broadcast — notice the for loop, which is Go's spelling of Java's while:

package pcbuffer

import "sync"

// CondBuffer mirrors the Java ReentrantLock+2-Condition design: a ring
// buffer guarded by one Mutex and two *sync.Cond built on it.
type CondBuffer[T any] struct {
    mu       sync.Mutex
    notFull  *sync.Cond
    notEmpty *sync.Cond
    items    []T
    head, n  int
    closed   bool
}

func NewCondBuffer[T any](capacity int) *CondBuffer[T] {
    b := &CondBuffer[T]{items: make([]T, capacity)}
    b.notFull = sync.NewCond(&b.mu)
    b.notEmpty = sync.NewCond(&b.mu)
    return b
}

func (b *CondBuffer[T]) Put(item T) {
    b.mu.Lock()
    defer b.mu.Unlock()
    for b.n == len(b.items) && !b.closed {   // for, not if -- see Movement 5
        b.notFull.Wait()
    }
    if b.closed { return }
    tail := (b.head + b.n) % len(b.items)
    b.items[tail] = item
    b.n++
    b.notEmpty.Broadcast()   // wake ALL waiting takers -- for must recheck
}

func (b *CondBuffer[T]) Take() (item T, ok bool) {
    b.mu.Lock()
    defer b.mu.Unlock()
    for b.n == 0 && !b.closed {               // for, not if -- see Movement 5
        b.notEmpty.Wait()
    }
    if b.n == 0 { return item, false }        // drained + closed (M4)
    item = b.items[b.head]
    var zero T
    b.items[b.head] = zero
    b.head = (b.head + 1) % len(b.items)
    b.n--
    b.notFull.Broadcast()
    return item, true
}

func (b *CondBuffer[T]) Close() {
    b.mu.Lock()
    defer b.mu.Unlock()
    b.closed = true
    b.notEmpty.Broadcast()
    b.notFull.Broadcast()
}

Both compile clean under go build ./... and go vet ./..., and both were exercised end to end before writing this page.

5. Break it — the test that fails

Change exactly one thing in each language: while/for becomes if. Everything else is identical. Scenario: a buffer of capacity 1, three consumer threads blocked on an empty buffer, one producer puts exactly one item and calls signalAll/Broadcast (which wakes all three, by design — that's not the bug, that's correct condition-variable usage). The bug is what each consumer does next.

// Java -- BuggyBoundedBuffer.take(), the ONLY change from the correct version:
public T take() throws InterruptedException {
    lock.lock();
    try {
        if (count == 0) notEmpty.await();   // BUG: if, not while -- no recheck after waking
        T item = (T) items[head];           // a second waiter reads a stale/empty slot here
        items[head] = null;
        head = (head + 1) % items.length;
        count--;                            // <-- goes negative when >1 waiter slips through
        notFull.signalAll();
        return item;
    } finally { lock.unlock(); }
}

Run it (three consumers race one producer putting a single item):

internal count field after run = -2
consumers that read a phantom (null) item = 2
BUG REPRODUCED: an `if`-guarded wait let more consumers through than items existed.
This is the lost-wakeup / stale-condition bug.

This is not a rare interleaving you need -race to notice by luck — it reproduces on every run, because with three waiters and one signalAll, ALL three always wake, and if never asks any of them to check whether the item is actually still there before the previous consumer already took it. The first consumer to re-acquire the lock gets the real item; the other two barrel through the exact same code path and read/decrement a buffer that's already empty. count goes negative — a silent, corrupted invariant, worse than a crash because nothing stops the program, it just quietly lies from then on.

The Go version reproduces identically, deterministically, under go test -race:

func (b *BuggyCondBuffer[T]) Take() T {
    b.mu.Lock()
    defer b.mu.Unlock()
    if b.n == 0 { b.notEmpty.Wait() }   // BUG: if, not for
    item := b.items[b.head]
    b.head = (b.head + 1) % len(b.items)
    b.n--                                // <-- goes negative
    b.notFull.Broadcast()
    return item
}

// $ go test -race -run TestBrokenWithoutForLoop -v
// condbuffer_test.go:33: BUG REPRODUCED: internal count went negative: -2
// --- PASS: TestBrokenWithoutForLoop (0.05s)

The fix is switching back to while/for: two of the three consumers re-check the condition after waking, see the buffer is genuinely empty again, and go back to sleep — correctly. Running the exact same three-consumers-one-item scenario against the correct BoundedBuffer/CondBuffer, followed by close() to release the two still-waiting consumers, gives:

items actually delivered = 1
consumers released by close() = 2
internal count field = 0
CORRECT: exactly one consumer got the item; the other two re-blocked on the
while-recheck and only woke again via close().

That's the entire lesson of this lab in two log lines: the loop is not defensive boilerplate, it's the correctness condition.

6. Optimise — with trade-offs

ApproachThroughputSimplicityBack-pressureUse when
Condition-variable blocking buffer (this lab / ArrayBlockingQueue)Good; one lock is contended on every opModerate — you own the wait/notify protocolNative and precise (blocks exactly at capacity)Default choice: bounded work queues, thread pools, moderate throughput (10,000-100,000 ops/sec)
Go channelsGood, comparable to the aboveHighest — the runtime owns the wait/notify protocol, you can't get the loop wrongNative (send/receive block at capacity)Any Go program; prefer over hand-rolled sync.Cond unless you need custom wake conditions
sync.Cond (Go, manual)Same ballpark as channelsLowest of the Go options — manual loop, no built-in timeout/cancellation (must hand-roll around a separate goroutine or accept it can't be interrupted)Native, but you implement it yourselfYou need a wait condition channels can't express directly (e.g. "wake when N-of-M sub-tasks are done"), and you accept the loss of built-in select/timeout
Lock-free ring buffer (Disruptor-style, e.g. LMAX Disruptor)Highest — no OS-level blocking, CAS-based sequence counters, mechanical-sympathy friendly (cache-line padding)Low — hardest to get right; wrong memory ordering is a silent correctness bug, not a crashVia a busy-spin/backoff wait strategy instead of OS blocking — trades CPU for latencyUltra-low-latency, single-digit-microsecond pipelines (trading systems, packet processing) where you can dedicate cores to spinning
Drop / best-effort queue (bounded, overwrite-oldest or reject-newest)Highest — producer never blocksHighNone — data loss instead of back-pressureMetrics/telemetry sampling where losing a data point under load beats blocking the producer

The real judgment call: a condition-variable buffer trades some throughput for exact, native back-pressure and a small, auditable surface area. A lock-free ring buffer buys you 5-10x the throughput at 1/10th the latency, but the price is that a subtle bug (a torn read, an ABA problem, a mis-ordered memory fence) can corrupt state silently instead of throwing — exactly the failure mode you just saw with a two-line change above, except now it's happening at the level of CPU memory ordering instead of a lock. Reach for lock-free only when you've profiled and the lock is provably the bottleneck; default to the boring, provably-correct version.

7. Defend under drilling

An interviewer will push on the design. These are the follow-ups that come up almost every time, with the answer a staff engineer gives — short, concrete, no hedging.

8. You can now defend

Re-authored/Deepened for this guide. Reference code compiled and executed (javac/java; go build, go vet, go test -race) before publishing; the break-it test reproduces the lost-wakeup bug deterministically on every run. See also: the Bounded Producer-Consumer & the Web Crawler pattern.

🤖 Don't fully get this? Learn it with Claude

Stuck on Build a Bounded Producer–Consumer? Open Claude, copy a block below, and it'll teach you this exact concept — visually and interactively.

🎨 Explain it visually

Build the mental picture, not memorization.

I just read a lesson on **Build a Bounded Producer–Consumer** (Hands-On Builds) and want to truly understand it. Explain Build a Bounded Producer–Consumer from first principles using ONE vivid real-world analogy and a visual mental model — draw it as ASCII art or a clear step-by-step diagram — with a concrete example using real numbers. Then ask me one question to check I got the mental picture, and wait for my reply. If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.
🤔 Walk me through it (interactive)

Socratic — adapts to where you're stuck.

Teach me **Build a Bounded Producer–Consumer** interactively. Ask me ONE guiding question at a time, wait for my answer, and adapt to my confusion — build the idea with me step by step instead of explaining it all at once. If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.
🧪 Quiz me & fix my gaps

Active recall exposes what you missed.

Quiz me on **Build a Bounded Producer–Consumer** with 5 questions, easy to tricky, ONE at a time. Tell me if each answer is right; at the end, explain clearly what I got wrong and why. If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.
🧠 Make it stick

Intuition + hook + flashcards for long-term memory.

Help me remember **Build a Bounded Producer–Consumer** for the long term: give the one-sentence intuition, a memorable hook/mnemonic, a tiny worked example, and 3 active-recall flashcards (Q -> A). If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.

📝 My notes

Build a Readers–Writer Lock →