Producer-Consumer Problem: Bounded Buffer with Locks and Modern Approaches

Producer-Consumer Problem: Bounded Buffer with Locks, Condition Variables, and Modern Approaches

The producer-consumer problem is the canonical concurrency interview question. Producers generate items; consumers process them; a bounded buffer mediates between them. The challenge is correct synchronization — producers must wait when the buffer is full, consumers must wait when it’s empty, and multiple producers/consumers must coordinate without race conditions or deadlock. This guide covers four standard approaches: condition variables (the textbook answer), semaphores, blocking queues (the practical answer in modern languages), and lock-free / async approaches. Working code in Python; the patterns translate directly to Java, C++, and Go.

Problem Statement

Design a thread-safe bounded buffer that supports:

• put(item) — producer adds an item; blocks if buffer is full
• get() — consumer removes and returns an item; blocks if buffer is empty

Multiple producers and consumers must operate concurrently without races, deadlock, or lost updates.

Approach 1: Condition Variables (Textbook Answer)

Use a single mutex protecting the buffer, plus two condition variables: one signaled when the buffer becomes non-full (wakes producers), one signaled when the buffer becomes non-empty (wakes consumers).

from collections import deque
from threading import Lock, Condition

class BoundedBuffer:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.buffer = deque()
        self.lock = Lock()
        self.not_full = Condition(self.lock)
        self.not_empty = Condition(self.lock)

    def put(self, item) -> None:
        with self.not_full:
            while len(self.buffer) == self.capacity:
                self.not_full.wait()
            self.buffer.append(item)
            self.not_empty.notify()

    def get(self):
        with self.not_empty:
            while not self.buffer:
                self.not_empty.wait()
            item = self.buffer.popleft()
            self.not_full.notify()
            return item


# Demo
import threading
import time
import random

buf = BoundedBuffer(capacity=5)

def producer(name, count):
    for i in range(count):
        item = f"{name}-{i}"
        buf.put(item)
        print(f"Produced {item}")
        time.sleep(random.random() * 0.1)

def consumer(name, count):
    for _ in range(count):
        item = buf.get()
        print(f"{name} consumed {item}")
        time.sleep(random.random() * 0.2)

threads = [
    threading.Thread(target=producer, args=("P1", 10)),
    threading.Thread(target=producer, args=("P2", 10)),
    threading.Thread(target=consumer, args=("C1", 10)),
    threading.Thread(target=consumer, args=("C2", 10)),
]
for t in threads:
    t.start()
for t in threads:
    t.join()

Why while instead of if in the wait loop?

Condition variables can have spurious wakeups (a thread wakes from wait() without being explicitly notified). They also wake all waiters when notify_all() is called, but only one will actually find the buffer ready. Using while rather than if ensures the condition is rechecked after wakeup — the standard rule in any condition-variable code.

Approach 2: Semaphores

Use two counting semaphores: empty_slots (counts free slots, decremented by producers) and full_slots (counts filled items, decremented by consumers). Plus a mutex for buffer access.

from threading import Semaphore, Lock
from collections import deque

class BoundedBufferSemaphore:
    def __init__(self, capacity: int):
        self.buffer = deque()
        self.empty_slots = Semaphore(capacity)
        self.full_slots = Semaphore(0)
        self.mutex = Lock()

    def put(self, item) -> None:
        self.empty_slots.acquire()
        with self.mutex:
            self.buffer.append(item)
        self.full_slots.release()

    def get(self):
        self.full_slots.acquire()
        with self.mutex:
            item = self.buffer.popleft()
        self.empty_slots.release()
        return item

Trade-off: Semaphores express “wait for available slot” naturally without explicit while-loops. The mutex still protects buffer mutations. Many textbooks (Tanenbaum, OS classics) prefer this formulation for its conceptual clarity. In practice, condition variables are equally common.

Approach 3: Blocking Queue (Practical Modern Answer)

Modern languages provide built-in thread-safe bounded queues. Don’t roll your own in production code; use the standard-library version. In Python:

from queue import Queue

class BoundedBufferStdlib:
    def __init__(self, capacity: int):
        self.queue = Queue(maxsize=capacity)

    def put(self, item) -> None:
        self.queue.put(item)  # blocks if full

    def get(self):
        return self.queue.get()  # blocks if empty

