Bounded Buffer: Multiple Producers and Consumers

1. Orchestrating the System

So far, we have the blueprints for the producer and consumer roles. Now we need to build the factory floor where they all work. This "factory floor" will be our main function, which is responsible for:

1. Initializing the single, shared buffer_t instance. This includes the mutex and both condition variables.
2. Creating and launching multiple producer threads.
3. Creating and launching multiple consumer threads.
4. Waiting for all threads to complete their work using pthread_join.
5. Cleaning up all shared resources (destroying the mutex and condition variables).

Visually, we are building the system depicted below, moving from the individual components to a complete, interacting system.

2. The main Function Skeleton

Based on your C experience, the structure of the main function will be familiar. The key is to manage the pthread objects and the shared buffer correctly. Here is a skeleton of what our main function will look like:

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h> // For sleep()

#define BUFFER_SIZE 5
#define NUM_PRODUCERS 2
#define NUM_CONSUMERS 3
#define ITEMS_TO_PRODUCE 10

// --- Assume buffer_t, producer(), and consumer() from previous lesson are here ---
// typedef struct { ... } buffer_t;
// void producer(buffer_t *b, ...);
// char consumer(buffer_t *b, ...);
// void init_buffer(buffer_t *b);

int main() {
    buffer_t shared_buffer;
    pthread_t producers[NUM_PRODUCERS];
    pthread_t consumers[NUM_CONSUMERS];

    // 1. Initialize the shared buffer and its synchronization primitives
    init_buffer(&shared_buffer); // A helper to init mutex, conds, etc.
    
    printf("Starting simulation...\n");

    // 2. Create producer threads
    for (int i = 0; i < NUM_PRODUCERS; i++) {
        // pthread_create for producers...
    }

    // 3. Create consumer threads
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        // pthread_create for consumers...
    }

    // 4. Wait for producers to finish
    for (int i = 0; i < NUM_PRODUCERS; i++) {
        pthread_join(producers[i], NULL);
    }
    printf("All producers have finished.\n");


    // 5. Wait for consumers to finish
    // How do we make consumers stop? We will address this.
    // For now, let's imagine they stop after the producers are done and buffer is empty.
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        pthread_join(consumers[i], NULL);
    }
    printf("All consumers have finished.\n");

    // 6. Destroy mutex and condition variables
    pthread_mutex_destroy(&shared_buffer.mutex);
    pthread_cond_destroy(&shared_buffer.less);
    pthread_cond_destroy(&shared_buffer.more);

    printf("Simulation finished.\n");
    return 0;
}

This structure is the standard way to manage a multithreaded application. The main thread acts as the coordinator, setting up the shared state, launching the workers, and then cleaning up after they are done.

3. The signal vs. broadcast Dilemma

Our implementation from the last lesson used pthread_cond_signal. This wakes up at most one waiting thread. In a single-producer, single-consumer scenario, this is perfect. But with multiple consumers, it can lead to subtle inefficiencies or even correctness issues.

Consider a scenario where the buffer has space for 10 items, but is currently empty. Three consumer threads are waiting on the more condition. A producer thread then adds 5 items to the buffer and calls pthread_cond_signal. Only one consumer wakes up, takes an item, and signals the producer. The other two consumers remain asleep, even though there are 4 more items available for them.

This is where pthread_cond_broadcast comes in. It wakes up all threads waiting on a condition variable.

To see a practical demonstration of this exact problem and its solution, the following video is excellent. It uses a "gas station" analogy (producers = fuelers, consumers = cars) to illustrate the point.

As the video demonstrates, using pthread_cond_broadcast allows all waiting threads to re-check the condition. The essential "wait-in-a-loop" pattern (while (condition) { pthread_cond_wait(...); }) ensures this is always safe. After the broadcast, one thread will win the race for the mutex, check the condition, and proceed. The other threads will then acquire the mutex one by one, re-check the condition, and either proceed if resources are still available or go back to waiting.

When to use signal vs. broadcast:

• pthread_cond_signal: Use when you know that only one waiting thread can possibly make progress. This is more efficient as it avoids the "thundering herd" problem of waking many threads unnecessarily. Our bounded buffer, where one item is added/removed at a time, is a good candidate for signal.
• pthread_cond_broadcast: Use when a state change might allow more than one waiting thread to make progress. A classic example is a resource manager: when a large chunk of a resource is freed, you broadcast so all waiting threads can re-evaluate if their request can now be met. It's also often used to signal termination to all worker threads.

For our primary implementation, pthread_cond_signal remains the more efficient choice, as each put operation enables only one get. However, understanding broadcast is critical for more complex coordination patterns and for implementing a graceful shutdown, which we will touch on next.

4. A Complete, Runnable Example

Let's now assemble a complete program. We'll stick with pthread_cond_signal for efficiency but design the program to terminate correctly.

A common challenge is signaling consumers to stop. If they are in an infinite loop (while(1)), they will wait forever on an empty buffer after the producers finish. A simple and effective solution is to use a special "sentinel" value (like -1 for integer data) or a separate shared bool flag to signal termination. When a consumer receives this sentinel, it knows to exit its loop.

Below is a complete program. It combines the buffer_t and functions from the previous lesson with the main function logic we've discussed. For simplicity, we'll have the producers put all their items in the buffer and then the main thread will wait a bit before we assume the consumers have finished, avoiding a complex shutdown mechanism for now.

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <assert.h>

#define BUFFER_SIZE 10
#define NUM_PRODUCERS 2
#define NUM_CONSUMERS 4
#define PRODUCTIONS_PER_PRODUCER 20

