thread_buffer_management_new.cpp

#include <sched.h>
#include <unistd.h>
#include <array>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstring>
#include <functional>
#include <iomanip>
#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();
    }
 
    bool is_empty()
    {
        std::lock_guard<std::mutex> lock(mutex);
        return size == 0;
    }
};
 
// 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);
    }
};
 
// Global total data tracker
std::atomic<uint64_t> globalTotalBytes{0};
const uint64_t        targetTotalBytes = 10L * 1024 * 1024 * 1024;  // 10 GB
std::atomic<bool>     kernelDone{false};  // Signal when kernelBufferWorker is finished
 
// Pin thread to specific CPU core
void pin_thread_to_core(size_t core_id)
{
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(core_id, &cpuset);
    pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
}
 
// General worker function for processing stages
void workerThread(RingBuffer<DataStruct>& inputQueue,
                  RingBuffer<DataStruct>& outputQueue,
                  BufferPool&             pool,
                  const char*             stage,
                  std::atomic<uint64_t>&  totalBytesProcessed,
                  size_t                  core_id)
{
    pin_thread_to_core(core_id);
    auto start_time = std::chrono::high_resolution_clock::now();
    while(!kernelDone || !inputQueue.is_empty())
    {
        std::vector<std::shared_ptr<DataStruct>> batch;
        {
            auto buffer = inputQueue.pop();
            if(buffer)
            {
                batch.push_back(buffer);
            }
        }
        for(auto& buffer : batch)
        {
            // Process the data and append metadata
            strncat(buffer->metadata,
                    stage,
                    sizeof(buffer->metadata) - strlen(buffer->metadata) - 1);
 
            // Update the total bytes processed
            uint64_t bytesProcessed = buffer->size * sizeof(int16_t);
            totalBytesProcessed += bytesProcessed;
 
            // Push the processed buffer to the next stage
            outputQueue.push(buffer);
        }
    }
    auto                          end_time = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> elapsed  = end_time - start_time;
    double gbits = totalBytesProcessed.load() * 8.0 / 1e9;  // Convert to Gbits
    std::cout << "[" << stage << "] Processing speed: " << std::fixed
              << std::setprecision(2) << (gbits / elapsed.count()) << " Gbit/s\n";
}
 
// Kernel buffer simulation function
void kernelBufferWorker(RingBuffer<DataStruct>& outputQueue,
                        BufferPool&             pool,
                        size_t                  core_id)
{
    pin_thread_to_core(core_id);
    while(globalTotalBytes < targetTotalBytes)
    {
        auto buffer  = pool.acquire();
        buffer->size = (rand() % (60 * 1024 * 1024 / sizeof(int16_t))) +
                       1000;  // Simulate variable size
        memset(buffer->data.data(), rand() % 100, buffer->size * sizeof(int16_t));
        strncpy(buffer->metadata,
                "Generated by kernelBufferWorker",
                sizeof(buffer->metadata));
 
        uint64_t bytesGenerated = buffer->size * sizeof(int16_t);
        globalTotalBytes += bytesGenerated;
 
        outputQueue.push(buffer);
    }
    kernelDone = true;  // Signal that kernel is done producing data
}
 
// Consumer thread to ensure the final buffer is emptied
void consumerThread(RingBuffer<DataStruct>& inputQueue, BufferPool& pool, size_t core_id)
{
    pin_thread_to_core(core_id);
    while(!kernelDone || !inputQueue.is_empty())
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            continue;  // Wait for remaining data
 
        // Simulate consumption of the final data
        pool.release(buffer);
    }
    std::cout << "[Consumer] Final buffer emptied.\n";
}
 
int main()
{
    const size_t max_queue_size   = 20;  // Increased maximum size for each ring buffer
    const size_t buffer_pool_size = 30;  // Increased 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);
 
    // Track total bytes processed for each thread
    std::atomic<uint64_t> totalBytesKernel{0};
    std::atomic<uint64_t> totalBytesCheck{0};
    std::atomic<uint64_t> totalBytesForm{0};
    std::atomic<uint64_t> totalBytesRaw{0};
    std::atomic<uint64_t> totalBytesZS{0};
    std::atomic<uint64_t> totalBytesMWD{0};
 
    // Launch worker threads for each stage
    std::thread kernelThread(
        kernelBufferWorker, std::ref(kernelBufferQueue), std::ref(pool), 0);
    std::thread checkDataThread(workerThread,
                                std::ref(kernelBufferQueue),
                                std::ref(checkDataQueue),
                                std::ref(pool),
                                " | Checked",
                                std::ref(totalBytesCheck),
                                1);
    std::thread formEventsThread(workerThread,
                                 std::ref(checkDataQueue),
                                 std::ref(formEventsQueue),
                                 std::ref(pool),
                                 " | Events Formed",
                                 std::ref(totalBytesForm),
                                 2);
    std::thread rawProcessThread(workerThread,
                                 std::ref(formEventsQueue),
                                 std::ref(rawBufferQueue),
                                 std::ref(pool),
                                 " | Algorithm Processed",
                                 std::ref(totalBytesRaw),
                                 3);
    std::thread zsProcessThread(workerThread,
                                std::ref(rawBufferQueue),
                                std::ref(zsBufferQueue),
                                std::ref(pool),
                                " | ZS Processed",
                                std::ref(totalBytesZS),
                                4);
    std::thread mwdProcessThread(workerThread,
                                 std::ref(zsBufferQueue),
                                 std::ref(mwdBufferQueue),
                                 std::ref(pool),
                                 " | MWD Processed",
                                 std::ref(totalBytesMWD),
                                 5);
    std::thread finalConsumerThread(
        consumerThread, std::ref(mwdBufferQueue), std::ref(pool), 6);
 
    // Wait for threads to complete
    kernelThread.join();
    checkDataThread.join();
    formEventsThread.join();
    rawProcessThread.join();
    zsProcessThread.join();
    mwdProcessThread.join();
    finalConsumerThread.join();
 
    // Shutdown all queues
    kernelBufferQueue.shutdown();
    checkDataQueue.shutdown();
    formEventsQueue.shutdown();
    rawBufferQueue.shutdown();
    zsBufferQueue.shutdown();
    mwdBufferQueue.shutdown();
 
    return 0;
}