thread_buffer_management.cpp

#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
{
    int16_t     data[60 * 1024 * 1024 / sizeof(int16_t)];  // 60 MB worth of int16_t data
    size_t      size;                                      // Actual size of the data used
    std::string metadata;                                  // Additional metadata
};
 
// Thread-safe queue for buffer management
template<typename T>
class ThreadSafeQueue
{
  private:
    std::queue<std::shared_ptr<T>> queue;
    std::mutex                     mutex;
    std::condition_variable        cv;
    bool                           stop = false;
    size_t                         max_size;  // Maximum queue size
 
  public:
    explicit ThreadSafeQueue(size_t max_size) : max_size(max_size) {}
 
    void push(const std::shared_ptr<T>& item)
    {
        {
            std::unique_lock<std::mutex> lock(mutex);
            cv.wait(lock, [this]() { return stop || queue.size() < max_size; });
            if(stop)
                return;  // Stop pushing if the queue is shutting down
            queue.push(item);
        }
        cv.notify_all();
    }
 
    std::shared_ptr<T> pop()
    {
        std::unique_lock<std::mutex> lock(mutex);
        cv.wait(lock, [this]() { return stop || !queue.empty(); });
        if(queue.empty())
        {
            return nullptr;
        }
        auto item = queue.front();
        queue.pop();
        cv.notify_all();
        return item;
    }
 
    void shutdown()
    {
        {
            std::lock_guard<std::mutex> lock(mutex);
            stop = true;
        }
        cv.notify_all();
    }
};
 
// Function for Kernel Buffer and UDP Server
auto kernelBufferThread = [](ThreadSafeQueue<DataStruct>& bufferQueue) {
    while(true)
    {
        auto buffer  = std::make_shared<DataStruct>();
        buffer->size = (rand() % (60 * 1024 * 1024 / sizeof(int16_t))) +
                       1000;  // Simulate variable size between 1,000 and max capacity
        std::fill(buffer->data,
                  buffer->data + buffer->size,
                  rand() % 100);                               // Fill with random data
        buffer->metadata = "Generated by kernelBufferThread";  // Example metadata
        bufferQueue.push(buffer);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
};
 
// Function for Data Checking
auto checkDataThread = [](ThreadSafeQueue<DataStruct>& inputQueue,
                          ThreadSafeQueue<DataStruct>& outputQueue) {
    while(true)
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            break;  // Exit if queue is shutting down
        // Simulate data checking (e.g., verifying integrity)
        buffer->metadata += " | Checked";
        outputQueue.push(buffer);
    }
};
 
// Function for Forming Events
auto formEventsThread = [](ThreadSafeQueue<DataStruct>& inputQueue,
                           ThreadSafeQueue<DataStruct>& outputQueue) {
    while(true)
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            break;  // Exit if queue is shutting down
        // Simulate event formation (e.g., extracting relevant data)
        buffer->metadata += " | Events Formed";
        outputQueue.push(buffer);
    }
};
 
// Function for Processing Algorithms
auto processAlgorithmThread = [](ThreadSafeQueue<DataStruct>& inputQueue,
                                 ThreadSafeQueue<DataStruct>& outputQueue) {
    while(true)
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            break;  // Exit if queue is shutting down
        // Simulate algorithm processing (prescale, ZS, MWD, etc.)
        buffer->metadata += " | Algorithm Processed";
        outputQueue.push(buffer);
    }
};
 
int main()
{
    const size_t max_queue_size = 10;  // Example maximum size for each queue
 
    // Define thread-safe queues for each step
    ThreadSafeQueue<DataStruct> kernelBufferQueue(max_queue_size);
    ThreadSafeQueue<DataStruct> checkDataQueue(max_queue_size);
    ThreadSafeQueue<DataStruct> formEventsQueue(max_queue_size);
    ThreadSafeQueue<DataStruct> rawBufferQueue(max_queue_size);
    ThreadSafeQueue<DataStruct> zsBufferQueue(max_queue_size);
    ThreadSafeQueue<DataStruct> mwdBufferQueue(max_queue_size);
 
    // Launch threads for data processing
    std::thread kernelThread(kernelBufferThread, std::ref(kernelBufferQueue));
    std::thread checkData(
        checkDataThread, std::ref(kernelBufferQueue), std::ref(checkDataQueue));
    std::thread formEvents(
        formEventsThread, std::ref(checkDataQueue), std::ref(formEventsQueue));
    std::thread rawProcess(
        processAlgorithmThread, std::ref(formEventsQueue), std::ref(rawBufferQueue));
    std::thread zsProcess(
        processAlgorithmThread, std::ref(rawBufferQueue), std::ref(zsBufferQueue));
    std::thread mwdProcess(
        processAlgorithmThread, std::ref(zsBufferQueue), std::ref(mwdBufferQueue));
 
    // 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;
}