thread_buffer_management_bpool.cpp

#include <algorithm>
#include <array>
#include <atomic>
#include <condition_variable>
#include <cstring>
#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);
    }
};
 
// Function for Kernel Buffer and UDP Server
auto kernelBufferThread = [](RingBuffer<DataStruct>& bufferQueue, BufferPool& pool) {
    while(true)
    {
        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);
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
};
 
// Function for Data Checking
auto checkDataThread = [](RingBuffer<DataStruct>& inputQueue,
                          RingBuffer<DataStruct>& outputQueue,
                          BufferPool&             pool) {
    while(true)
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            break;
        strncpy(buffer->metadata + strlen(buffer->metadata),
                " | Checked",
                sizeof(buffer->metadata) - strlen(buffer->metadata));
        outputQueue.push(buffer);
    }
};
 
// Function for Forming Events
auto formEventsThread = [](RingBuffer<DataStruct>& inputQueue,
                           RingBuffer<DataStruct>& outputQueue,
                           BufferPool&             pool) {
    while(true)
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            break;
        strncpy(buffer->metadata + strlen(buffer->metadata),
                " | Events Formed",
                sizeof(buffer->metadata) - strlen(buffer->metadata));
        outputQueue.push(buffer);
    }
};
 
// Function for Processing Algorithms
auto processAlgorithmThread = [](RingBuffer<DataStruct>& inputQueue,
                                 RingBuffer<DataStruct>& outputQueue,
                                 BufferPool&             pool) {
    while(true)
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            break;
        strncpy(buffer->metadata + strlen(buffer->metadata),
                " | Algorithm Processed",
                sizeof(buffer->metadata) - strlen(buffer->metadata));
        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
 
    // 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);
 
    // Launch threads for data processing
    std::thread kernelThread(
        kernelBufferThread, std::ref(kernelBufferQueue), std::ref(pool));
    std::thread checkData(checkDataThread,
                          std::ref(kernelBufferQueue),
                          std::ref(checkDataQueue),
                          std::ref(pool));
    std::thread formEvents(formEventsThread,
                           std::ref(checkDataQueue),
                           std::ref(formEventsQueue),
                           std::ref(pool));
    std::thread rawProcess(processAlgorithmThread,
                           std::ref(formEventsQueue),
                           std::ref(rawBufferQueue),
                           std::ref(pool));
    std::thread zsProcess(processAlgorithmThread,
                          std::ref(rawBufferQueue),
                          std::ref(zsBufferQueue),
                          std::ref(pool));
    std::thread mwdProcess(processAlgorithmThread,
                           std::ref(zsBufferQueue),
                           std::ref(mwdBufferQueue),
                           std::ref(pool));
 
    // Simulate running for a limited time
    std::this_thread::sleep_for(std::chrono::seconds(10));
 
    // Shutdown all queues
    kernelBufferQueue.shutdown();
    checkDataQueue.shutdown();
    formEventsQueue.shutdown();
    rawBufferQueue.shutdown();
    zsBufferQueue.shutdown();
    mwdBufferQueue.shutdown();
 
    // Join threads
    kernelThread.join();
    checkData.join();
    formEvents.join();
    rawProcess.join();
    zsProcess.join();
    mwdProcess.join();
 
    return 0;
}