In Java: BlockingQueue (typically ArrayBlockingQueue or LinkedBlockingQueue). In Go: a buffered channel ch := make(chan T, capacity). In C++: std::queue with std::condition_variable and std::mutex, or a third-party MPMC queue.

For interview purposes, mention the stdlib option even if you implement from scratch — it shows production awareness.

Approach 4: Async / Coroutines (Modern Concurrency)

For I/O-bound producers and consumers, async/await with an asyncio queue avoids thread overhead entirely:

import asyncio

async def producer(queue: asyncio.Queue, name: str, count: int):
    for i in range(count):
        item = f"{name}-{i}"
        await queue.put(item)  # awaits if full

async def consumer(queue: asyncio.Queue, name: str, count: int):
    for _ in range(count):
        item = await queue.get()  # awaits if empty
        queue.task_done()


async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue, "P1", 10),
        producer(queue, "P2", 10),
        consumer(queue, "C1", 10),
        consumer(queue, "C2", 10),
    )

asyncio.run(main())

Async producer-consumer is appropriate when the work itself is I/O-bound (network, file, database). For CPU-bound work, threading or multiprocessing is required because of Python’s GIL.

Common Pitfalls

Race conditions from forgotten locks

Any read or write of shared state (the buffer) must hold the mutex. Forgetting the lock around a read or a check is a classic bug — the buffer state can change between check and use.

Deadlock from wrong notification

If put notifies not_full instead of not_empty, consumers never wake. Always notify the condition that downstream waiters are blocked on.

Lost wakeups

Calling notify() while the buffer is being mutated and the lock is briefly released can leave a waiter blocked. Holding the lock while calling notify (Python’s Condition does this automatically with with) prevents this.

Wrong order of acquires

In the semaphore version, empty_slots.acquire() must come before mutex.acquire() in producers. Reversing them can cause deadlock when the buffer is full and consumers need both semaphores to make progress.

Thundering herd

Using notify_all() when only one item was added wakes all consumers, all of whom recheck the condition and most go back to sleep. Use notify() when you’ve added or removed exactly one item.

Variations and Extensions

Multiple producers, multiple consumers (MPMC)

The implementations above already handle this. The mutex serializes access; condition variables wake one or more waiters as appropriate.

Priority producer-consumer

Replace the deque with a heap; producers insert with priority; consumers always get the highest-priority item. Python’s queue.PriorityQueue handles this with the same blocking semantics.

Bounded buffer with timeouts

Add timeout parameters to put and get. Both stdlib and async versions support this. Useful when you want to fail-fast rather than block indefinitely.

Lock-free queues

For extreme performance, lock-free MPMC queues (e.g., based on Michael-Scott algorithm or LMAX Disruptor) avoid mutex overhead entirely. Substantially harder to implement correctly; usually use a battle-tested library rather than rolling your own.

Frequently Asked Questions

See also: Fair Server Processing • Implement a Rate Limiter • Producer-Consumer Problem

class BoundedBuffer {
    constructor(capacity) {
        this.capacity = capacity;
        this.buffer = [];
        this.mutex = new Mutex();
        this.notFull = new ConditionVariable();
        this.notEmpty = new ConditionVariable();
    }

    async produce(item) {
        await this.mutex.lock();

        try {
            // Wait while buffer is full
            while (this.buffer.length >= this.capacity) {
                await this.notFull.wait(this.mutex);
            }

            // Add item
            this.buffer.push(item);
            console.log(`Produced: ${item}, Buffer: [${this.buffer}]`);

            // Signal that buffer is not empty
            this.notEmpty.signal();
        } finally {
            this.mutex.unlock();
        }
    }

    async consume() {
        await this.mutex.lock();

        try {
            // Wait while buffer is empty
            while (this.buffer.length === 0) {
                await this.notEmpty.wait(this.mutex);
            }

            // Remove item
            const item = this.buffer.shift();
            console.log(`Consumed: ${item}, Buffer: [${this.buffer}]`);

            // Signal that buffer is not full
            this.notFull.signal();

            return item;
        } finally {
            this.mutex.unlock();
        }
    }
}

