Semaphore Initialization

The sem_init() function accepts three parameters:

• Parameter 1: Pointer to the semaphore variable
• Parameter 2: Sharing mode (0 for threads within a process, non-zero for processes)
• Parameter 3: Initial semaphore count

Semaphore vs Mutex Acquisition Order

Acquiring the semaphore before the mutex is critical for performance. This ordering allows threads to block at the semaphore level rather than contending for the lock imediately. If the mutex were acquired first, both semaphore and lock operations would execute sequentially, eliminating any concurrency benefits.

Multi-Producer Multi-Consumer vs Single-Producer Single-Consumer

Both models guarantee exclusive access—one producer pushes while one consumer pops at any given time. The multi-producer/multi-consumer variant excels at overlapping production and consumptoin workloads. Since task creation and task processing typically dominate execution time, parallelizing these phases significantly improves throughput.

Implementation

computation_task.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>

class ComputationTask
{
public:
    ComputationTask()
    {
    }

    ComputationTask(int a, int b) : _operandA(a), _operandB(b), _computedResult(0)
    {
    }

    void Execute()
    {
        _computedResult = _operandA + _operandB;
    }

    void operator()()
    {
        Execute();
    }

    std::string ToString()
    {
        return std::to_string(_operandA) + " + " + std::to_string(_operandB) + " = ?";
    }

    std::string GetResult()
    {
        return std::to_string(_operandA) + " + " + std::to_string(_operandB) 
               + " = " + std::to_string(_computedResult);
    }

private:
    int _operandA;
    int _operandB;
    int _computedResult;
};

circular_buffer.hpp

#pragma once

#include <iostream>
#include <vector>
#include <pthread.h>
#include <semaphore.h>

template <typename T>
class CircularBuffer
{
private:
    void Acquire(sem_t &semaphore)
    {
        sem_wait(&semaphore);
    }

    void Release(sem_t &semaphore)
    {
        sem_post(&semaphore);
    }

public:
    CircularBuffer(size_t capacity)
        : _storage(capacity)
        , _capacity(capacity)
        , _readPos(0)
        , _writePos(0)
    {
        sem_init(&_itemCount, 0, 0);
        sem_init(&_slotCount, 0, capacity);

        pthread_mutex_init(&_readMutex, nullptr);
        pthread_mutex_init(&_writeMutex, nullptr);
    }

    void Produce(T *element)
    {
        Acquire(_slotCount);
        pthread_mutex_lock(&_writeMutex);
        _storage[_writePos] = *element;
        _writePos = (_writePos + 1) % _capacity;
        pthread_mutex_unlock(&_writeMutex);
        Release(_itemCount);
    }

    void Consume(T *element)
    {
        Acquire(_itemCount);
        pthread_mutex_lock(&_readMutex);
        *element = _storage[_readPos];
        _readPos = (_readPos + 1) % _capacity;
        pthread_mutex_unlock(&_readMutex);
        Release(_slotCount);
    }

    ~CircularBuffer()
    {
        sem_destroy(&_itemCount);
        sem_destroy(&_slotCount);
        pthread_mutex_destroy(&_readMutex);
        pthread_mutex_destroy(&_writeMutex);
    }

private:
    std::vector<T> _storage;
    size_t _capacity;
    size_t _readPos;
    size_t _writePos;

    sem_t _itemCount;
    sem_t _slotCount;

    pthread_mutex_t _readMutex;
    pthread_mutex_t _writeMutex;
};

application.cpp

#include "circular_buffer.hpp"
#include "computation_task.hpp"
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <ctime>
#include <cstdlib>

void* ConsumerRoutine(void* args)
{
    CircularBuffer<ComputationTask>* buffer = 
        static_cast<CircularBuffer<ComputationTask>*>(args);
    
    while (true)
    {
        usleep(100000);
        ComputationTask task;
        buffer->Consume(&task);

        task();
        std::cout << "Consumed: " << task.GetResult() << std::endl;
    }
}

void* ProducerRoutine(void* args)
{
    CircularBuffer<ComputationTask>* buffer = 
        static_cast<CircularBuffer<ComputationTask>*>(args);
    
    while (true)
    {
        int x = rand() % 10 + 1;
        usleep(x * 10000);
        int y = rand() % 10 + 1;
        ComputationTask newTask(x, y);

        buffer->Produce(&newTask);
        std::cout << "Produced: " << newTask.ToString() << std::endl;
    }
}

int main()
{
    srand(time(nullptr));
    CircularBuffer<ComputationTask>* buffer = new CircularBuffer<ComputationTask>(5);
    
    pthread_t consumerThread1, consumerThread2, consumerThread3;
    pthread_t producerThread1, producerThread2, producerThread3, producerThread4;

    pthread_create(&consumerThread1, nullptr, ConsumerRoutine, buffer);
    pthread_create(&consumerThread2, nullptr, ConsumerRoutine, buffer);
    pthread_create(&consumerThread3, nullptr, ConsumerRoutine, buffer);

    pthread_create(&producerThread1, nullptr, ProducerRoutine, buffer);
    pthread_create(&producerThread2, nullptr, ProducerRoutine, buffer);
    pthread_create(&producerThread3, nullptr, ProducerRoutine, buffer);
    pthread_create(&producerThread4, nullptr, ProducerRoutine, buffer);

    pthread_join(consumerThread1, nullptr);
    pthread_join(consumerThread2, nullptr);
    pthread_join(consumerThread3, nullptr);
    pthread_join(producerThread1, nullptr);
    pthread_join(producerThread2, nullptr);
    pthread_join(producerThread3, nullptr);
    pthread_join(producerThread4, nullptr);

    delete buffer;
    return 0;
}