Producer-Consumer Problem: Synchronization with Semaphores, Monitors & Message Queues

Introduction: The Milk Tea Shop Analogy

Imagine ordering at a milk tea shop: if no drinks are ready, you wait (consumer waits); if prepared drinks pile up, where do they go (producer problem)? This illustrates the producer-consumer problem—how to balance production and consumption so the system runs efficiently.

Problem Description

Producers create items and place them into a bounded buffer; consumers remove items from the buffer. Both may operate at varying speeds. The buffer has limited capacity, and both put and take operations must be thread-safe.

Producer:
- Produce item
- Put into buffer
- Speed may be fast or slow

Consumer:
- Take item from buffer
- Consume item
- Speed may be fast or slow

Buffer:
- Limited capacity
- Put and take must be safe

┌─────────┐      ┌─────────────────┐      ┌─────────┐
│Producer A│ ───→ │      Buffer     │ ───→ │Consumer X│
│Producer B│      │  (Bounded Queue)│      │Consumer Y│
│Producer C│      │                 │      │Consumer Z│
└─────────┘      └─────────────────┘      └─────────┘

Why Is This a Problem?

Problem 1: Buffer Full

Producer keeps putting
Consumer takes slowly
Buffer becomes full!
Producer: wait for consumer

Problem 2: Buffer Empty

Consumer keeps taking
Producer produces slowly
Buffer becomes empty!
Consumer: wait for producer

Problem 3: Race Condition

Producer A and B try to put simultaneously:
- A sees empty slot, prepares to put at position 3
- B also sees empty slot, prepares to put at position 3
- A puts: item A
- B puts: item B (overwrites A!)

Or:
- A reads count=5
- B reads count=5
- A writes count=6
- B writes count=6
- Actual should be 7! → Lost one item!

Constraints

1. Mutual Exclusion
   - Only one process may access buffer at a time
   - Prevents race conditions

2. Full Condition
   - When buffer is full, producer cannot put
   - Must wait

3. Empty Condition
   - When buffer is empty, consumer cannot take
   - Must wait

4. Ordering (optional)
   - FIFO order

Solution 1: Semaphores

Three semaphores coordinate access:

#define N 100
Semaphore empty = N;   // Empty slots, initially N
Semaphore full = 0;    // Filled slots, initially 0
Semaphore mutex = 1;   // Mutual exclusion for buffer

int buffer[N];
int in = 0, out = 0;

// Producer
void producer() {
    while (TRUE) {
        int item = produce();  // Produce an item
        P(&empty);             // Wait for empty slot (P1)
        P(&mutex);             // Enter critical section (P2)
        buffer[in] = item;     // Put item
        in = (in + 1) % N;
        V(&mutex);             // Leave critical section (V2)
        V(&full);              // Increment full count, wake consumer (V1)
    }
}

// Consumer
void consumer() {
    while (TRUE) {
        P(&full);              // Wait for item (P1)
        P(&mutex);             // Enter critical section (P2)
        int item = buffer[out];// Take item
        out = (out + 1) % N;
        V(&mutex);             // Leave critical section (V2)
        V(&empty);             // Increment empty count, wake producer (V1)
        consume(item);         // Consume
    }
}

Roles of the Three Semaphores

empty (N): Empty slot count
    Producer uses: wait for empty slot
    Consumer uses: wake producer

full (0): Filled slot count
    Consumer uses: wait for item
    Producer uses: wake consumer

mutex (1): Mutual exclusion lock
    Producer uses: protect critical section
    Consumer uses: protect critical section

Execution Sequence

Initial: empty=N, full=0, mutex=1

Time 1: Producer P(empty) → empty=N-1
Time 2: Producer P(mutex) → mutex=0
Time 3: Producer puts item
Time 4: Producer V(mutex) → mutex=1
Time 5: Producer V(full) → full=1

Time 6: Consumer P(full) → full=0
Time 7: Consumer P(mutex) → mutex=0
Time 8: Consumer takes item
Time 9: Consumer V(mutex) → mutex=1
Time 10: Consumer V(empty) → empty=N

Solution 2: Monitors

Monitors encapsulate shared data and synchronization, providing automatic mutual exclusion and condition variables.

monitor ProducerConsumer {
    int buffer[N];
    int count = 0;          // Items in buffer
    int in = 0, out = 0;
    condition not_full;     // Buffer not full
    condition not_empty;    // Buffer not empty

    procedure add(item) {
        if (count == N) {
            not_full.wait(); // Wait for empty slot
        }
        buffer[in] = item;
        in = (in + 1) % N;
        count++;
        not_empty.signal();  // Wake consumer
    }

    procedure remove() returns int {
        if (count == 0) {
            not_empty.wait(); // Wait for item
        }
        int item = buffer[out];
        out = (out + 1) % N;
        count--;
        not_full.signal();   // Wake producer
        return item;
    }
}

// Producer
void producer() {
    while (TRUE) {
        int item = produce();
        ProducerConsumer.add(item);
    }
}

