The producer-consumer problem is a classic synchronization challenge in concurrent programimng. This article demonstrates various implementations using C++11 threading and synchronization primitives.

Shared Code

#include <cstdlib>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

constexpr int BUFFER_CAPACITY = 10;
constexpr int TOTAL_ITEMS = 1000;
std::mutex console_mutex;

Single Producer Single Consmuer

struct CircularBuffer {
    int data[BUFFER_CAPACITY];
    size_t read_pos = 0;
    size_t write_pos = 0;
    std::mutex buffer_mutex;
    std::condition_variable not_full;
    std::condition_variable not_empty;
};

CircularBuffer global_buffer;

void produce(CircularBuffer* buf, int value) {
    std::unique_lock<std::mutex> lock(buf->buffer_mutex);
    while (((buf->write_pos + 1) % BUFFER_CAPACITY) == buf->read_pos) {
        {
            std::lock_guard<std::mutex> console_lock(console_mutex);
            std::cout << "Buffer full, waiting..." << std::endl;
        }
        buf->not_full.wait(lock);
    }
    
    buf->data[buf->write_pos] = value;
    buf->write_pos = (buf->write_pos + 1) % BUFFER_CAPACITY;
    buf->not_empty.notify_all();
}

int consume(CircularBuffer* buf) {
    std::unique_lock<std::mutex> lock(buf->buffer_mutex);
    while (buf->write_pos == buf->read_pos) {
        {
            std::lock_guard<std::mutex> console_lock(console_mutex);
            std::cout << "Buffer empty, waiting..." << std::endl;
        }
        buf->not_empty.wait(lock);
    }
    
    int value = buf->data[buf->read_pos];
    buf->read_pos = (buf->read_pos + 1) % BUFFER_CAPACITY;
    buf->not_full.notify_all();
    return value;
}

void producer_thread() {
    for (int i = 1; i <= TOTAL_ITEMS; ++i) {
        produce(&global_buffer, i);
        {
            std::lock_guard<std::mutex> lock(console_mutex);
            std::cout << "Produced item: " << i << std::endl;
        }
    }
}

void consumer_thread() {
    int consumed_count = 0;
    while (consumed_count < TOTAL_ITEMS) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        int item = consume(&global_buffer);
        {
            std::lock_guard<std::mutex> lock(console_mutex);
            std::cout << "Consumed item: " << item << std::endl;
        }
        ++consumed_count;
    }
}

Multiple Consumers Single Producer

struct SharedBuffer {
    int items[BUFFER_CAPACITY];
    size_t read_index = 0;
    size_t write_index = 0;
    size_t produced_count = 0;
    std::mutex buffer_lock;
    std::mutex counter_lock;
    std::condition_variable space_available;
    std::condition_variable data_available;
};

SharedBuffer shared_buffer;

void multi_produce(SharedBuffer* buf, int value) {
    std::unique_lock<std::mutex> lock(buf->buffer_lock);
    while (((buf->write_index + 1) % BUFFER_CAPACITY) == buf->read_index) {
        buf->space_available.wait(lock);
    }
    
    buf->items[buf->write_index] = value;
    buf->write_index = (buf->write_index + 1) % BUFFER_CAPACITY;
    buf->data_available.notify_all();
}

int multi_consume(SharedBuffer* buf) {
    std::unique_lock<std::mutex> lock(buf->buffer_lock);
    while (buf->write_index == buf->read_index) {
        buf->data_available.wait(lock);
    }
    
    int value = buf->items[buf->read_index];
    buf->read_index = (buf->read_index + 1) % BUFFER_CAPACITY;
    buf->space_available.notify_all();
    return value;
}

void producer_worker() {
    for (int i = 1; i <= TOTAL_ITEMS; ++i) {
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
        multi_produce(&shared_buffer, i);
        {
            std::lock_guard<std::mutex> lock(console_mutex);
            std::cout << "Producer " << std::this_thread::get_id() 
                     << " produced: " << i << std::endl;
        }
    }
}

void consumer_worker() {
    while (true) {
        std::unique_lock<std::mutex> lock(shared_buffer.counter_lock);
        if (shared_buffer.produced_count >= TOTAL_ITEMS) break;
        
        int item = multi_consume(&shared_buffer);
        ++shared_buffer.produced_count;
        lock.unlock();
        
        {
            std::lock_guard<std::mutex> console_lock(console_mutex);
            std::cout << "Consumer " << std::this_thread::get_id() 
                     << " consumed: " << item << std::endl;
        }
    }
}

Implemantation Notes

• Buffer capacity of N can actually hold N-1 items due to circular buffer implementation constraints
• Console output requires synchronization across threads
• Condition variables prevent busy waiting and enable efficient thread coordination
• Proper lock management ensures thread safety and prevents deadlocks