Low-level concurrency primitives form the foundation of high-performance asynchronous architectures. By wrapping POSIX threading APIs, developers can construct predictable synchronization mechanisms tailored for Linux environments. These abstractions ennable the creation of producer-consumer pipelines where a bounded queue mediates task distribution acrosss a fixed set of worker threads.

POSIX-Based Synchronization and Consumer Queue Architecture

Custom RAII wrappers encapsulate semaphore, mutex, and condition variable lifecycles, preventing resource leaks and standardizing error handling. The underlying data structure employs a circular array to simulate a blocking queue, utilizing index pointers for constant-time insertion and extraction. Condition variables coordinate thread wake-ups when the buffer state changes.

#include <exception>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>
#include <list>
#include <iostream>

class Semaphore {
public:
    explicit Semaphore(int initial = 0) {
        if (sem_init(&handle_, 0, initial) != 0) {
            throw std::runtime_error("Semaphore initialization failed");
        }
    }
    ~Semaphore() { sem_destroy(&handle_); }

    bool acquire() { return sem_wait(&handle_) == 0; }
    bool release() { return sem_post(&handle_) == 0; }

private:
    sem_t handle_;
    Semaphore(const Semaphore&) = delete;
    Semaphore& operator=(const Semaphore&) = delete;
};

class MutexLock {
public:
    MutexLock() {
        if (pthread_mutex_init(&mtx_, nullptr) != 0) {
            throw std::runtime_error("Mutex initialization failed");
        }
    }
    ~MutexLock() { pthread_mutex_destroy(&mtx_); }

    void lock() { pthread_mutex_lock(&mtx_); }
    void unlock() { pthread_mutex_unlock(&mtx_); }

    pthread_mutex_t* getRaw() { return &mtx_; }

private:
    pthread_mutex_t mtx_;
    MutexLock(const MutexLock&) = delete;
    MutexLock& operator=(const MutexLock&) = delete;
};

class ConditionVar {
public:
    ConditionVar() {
        if (pthread_cond_init(&cv_, nullptr) != 0) {
            throw std::runtime_error("Condition variable initialization failed");
        }
    }
    ~ConditionVar() { pthread_cond_destroy(&cv_); }

    bool waitFor(MutexLock& mutex) {
        pthread_mutex_lock(mutex.getRaw());
        int ret = pthread_cond_wait(&cv_, mutex.getRaw());
        pthread_mutex_unlock(mutex.getRaw());
        return ret == 0;
    }

    bool waitForTimeout(MutexLock& mutex, struct timespec& ts) {
        pthread_mutex_lock(mutex.getRaw());
        int ret = pthread_cond_timedwait(&cv_, mutex.getRaw(), &ts);
        pthread_mutex_unlock(mutex.getRaw());
        return ret == 0;
    }

    bool notifyOne() { return pthread_cond_signal(&cv_) == 0; }
    bool notifyAll() { return pthread_cond_broadcast(&cv_) == 0; }

private:
    pthread_cond_t cv_;
    ConditionVar(const ConditionVar&) = delete;
    ConditionVar& operator=(const ConditionVar&) = delete;
};

template <typename TaskType>
class PosixThreadPool {
public:
    PosixThreadPool(int workerCount, int maxCapacity)
        : workerCount_(workerCount), maxCapacity_(maxCapacity), stopFlag_(false), threads_(nullptr)
    {
        if (workerCount_ <= 0 || maxCapacity_ <= 0) {
            throw std::invalid_argument("Invalid configuration parameters");
        }
        threads_ = new pthread_t[workerCount_];
        for (int i = 0; i < workerCount_; ++i) {
            if (pthread_create(&threads_[i], nullptr, runnerEntry, this) != 0) {
                delete[] threads_;
                throw std::runtime_error("Thread creation failed");
            }
            pthread_detach(threads_[i]);
        }
    }

    ~PosixThreadPool() {
        stopFlag_ = true;
        queueCond_.notifyAll();
        delete[] threads_;
    }

    bool submit(TaskType* task) {
        queueMtx_.lock();
        if (workBuffer_.size() >= maxCapacity_) {
            queueMtx_.unlock();
            return false;
        }
        workBuffer_.push_back(task);
        queueMtx_.unlock();
        queueSem_.release();
        return true;
    }

    static void* runnerEntry(void* arg) {
        PosixThreadPool* pool = static_cast<PosixThreadPool*>(arg);
        pool->executeLoop();
        return nullptr;
    }

    void executeLoop() {
        while (!stopFlag_) {
            queueSem_.acquire();
            queueMtx_.lock();
            if (workBuffer_.empty()) {
                queueMtx_.unlock();
                continue;
            }
            TaskType* currentTask = workBuffer_.front();
            workBuffer_.pop_front();
            queueMtx_.unlock();

            if (currentTask) {
                currentTask->process();
            }
        }
    }

private:
    int workerCount_;
    int maxCapacity_;
    bool stopFlag_;
    pthread_t* threads_;
    MutexLock queueMtx_;
    Semaphore queueSem_;
    ConditionVar queueCond_;
    std::list<TaskType*> workBuffer_;
};

Modern C++ Concurrent Abstraction

Standard library concurrency components eliminate manual thread management and reduce boilerplate overhead. C++11 and later introduce high-level constructs such as std::thread, std::condition_variable, std::atomic, and std::packaged_task. These primitives enable self-managing thread pools that support asynchronous task sbumission, result retrieval via futures, and graceful shutdown sequences without relying on platform-specific APIs.

#include <iostream>
#include <thread>
#include <mutex>
#include <future>
#include <atomic>
#include <queue>
#include <vector>
#include <condition_variable>
#include <functional>

class StdThreadPool {
private:
    std::atomic<bool> running_{true};
    std::queue<std::packaged_task<void()>> taskQueue_;
    std::vector<std::thread> workerThreads_;
    std::mutex queueMtx_;
    std::condition_variable cv_;

    void workerRoutine() {
        while (true) {
            std::packaged_task<void()> task;
            {
                std::unique_lock<std::mutex> lock(queueMtx_);
                cv_.wait(lock, [this]() {
                    return !running_.load() || !taskQueue_.empty();
                });

                if (running_.load() && taskQueue_.empty()) {
                    return;
                }
                task = std::move(taskQueue_.front());
                taskQueue_.pop();
            }
            task();
        }
    }

public:
    explicit StdThreadPool(size_t threadCount = std::thread::hardware_concurrency()) {
        size_t actualCount = (threadCount > 0) ? threadCount : 2;
        for (size_t i = 0; i < actualCount; ++i) {
            workerThreads_.emplace_back([this]() { workerRoutine(); });
        }
    }

    ~StdThreadPool() {
        shutdown();
    }

    void shutdown() {
        {
            std::lock_guard<std::mutex> lock(queueMtx_);
            running_.store(false);
        }
        cv_.notify_all();
        for (auto& t : workerThreads_) {
            if (t.joinable()) {
                t.join();
            }
        }
    }

    template <typename Func, typename... Args>
    auto submit(Func&& func, Args&&... args) -> std::future<decltype(func(args...))> {
        using ReturnType = decltype(func(args...));

        if (!running_.load()) {
            return std::future<ReturnType>{};
        }

        auto task = std::make_shared<std::packaged_task<ReturnType()>>(
            std::bind(std::forward<Func>(func), std::forward<Args>(args)...)
        );

        std::future<ReturnType> result = task->get_future();
        {
            std::lock_guard<std::mutex> lock(queueMtx_);
            taskQueue_.emplace([task]() { (*task)(); });
        }
        cv_.notify_one();
        return result;
    }
};