// Simplified Mutex and ConditionVariable classes
class Mutex {
    constructor() {
        this.locked = false;
        this.waiting = [];
    }

    async lock() {
        while (this.locked) {
            await new Promise(resolve => this.waiting.push(resolve));
        }
        this.locked = true;
    }

    unlock() {
        this.locked = false;
        if (this.waiting.length > 0) {
            const resolve = this.waiting.shift();
            resolve();
        }
    }
}

class ConditionVariable {
    constructor() {
        this.waiting = [];
    }

    async wait(mutex) {
        const promise = new Promise(resolve => this.waiting.push(resolve));
        mutex.unlock();
        await promise;
        await mutex.lock();
    }

    signal() {
        if (this.waiting.length > 0) {
            const resolve = this.waiting.shift();
            resolve();
        }
    }

    broadcast() {
        while (this.waiting.length > 0) {
            const resolve = this.waiting.shift();
            resolve();
        }
    }
}

// Test
async function test() {
    const buffer = new BoundedBuffer(3);

    // Start producers
    async function producer(id, count) {
        for (let i = 0; i < count; i++) {
            await buffer.produce(`P${id}-${i}`);
            await sleep(Math.random() * 100);
        }
    }

    // Start consumers
    async function consumer(id, count) {
        for (let i = 0; i < count; i++) {
            await buffer.consume();
            await sleep(Math.random() * 150);
        }
    }

    function sleep(ms) {
        return new Promise(resolve => setTimeout(resolve, ms));
    }

    // Run 2 producers and 2 consumers
    await Promise.all([
        producer(1, 5),
        producer(2, 5),
        consumer(1, 5),
        consumer(2, 5)
    ]);
}

test();

class BoundedBufferWithSemaphores {
    constructor(capacity) {
        this.capacity = capacity;
        this.buffer = [];
        this.mutex = new Semaphore(1);        // Binary semaphore (mutex)
        this.empty = new Semaphore(capacity); // Count empty slots
        this.full = new Semaphore(0);         // Count full slots
    }

    async produce(item) {
        await this.empty.acquire();  // Wait for empty slot
        await this.mutex.acquire();  // Lock buffer

        this.buffer.push(item);
        console.log(`Produced: ${item}`);

        this.mutex.release();
        this.full.release();  // Signal item available
    }

    async consume() {
        await this.full.acquire();   // Wait for item
        await this.mutex.acquire();  // Lock buffer

        const item = this.buffer.shift();
        console.log(`Consumed: ${item}`);

        this.mutex.release();
        this.empty.release();  // Signal slot available

        return item;
    }
}

class Semaphore {
    constructor(count) {
        this.count = count;
        this.waiting = [];
    }

    async acquire() {
        if (this.count > 0) {
            this.count--;
        } else {
            await new Promise(resolve => this.waiting.push(resolve));
        }
    }

    release() {
        if (this.waiting.length > 0) {
            const resolve = this.waiting.shift();
            resolve();
        } else {
            this.count++;
        }
    }
}

class BoundedBuffer {
    private final Queue buffer;
    private final int capacity;
    private final Object lock = new Object();

    public BoundedBuffer(int capacity) {
        this.capacity = capacity;
        this.buffer = new LinkedList<>();
    }

    public void produce(T item) throws InterruptedException {
        synchronized (lock) {
            while (buffer.size() >= capacity) {
                lock.wait();  // Wait until not full
            }

            buffer.add(item);
            System.out.println("Produced: " + item);

            lock.notifyAll();  // Wake up consumers
        }
    }

    public T consume() throws InterruptedException {
        synchronized (lock) {
            while (buffer.isEmpty()) {
                lock.wait();  // Wait until not empty
            }

            T item = buffer.remove();
            System.out.println("Consumed: " + item);

            lock.notifyAll();  // Wake up producers

            return item;
        }
    }
}

// WRONG - uses if
if (buffer.isEmpty()) {
    wait();
}
// After wakeup, might still be empty!
item = buffer.remove();  // CRASH

// CORRECT - uses while
while (buffer.isEmpty()) {
    wait();
}
// After wakeup, check again
item = buffer.remove();  // Safe