best_so_far_17-01-25.cc

#define _GNU_SOURCE
#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>
 
#include "ring_buffer.hh"
 
// // 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) {}
 
//     ~RingBuffer() { // Destructor for logging
//         std::cout << "RingBuffer destructor called.\n";
//     }
 
//     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));
        }
    }
 
    ~BufferPool()
    {  // Destructor for logging
        std::cout << "BufferPool destructor called.\n";
    }
 
    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
std::vector<std::atomic<bool>> threadDone(
    7);  // Flags to signal when each thread has 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,
                  std::function<void(std::shared_ptr<DataStruct>&)> operation,
                  size_t                                            threadIndex)
{
    pin_thread_to_core(core_id);
    auto start_time = std::chrono::high_resolution_clock::now();
 
    while(!kernelDone || !inputQueue.is_empty() || !threadDone[threadIndex - 1].load())
    {
        auto buffer = inputQueue.pop();
        if(buffer)
        {
            operation(buffer);  // Perform the operation on the buffer
 
            // Append stage-specific metadata
            strncat(buffer->metadata,
                    stage,
                    sizeof(buffer->metadata) - strlen(buffer->metadata) - 1);
 
            uint64_t bytesProcessed = buffer->size * sizeof(int16_t);
            totalBytesProcessed += bytesProcessed;
 
            outputQueue.push(buffer);  // Push the processed buffer to the next stage
        }
    }
 
    // Signal completion to the next thread
    threadDone[threadIndex].store(true);
 
    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
 
    // Signal the first worker thread
    threadDone[0].store(true);
}
 
// Consumer thread to ensure the final buffer is emptied
void consumerThread(RingBuffer<DataStruct>& inputQueue,
                    BufferPool&             pool,
                    size_t                  core_id,
                    size_t                  threadIndex)
{
    pin_thread_to_core(core_id);
    while(!kernelDone || !inputQueue.is_empty() || !threadDone[threadIndex - 1].load())
    {
        auto buffer = inputQueue.pop();
        if(!buffer)
            continue;  // Wait for remaining data
        pool.release(buffer);
    }
 
    // Signal completion
    threadDone[threadIndex].store(true);
 
    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};
 
    // Initialize thread completion flags
    //    threadDone = std::vector<std::atomic_bool>(7, false);
    // threadDone.resize(7);
    for(auto& flag : threadDone)
    {
        flag.store(false);
    }
 
    // Example operations for each worker thread
    auto operationExample = [](std::shared_ptr<DataStruct>& buffer) {
        for(size_t i = 0; i < buffer->size; ++i)
        {
            buffer->data[i] += 1;  // Increment each data element by 1
        }
    };
 
    // 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,
                                operationExample,
                                1);
    std::thread formEventsThread(workerThread,
                                 std::ref(checkDataQueue),
                                 std::ref(formEventsQueue),
                                 std::ref(pool),
                                 " | Events Formed",
                                 std::ref(totalBytesForm),
                                 2,
                                 operationExample,
                                 2);
    std::thread rawProcessThread(workerThread,
                                 std::ref(formEventsQueue),
                                 std::ref(rawBufferQueue),
                                 std::ref(pool),
                                 " | Algorithm Processed",
                                 std::ref(totalBytesRaw),
                                 3,
                                 operationExample,
                                 3);
    std::thread zsProcessThread(workerThread,
                                std::ref(rawBufferQueue),
                                std::ref(zsBufferQueue),
                                std::ref(pool),
                                " | ZS Processed",
                                std::ref(totalBytesZS),
                                4,
                                operationExample,
                                4);
    std::thread mwdProcessThread(workerThread,
                                 std::ref(zsBufferQueue),
                                 std::ref(mwdBufferQueue),
                                 std::ref(pool),
                                 " | MWD Processed",
                                 std::ref(totalBytesMWD),
                                 5,
                                 operationExample,
                                 5);
    std::thread finalConsumerThread(
        consumerThread, std::ref(mwdBufferQueue), std::ref(pool), 6, 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;
}