// Consumer
void consumer() {
    while (TRUE) {
        int item = ProducerConsumer.remove();
        consume(item);
    }
}

Monitor Advantages

✅ Automatic mutual exclusion
✅ Condition variables for waiting
✅ Easier to understand

Compiler/runtime guarantees:
- Acquire lock on monitor entry
- Release lock on monitor exit
- wait automatically releases lock, signal re-acquires after wakeup

Solution 3: Message Passing

Using a message queue (inter-process communication) where the buffer is the queue itself.

// Using message queue (IPC)
// Buffer = message queue
// N = queue capacity

mailbox_t mailbox;  // Mailbox (buffer)

// Producer
void producer() {
    while (TRUE) {
        int item = produce();
        send(mailbox, item);  // Send message (blocks if full)
    }
}

// Consumer
void consumer() {
    while (TRUE) {
        int item;
        receive(mailbox, &item); // Receive message (blocks if empty)
        consume(item);
    }
}

Message Queue Characteristics

Pros:
✅ Built-in buffer
✅ Naturally supports distributed systems

Cons:
✅ Extra overhead
✅ Less efficient than shared memory

Variants

1. Multiple Producers and Consumers

Multiple producers compete for empty
Multiple consumers compete for full
Mutex protects shared buffer

2. Unbounded Buffer

Buffer unlimited size
No empty semaphore needed
Producer never waits

3. Single-Slot Buffer

Buffer holds only one item
empty=1, full=0
Simplified mutual exclusion problem

4. Reader-Priority / Writer-Priority (Readers-Writers)

Readers-Writers problem:
Readers: read only
Writers: read and write

Reader priority:
- Readers proceed, writers wait
- May starve writers

Writer priority:
- Writers proceed, new readers wait
- May starve readers

Real-World Applications

1. Print Queue

Producer: Application (print jobs)
Buffer: Print queue
Consumer: Printer

Print queue management is classic producer-consumer

2. Web Server

Producer: HTTP requests
Buffer: Request queue
Consumer: Worker thread pool

Server handles concurrent requests

3. Message Queue Middleware

RabbitMQ, Kafka, RocketMQ

All implement producer-consumer pattern
Support distributed, high availability

4. Pipes

Linux pipe:
Producer process → Pipe → Consumer process

cat file.txt | grep pattern

5. Python Queue Example

# Python: Queue implementation
from queue import Queue

q = Queue(maxsize=10)  # Buffer size 10

def producer():
    while True:
        item = produce()
        q.put(item)      # Blocks if full

def consumer():
    while True:
        item = q.get()   # Blocks if empty
        process(item)

Code Verification: C Pthreads with Semaphores

#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>

#define N 5
#define ITERATIONS 10

sem_t empty, full, mutex;
int buffer[N];
int in = 0, out = 0;

void *producer(void *arg) {
    for (int i = 0; i < ITERATIONS; i++) {
        int item = rand() % 100;
        sem_wait(&empty);
        sem_wait(&mutex);
        buffer[in] = item;
        printf("Producer %ld put: %d at %d
", (long)arg, item, in);
        in = (in + 1) % N;
        sem_post(&mutex);
        sem_post(&full);
        usleep(rand() % 100000);
    }
    return NULL;
}

void *consumer(void *arg) {
    for (int i = 0; i < ITERATIONS; i++) {
        sem_wait(&full);
        sem_wait(&mutex);
        int item = buffer[out];
        printf("Consumer %ld got: %d from %d
", (long)arg, item, out);
        out = (out + 1) % N;
        sem_post(&mutex);
        sem_post(&empty);
        usleep(rand() % 100000);
    }
    return NULL;
}

int main() {
    sem_init(&empty, 0, N);
    sem_init(&full, 0, 0);
    sem_init(&mutex, 0, 1);

    pthread_t p1, p2, c1, c2;
    pthread_create(&p1, NULL, producer, (void*)1);
    pthread_create(&p2, NULL, producer, (void*)2);
    pthread_create(&c1, NULL, consumer, (void*)1);
    pthread_create(&c2, NULL, consumer, (void*)2);

    pthread_join(p1, NULL);
    pthread_join(p2, NULL);
    pthread_join(c1, NULL);
    pthread_join(c2, NULL);

    sem_destroy(&empty);
    sem_destroy(&full);
    sem_destroy(&mutex);
    return 0;
}

Summary: The Art of Balance

Producer-Consumer = Balance production and consumption

Core Issues:
1. Mutual Exclusion: Protect shared buffer
2. Synchronization: Coordinate producer and consumer

Solutions:
1. Semaphores: Three semaphores
2. Monitors: Encapsulation + condition variables
3. Message Queues: Distributed solution

Real-World Uses:
- Print queues
- Web servers
- Message middleware
- Pipe communication

Remember: The producer-consumer problem is a classic of concurrent programming; mastering it solves most multi-threaded synchronization challenges!

Key Points:

Producers and consumers share a buffer

Producer waits when buffer full; consumer waits when buffer empty

Mutex protects critical section

Empty/full semaphores implement synchronization

Monitors provide higher-level abstraction