Ringbuffer.cpp

#include<iostream>
#include<thread>
#include<atomic>
#include<vector>
#include<cstdint>
#include<memory>
#include<chrono>


//64 bytes not bits

static constexpr size_t CACHE_LINE_SIZE=64;


//array implementation
    template <typename T>
    struct Queue{
        size_t Size;
        std::vector<T> queue;
        std::atomic<size_t> Head_index;
        char padding[CACHE_LINE_SIZE-sizeof(std::atomic<size_t>)];
        std::atomic<size_t> Tail_index;
        
        //init
        Queue(size_t size){
            Size=size;
            queue.resize(size);
            Head_index.store(0);
            Tail_index.store(0);
        }
        //enqueue 
        bool EnQueue(T value){
        
        //check if full
            if(isFull()){
                return false; 
            }

            size_t index=Tail_index.load();
            size_t expected_index=(index+1)%Size;
            while(!Tail_index.compare_exchange_strong(index,expected_index)){
                //update incase the index and expected index got changed in the other Thread
                size_t index=Tail_index.load();
                size_t expected_index=(index+1)%Size;
                if(isFull()){ //incase it got full by another thread
                    return false;
                }
            }

            queue[index]=value;
            return true;
        }


    //dequeue
    bool Dequeue(T& val){
        if(isEmpty()){
           return false;
        }

        size_t index=Head_index.load();
        size_t expected_index=(index+1)%Size;

        while(!Head_index.compare_exchange_strong(index,expected_index)){
            //update incase the index and expected index got changed in the other Thread
            
            index=Head_index.load();
            expected_index=(index+1)%Size;
    
            if(isEmpty()){ //incase it got full by another thread
                return false;
            }
        }
        val =queue[index];
        return true;
    }

    //empty
    bool isEmpty(){
        auto Headval=Head_index.load();
        auto Tailval=Tail_index.load();
        if( Headval==Tailval ){
            return true;
        }
        return false;
    }

    bool isFull(){
        size_t Headval=Head_index.load();
        size_t Tailval=Tail_index.load();

        if((Tailval+1)%Size ==Headval ){
            return true;
        }
        return false;
    }
};

double producer_thread(Queue<uint32_t>& queue, uint32_t operations_per_thread) {
    auto start_time = std::chrono::high_resolution_clock::now();
    uint32_t val;
    for (uint32_t i = 0; i < operations_per_thread; ++i) {
        uint32_t value=i;
        // while (!queue.DeQueue(value)) {
        //     // Spin-wait if the queue is full. This is common in benchmarking
        //     // to keep the thread busy and measuring performance under contention.
        // }
        while(!queue.EnQueue(val)){ }
    }

    auto end_time = std::chrono::high_resolution_clock::now();
    return std::chrono::duration<double, std::nano>(end_time - start_time).count();
}
double consumer_thread(Queue<uint32_t>& queue,  uint32_t operations_per_thread) {
    auto start_time = std::chrono::high_resolution_clock::now();
    uint32_t val;
    for (uint32_t i = 0; i < operations_per_thread; ++i) {
        uint32_t value=i;
        // while (!queue.EnQueue(value)) {
        //     // Spin-wait if the queue is full. This is common in benchmarking
        //     // to keep the thread busy and measuring performance under contention.
        // }
        while(!queue.Dequeue(val)){ }
    }

    auto end_time = std::chrono::high_resolution_clock::now();
    return std::chrono::duration<double, std::nano>(end_time - start_time).count();
}


int main() {
    const size_t num_threads = 6;
    const uint32_t operations_per_thread = 300000; 

    const size_t num_producers=3;
    const size_t num_consumers=3;

    Queue<uint32_t> queue((num_threads*operations_per_thread) + 400000);
    std::vector<std::thread> threads;
    //-------------------This is for the max contention testing
    //--------------------------------------------
    //std::vector<double> thread_times(num_threads);

    std::vector<double>producer_times(num_producers);
    std::vector<double>consumer_times(num_consumers);
    

    std::cout << "Starting benchmark with " << num_threads << " threads and "
              << operations_per_thread << " operations per thread.\n";

    auto total_start_time = std::chrono::high_resolution_clock::now();

    //---------This is the test for max contention with all threads doing both producing and consuming
    //----------------------------------------------------
    // // Create and launch all worker threads.
    // for (size_t i = 0; i < num_threads; ++i) {
    //     // The lambda function captures queue, i, and a reference to the specific
    //     // slot in thread_times to store its result.
    //     threads.emplace_back([&queue, i, operations_per_thread, &thread_times]() {
    //         thread_times[i] = producer_thread(queue, i, operations_per_thread);
    //         thread_times[i+1]=consumer_thread(queue,i,operations_per_thread);
    //     });
    // }


    // Create and launch producer threads
    for (size_t i = 0; i < num_producers; ++i) {
        threads.emplace_back([&queue, operations_per_thread, &producer_times, i]() {
            producer_times[i] = producer_thread(queue, operations_per_thread);
        });
    }

    // Create and launch consumer threads
    for (size_t i = 0; i < num_consumers; ++i) {
        threads.emplace_back([&queue, operations_per_thread, &consumer_times, i]() {
            consumer_times[i] = consumer_thread(queue, operations_per_thread);
        });
    }

    for (auto& t : threads) {
        t.join();
    }


    //------------------------ORIGINAL TESTING
    //-------------------------------------------------------------------------------------------
    // auto total_end_time = std::chrono::high_resolution_clock::now();
    // double total_time_ns = std::chrono::duration<double, std::nano>(total_end_time - total_start_time).count();

    // uint32_t total_operations = num_threads * operations_per_thread;
    // double average_time_per_op_ns = total_time_ns / total_operations;

    // std::cout << "\nBenchmark Results:\n";
    // std::cout << "Total time elapsed: " << total_time_ns / 1e6 << " ms\n";
    // std::cout << "Total operations: " << total_operations << "\n";
    // std::cout << "Total throughput: " << total_operations / (total_time_ns / 1e9) << " ops/sec\n";
    // std::cout << "Average time per operation: " << average_time_per_op_ns << " ns\n";
    
    // double sum_thread_times = 0.0;
    // for (size_t i = 0; i < num_threads; ++i) {
    //     std::cout << "Thread " << i << " time: " << thread_times[i] / 1e6 << " ms\n";
    //     sum_thread_times += thread_times[i];
    // }
    // std::cout << "Average thread time: " << (sum_thread_times / num_threads) / 1e6 << " ms\n";

    auto total_end_time = std::chrono::high_resolution_clock::now();
    double total_time_ns = std::chrono::duration<double, std::nano>(total_end_time - total_start_time).count();

    uint32_t total_operations = (num_producers + num_consumers) * operations_per_thread;
    double average_time_per_op_ns = total_time_ns / total_operations;

    std::cout << "\nBenchmark Results:\n";
    std::cout << "Total time elapsed: " << total_time_ns / 1e6 << " ms\n";
    std::cout << "Total operations: " << total_operations << "\n";
    std::cout << "Total throughput: " << total_operations / (total_time_ns / 1e9) << " ops/sec\n";
    std::cout << "Average time per operation: " << average_time_per_op_ns << " ns\n";

    
    return 0;
}