// The shared buffer structure
typedef struct {
    int buf[BUFFER_SIZE];
    int occupied;
    int nextin;
    int nextout;
    pthread_mutex_t mutex;
    pthread_cond_t more; // a.k.a 'not_empty'
    pthread_cond_t less; // a.k.a 'not_full'
} buffer_t;

// The argument struct for each thread
typedef struct {
    int thread_id;
    buffer_t *buffer;
} thread_args_t;

void init_buffer(buffer_t *b) {
    b->occupied = 0;
    b->nextin = 0;
    b->nextout = 0;
    pthread_mutex_init(&b->mutex, NULL);
    pthread_cond_init(&b->more, NULL);
    pthread_cond_init(&b->less, NULL);
}

// Producer function from previous lesson
void* producer(void* arg) {
    thread_args_t* args = (thread_args_t*)arg;
    buffer_t* b = args->buffer;
    int id = args->thread_id;

    for (int i = 0; i < PRODUCTIONS_PER_PRODUCER; i++) {
        int item = (id * PRODUCTIONS_PER_PRODUCER) + i; // Unique item
        
        pthread_mutex_lock(&b->mutex);
        while (b->occupied >= BUFFER_SIZE) {
            pthread_cond_wait(&b->less, &b->mutex);
        }

        assert(b->occupied < BUFFER_SIZE);
        
        printf("Producer %d: putting item %d\n", id, item);
        b->buf[b->nextin] = item;
        b->nextin = (b->nextin + 1) % BUFFER_SIZE;
        b->occupied++;

        pthread_cond_signal(&b->more);
        pthread_mutex_unlock(&b->mutex);

        // sleep(1); // Optional: slow down producer to see consumers work
    }
    printf("Producer %d finished.\n", id);
    return NULL;
}

// Consumer function from previous lesson
void* consumer(void* arg) {
    thread_args_t* args = (thread_args_t*)arg;
    buffer_t* b = args->buffer;
    int id = args->thread_id;
    
    // Total items to consume is known
    int total_items = NUM_PRODUCERS * PRODUCTIONS_PER_PRODUCER;
    int items_per_consumer = total_items / NUM_CONSUMERS;
    // Simple way to distribute work. A real system might be more dynamic.
    
    for (int i = 0; i < items_per_consumer; i++) {
        pthread_mutex_lock(&b->mutex);
        while(b->occupied <= 0) {
            pthread_cond_wait(&b->more, &b->mutex);
        }

        assert(b->occupied > 0);
        
        int item = b->buf[b->nextout];
        b->nextout = (b->nextout + 1) % BUFFER_SIZE;
        b->occupied--;

        printf("Consumer %d: got item %d\n", id, item);

        pthread_cond_signal(&b->less);
        pthread_mutex_unlock(&b->mutex);
    }
    printf("Consumer %d finished.\n", id);
    return NULL;
}


int main() {
    buffer_t shared_buffer;
    pthread_t producers[NUM_PRODUCERS];
    pthread_t consumers[NUM_CONSUMERS];
    thread_args_t producer_args[NUM_PRODUCERS];
    thread_args_t consumer_args[NUM_CONSUMERS];

    init_buffer(&shared_buffer);

    for (int i = 0; i < NUM_PRODUCERS; i++) {
        producer_args[i] = (thread_args_t){ .thread_id = i, .buffer = &shared_buffer };
        pthread_create(&producers[i], NULL, producer, &producer_args[i]);
    }

    for (int i = 0; i < NUM_CONSUMERS; i++) {
        consumer_args[i] = (thread_args_t){ .thread_id = i, .buffer = &shared_buffer };
        pthread_create(&consumers[i], NULL, consumer, &consumer_args[i]);
    }

    for (int i = 0; i < NUM_PRODUCERS; i++) {
        pthread_join(producers[i], NULL);
    }

    for (int i = 0; i < NUM_CONSUMERS; i++) {
        pthread_join(consumers[i], NULL);
    }
    
    pthread_mutex_destroy(&shared_buffer.mutex);
    pthread_cond_destroy(&b->more);
    pthread_cond_destroy(&b->less);

    printf("Main: All threads finished. Final buffer occupancy: %d\n", shared_buffer.occupied);
    return 0;
}

Note: In this example, the consumers only stop after consuming a fixed number of items. This works because the total number of items is known beforehand. A more robust shutdown (e.g., for a server that runs indefinitely) would require a more sophisticated mechanism, often involving pthread_cond_broadcast and a shared boolean flag.

Conclusion

Congratulations on completing the module on condition variables! You have now built and orchestrated a complete, multi-threaded producer-consumer system—one of the foundational patterns in concurrent programming.

Key Takeaways:

• A complete multithreaded program requires a main coordinating thread to initialize shared state, create worker threads, join them, and clean up resources.
• All threads must operate on the same instance of the shared data, passed typically via a pointer in the thread creation call.
• pthread_cond_signal wakes one thread, which is efficient when only one thread can make progress.
• pthread_cond_broadcast wakes all waiting threads, which is necessary when a state change could unblock multiple threads, but it can be less efficient due to contention (the "thundering herd").
• The "wait-in-a-loop" pattern is what makes using pthread_cond_broadcast safe and correct.

Preview of the Next Lesson:
We have now mastered synchronization with mutexes and condition variables. In the next module, we will explore other powerful synchronization primitives. We will start by learning how to initialize, use, and destroy read-write locks (pthread_rwlock), which provide a significant performance boost for data structures that are read far more often than they are written.