thread_buffer_management_btpool.cpp

#include <algorithm>
#include <array>
#include <atomic>
#include <condition_variable>
#include <cstring>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>
 
// Data structure representing a buffer with metadata
struct DataStruct
{
    std::vector<int16_t> data;           // Dynamically sized data buffer
    size_t               size;           // Actual size of the data used
    char                 metadata[128];  // Fixed-size metadata array
 
    DataStruct(size_t buffer_size) : data(buffer_size), size(0)
    {
        std::fill(std::begin(metadata), std::end(metadata), '\0');
    }
};
 
// Ring buffer for thread-safe buffer management
template<typename T>
class RingBuffer
{
  private:
    std::vector<std::shared_ptr<T>> buffer;
    size_t                          head = 0;
    size_t                          tail = 0;
    size_t                          capacity;
    std::atomic<size_t>             size = 0;
    std::mutex                      mutex;
    std::condition_variable         cv_push;
    std::condition_variable         cv_pop;
    bool                            stop = false;
 
  public:
    explicit RingBuffer(size_t max_size) : capacity(max_size), buffer(max_size) {}
 
    void push(const std::shared_ptr<T>& item)
    {
        std::unique_lock<std::mutex> lock(mutex);
        cv_push.wait(lock, [this]() { return stop || size < capacity; });
        if(stop)
            return;
        buffer[head] = item;
        head         = (head + 1) % capacity;
        ++size;
        cv_pop.notify_one();
    }
 
    std::shared_ptr<T> pop()
    {
        std::unique_lock<std::mutex> lock(mutex);
        cv_pop.wait(lock, [this]() { return stop || size > 0; });
        if(size == 0)
            return nullptr;
        auto item = buffer[tail];
        tail      = (tail + 1) % capacity;
        --size;
        cv_push.notify_one();
        return item;
    }
 
    void shutdown()
    {
        {
            std::lock_guard<std::mutex> lock(mutex);
            stop = true;
        }
        cv_push.notify_all();
        cv_pop.notify_all();
    }
};
 
// Pre-allocated buffer pool for memory reuse
class BufferPool
{
  private:
    std::vector<std::shared_ptr<DataStruct>> pool;
    std::mutex                               mutex;
 
  public:
    BufferPool(size_t pool_size, size_t buffer_size)
    {
        for(size_t i = 0; i < pool_size; ++i)
        {
            pool.emplace_back(std::make_shared<DataStruct>(buffer_size));
        }
    }
 
    std::shared_ptr<DataStruct> acquire()
    {
        std::lock_guard<std::mutex> lock(mutex);
        if(pool.empty())
        {
            return std::make_shared<DataStruct>(60 * 1024 * 1024 / sizeof(int16_t));
        }
        else
        {
            auto buffer = pool.back();
            pool.pop_back();
            return buffer;
        }
    }
 
    void release(std::shared_ptr<DataStruct> buffer)
    {
        std::lock_guard<std::mutex> lock(mutex);
        pool.push_back(buffer);
    }
};
 
// Thread pool for task management
class ThreadPool
{
  private:
    std::vector<std::thread>          workers;
    std::queue<std::function<void()>> tasks;
    std::mutex                        mutex;
    std::condition_variable           cv;
    bool                              stop = false;
 
  public:
    explicit ThreadPool(size_t thread_count)
    {
        for(size_t i = 0; i < thread_count; ++i)
        {
            workers.emplace_back([this] {
                while(true)
                {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(mutex);
                        cv.wait(lock, [this]() { return stop || !tasks.empty(); });
                        if(stop && tasks.empty())
                            return;
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            });
        }
    }
 
    void enqueue(std::function<void()> task)
    {
        {
            std::lock_guard<std::mutex> lock(mutex);
            tasks.emplace(std::move(task));
        }
        cv.notify_one();
    }
 
    void shutdown()
    {
        {
            std::lock_guard<std::mutex> lock(mutex);
            stop = true;
        }
        cv.notify_all();
        for(auto& worker : workers)
        {
            worker.join();
        }
    }
};
 
// Kernel buffer simulation task
void kernelBufferTask(RingBuffer<DataStruct>& bufferQueue, BufferPool& pool)
{
    auto buffer = pool.acquire();
    buffer->size =
        (rand() % (60 * 1024 * 1024 / sizeof(int16_t))) + 1000;  // Simulate variable size
    std::fill(buffer->data.begin(), buffer->data.begin() + buffer->size, rand() % 100);
    strncpy(
        buffer->metadata, "Generated by kernelBufferThread", sizeof(buffer->metadata));
    bufferQueue.push(buffer);
}
 
// General processing task
void processingTask(RingBuffer<DataStruct>& inputQueue,
                    RingBuffer<DataStruct>& outputQueue,
                    BufferPool&             pool,
                    const char*             stage)
{
    auto buffer = inputQueue.pop();
    if(buffer)
    {
        strncat(buffer->metadata,
                stage,
                sizeof(buffer->metadata) - strlen(buffer->metadata) - 1);
        outputQueue.push(buffer);
    }
}
 
int main()
{
    const size_t max_queue_size   = 10;  // Maximum size for each ring buffer
    const size_t buffer_pool_size = 20;  // Pre-allocated buffer pool size
    const size_t buffer_size =
        60 * 1024 * 1024 / sizeof(int16_t);  // Buffer size in elements
    const size_t thread_pool_size = 6;       // Number of threads in the pool
 
    // Create buffer pool
    BufferPool pool(buffer_pool_size, buffer_size);
 
    // Define ring buffers for each step
    RingBuffer<DataStruct> kernelBufferQueue(max_queue_size);
    RingBuffer<DataStruct> checkDataQueue(max_queue_size);
    RingBuffer<DataStruct> formEventsQueue(max_queue_size);
    RingBuffer<DataStruct> rawBufferQueue(max_queue_size);
    RingBuffer<DataStruct> zsBufferQueue(max_queue_size);
    RingBuffer<DataStruct> mwdBufferQueue(max_queue_size);
 
    // Create thread pool
    ThreadPool threadPool(thread_pool_size);
 
    // Enqueue tasks
    for(int i = 0; i < 100; ++i)
    {
        threadPool.enqueue([&]() { kernelBufferTask(kernelBufferQueue, pool); });
        threadPool.enqueue([&]() {
            processingTask(kernelBufferQueue, checkDataQueue, pool, " | Checked");
        });
        threadPool.enqueue([&]() {
            processingTask(checkDataQueue, formEventsQueue, pool, " | Events Formed");
        });
        threadPool.enqueue([&]() {
            processingTask(
                formEventsQueue, rawBufferQueue, pool, " | Algorithm Processed");
        });
        threadPool.enqueue([&]() {
            processingTask(rawBufferQueue, zsBufferQueue, pool, " | ZS Processed");
        });
        threadPool.enqueue([&]() {
            processingTask(zsBufferQueue, mwdBufferQueue, pool, " | MWD Processed");
        });
    }
 
    // Simulate running for a limited time
    std::this_thread::sleep_for(std::chrono::seconds(10));
 
    // Shutdown thread pool
    threadPool.shutdown();
 
    // Shutdown all queues
    kernelBufferQueue.shutdown();
    checkDataQueue.shutdown();
    formEventsQueue.shutdown();
    rawBufferQueue.shutdown();
    zsBufferQueue.shutdown();
    mwdBufferQueue.shutdown();
 
    return 0;
}