Add comparative queue benchmark with moodycamel

joz-k · joz-k · commit 708632861a56 · 2025-08-11T17:33:19.000+02:00
Introduces a new benchmark (queue_comparison_benchmark.cpp) to compare the performance of LockFreeSpscQueue with moodycamel::ReaderWriterQueue using Google Benchmark. Updates CMake to add a SPSC_QUEUE_BUILD_BENCHMARK_COMPARE option, fetches moodycamel's queue if enabled, and builds the new benchmark executable. Existing benchmark logic is refactored to support conditional builds for standard and comparative benchmarks.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -27,9 +27,13 @@ target_include_directories(spsc_queue INTERFACE
 option(SPSC_QUEUE_BUILD_TESTS "Build the project's tests" ON)
 option(SPSC_QUEUE_BUILD_EXAMPLES "Build the project's examples" ON)
 option(SPSC_QUEUE_BUILD_BENCHMARKS "Build the project's benchmarks" OFF)
+option(SPSC_QUEUE_BUILD_BENCHMARK_COMPARE "Build comparative benchmarks against other libraries" OFF)
 
 # For MacOS
-if(APPLE AND (SPSC_QUEUE_BUILD_TESTS OR SPSC_QUEUE_BUILD_EXAMPLES OR SPSC_QUEUE_BUILD_BENCHMARKS))
+if(APPLE AND (SPSC_QUEUE_BUILD_TESTS
+              OR SPSC_QUEUE_BUILD_EXAMPLES
+              OR SPSC_QUEUE_BUILD_BENCHMARKS
+              OR SPSC_QUEUE_BUILD_BENCHMARK_COMPARE))
     # Enable experimental Clang stdlib features (for std::jthreads) when
     # building tests, examples or benchmark
     # As of Apple Clang version 17.0.0, this is still required
@@ -46,7 +50,7 @@ if(SPSC_QUEUE_BUILD_EXAMPLES)
     add_subdirectory(examples)
 endif()
 
-if(SPSC_QUEUE_BUILD_BENCHMARKS)
+if(SPSC_QUEUE_BUILD_BENCHMARKS OR SPSC_QUEUE_BUILD_BENCHMARK_COMPARE)
     add_subdirectory(benchmarks)
 endif()
 
diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt
@@ -1,6 +1,6 @@
 # This file handles the building of performance benchmarks.
 
-# 1. Fetch Google Benchmark
+# 1. Fetch Google Benchmark (needed for all benchmarks)
 # Use FetchContent to download and prepare Google Benchmark. This is the modern
 # CMake approach and does not require the user to have it installed system-wide.
 include(FetchContent)
@@ -20,13 +20,35 @@ FetchContent_MakeAvailable(GoogleBenchmark)
 # 2. Find the threading library
 find_package(Threads REQUIRED)
 
-# 3. Define the benchmark executable
-add_executable(queue_benchmark queue_benchmark.cpp)
+# 3. Build the standard, internal benchmark if enabled
+if(SPSC_QUEUE_BUILD_BENCHMARKS)
+    add_executable(queue_benchmark queue_benchmark.cpp)
+    target_link_libraries(queue_benchmark PRIVATE
+        spsc_queue
+        benchmark::benchmark
+        Threads::Threads
+    )
+endif()
+
+# 4. Build the new comparative benchmark if enabled
+if(SPSC_QUEUE_BUILD_BENCHMARK_COMPARE)
+    # Fetch moodycamel::ConcurrentQueue
+    FetchContent_Declare(
+        MoodycamelConcurrentQueue
+        GIT_REPOSITORY https://github.com/cameron314/readerwriterqueue.git
+        GIT_TAG        v1.0.7 # Use a specific tag for stability
+    )
+    FetchContent_MakeAvailable(MoodycamelConcurrentQueue)
+
+    # Define the comparative benchmark executable
+    add_executable(queue_comparison_benchmark queue_comparison_benchmark.cpp)
+
+    # Link it against our queue, the moodycamel queue, Google Benchmark, and threads
+    target_link_libraries(queue_comparison_benchmark PRIVATE
+        spsc_queue
+        readerwriterqueue
+        benchmark::benchmark
+        Threads::Threads
+    )
+endif()
 
-# 4. Link the executable against our queue, Google Benchmark, and threads.
-#    The `benchmark::benchmark` target is created by FetchContent_MakeAvailable.
-target_link_libraries(queue_benchmark PRIVATE
-    spsc_queue
-    benchmark::benchmark
-    Threads::Threads
-)
diff --git a/benchmarks/queue_comparison_benchmark.cpp b/benchmarks/queue_comparison_benchmark.cpp
@@ -0,0 +1,285 @@
+#include "LockFreeSpscQueue.h"
+#include "readerwriterqueue.h" // Moodycamel's queue header
+#include <benchmark/benchmark.h>
+#include <thread>
+#include <vector>
+#include <atomic>
+#include <numeric>
+#include <random>
+
+using DataType = int64_t;
+constexpr size_t RandomDataSize       = 4001; // The prime to make patterns less obvious
+constexpr size_t ItemsPerIteration    = 100'025;
+constexpr size_t DefaultQueueCapacity = 65536; // 2^16
+
+// A capacity guaranteed to be larger than ItemsPerIteration.
+constexpr size_t LargeQueueCapacity   = 262144; // 2^18
+
+// Shared Test Data Generation
+const std::vector<DataType>& get_random_data() {
+    static const auto data = []{
+        std::vector<DataType> d(RandomDataSize);
+        std::mt19937_64 rng(33317); // Fixed seed for test stability
+        std::uniform_int_distribution<DataType> dist;
+        for (auto& val : d) { val = dist(rng); }
+        return d;
+    }();
+    return data;
+}
+
+// Benchmark Group 1: Single Item, Default "Stalling" Capacity
+
+static void BM_ThisQueue_SingleItem_DefaultBuffer(benchmark::State& state) {
+    const auto& random_data = get_random_data();
+    std::vector<DataType> shared_buffer(DefaultQueueCapacity);
+    LockFreeSpscQueue<DataType> queue(shared_buffer);
+
+    std::atomic<bool> verification_failed  = false;
+    std::atomic<bool> consumer_should_stop = false;
+    std::jthread consumer([&] {
+        size_t i = 0;
+        while (!consumer_should_stop.load(std::memory_order_relaxed)) {
+            auto scope = queue.prepare_read(1);
+            if (scope.get_items_read() == 1) {
+                if (scope.get_block1()[0] != random_data[i % RandomDataSize]) verification_failed.store(true);
+                i++;
+            } else {
+                std::this_thread::yield();
+            }
+        }
+    });
+
+    size_t total_written = 0;
+    for (auto _ : state) {
+        for (size_t n = 0; n < ItemsPerIteration; ++n) {
+            if (verification_failed.load(std::memory_order_relaxed)) {
+                state.SkipWithError("Verification failed!"); return;
+            }
+            const auto& item_to_write = random_data[total_written % RandomDataSize];
+            while (true) {
+                auto scope = queue.prepare_write(1);
+                if (scope.get_items_written() == 1) {
+                    scope.get_block1()[0] = item_to_write;
+                    break;
+                }
+            }
+            total_written++;
+        }
+    }
+    consumer_should_stop.store(true, std::memory_order_relaxed);
+    state.SetItemsProcessed(total_written);
+}
+BENCHMARK(BM_ThisQueue_SingleItem_DefaultBuffer)->Unit(benchmark::kMillisecond)->UseRealTime();
+
+static void BM_Moodycamel_SingleItem_DefaultBuffer(benchmark::State& state) {
+    const auto& random_data = get_random_data();
+    moodycamel::ReaderWriterQueue<DataType> queue(DefaultQueueCapacity);
+    std::atomic<bool> verification_failed  = false;
+    std::atomic<bool> consumer_should_stop = false;
+    std::jthread consumer([&] {
+        size_t i = 0;
+        DataType item;
+        while (!consumer_should_stop.load(std::memory_order_relaxed)) {
+            if (queue.try_dequeue(item)) {
+                if (item != random_data[i % RandomDataSize]) verification_failed.store(true);
+                i++;
+            } else {
+                std::this_thread::yield();
+            }
+        }
+    });
+
+    size_t total_written = 0;
+    for (auto _ : state) {
+        for (size_t n = 0; n < ItemsPerIteration; ++n) {
+            if (verification_failed.load(std::memory_order_relaxed)) {
+                state.SkipWithError("Verification failed!"); return;
+            }
+            const auto& item_to_write = random_data[total_written % RandomDataSize];
+            while (!queue.try_enqueue(item_to_write)) {}
+            total_written++;
+        }
+    }
+    consumer_should_stop.store(true, std::memory_order_relaxed);
+    state.SetItemsProcessed(total_written);
+}
+BENCHMARK(BM_Moodycamel_SingleItem_DefaultBuffer)->Unit(benchmark::kMillisecond)->UseRealTime();
+
+
+// Benchmark Group 2: Single Item, Large, hopefully "Never-Full" Capacity
+
+static void BM_OurQueue_SingleItem_LargeBuffer(benchmark::State& state) {
+    const auto& random_data = get_random_data();
+    std::vector<DataType> shared_buffer(LargeQueueCapacity);
+    LockFreeSpscQueue<DataType> queue(shared_buffer);
+
+    std::atomic<bool> verification_failed = false;
+    std::atomic<bool> consumer_should_stop = false;
+    std::jthread consumer([&] {
+        size_t i = 0;
+        while (!consumer_should_stop.load(std::memory_order_relaxed)) {
+            auto scope = queue.prepare_read(1);
+            if (scope.get_items_read() == 1) {
+                if (scope.get_block1()[0] != random_data[i % RandomDataSize]) verification_failed.store(true);
+                i++;
+            } else {
+                std::this_thread::yield();
+            }
+        }
+    });
+
+    size_t total_written = 0;
+    for (auto _ : state) {
+        for (size_t n = 0; n < ItemsPerIteration; ++n) {
+            if (verification_failed.load(std::memory_order_relaxed)) {
+                state.SkipWithError("Verification failed!"); return;
+            }
+            const auto& item_to_write = random_data[total_written % RandomDataSize];
+            while (true) {
+                auto scope = queue.prepare_write(1);
+                if (scope.get_items_written() == 1) {
+                    scope.get_block1()[0] = item_to_write;
+                    break;
+                }
+                // This spin-wait should theoretically never be hit when the buffer is large.
+            }
+            total_written++;
+        }
+    }
+    consumer_should_stop.store(true, std::memory_order_relaxed);
+    state.SetItemsProcessed(total_written);
+}
+BENCHMARK(BM_OurQueue_SingleItem_LargeBuffer)->Unit(benchmark::kMillisecond)->UseRealTime();
+
+static void BM_Moodycamel_SingleItem_LargeBuffer(benchmark::State& state) {
+    const auto& random_data = get_random_data();
+    moodycamel::ReaderWriterQueue<DataType> queue(LargeQueueCapacity);
+    std::atomic<bool> verification_failed = false;
+    std::atomic<bool> consumer_should_stop = false;
+    std::jthread consumer([&] {
+        size_t i = 0;
+        DataType item;
+        while (!consumer_should_stop.load(std::memory_order_relaxed)) {
+            if (queue.try_dequeue(item)) {
+                if (item != random_data[i % RandomDataSize]) verification_failed.store(true);
+                i++;
+            } else {
+                std::this_thread::yield();
+            }
+        }
+    });
+
+    size_t total_written = 0;
+    for (auto _ : state) {
+        for (size_t n = 0; n < ItemsPerIteration; ++n) {
+            if (verification_failed.load(std::memory_order_relaxed)) {
+                state.SkipWithError("Verification failed!"); return;
+            }
+            const auto& item_to_write = random_data[total_written % RandomDataSize];
+            while (!queue.try_enqueue(item_to_write)) {
+                // This spin-wait should theoretically never be hit.
+            }
+            total_written++;
+        }
+    }
+    consumer_should_stop.store(true, std::memory_order_relaxed);
+    state.SetItemsProcessed(total_written);
+}
+BENCHMARK(BM_Moodycamel_SingleItem_LargeBuffer)->Unit(benchmark::kMillisecond)->UseRealTime();
+
+
+// Benchmark Group 3: Batch/Bulk Transfers (with Default Capacity)
+
+static void BM_ThisQueue_Batch(benchmark::State& state) {
+    const size_t batch_size = state.range(0);
+    const auto& random_data = get_random_data();
+    std::vector<DataType> shared_buffer(DefaultQueueCapacity);
+    LockFreeSpscQueue<DataType> queue(shared_buffer);
+
+    std::atomic<bool> verification_failed = false;
+    std::atomic<bool> consumer_should_stop = false;
+    std::jthread consumer([&]{
+        size_t received_count = 0;
+        while (!consumer_should_stop.load(std::memory_order_relaxed)) {
+            const size_t items_read = queue.try_read(batch_size, [&](auto b1, auto b2){
+                for (const auto& item : b1) if (item != random_data[(received_count++) % RandomDataSize]) verification_failed.store(true);
+                for (const auto& item : b2) if (item != random_data[(received_count++) % RandomDataSize]) verification_failed.store(true);
+            });
+            if (items_read == 0) std::this_thread::yield();
+        }
+    });
+
+    size_t total_written = 0;
+    for (auto _ : state) {
+        for (size_t n = 0; n < ItemsPerIteration; ) {
+            if (verification_failed.load(std::memory_order_relaxed)) {
+                state.SkipWithError("Verification failed!"); consumer_should_stop.store(true); return;
+            }
+
+            const size_t current_rand_idx = total_written % RandomDataSize;
+            const size_t items_to_rand_end = RandomDataSize - current_rand_idx;
+            size_t remaining_in_iter = ItemsPerIteration - n;
+            size_t batch_to_send_size = std::min({batch_size, remaining_in_iter, items_to_rand_end});
+
+            std::span<const DataType> sub_batch(&random_data[current_rand_idx], batch_to_send_size);
+
+            size_t written_this_batch = 0;
+            while(written_this_batch < sub_batch.size()) {
+                 written_this_batch += queue.try_write(sub_batch.size() - written_this_batch, [&](auto b1, auto b2){
+                    std::copy_n(sub_batch.begin() + written_this_batch, b1.size(), b1.begin());
+                    if (!b2.empty()) std::copy_n(sub_batch.begin() + written_this_batch + b1.size(), b2.size(), b2.begin());
+                });
+            }
+            n += written_this_batch;
+            total_written += written_this_batch;
+        }
+    }
+    consumer_should_stop.store(true, std::memory_order_relaxed);
+    state.SetItemsProcessed(total_written);
+}
+BENCHMARK(BM_ThisQueue_Batch)->Arg(4)->Arg(16)->Arg(64)->Arg(256)->Unit(benchmark::kMillisecond)->UseRealTime();
+
+
+static void BM_Moodycamel_Batch(benchmark::State& state) {
+    const size_t batch_size = state.range(0);
+    const auto& random_data = get_random_data();
+    moodycamel::ReaderWriterQueue<DataType> queue(DefaultQueueCapacity);
+
+    std::atomic<bool> verification_failed  = false;
+    std::atomic<bool> consumer_should_stop = false;
+    std::jthread consumer([&]{
+        size_t received_count = 0;
+        DataType item;
+        while (!consumer_should_stop.load(std::memory_order_relaxed)) {
+            bool dequeued_something = false;
+            for (size_t i = 0; i < batch_size; ++i) {
+                if (queue.try_dequeue(item)) {
+                    if (item != random_data[(received_count++) % RandomDataSize]) verification_failed.store(true);
+                    dequeued_something = true;
+                } else {
+                    break;
+                }
+            }
+            if (!dequeued_something) std::this_thread::yield();
+        }
+    });
+
+    size_t total_written = 0;
+    for (auto _ : state) {
+        for (size_t n = 0; n < ItemsPerIteration; n += batch_size) {
+            if (verification_failed.load(std::memory_order_relaxed)) {
+                state.SkipWithError("Verification failed!"); consumer_should_stop.store(true); return;
+            }
+            for (size_t i = 0; i < batch_size; ++i) {
+                const auto& item_to_write = random_data[(total_written + i) % RandomDataSize];
+                while (!queue.try_enqueue(item_to_write)) {}
+            }
+            total_written += batch_size;
+        }
+    }
+    consumer_should_stop.store(true, std::memory_order_relaxed);
+    state.SetItemsProcessed(total_written);
+}
+BENCHMARK(BM_Moodycamel_Batch)->Arg(4)->Arg(16)->Arg(64)->Arg(256)->Unit(benchmark::kMillisecond)->UseRealTime();
+
+BENCHMARK_MAIN();
