From d3700fc72abcb8239a388f2e2d959202daf1ee89 Mon Sep 17 00:00:00 2001 From: Debra Date: Sun, 16 Aug 2026 16:13:41 +0800 Subject: [PATCH 1/9] feat: add MutexQueue v1 with tests and benchmarks Bounded mutex+condvar FIFO ring (v1 baseline), its GoogleTest unit and TSan stress tests, and Google Benchmark targets. The CMake scaffolding from the tooling PR picks these directories up automatically. Co-Authored-By: Claude Fable 5 --- bench/.clang-tidy | 10 +++ bench/CMakeLists.txt | 6 ++ bench/queue_bench.cpp | 83 +++++++++++++++++++++ include/cq/mutex_queue.hpp | 97 ++++++++++++++++++++++++ include/cq/mutex_queue.ipp | 124 ++++++++++++++++++++++++++++++ tests/.clang-tidy | 4 + tests/CMakeLists.txt | 9 +++ tests/mutex_queue_test.cpp | 149 +++++++++++++++++++++++++++++++++++++ tests/stress_test.cpp | 75 +++++++++++++++++++ 9 files changed, 557 insertions(+) create mode 100644 bench/.clang-tidy create mode 100644 bench/CMakeLists.txt create mode 100644 bench/queue_bench.cpp create mode 100644 include/cq/mutex_queue.hpp create mode 100644 include/cq/mutex_queue.ipp create mode 100644 tests/.clang-tidy create mode 100644 tests/CMakeLists.txt create mode 100644 tests/mutex_queue_test.cpp create mode 100644 tests/stress_test.cpp diff --git a/bench/.clang-tidy b/bench/.clang-tidy new file mode 100644 index 0000000..c33661a --- /dev/null +++ b/bench/.clang-tidy @@ -0,0 +1,10 @@ +--- +InheritParentConfig: true +# The canonical Google Benchmark loop `for (auto _ : state)` trips DeadStores +# in every benchmark file; suppress it here instead of per-line NOLINTs. +Checks: '-clang-analyzer-deadcode.DeadStores' +CheckOptions: + # Allow the `for (auto _ : state)` benchmark idiom. + readability-identifier-length.IgnoredVariableNames: '^_$' + # Google Benchmark's BM_PascalCase convention. + readability-identifier-naming.FunctionIgnoredRegexp: '^BM_.*' diff --git a/bench/CMakeLists.txt b/bench/CMakeLists.txt new file mode 100644 index 0000000..3e0d49f --- /dev/null +++ b/bench/CMakeLists.txt @@ -0,0 +1,6 @@ +add_executable(queue_bench queue_bench.cpp) +target_link_libraries(queue_bench PRIVATE cq::cq cq_warnings benchmark::benchmark_main) + +# One-iteration smoke run so CI (and `ctest`) picks up every future benchmark +# target without hardcoding binary paths. +add_test(NAME bench_smoke COMMAND queue_bench --benchmark_min_time=1x) diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp new file mode 100644 index 0000000..7ac75ea --- /dev/null +++ b/bench/queue_bench.cpp @@ -0,0 +1,83 @@ +// Throughput benchmarks for cq::MutexQueue (v1 baseline). +// +// Each benchmark uses Google Benchmark's multi-thread support: the first half +// of the threads produce, the second half consume. Reported items/s is the +// end-to-end transfer rate through the queue. +// +// Run: ./queue_bench --benchmark_repetitions=10 + +#include + +#include +#include +#include + +#include + +namespace { + +constexpr std::size_t kCapacity = 1024; +constexpr std::int64_t kItemsPerThreadPair = 100'000; + +// Created/destroyed by the Setup/Teardown hooks below, which run once per +// repetition outside the threaded region. +std::optional> shared_queue; + +void setup_queue(const benchmark::State& /*state*/) { shared_queue.emplace(kCapacity); } +void teardown_queue(const benchmark::State& /*state*/) { shared_queue.reset(); } + +// state.threads() is 2 * pairs: thread_index [0, pairs) produce, the rest consume. +void BM_MutexQueueThroughput(benchmark::State& state) { + const int pairs = state.threads() / 2; + const bool is_producer = state.thread_index() < pairs; + if (!shared_queue.has_value()) { // a registration forgot its ->Setup hook + state.SkipWithError("shared_queue not initialized"); + return; + } + auto& queue = *shared_queue; // hoisted out of the measured loop + + for (auto _ : state) { + if (is_producer) { + for (std::int64_t i = 0; i < kItemsPerThreadPair; ++i) { + benchmark::DoNotOptimize(queue.push(static_cast(i))); + } + } else { + std::uint64_t value = 0; + for (std::int64_t i = 0; i < kItemsPerThreadPair; ++i) { + benchmark::DoNotOptimize(queue.pop(value)); + } + } + } + + // Count the producer side only: Google Benchmark sums the counter across + // threads, and each item passes through one producer and one consumer. + if (is_producer) { + state.SetItemsProcessed(state.iterations() * kItemsPerThreadPair); + } +} + +// SPSC: 1 producer + 1 consumer; MPMC: 4 + 4. Google Benchmark appends the +// /threads:N suffix to the reported name. +constexpr int kSpscThreads = 2; +constexpr int kMpmcThreads = 8; +BENCHMARK(BM_MutexQueueThroughput) + ->Setup(setup_queue) + ->Teardown(teardown_queue) + ->Threads(kSpscThreads) + ->Threads(kMpmcThreads) + ->UseRealTime() + ->Name("MutexQueue/throughput"); + +// Uncontended single-thread round trip: the queue's raw locked cost. +void BM_MutexQueuePushPopSingleThread(benchmark::State& state) { + cq::MutexQueue queue(kCapacity); + std::uint64_t value = 0; + for (auto _ : state) { + benchmark::DoNotOptimize(queue.push(1)); + benchmark::DoNotOptimize(queue.pop(value)); + } + state.SetItemsProcessed(state.iterations()); +} +BENCHMARK(BM_MutexQueuePushPopSingleThread)->Name("MutexQueue/single_thread_roundtrip"); + +} // namespace diff --git a/include/cq/mutex_queue.hpp b/include/cq/mutex_queue.hpp new file mode 100644 index 0000000..8ff672e --- /dev/null +++ b/include/cq/mutex_queue.hpp @@ -0,0 +1,97 @@ +#ifndef CQ_MUTEX_QUEUE_HPP_ +#define CQ_MUTEX_QUEUE_HPP_ + +#include +#include +#include +#include + +namespace cq { + +/// v1 baseline: bounded FIFO ring guarded by a single std::mutex, with +/// not_full / not_empty condition variables and close() shutdown semantics. +/// +/// - push()/pop() block; try_push()/try_pop() never block. +/// - close() is idempotent and wakes every blocked producer and consumer. +/// After close(), push() refuses new values; pop() drains what remains, +/// then returns false. +/// +/// Notifications are unconditional (fired even when no thread waits) — +/// deliberate v1 simplicity; the benchmarks measure that cost as part of +/// the baseline. +/// +/// Member function definitions live in mutex_queue.ipp, included below. +/// +/// @tparam T Element type. Must be DefaultConstructible (ring slots are +/// constructed up front) and MoveAssignable. +template +class MutexQueue { + public: + /// @param capacity Fixed number of ring slots; never resized. + /// @throws std::invalid_argument if capacity is 0. + explicit MutexQueue(std::size_t capacity); + + // Not copyable or movable: blocked producers/consumers hold references to + // mutex_ and the condition variables, so the queue needs a stable address. + // Share it by reference (or shared_ptr) instead. + MutexQueue(const MutexQueue&) = delete; + MutexQueue& operator=(const MutexQueue&) = delete; + MutexQueue(MutexQueue&&) = delete; + MutexQueue& operator=(MutexQueue&&) = delete; + + /// Enqueues a value, blocking while the queue is full. + /// @param value Element to enqueue; consumed even when the push fails. + /// @return false if the queue is closed (the value is dropped). + bool push(T value); + + /// Enqueues a value without blocking. + /// @param value Element to enqueue; consumed even when the push fails. + /// @return false if the queue is full or closed. + bool try_push(T value); + + /// Dequeues into out, blocking while the queue is empty and open. + /// @param[out] out Receives the dequeued element on success. + /// @return false once the queue is closed and drained. + bool pop(T& out); + + /// Dequeues into out without blocking. + /// @param[out] out Receives the dequeued element on success. + /// @return false if the queue is empty. + bool try_pop(T& out); + + /// Closes the queue and wakes all blocked producers and consumers. + /// Idempotent. After close(), push() refuses new values; pop() drains + /// what remains. + void close(); + + /// @return true once close() has been called. + [[nodiscard]] bool closed() const; + + /// @return Current number of queued elements. + [[nodiscard]] std::size_t size() const; + + /// @return Fixed capacity set at construction. + [[nodiscard]] std::size_t capacity() const; + + private: + // The *_locked helpers require mutex_ to be held by the caller. + void enqueue_locked(T&& value); + void dequeue_locked(T& out); + + [[nodiscard]] std::size_t next(std::size_t index) const; + + mutable std::mutex mutex_; + std::condition_variable not_full_; + std::condition_variable not_empty_; + std::vector buffer_; + std::size_t head_ = 0; + std::size_t tail_ = 0; + std::size_t size_ = 0; + bool closed_ = false; +}; + +} // namespace cq + +#include "cq/mutex_queue.ipp" // IWYU pragma: keep + +#endif // CQ_MUTEX_QUEUE_HPP_ diff --git a/include/cq/mutex_queue.ipp b/include/cq/mutex_queue.ipp new file mode 100644 index 0000000..0c2d68c --- /dev/null +++ b/include/cq/mutex_queue.ipp @@ -0,0 +1,124 @@ +// Member function definitions for cq::MutexQueue. Included at the bottom of +// mutex_queue.hpp — templates must be visible at every instantiation point, +// so this file cannot be compiled as a standalone translation unit. +#ifndef CQ_MUTEX_QUEUE_IPP_ +#define CQ_MUTEX_QUEUE_IPP_ + +#include +#include +#include +#include + +namespace cq { + +template +MutexQueue::MutexQueue(std::size_t capacity) : buffer_(capacity) { + if (capacity == 0) { + throw std::invalid_argument("MutexQueue capacity must be > 0"); + } +} + +template +bool MutexQueue::push(T value) { + { + std::unique_lock lock(mutex_); + not_full_.wait(lock, [&] { return closed_ || size_ < buffer_.size(); }); + if (closed_) { + return false; + } + enqueue_locked(std::move(value)); + } + not_empty_.notify_one(); + return true; +} + +template +bool MutexQueue::try_push(T value) { + { + const std::lock_guard lock(mutex_); + if (closed_ || size_ == buffer_.size()) { + return false; + } + enqueue_locked(std::move(value)); + } + not_empty_.notify_one(); + return true; +} + +template +bool MutexQueue::pop(T& out) { + { + std::unique_lock lock(mutex_); + not_empty_.wait(lock, [&] { return closed_ || size_ > 0; }); + if (size_ == 0) { + return false; // closed and drained + } + dequeue_locked(out); + } + not_full_.notify_one(); + return true; +} + +template +bool MutexQueue::try_pop(T& out) { + { + const std::lock_guard lock(mutex_); + if (size_ == 0) { + return false; + } + dequeue_locked(out); + } + not_full_.notify_one(); + return true; +} + +template +void MutexQueue::close() { + { + const std::lock_guard lock(mutex_); + closed_ = true; + } + not_full_.notify_all(); + not_empty_.notify_all(); +} + +template +bool MutexQueue::closed() const { + const std::lock_guard lock(mutex_); + return closed_; +} + +template +std::size_t MutexQueue::size() const { + const std::lock_guard lock(mutex_); + return size_; +} + +// buffer_ is never resized after construction, so no lock is needed. +template +std::size_t MutexQueue::capacity() const { + return buffer_.size(); +} + +template +void MutexQueue::enqueue_locked(T&& value) { + buffer_[tail_] = std::move(value); + tail_ = next(tail_); + ++size_; +} + +template +void MutexQueue::dequeue_locked(T& out) { + out = std::move(buffer_[head_]); + head_ = next(head_); + --size_; +} + +template +std::size_t MutexQueue::next(std::size_t index) const { + return index + 1 == buffer_.size() ? 0 : index + 1; +} + +} // namespace cq + +#endif // CQ_MUTEX_QUEUE_IPP_ diff --git a/tests/.clang-tidy b/tests/.clang-tidy new file mode 100644 index 0000000..f63cba8 --- /dev/null +++ b/tests/.clang-tidy @@ -0,0 +1,4 @@ +--- +InheritParentConfig: true +# Test-only relaxation: short names (q, p, c) are idiomatic in tests. +Checks: '-readability-identifier-length' diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt new file mode 100644 index 0000000..4574b88 --- /dev/null +++ b/tests/CMakeLists.txt @@ -0,0 +1,9 @@ +add_executable(queue_tests + mutex_queue_test.cpp + stress_test.cpp) +target_link_libraries(queue_tests PRIVATE cq::cq cq_warnings cq_sanitizers GTest::gtest_main) + +include(GoogleTest) +# PRE_TEST: discover at ctest time instead of running the (TSan-instrumented) +# binary on every build. +gtest_discover_tests(queue_tests DISCOVERY_TIMEOUT 60 DISCOVERY_MODE PRE_TEST) diff --git a/tests/mutex_queue_test.cpp b/tests/mutex_queue_test.cpp new file mode 100644 index 0000000..b2986c5 --- /dev/null +++ b/tests/mutex_queue_test.cpp @@ -0,0 +1,149 @@ +// Unit tests for cq::MutexQueue (v1: mutex + condition_variable bounded queue). + +#include + +#include +#include +#include +#include + +#include + +namespace cq { +namespace { + +// Long enough for a spawned thread to reach its blocking call; the tests stay +// correct (just less interesting) if it ever proves too short. +constexpr auto kSettleTime = std::chrono::milliseconds(20); + +// Runs blocked_op on its own thread, gives it kSettleTime to reach its +// blocking call, runs unblock, and returns blocked_op's result after joining. +bool run_blocked(auto&& blocked_op, auto&& unblock) { + bool result = false; + std::jthread worker([&] { result = blocked_op(); }); + std::this_thread::sleep_for(kSettleTime); + unblock(); + worker.join(); + return result; +} + +TEST(MutexQueue, StartsEmptyWithGivenCapacity) { + const MutexQueue q(4); + EXPECT_EQ(q.capacity(), 4U); + EXPECT_EQ(q.size(), 0U); + EXPECT_FALSE(q.closed()); +} + +TEST(MutexQueue, ZeroCapacityThrows) { EXPECT_THROW(MutexQueue(0), std::invalid_argument); } + +TEST(MutexQueue, PopsInFifoOrder) { + MutexQueue q(4); + ASSERT_TRUE(q.push(1)); + ASSERT_TRUE(q.push(2)); + ASSERT_TRUE(q.push(3)); + + int out = 0; + ASSERT_TRUE(q.pop(out)); + EXPECT_EQ(out, 1); + ASSERT_TRUE(q.pop(out)); + EXPECT_EQ(out, 2); + ASSERT_TRUE(q.pop(out)); + EXPECT_EQ(out, 3); + EXPECT_EQ(q.size(), 0U); +} + +TEST(MutexQueue, WrapsAroundRingBoundary) { + constexpr int kRoundTrips = 10; + MutexQueue q(2); + int out = 0; + for (int i = 0; i < kRoundTrips; ++i) { + ASSERT_TRUE(q.push(i)); + ASSERT_TRUE(q.pop(out)); + EXPECT_EQ(out, i); + } +} + +TEST(MutexQueue, TryPushFailsWhenFull) { + MutexQueue q(2); + EXPECT_TRUE(q.try_push(1)); + EXPECT_TRUE(q.try_push(2)); + EXPECT_FALSE(q.try_push(3)); + EXPECT_EQ(q.size(), 2U); +} + +TEST(MutexQueue, TryPopFailsWhenEmpty) { + MutexQueue q(2); + int out = 0; + EXPECT_FALSE(q.try_pop(out)); +} + +TEST(MutexQueue, SupportsMoveOnlyTypes) { + MutexQueue> q(2); + ASSERT_TRUE(q.push(std::make_unique(42))); + + std::unique_ptr out; + ASSERT_TRUE(q.pop(out)); + ASSERT_NE(out, nullptr); + EXPECT_EQ(*out, 42); +} + +TEST(MutexQueue, PopBlocksUntilPush) { + MutexQueue q(1); + int out = 0; + EXPECT_TRUE(run_blocked([&] { return q.pop(out); }, // + [&] { EXPECT_TRUE(q.push(7)); })); + EXPECT_EQ(out, 7); +} + +TEST(MutexQueue, PushBlocksUntilPopWhenFull) { + MutexQueue q(1); + ASSERT_TRUE(q.push(1)); + int out = 0; + EXPECT_TRUE(run_blocked([&] { return q.push(2); }, + [&] { + EXPECT_TRUE(q.pop(out)); + EXPECT_EQ(out, 1); + })); + ASSERT_TRUE(q.pop(out)); + EXPECT_EQ(out, 2); +} + +TEST(MutexQueue, PushAfterCloseFails) { + MutexQueue q(2); + q.close(); + EXPECT_TRUE(q.closed()); + EXPECT_FALSE(q.push(1)); + EXPECT_FALSE(q.try_push(1)); +} + +TEST(MutexQueue, PopDrainsRemainingItemsAfterClose) { + MutexQueue q(4); + ASSERT_TRUE(q.push(1)); + ASSERT_TRUE(q.push(2)); + q.close(); + + int out = 0; + EXPECT_TRUE(q.pop(out)); + EXPECT_EQ(out, 1); + EXPECT_TRUE(q.try_pop(out)); + EXPECT_EQ(out, 2); + EXPECT_FALSE(q.pop(out)); + EXPECT_FALSE(q.try_pop(out)); +} + +TEST(MutexQueue, CloseWakesBlockedPop) { + MutexQueue q(1); + int out = 0; + EXPECT_FALSE(run_blocked([&] { return q.pop(out); }, // + [&] { q.close(); })); +} + +TEST(MutexQueue, CloseWakesBlockedPush) { + MutexQueue q(1); + ASSERT_TRUE(q.push(1)); + EXPECT_FALSE(run_blocked([&] { return q.push(2); }, // + [&] { q.close(); })); +} + +} // namespace +} // namespace cq diff --git a/tests/stress_test.cpp b/tests/stress_test.cpp new file mode 100644 index 0000000..f9ab82e --- /dev/null +++ b/tests/stress_test.cpp @@ -0,0 +1,75 @@ +// Checksum stress test (see README correctness policy): N producers push known +// values, consumers' totals must reconcile. Run under ThreadSanitizer. + +#include + +#include +#include +#include +#include + +#include + +namespace cq { +namespace { + +// Launches count threads, each running fn(thread_index). +std::vector spawn_threads(int count, auto fn) { + std::vector threads; + threads.reserve(static_cast(count)); + for (int i = 0; i < count; ++i) { + threads.emplace_back([fn, i] { fn(i); }); + } + return threads; +} + +TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { + constexpr int kProducers = 4; + constexpr int kConsumers = 4; + constexpr int kItemsPerProducer = 25'000; + constexpr std::uint64_t kTotalItems = static_cast(kProducers) * kItemsPerProducer; + // Much smaller than the item count so the ring wraps and fills constantly. + constexpr std::size_t kQueueCapacity = 64; + + MutexQueue q(kQueueCapacity); + + auto producers = spawn_threads(kProducers, [&q](int p) { + for (int i = 0; i < kItemsPerProducer; ++i) { + const auto value = + (static_cast(p) * kItemsPerProducer) + static_cast(i) + 1; + ASSERT_TRUE(q.push(value)); + } + }); + + std::atomic consumed_sum{0}; + std::atomic consumed_count{0}; + auto consumers = spawn_threads(kConsumers, [&](int /*c*/) { + std::uint64_t local_sum = 0; + std::uint64_t local_count = 0; + std::uint64_t value = 0; + while (q.pop(value)) { + local_sum += value; + ++local_count; + } + consumed_sum.fetch_add(local_sum, std::memory_order_relaxed); + consumed_count.fetch_add(local_count, std::memory_order_relaxed); + }); + + for (auto& t : producers) { + t.join(); + } + q.close(); + for (auto& t : consumers) { + t.join(); + } + + // Sum of 1..kTotalItems: each producer p pushes the contiguous block + // [p*kItems+1, (p+1)*kItems]. + const std::uint64_t expected_sum = kTotalItems * (kTotalItems + 1) / 2; + EXPECT_EQ(consumed_count.load(), kTotalItems); + EXPECT_EQ(consumed_sum.load(), expected_sum); + EXPECT_EQ(q.size(), 0U); +} + +} // namespace +} // namespace cq From fc8abd9ec97111ca91916c8a9fccde186b328c51 Mon Sep 17 00:00:00 2001 From: Debra Date: Sun, 16 Aug 2026 16:25:15 +0800 Subject: [PATCH 2/9] refactor: apply simplify-review findings to queue and benchmarks - factor the hand-negated full/empty conditions into can_enqueue_locked / can_dequeue_locked so each contract exists once - drop the now-unneeded EXISTS scaffolding guards and make ctest fail on zero discovered tests (--no-tests=error): a missing directory should fail loudly now that the sources exist - shared benchmark queue: optional + defensive guard -> unique_ptr (bugprone-unchecked-optional-access wants the guard the review flagged as dead; unique_ptr needs neither) - close() contract documented once, on close() - '^$_' identifier carve-out now inherited from the root .clang-tidy Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 4 ++-- CMakeLists.txt | 7 ++----- bench/.clang-tidy | 2 -- bench/queue_bench.cpp | 12 +++++------- include/cq/mutex_queue.hpp | 6 +++--- include/cq/mutex_queue.ipp | 20 +++++++++++++++----- 6 files changed, 27 insertions(+), 24 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6272268..55995cc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,7 +30,7 @@ jobs: - name: Build run: cmake --build build-tsan - name: Test - run: ctest --test-dir build-tsan --output-on-failure + run: ctest --test-dir build-tsan --output-on-failure --no-tests=error bench-build: name: Benchmarks (Release, smoke-run) @@ -42,7 +42,7 @@ jobs: - name: Build run: cmake --build build-rel - name: Smoke-run benchmarks - run: ctest --test-dir build-rel --output-on-failure + run: ctest --test-dir build-rel --output-on-failure --no-tests=error lint: name: clang-format & clang-tidy diff --git a/CMakeLists.txt b/CMakeLists.txt index c2bcc25..d876fb4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -38,9 +38,7 @@ endif() include(FetchContent) enable_testing() -# The tests/ and bench/ sources land in follow-up PRs; each block is a no-op -# until its directory exists so the build scaffolding can merge first. -if(CQ_BUILD_TESTS AND EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/tests/CMakeLists.txt") +if(CQ_BUILD_TESTS) FetchContent_Declare( googletest URL https://github.com/google/googletest/archive/refs/tags/v1.15.2.tar.gz @@ -52,8 +50,7 @@ if(CQ_BUILD_TESTS AND EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/tests/CMakeLists.txt") add_subdirectory(tests) endif() -if(CQ_BUILD_BENCHMARKS AND NOT ENABLE_TSAN - AND EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/bench/CMakeLists.txt") +if(CQ_BUILD_BENCHMARKS AND NOT ENABLE_TSAN) FetchContent_Declare( benchmark URL https://github.com/google/benchmark/archive/refs/tags/v1.9.1.tar.gz diff --git a/bench/.clang-tidy b/bench/.clang-tidy index c33661a..cdfd8a4 100644 --- a/bench/.clang-tidy +++ b/bench/.clang-tidy @@ -4,7 +4,5 @@ InheritParentConfig: true # in every benchmark file; suppress it here instead of per-line NOLINTs. Checks: '-clang-analyzer-deadcode.DeadStores' CheckOptions: - # Allow the `for (auto _ : state)` benchmark idiom. - readability-identifier-length.IgnoredVariableNames: '^_$' # Google Benchmark's BM_PascalCase convention. readability-identifier-naming.FunctionIgnoredRegexp: '^BM_.*' diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp index 7ac75ea..f175a9c 100644 --- a/bench/queue_bench.cpp +++ b/bench/queue_bench.cpp @@ -10,7 +10,7 @@ #include #include -#include +#include #include @@ -21,19 +21,17 @@ constexpr std::int64_t kItemsPerThreadPair = 100'000; // Created/destroyed by the Setup/Teardown hooks below, which run once per // repetition outside the threaded region. -std::optional> shared_queue; +std::unique_ptr> shared_queue; -void setup_queue(const benchmark::State& /*state*/) { shared_queue.emplace(kCapacity); } +void setup_queue(const benchmark::State& /*state*/) { + shared_queue = std::make_unique>(kCapacity); +} void teardown_queue(const benchmark::State& /*state*/) { shared_queue.reset(); } // state.threads() is 2 * pairs: thread_index [0, pairs) produce, the rest consume. void BM_MutexQueueThroughput(benchmark::State& state) { const int pairs = state.threads() / 2; const bool is_producer = state.thread_index() < pairs; - if (!shared_queue.has_value()) { // a registration forgot its ->Setup hook - state.SkipWithError("shared_queue not initialized"); - return; - } auto& queue = *shared_queue; // hoisted out of the measured loop for (auto _ : state) { diff --git a/include/cq/mutex_queue.hpp b/include/cq/mutex_queue.hpp index 8ff672e..cfbc391 100644 --- a/include/cq/mutex_queue.hpp +++ b/include/cq/mutex_queue.hpp @@ -12,9 +12,7 @@ namespace cq { /// not_full / not_empty condition variables and close() shutdown semantics. /// /// - push()/pop() block; try_push()/try_pop() never block. -/// - close() is idempotent and wakes every blocked producer and consumer. -/// After close(), push() refuses new values; pop() drains what remains, -/// then returns false. +/// - close() shuts the queue down; see close() for the full contract. /// /// Notifications are unconditional (fired even when no thread waits) — /// deliberate v1 simplicity; the benchmarks measure that cost as part of @@ -75,6 +73,8 @@ class MutexQueue { private: // The *_locked helpers require mutex_ to be held by the caller. + [[nodiscard]] bool can_enqueue_locked() const; // open and not full + [[nodiscard]] bool can_dequeue_locked() const; // not empty void enqueue_locked(T&& value); void dequeue_locked(T& out); diff --git a/include/cq/mutex_queue.ipp b/include/cq/mutex_queue.ipp index 0c2d68c..ecf1702 100644 --- a/include/cq/mutex_queue.ipp +++ b/include/cq/mutex_queue.ipp @@ -22,7 +22,7 @@ template bool MutexQueue::push(T value) { { std::unique_lock lock(mutex_); - not_full_.wait(lock, [&] { return closed_ || size_ < buffer_.size(); }); + not_full_.wait(lock, [&] { return closed_ || can_enqueue_locked(); }); if (closed_) { return false; } @@ -36,7 +36,7 @@ template bool MutexQueue::try_push(T value) { { const std::lock_guard lock(mutex_); - if (closed_ || size_ == buffer_.size()) { + if (!can_enqueue_locked()) { return false; } enqueue_locked(std::move(value)); @@ -49,8 +49,8 @@ template bool MutexQueue::pop(T& out) { { std::unique_lock lock(mutex_); - not_empty_.wait(lock, [&] { return closed_ || size_ > 0; }); - if (size_ == 0) { + not_empty_.wait(lock, [&] { return closed_ || can_dequeue_locked(); }); + if (!can_dequeue_locked()) { return false; // closed and drained } dequeue_locked(out); @@ -63,7 +63,7 @@ template bool MutexQueue::try_pop(T& out) { { const std::lock_guard lock(mutex_); - if (size_ == 0) { + if (!can_dequeue_locked()) { return false; } dequeue_locked(out); @@ -100,6 +100,16 @@ std::size_t MutexQueue::capacity() const { return buffer_.size(); } +template +bool MutexQueue::can_enqueue_locked() const { + return !closed_ && size_ < buffer_.size(); +} + +template +bool MutexQueue::can_dequeue_locked() const { + return size_ > 0; +} + template void MutexQueue::enqueue_locked(T&& value) { buffer_[tail_] = std::move(value); From c3142598426841e86fa8f59108c0abad7699d980 Mon Sep 17 00:00:00 2001 From: Debra Date: Sun, 16 Aug 2026 17:18:16 +0800 Subject: [PATCH 3/9] refactor: apply review cleanups to queue, tests, and benchmarks Queue: - Inline the can_enqueue/can_dequeue predicates: push's wait no longer re-tests closed_ inside the predicate, pop no longer evaluates the predicate twice per dequeue, and closure is checked exactly once per entry point - close() skips both notify_all()s on repeat calls (std::exchange) - Class doc states the thread-safety contract (STYLE.md requires it); drop the .ipp build-layout note from the published API docs Tests: - Shared tests/queue_test_util.hpp holds kSettleTime, run_blocked, and spawn_threads for both TUs (and future queue variants' tests) - tests/.clang-tidy narrows identifier-length via Ignored*Names regexes instead of disabling the whole check Benchmarks: - One queue op per harness iteration: Google Benchmark regains control of run length (the 100k inner loop defeated --benchmark_min_time and made the CI smoke run execute ~500k ops), and the producer/consumer branch moves out of the timed region - Guard odd ->Threads() counts with SkipWithError instead of a comment CMake: - cq INTERFACE links Threads::Threads (the library that creates the thread requirement owns it, not each consumer) - gtest PRE_TEST discovery set project-wide instead of per-target Co-Authored-By: Claude Fable 5 --- CMakeLists.txt | 8 ++++++- bench/CMakeLists.txt | 4 ++-- bench/queue_bench.cpp | 43 +++++++++++++++++++------------------- include/cq/mutex_queue.hpp | 5 ++--- include/cq/mutex_queue.ipp | 24 +++++++-------------- tests/.clang-tidy | 8 +++++-- tests/CMakeLists.txt | 5 ++--- tests/mutex_queue_test.cpp | 19 ++--------------- tests/queue_test_util.hpp | 41 ++++++++++++++++++++++++++++++++++++ tests/stress_test.cpp | 14 ++----------- 10 files changed, 93 insertions(+), 78 deletions(-) create mode 100644 tests/queue_test_util.hpp diff --git a/CMakeLists.txt b/CMakeLists.txt index d876fb4..c9c17be 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -12,11 +12,14 @@ option(CQ_BUILD_BENCHMARKS "Build Google Benchmark targets" ON) # Every build tree gets a compile_commands.json for clangd/clang-tidy. set(CMAKE_EXPORT_COMPILE_COMMANDS ON) -# Header-only library; consumers inherit C++20 via the compile feature. +# Header-only library; consumers inherit C++20 via the compile feature and +# the thread requirement from the library that creates it. +find_package(Threads REQUIRED) add_library(cq INTERFACE) add_library(cq::cq ALIAS cq) target_include_directories(cq INTERFACE ${CMAKE_CURRENT_SOURCE_DIR}/include) target_compile_features(cq INTERFACE cxx_std_20) +target_link_libraries(cq INTERFACE Threads::Threads) add_library(cq_warnings INTERFACE) target_compile_options(cq_warnings INTERFACE @@ -37,6 +40,9 @@ endif() include(FetchContent) enable_testing() +# Project-wide: never execute a (possibly TSan-instrumented) test binary at +# build time just to enumerate tests. +set(CMAKE_GTEST_DISCOVER_TESTS_DISCOVERY_MODE PRE_TEST) if(CQ_BUILD_TESTS) FetchContent_Declare( diff --git a/bench/CMakeLists.txt b/bench/CMakeLists.txt index 3e0d49f..744d6bb 100644 --- a/bench/CMakeLists.txt +++ b/bench/CMakeLists.txt @@ -1,6 +1,6 @@ add_executable(queue_bench queue_bench.cpp) target_link_libraries(queue_bench PRIVATE cq::cq cq_warnings benchmark::benchmark_main) -# One-iteration smoke run so CI (and `ctest`) picks up every future benchmark -# target without hardcoding binary paths. +# One-iteration smoke run; CI's Release job runs it via ctest. A future +# benchmark executable needs its own add_test line here. add_test(NAME bench_smoke COMMAND queue_bench --benchmark_min_time=1x) diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp index f175a9c..469d81f 100644 --- a/bench/queue_bench.cpp +++ b/bench/queue_bench.cpp @@ -1,8 +1,10 @@ // Throughput benchmarks for cq::MutexQueue (v1 baseline). // // Each benchmark uses Google Benchmark's multi-thread support: the first half -// of the threads produce, the second half consume. Reported items/s is the -// end-to-end transfer rate through the queue. +// of the threads produce, the second half consume, one queue op per benchmark +// iteration. Every thread runs the same iteration count, so pushes and pops +// balance and the harness keeps full control of run length. Reported items/s +// is the end-to-end transfer rate through the queue. // // Run: ./queue_bench --benchmark_repetitions=10 @@ -17,7 +19,6 @@ namespace { constexpr std::size_t kCapacity = 1024; -constexpr std::int64_t kItemsPerThreadPair = 100'000; // Created/destroyed by the Setup/Teardown hooks below, which run once per // repetition outside the threaded region. @@ -28,29 +29,27 @@ void setup_queue(const benchmark::State& /*state*/) { } void teardown_queue(const benchmark::State& /*state*/) { shared_queue.reset(); } -// state.threads() is 2 * pairs: thread_index [0, pairs) produce, the rest consume. void BM_MutexQueueThroughput(benchmark::State& state) { - const int pairs = state.threads() / 2; - const bool is_producer = state.thread_index() < pairs; - auto& queue = *shared_queue; // hoisted out of the measured loop - - for (auto _ : state) { - if (is_producer) { - for (std::int64_t i = 0; i < kItemsPerThreadPair; ++i) { - benchmark::DoNotOptimize(queue.push(static_cast(i))); - } - } else { - std::uint64_t value = 0; - for (std::int64_t i = 0; i < kItemsPerThreadPair; ++i) { - benchmark::DoNotOptimize(queue.pop(value)); - } - } + if (state.threads() % 2 != 0) { + state.SkipWithError("thread count must be even (producer/consumer pairs)"); + return; } + const bool is_producer = state.thread_index() < state.threads() / 2; + auto& queue = *shared_queue; - // Count the producer side only: Google Benchmark sums the counter across - // threads, and each item passes through one producer and one consumer. if (is_producer) { - state.SetItemsProcessed(state.iterations() * kItemsPerThreadPair); + std::uint64_t item = 0; + for (auto _ : state) { + benchmark::DoNotOptimize(queue.push(item++)); + } + // Count the producer side only: Google Benchmark sums the counter across + // threads, and each item passes through one producer and one consumer. + state.SetItemsProcessed(state.iterations()); + } else { + std::uint64_t value = 0; + for (auto _ : state) { + benchmark::DoNotOptimize(queue.pop(value)); + } } } diff --git a/include/cq/mutex_queue.hpp b/include/cq/mutex_queue.hpp index cfbc391..379bbb9 100644 --- a/include/cq/mutex_queue.hpp +++ b/include/cq/mutex_queue.hpp @@ -18,7 +18,8 @@ namespace cq { /// deliberate v1 simplicity; the benchmarks measure that cost as part of /// the baseline. /// -/// Member function definitions live in mutex_queue.ipp, included below. +/// Thread-safety: all member functions may be called concurrently from any +/// number of producer and consumer threads. /// /// @tparam T Element type. Must be DefaultConstructible (ring slots are /// constructed up front) and MoveAssignable. @@ -73,8 +74,6 @@ class MutexQueue { private: // The *_locked helpers require mutex_ to be held by the caller. - [[nodiscard]] bool can_enqueue_locked() const; // open and not full - [[nodiscard]] bool can_dequeue_locked() const; // not empty void enqueue_locked(T&& value); void dequeue_locked(T& out); diff --git a/include/cq/mutex_queue.ipp b/include/cq/mutex_queue.ipp index ecf1702..a2d6b28 100644 --- a/include/cq/mutex_queue.ipp +++ b/include/cq/mutex_queue.ipp @@ -22,7 +22,7 @@ template bool MutexQueue::push(T value) { { std::unique_lock lock(mutex_); - not_full_.wait(lock, [&] { return closed_ || can_enqueue_locked(); }); + not_full_.wait(lock, [&] { return closed_ || size_ < buffer_.size(); }); if (closed_) { return false; } @@ -36,7 +36,7 @@ template bool MutexQueue::try_push(T value) { { const std::lock_guard lock(mutex_); - if (!can_enqueue_locked()) { + if (closed_ || size_ == buffer_.size()) { return false; } enqueue_locked(std::move(value)); @@ -49,8 +49,8 @@ template bool MutexQueue::pop(T& out) { { std::unique_lock lock(mutex_); - not_empty_.wait(lock, [&] { return closed_ || can_dequeue_locked(); }); - if (!can_dequeue_locked()) { + not_empty_.wait(lock, [&] { return closed_ || size_ > 0; }); + if (size_ == 0) { return false; // closed and drained } dequeue_locked(out); @@ -63,7 +63,7 @@ template bool MutexQueue::try_pop(T& out) { { const std::lock_guard lock(mutex_); - if (!can_dequeue_locked()) { + if (size_ == 0) { return false; } dequeue_locked(out); @@ -76,7 +76,9 @@ template void MutexQueue::close() { { const std::lock_guard lock(mutex_); - closed_ = true; + if (std::exchange(closed_, true)) { + return; // already closed; waiters were woken the first time + } } not_full_.notify_all(); not_empty_.notify_all(); @@ -100,16 +102,6 @@ std::size_t MutexQueue::capacity() const { return buffer_.size(); } -template -bool MutexQueue::can_enqueue_locked() const { - return !closed_ && size_ < buffer_.size(); -} - -template -bool MutexQueue::can_dequeue_locked() const { - return size_ > 0; -} - template void MutexQueue::enqueue_locked(T&& value) { buffer_[tail_] = std::move(value); diff --git a/tests/.clang-tidy b/tests/.clang-tidy index f63cba8..f91fbde 100644 --- a/tests/.clang-tidy +++ b/tests/.clang-tidy @@ -1,4 +1,8 @@ --- InheritParentConfig: true -# Test-only relaxation: short names (q, p, c) are idiomatic in tests. -Checks: '-readability-identifier-length' +CheckOptions: + # Test-only relaxation: these short names are idiomatic in tests (queue, + # producer/consumer indices, thread handles). Child CheckOptions replace the + # parent's value, so `_` is re-listed. + readability-identifier-length.IgnoredVariableNames: '^(_|q|p|c|i|t)$' + readability-identifier-length.IgnoredParameterNames: '^(p|c|i)$' diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 4574b88..4f53177 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -4,6 +4,5 @@ add_executable(queue_tests target_link_libraries(queue_tests PRIVATE cq::cq cq_warnings cq_sanitizers GTest::gtest_main) include(GoogleTest) -# PRE_TEST: discover at ctest time instead of running the (TSan-instrumented) -# binary on every build. -gtest_discover_tests(queue_tests DISCOVERY_TIMEOUT 60 DISCOVERY_MODE PRE_TEST) +# Discovery mode is set project-wide (PRE_TEST) in the root CMakeLists. +gtest_discover_tests(queue_tests DISCOVERY_TIMEOUT 60) diff --git a/tests/mutex_queue_test.cpp b/tests/mutex_queue_test.cpp index b2986c5..0367ab4 100644 --- a/tests/mutex_queue_test.cpp +++ b/tests/mutex_queue_test.cpp @@ -2,31 +2,16 @@ #include -#include #include #include -#include #include +#include "queue_test_util.hpp" + namespace cq { namespace { -// Long enough for a spawned thread to reach its blocking call; the tests stay -// correct (just less interesting) if it ever proves too short. -constexpr auto kSettleTime = std::chrono::milliseconds(20); - -// Runs blocked_op on its own thread, gives it kSettleTime to reach its -// blocking call, runs unblock, and returns blocked_op's result after joining. -bool run_blocked(auto&& blocked_op, auto&& unblock) { - bool result = false; - std::jthread worker([&] { result = blocked_op(); }); - std::this_thread::sleep_for(kSettleTime); - unblock(); - worker.join(); - return result; -} - TEST(MutexQueue, StartsEmptyWithGivenCapacity) { const MutexQueue q(4); EXPECT_EQ(q.capacity(), 4U); diff --git a/tests/queue_test_util.hpp b/tests/queue_test_util.hpp new file mode 100644 index 0000000..70e2485 --- /dev/null +++ b/tests/queue_test_util.hpp @@ -0,0 +1,41 @@ +// Thread harness shared by the queue test files (and by future queue +// variants' tests). +#ifndef CQ_TESTS_QUEUE_TEST_UTIL_HPP_ +#define CQ_TESTS_QUEUE_TEST_UTIL_HPP_ + +#include +#include +#include +#include + +namespace cq { + +// Long enough for a spawned thread to reach its blocking call; the tests stay +// correct (just less interesting) if it ever proves too short. +constexpr auto kSettleTime = std::chrono::milliseconds(20); + +// Runs blocked_op on its own thread, gives it kSettleTime to reach its +// blocking call, runs unblock, and returns blocked_op's result after joining. +// The explicit join is load-bearing: result must not be read before it. +bool run_blocked(auto&& blocked_op, auto&& unblock) { + bool result = false; + std::jthread worker([&] { result = blocked_op(); }); + std::this_thread::sleep_for(kSettleTime); + unblock(); + worker.join(); + return result; +} + +// Launches count threads, each running body(thread_index). +std::vector spawn_threads(int count, auto body) { + std::vector threads; + threads.reserve(static_cast(count)); + for (int i = 0; i < count; ++i) { + threads.emplace_back([body, i] { body(i); }); + } + return threads; +} + +} // namespace cq + +#endif // CQ_TESTS_QUEUE_TEST_UTIL_HPP_ diff --git a/tests/stress_test.cpp b/tests/stress_test.cpp index f9ab82e..c068baa 100644 --- a/tests/stress_test.cpp +++ b/tests/stress_test.cpp @@ -5,24 +5,14 @@ #include #include -#include -#include #include +#include "queue_test_util.hpp" + namespace cq { namespace { -// Launches count threads, each running fn(thread_index). -std::vector spawn_threads(int count, auto fn) { - std::vector threads; - threads.reserve(static_cast(count)); - for (int i = 0; i < count; ++i) { - threads.emplace_back([fn, i] { fn(i); }); - } - return threads; -} - TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { constexpr int kProducers = 4; constexpr int kConsumers = 4; From 73b848adfdcdb2c54aee5701e99244ea5207d65a Mon Sep 17 00:00:00 2001 From: Debra Date: Sun, 16 Aug 2026 17:54:25 +0800 Subject: [PATCH 4/9] =?UTF-8?q?refactor:=20round-2=20review=20cleanups=20?= =?UTF-8?q?=E2=80=94=20test=20harness=20rigor,=20benchmark=20accuracy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tests: - run_blocked now asserts the operation was actually still blocked before unblocking (atomic finished flag); previously the four blocking tests passed even if the op returned immediately without blocking - join_all helper (inline; the header is multi-TU) makes the stress test's shutdown ordering read as intent: join_all(producers); close(); join_all(consumers) - Stress producers assert once per thread instead of expanding an AssertionResult per item (25k per thread under TSan) - tests/.clang-tidy ignore lists trimmed to names the code actually uses Benchmarks (both defects found by measurement): - SetItemsProcessed now reported from every thread; producer-only reporting inflated items/s by ~100000/threads (measured 1.33M/s reported vs 106 items/s actual at threads:8) - setup prefills the ring half-full so measurement starts in steady state instead of on a condition-variable wakeup storm - File header documents that thread spawn sits inside the timed region and recommends --benchmark_min_time=1s for meaningful numbers CMake: comment accuracy fixes (garbled INTERFACE note, discovery-timeout rationale, drop speculative sentence). Co-Authored-By: Claude Fable 5 --- CMakeLists.txt | 4 ++-- bench/CMakeLists.txt | 3 +-- bench/queue_bench.cpp | 17 ++++++++++++----- tests/.clang-tidy | 10 +++++----- tests/CMakeLists.txt | 2 +- tests/queue_test_util.hpp | 24 +++++++++++++++++++++--- tests/stress_test.cpp | 18 +++++++++--------- 7 files changed, 51 insertions(+), 27 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index c9c17be..5c7f169 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -12,8 +12,8 @@ option(CQ_BUILD_BENCHMARKS "Build Google Benchmark targets" ON) # Every build tree gets a compile_commands.json for clangd/clang-tidy. set(CMAKE_EXPORT_COMPILE_COMMANDS ON) -# Header-only library; consumers inherit C++20 via the compile feature and -# the thread requirement from the library that creates it. +# Header-only library; consumers inherit C++20 and the thread dependency +# through the INTERFACE properties. find_package(Threads REQUIRED) add_library(cq INTERFACE) add_library(cq::cq ALIAS cq) diff --git a/bench/CMakeLists.txt b/bench/CMakeLists.txt index 744d6bb..7dbb97a 100644 --- a/bench/CMakeLists.txt +++ b/bench/CMakeLists.txt @@ -1,6 +1,5 @@ add_executable(queue_bench queue_bench.cpp) target_link_libraries(queue_bench PRIVATE cq::cq cq_warnings benchmark::benchmark_main) -# One-iteration smoke run; CI's Release job runs it via ctest. A future -# benchmark executable needs its own add_test line here. +# One-iteration smoke run; CI's Release job runs it via ctest. add_test(NAME bench_smoke COMMAND queue_bench --benchmark_min_time=1x) diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp index 469d81f..405fb67 100644 --- a/bench/queue_bench.cpp +++ b/bench/queue_bench.cpp @@ -4,9 +4,11 @@ // of the threads produce, the second half consume, one queue op per benchmark // iteration. Every thread runs the same iteration count, so pushes and pops // balance and the harness keeps full control of run length. Reported items/s -// is the end-to-end transfer rate through the queue. +// is total queue ops per second (a push and its pop count as two). // -// Run: ./queue_bench --benchmark_repetitions=10 +// Thread spawn and the start barrier sit inside the timed region, so a +// min_time well above that (~50ms) is required for meaningful numbers: +// Run: ./queue_bench --benchmark_min_time=1s --benchmark_repetitions=10 #include @@ -26,6 +28,11 @@ std::unique_ptr> shared_queue; void setup_queue(const benchmark::State& /*state*/) { shared_queue = std::make_unique>(kCapacity); + // Half-full start: neither side begins blocked on a condition variable, so + // the measurement starts in steady state. + for (std::uint64_t i = 0; i < kCapacity / 2; ++i) { + shared_queue->try_push(i); + } } void teardown_queue(const benchmark::State& /*state*/) { shared_queue.reset(); } @@ -42,15 +49,15 @@ void BM_MutexQueueThroughput(benchmark::State& state) { for (auto _ : state) { benchmark::DoNotOptimize(queue.push(item++)); } - // Count the producer side only: Google Benchmark sums the counter across - // threads, and each item passes through one producer and one consumer. - state.SetItemsProcessed(state.iterations()); } else { std::uint64_t value = 0; for (auto _ : state) { benchmark::DoNotOptimize(queue.pop(value)); } } + // Every thread reports its op count, matching the items/s definition in the + // file header. Reporting from a subset of threads skews the computed rate. + state.SetItemsProcessed(state.iterations()); } // SPSC: 1 producer + 1 consumer; MPMC: 4 + 4. Google Benchmark appends the diff --git a/tests/.clang-tidy b/tests/.clang-tidy index f91fbde..60c55b9 100644 --- a/tests/.clang-tidy +++ b/tests/.clang-tidy @@ -1,8 +1,8 @@ --- InheritParentConfig: true CheckOptions: - # Test-only relaxation: these short names are idiomatic in tests (queue, - # producer/consumer indices, thread handles). Child CheckOptions replace the - # parent's value, so `_` is re-listed. - readability-identifier-length.IgnoredVariableNames: '^(_|q|p|c|i|t)$' - readability-identifier-length.IgnoredParameterNames: '^(p|c|i)$' + # Test-only relaxation: the short names actually used by these tests + # (queue, loop index, thread handle, producer index). Child CheckOptions + # replace the parent's value rather than extend it. + readability-identifier-length.IgnoredVariableNames: '^(q|i|t)$' + readability-identifier-length.IgnoredParameterNames: '^p$' diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 4f53177..0af5275 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -4,5 +4,5 @@ add_executable(queue_tests target_link_libraries(queue_tests PRIVATE cq::cq cq_warnings cq_sanitizers GTest::gtest_main) include(GoogleTest) -# Discovery mode is set project-wide (PRE_TEST) in the root CMakeLists. +# 60s: TSan-instrumented binaries take ~0.25s per process just to start. gtest_discover_tests(queue_tests DISCOVERY_TIMEOUT 60) diff --git a/tests/queue_test_util.hpp b/tests/queue_test_util.hpp index 70e2485..09a2dce 100644 --- a/tests/queue_test_util.hpp +++ b/tests/queue_test_util.hpp @@ -3,11 +3,14 @@ #ifndef CQ_TESTS_QUEUE_TEST_UTIL_HPP_ #define CQ_TESTS_QUEUE_TEST_UTIL_HPP_ +#include #include #include #include #include +#include + namespace cq { // Long enough for a spawned thread to reach its blocking call; the tests stay @@ -15,12 +18,19 @@ namespace cq { constexpr auto kSettleTime = std::chrono::milliseconds(20); // Runs blocked_op on its own thread, gives it kSettleTime to reach its -// blocking call, runs unblock, and returns blocked_op's result after joining. -// The explicit join is load-bearing: result must not be read before it. +// blocking call, checks it really is still blocked, runs unblock, and returns +// blocked_op's result after joining. The explicit join is load-bearing: +// result must not be read before it. bool run_blocked(auto&& blocked_op, auto&& unblock) { bool result = false; - std::jthread worker([&] { result = blocked_op(); }); + std::atomic finished{false}; + std::jthread worker([&] { + result = blocked_op(); + finished.store(true, std::memory_order_release); + }); std::this_thread::sleep_for(kSettleTime); + EXPECT_FALSE(finished.load(std::memory_order_acquire)) + << "operation returned without ever blocking"; unblock(); worker.join(); return result; @@ -36,6 +46,14 @@ std::vector spawn_threads(int count, auto body) { return threads; } +// inline: unlike the auto-parameter helpers above this is not a template, and +// the header is included from more than one TU. +inline void join_all(std::vector& threads) { + for (auto& t : threads) { + t.join(); + } +} + } // namespace cq #endif // CQ_TESTS_QUEUE_TEST_UTIL_HPP_ diff --git a/tests/stress_test.cpp b/tests/stress_test.cpp index c068baa..7c7596e 100644 --- a/tests/stress_test.cpp +++ b/tests/stress_test.cpp @@ -24,11 +24,15 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { MutexQueue q(kQueueCapacity); auto producers = spawn_threads(kProducers, [&q](int p) { - for (int i = 0; i < kItemsPerProducer; ++i) { + // One assertion per producer, not per item: each ASSERT expands to a full + // AssertionResult, which is measurable 25k times per thread under TSan. + bool all_pushed = true; + for (int i = 0; all_pushed && i < kItemsPerProducer; ++i) { const auto value = (static_cast(p) * kItemsPerProducer) + static_cast(i) + 1; - ASSERT_TRUE(q.push(value)); + all_pushed = q.push(value); } + EXPECT_TRUE(all_pushed); }); std::atomic consumed_sum{0}; @@ -45,13 +49,9 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { consumed_count.fetch_add(local_count, std::memory_order_relaxed); }); - for (auto& t : producers) { - t.join(); - } - q.close(); - for (auto& t : consumers) { - t.join(); - } + join_all(producers); + q.close(); // all items in; wake the consumers so they drain and exit + join_all(consumers); // Sum of 1..kTotalItems: each producer p pushes the contiguous block // [p*kItems+1, (p+1)*kItems]. From 948c186701f9efc1c07b9a207f7a24cdb14b6fd8 Mon Sep 17 00:00:00 2001 From: Debra Date: Sun, 16 Aug 2026 21:16:00 +0800 Subject: [PATCH 5/9] docs+api: harden the public contract of MutexQueue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - [[nodiscard]] on push/try_push/pop/try_pop — the operations whose discarded result is always a bug (reading out after an ignored try_pop, silent data loss after an ignored try_push); accessors already had it - Class doc gains the two missing contracts STYLE.md requires: lifetime (close() + join before destruction; destroying with blocked threads is UB) and exception behavior (throwing T move-assign leaves the queue consistent) - size()/closed() documented as advisory snapshots that must not drive control flow — the try_ operations are the atomic alternative - try_pop doc states out is untouched on failure - Bench prefill explicitly discards try_push's result with a cannot-fail note Co-Authored-By: Claude Fable 5 --- bench/queue_bench.cpp | 5 +++-- include/cq/mutex_queue.hpp | 31 ++++++++++++++++++++++--------- 2 files changed, 25 insertions(+), 11 deletions(-) diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp index 405fb67..b4f47c6 100644 --- a/bench/queue_bench.cpp +++ b/bench/queue_bench.cpp @@ -29,9 +29,10 @@ std::unique_ptr> shared_queue; void setup_queue(const benchmark::State& /*state*/) { shared_queue = std::make_unique>(kCapacity); // Half-full start: neither side begins blocked on a condition variable, so - // the measurement starts in steady state. + // the measurement starts in steady state. Cannot fail: the queue is fresh + // and i stays below capacity. for (std::uint64_t i = 0; i < kCapacity / 2; ++i) { - shared_queue->try_push(i); + (void)shared_queue->try_push(i); } } void teardown_queue(const benchmark::State& /*state*/) { shared_queue.reset(); } diff --git a/include/cq/mutex_queue.hpp b/include/cq/mutex_queue.hpp index 379bbb9..bd2d64b 100644 --- a/include/cq/mutex_queue.hpp +++ b/include/cq/mutex_queue.hpp @@ -18,8 +18,16 @@ namespace cq { /// deliberate v1 simplicity; the benchmarks measure that cost as part of /// the baseline. /// -/// Thread-safety: all member functions may be called concurrently from any -/// number of producer and consumer threads. +/// Thread-safety: after construction, all member functions may be called +/// concurrently from any number of producer and consumer threads. +/// +/// Lifetime: the queue must outlive every thread using it — call close() +/// and join all producers/consumers before destruction. Destroying the +/// queue while a thread is blocked in push()/pop() is undefined behavior. +/// +/// Exceptions: if T's move assignment throws, the failing push()/try_push() +/// enqueues nothing and the failing pop()/try_pop() leaves the element +/// queued — the queue itself stays consistent. /// /// @tparam T Element type. Must be DefaultConstructible (ring slots are /// constructed up front) and MoveAssignable. @@ -41,32 +49,37 @@ class MutexQueue { /// Enqueues a value, blocking while the queue is full. /// @param value Element to enqueue; consumed even when the push fails. /// @return false if the queue is closed (the value is dropped). - bool push(T value); + [[nodiscard]] bool push(T value); /// Enqueues a value without blocking. /// @param value Element to enqueue; consumed even when the push fails. /// @return false if the queue is full or closed. - bool try_push(T value); + [[nodiscard]] bool try_push(T value); /// Dequeues into out, blocking while the queue is empty and open. /// @param[out] out Receives the dequeued element on success. /// @return false once the queue is closed and drained. - bool pop(T& out); + [[nodiscard]] bool pop(T& out); /// Dequeues into out without blocking. - /// @param[out] out Receives the dequeued element on success. + /// @param[out] out Receives the dequeued element on success; untouched on + /// failure. /// @return false if the queue is empty. - bool try_pop(T& out); + [[nodiscard]] bool try_pop(T& out); /// Closes the queue and wakes all blocked producers and consumers. /// Idempotent. After close(), push() refuses new values; pop() drains /// what remains. void close(); - /// @return true once close() has been called. + /// @return true once close() has been called. Advisory snapshot: may be + /// stale by the time it returns; do not build control flow on it — use + /// the return values of push()/pop() instead. [[nodiscard]] bool closed() const; - /// @return Current number of queued elements. + /// @return Current number of queued elements. Advisory snapshot: may be + /// stale by the time it returns; meant for monitoring and tests, not + /// for emptiness/fullness decisions — use try_push()/try_pop(). [[nodiscard]] std::size_t size() const; /// @return Fixed capacity set at construction. From ed7cbe708cbad5015f81cae742f0af01275fc50c Mon Sep 17 00:00:00 2001 From: Debra Date: Sun, 16 Aug 2026 21:24:40 +0800 Subject: [PATCH 6/9] =?UTF-8?q?refactor:=20round-3=20review=20cleanups=20?= =?UTF-8?q?=E2=80=94=20fix=20benchmark=20rate=20math,=20simplify=20harness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Benchmarks (measured: reported scaling was wrong by a factor of threads): - SetItemsProcessed(iterations * threads): Benchmark sums real time across threads, so per-thread iteration counts divided the true aggregate rate by the thread count (2->8 slowdown reported 6.9x, actually 1.9x) - Single-thread roundtrip counts push+pop as two ops, making its unit identical to the threaded benchmark - Odd-thread guard moved from a per-run runtime check to a static_assert beside the thread-count constants - Prefill self-checks via one postcondition instead of 512 (void) casts Tests: - run_blocked rewritten on std::async/std::future: wait_for IS the still-blocked check and get() IS the join — drops the atomic flag, the memory orders, and the hand-rolled comment; return type now deduced - kSettleTime 20ms -> 5ms (measured: the sleeps were 66% of the Debug suite wall time; 5ms is still ~250x the spawn-to-block latency) - Stress producers report failures via ADD_FAILURE on the failure path only; [[nodiscard]] on both test helpers - tests/.clang-tidy states the policy (check off) instead of allowlisting each identifier in use Docs: dedupe the class comment (snapshot caveat said once at class level, notification note moved beside the notify calls in the .ipp, lifetime rationale said once); gtest discovery settings co-located on the single gtest_discover_tests call. Co-Authored-By: Claude Fable 5 --- CMakeLists.txt | 3 --- bench/queue_bench.cpp | 27 ++++++++++++++++----------- include/cq/mutex_queue.hpp | 22 +++++++--------------- include/cq/mutex_queue.ipp | 3 +++ tests/.clang-tidy | 9 +++------ tests/CMakeLists.txt | 5 +++-- tests/queue_test_util.hpp | 32 ++++++++++++-------------------- tests/stress_test.cpp | 12 ++++++------ 8 files changed, 50 insertions(+), 63 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 5c7f169..7c0d689 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -40,9 +40,6 @@ endif() include(FetchContent) enable_testing() -# Project-wide: never execute a (possibly TSan-instrumented) test binary at -# build time just to enumerate tests. -set(CMAKE_GTEST_DISCOVER_TESTS_DISCOVERY_MODE PRE_TEST) if(CQ_BUILD_TESTS) FetchContent_Declare( diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp index b4f47c6..c95ff5d 100644 --- a/bench/queue_bench.cpp +++ b/bench/queue_bench.cpp @@ -14,6 +14,7 @@ #include #include +#include #include #include @@ -29,19 +30,18 @@ std::unique_ptr> shared_queue; void setup_queue(const benchmark::State& /*state*/) { shared_queue = std::make_unique>(kCapacity); // Half-full start: neither side begins blocked on a condition variable, so - // the measurement starts in steady state. Cannot fail: the queue is fresh - // and i stays below capacity. + // the measurement starts in steady state. + bool prefilled = true; for (std::uint64_t i = 0; i < kCapacity / 2; ++i) { - (void)shared_queue->try_push(i); + prefilled = prefilled && shared_queue->try_push(i); + } + if (!prefilled) { + std::abort(); // unreachable: fresh queue, stays below capacity } } void teardown_queue(const benchmark::State& /*state*/) { shared_queue.reset(); } void BM_MutexQueueThroughput(benchmark::State& state) { - if (state.threads() % 2 != 0) { - state.SkipWithError("thread count must be even (producer/consumer pairs)"); - return; - } const bool is_producer = state.thread_index() < state.threads() / 2; auto& queue = *shared_queue; @@ -56,15 +56,18 @@ void BM_MutexQueueThroughput(benchmark::State& state) { benchmark::DoNotOptimize(queue.pop(value)); } } - // Every thread reports its op count, matching the items/s definition in the - // file header. Reporting from a subset of threads skews the computed rate. - state.SetItemsProcessed(state.iterations()); + // Benchmark accumulates real time as the sum over threads, so each thread + // reports iterations * threads: the sum divided by (threads * wall) is then + // the aggregate ops/s the file header promises. + state.SetItemsProcessed(state.iterations() * state.threads()); } // SPSC: 1 producer + 1 consumer; MPMC: 4 + 4. Google Benchmark appends the // /threads:N suffix to the reported name. constexpr int kSpscThreads = 2; constexpr int kMpmcThreads = 8; +static_assert(kSpscThreads % 2 == 0 && kMpmcThreads % 2 == 0, + "producer/consumer pairing needs an even thread count"); BENCHMARK(BM_MutexQueueThroughput) ->Setup(setup_queue) ->Teardown(teardown_queue) @@ -81,7 +84,9 @@ void BM_MutexQueuePushPopSingleThread(benchmark::State& state) { benchmark::DoNotOptimize(queue.push(1)); benchmark::DoNotOptimize(queue.pop(value)); } - state.SetItemsProcessed(state.iterations()); + // A push and its pop are two queue ops — keep the unit identical to the + // threaded benchmark so the rates compare directly. + state.SetItemsProcessed(state.iterations() * 2); } BENCHMARK(BM_MutexQueuePushPopSingleThread)->Name("MutexQueue/single_thread_roundtrip"); diff --git a/include/cq/mutex_queue.hpp b/include/cq/mutex_queue.hpp index bd2d64b..8cc4f9e 100644 --- a/include/cq/mutex_queue.hpp +++ b/include/cq/mutex_queue.hpp @@ -12,14 +12,11 @@ namespace cq { /// not_full / not_empty condition variables and close() shutdown semantics. /// /// - push()/pop() block; try_push()/try_pop() never block. -/// - close() shuts the queue down; see close() for the full contract. -/// -/// Notifications are unconditional (fired even when no thread waits) — -/// deliberate v1 simplicity; the benchmarks measure that cost as part of -/// the baseline. /// /// Thread-safety: after construction, all member functions may be called -/// concurrently from any number of producer and consumer threads. +/// concurrently from any number of producer and consumer threads. closed() +/// and size() return advisory snapshots — drive control flow off the +/// push/pop return values instead. /// /// Lifetime: the queue must outlive every thread using it — call close() /// and join all producers/consumers before destruction. Destroying the @@ -38,9 +35,8 @@ class MutexQueue { /// @throws std::invalid_argument if capacity is 0. explicit MutexQueue(std::size_t capacity); - // Not copyable or movable: blocked producers/consumers hold references to - // mutex_ and the condition variables, so the queue needs a stable address. - // Share it by reference (or shared_ptr) instead. + // Not copyable or movable: waiters hold references to mutex_ and the + // condition variables; share the queue by reference instead. MutexQueue(const MutexQueue&) = delete; MutexQueue& operator=(const MutexQueue&) = delete; MutexQueue(MutexQueue&&) = delete; @@ -72,14 +68,10 @@ class MutexQueue { /// what remains. void close(); - /// @return true once close() has been called. Advisory snapshot: may be - /// stale by the time it returns; do not build control flow on it — use - /// the return values of push()/pop() instead. + /// @return true once close() has been called (advisory snapshot). [[nodiscard]] bool closed() const; - /// @return Current number of queued elements. Advisory snapshot: may be - /// stale by the time it returns; meant for monitoring and tests, not - /// for emptiness/fullness decisions — use try_push()/try_pop(). + /// @return Current number of queued elements (advisory snapshot). [[nodiscard]] std::size_t size() const; /// @return Fixed capacity set at construction. diff --git a/include/cq/mutex_queue.ipp b/include/cq/mutex_queue.ipp index a2d6b28..b879a77 100644 --- a/include/cq/mutex_queue.ipp +++ b/include/cq/mutex_queue.ipp @@ -18,6 +18,9 @@ MutexQueue::MutexQueue(std::size_t capacity) : buffer_(capacity) { } } +// Notifications throughout are unconditional (fired even when no thread +// waits) — deliberate v1 simplicity; the benchmarks measure that cost as +// part of the baseline. template bool MutexQueue::push(T value) { { diff --git a/tests/.clang-tidy b/tests/.clang-tidy index 60c55b9..5fa3960 100644 --- a/tests/.clang-tidy +++ b/tests/.clang-tidy @@ -1,8 +1,5 @@ --- InheritParentConfig: true -CheckOptions: - # Test-only relaxation: the short names actually used by these tests - # (queue, loop index, thread handle, producer index). Child CheckOptions - # replace the parent's value rather than extend it. - readability-identifier-length.IgnoredVariableNames: '^(q|i|t)$' - readability-identifier-length.IgnoredParameterNames: '^p$' +# Short names (q, p, i, t, ...) are idiomatic throughout tests; disable the +# check as directory policy rather than allowlisting each name. +Checks: '-readability-identifier-length' diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 0af5275..efd8fe3 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -4,5 +4,6 @@ add_executable(queue_tests target_link_libraries(queue_tests PRIVATE cq::cq cq_warnings cq_sanitizers GTest::gtest_main) include(GoogleTest) -# 60s: TSan-instrumented binaries take ~0.25s per process just to start. -gtest_discover_tests(queue_tests DISCOVERY_TIMEOUT 60) +# PRE_TEST + a generous timeout: never run the (possibly TSan-instrumented) +# binary at build time, and allow for its slow (~0.25s) process startup. +gtest_discover_tests(queue_tests DISCOVERY_MODE PRE_TEST DISCOVERY_TIMEOUT 60) diff --git a/tests/queue_test_util.hpp b/tests/queue_test_util.hpp index 09a2dce..5bbf850 100644 --- a/tests/queue_test_util.hpp +++ b/tests/queue_test_util.hpp @@ -3,9 +3,9 @@ #ifndef CQ_TESTS_QUEUE_TEST_UTIL_HPP_ #define CQ_TESTS_QUEUE_TEST_UTIL_HPP_ -#include #include #include +#include #include #include @@ -15,29 +15,21 @@ namespace cq { // Long enough for a spawned thread to reach its blocking call; the tests stay // correct (just less interesting) if it ever proves too short. -constexpr auto kSettleTime = std::chrono::milliseconds(20); - -// Runs blocked_op on its own thread, gives it kSettleTime to reach its -// blocking call, checks it really is still blocked, runs unblock, and returns -// blocked_op's result after joining. The explicit join is load-bearing: -// result must not be read before it. -bool run_blocked(auto&& blocked_op, auto&& unblock) { - bool result = false; - std::atomic finished{false}; - std::jthread worker([&] { - result = blocked_op(); - finished.store(true, std::memory_order_release); - }); - std::this_thread::sleep_for(kSettleTime); - EXPECT_FALSE(finished.load(std::memory_order_acquire)) +constexpr auto kSettleTime = std::chrono::milliseconds(5); + +// Runs blocked_op on its own thread, confirms it is still blocked after +// kSettleTime, runs unblock, and returns blocked_op's result. +[[nodiscard]] auto run_blocked(auto&& blocked_op, auto&& unblock) { + auto pending = std::async(std::launch::async, blocked_op); + EXPECT_EQ(pending.wait_for(kSettleTime), std::future_status::timeout) << "operation returned without ever blocking"; unblock(); - worker.join(); - return result; + return pending.get(); } -// Launches count threads, each running body(thread_index). -std::vector spawn_threads(int count, auto body) { +// Launches count threads, each running body(thread_index). Hold the returned +// vector: dropping it joins every thread immediately. +[[nodiscard]] std::vector spawn_threads(int count, auto body) { std::vector threads; threads.reserve(static_cast(count)); for (int i = 0; i < count; ++i) { diff --git a/tests/stress_test.cpp b/tests/stress_test.cpp index 7c7596e..9c44fe9 100644 --- a/tests/stress_test.cpp +++ b/tests/stress_test.cpp @@ -24,15 +24,15 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { MutexQueue q(kQueueCapacity); auto producers = spawn_threads(kProducers, [&q](int p) { - // One assertion per producer, not per item: each ASSERT expands to a full - // AssertionResult, which is measurable 25k times per thread under TSan. - bool all_pushed = true; - for (int i = 0; all_pushed && i < kItemsPerProducer; ++i) { + // Assert only on failure: a per-item ASSERT is measurable under TSan. + for (int i = 0; i < kItemsPerProducer; ++i) { const auto value = (static_cast(p) * kItemsPerProducer) + static_cast(i) + 1; - all_pushed = q.push(value); + if (!q.push(value)) { + ADD_FAILURE() << "producer " << p << " push failed at item " << i; + break; + } } - EXPECT_TRUE(all_pushed); }); std::atomic consumed_sum{0}; From f5d733ed734e360b57c8777446e58310d1a025cf Mon Sep 17 00:00:00 2001 From: Debra Date: Mon, 17 Aug 2026 07:31:30 +0800 Subject: [PATCH 7/9] =?UTF-8?q?refactor:=20round-4=20review=20cleanups=20?= =?UTF-8?q?=E2=80=94=20re-measure=20results,=20drop=20redundant=20harness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit README (the published numbers were measured before round 3 changed the benchmark's accounting, so every row and the story built on them was stale): - Re-measured on this machine: 105.1M ops/s single-thread round trip, 35.7M SPSC, 21.6M MPMC, with CV per row - The 2->8 thread slowdown is 1.65x, not the ~6x claimed; the SPSC run's CV is 1.4%, not ~18%, so the lock-convoy-jitter reading is gone - Define the unit once (an op is one push or one pop) and drop the 'counted on the producer side only' claim, which the benchmark never did - Note the measured cost of unconditional notify (~20% MPMC, ~26-37% SPSC) so v1 is not read as the fastest single-mutex design - Lint commands note CI's version pin Benchmarks: - MinTime(1s) on the registrations: run length is a property of the benchmark, so the documented command reproduces the table (the ctest smoke run still overrides with 1x) - Drop the file header's claim that thread spawn is inside the timed region — StartKeepRunning hits the start barrier before ResumeTiming Tests: - Delete join_all: spawn_threads already returns jthreads, whose destructor joins; the stress test says producers.clear() / consumers.clear() - spawn_threads uses jthread's forwarding constructor instead of a wrapper lambda per thread - Test helpers move to namespace cq::testing, out of the shipping namespace CMake/lint: - Sanitizer-vs-benchmark policy stated once, with a status message, instead of silently dropping benchmark targets from an explicit -DCQ_BUILD_BENCHMARKS=ON - The 'auto _' identifier carve-out moves to bench/.clang-tidy beside the other two benchmark-loop exceptions Co-Authored-By: Claude Fable 5 --- .clang-tidy | 2 -- CMakeLists.txt | 9 ++++++++- README.md | 36 ++++++++++++++++++++++++++++++++++-- bench/.clang-tidy | 2 ++ bench/queue_bench.cpp | 16 ++++++++++++---- include/cq/mutex_queue.hpp | 2 -- tests/mutex_queue_test.cpp | 22 +++++++++++----------- tests/queue_test_util.hpp | 16 ++++------------ tests/stress_test.cpp | 10 +++++----- 9 files changed, 76 insertions(+), 39 deletions(-) diff --git a/.clang-tidy b/.clang-tidy index 46f14a6..28b0d3a 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -19,8 +19,6 @@ CheckOptions: # Complexity contributed by macro expansion (GTest asserts, etc.) is not # the author's complexity. readability-function-cognitive-complexity.IgnoreMacros: 'true' - # `auto _` is the general placeholder idiom (a language feature in C++26). - readability-identifier-length.IgnoredVariableNames: '^_$' readability-identifier-naming.ClassCase: CamelCase readability-identifier-naming.StructCase: CamelCase readability-identifier-naming.EnumCase: CamelCase diff --git a/CMakeLists.txt b/CMakeLists.txt index 7c0d689..b30e177 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -41,6 +41,13 @@ endif() include(FetchContent) enable_testing() +# Sanitizer builds are test-only: benchmark numbers from an instrumented build +# are meaningless. Say so rather than silently producing no benchmark targets. +if(ENABLE_TSAN AND CQ_BUILD_BENCHMARKS) + message(STATUS "ENABLE_TSAN is on: disabling benchmarks (timings under TSan are meaningless)") + set(CQ_BUILD_BENCHMARKS OFF) +endif() + if(CQ_BUILD_TESTS) FetchContent_Declare( googletest @@ -53,7 +60,7 @@ if(CQ_BUILD_TESTS) add_subdirectory(tests) endif() -if(CQ_BUILD_BENCHMARKS AND NOT ENABLE_TSAN) +if(CQ_BUILD_BENCHMARKS) FetchContent_Declare( benchmark URL https://github.com/google/benchmark/archive/refs/tags/v1.9.1.tar.gz diff --git a/README.md b/README.md index 45feca6..0139997 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,12 @@ cmake --build build-tsan && ctest --test-dir build-tsan --output-on-failure # benchmarks (Release, no sanitizers) cmake -B build-rel -DCMAKE_BUILD_TYPE=Release -cmake --build build-rel && ./build-rel/queue_bench --benchmark_repetitions=10 +cmake --build build-rel && ./build-rel/bench/queue_bench --benchmark_repetitions=10 + +# lint (same file sets CI checks; CI pins version 18, so match it locally +# — a different major version formats differently and CI will disagree) +git ls-files '*.hpp' '*.ipp' '*.cpp' | xargs clang-format --dry-run --Werror +git ls-files '*.cpp' | xargs clang-tidy -p build-rel ``` ## Correctness policy @@ -50,4 +55,31 @@ cmake --build build-rel && ./build-rel/queue_bench --benchmark_repetitions=10 ## Results -_To be filled in as v1 → v3 land._ +Machine: Apple M2 Pro (12 cores), 32 GB, macOS 26. Release build, +`--benchmark_repetitions=10` (each run is `MinTime` 1s, set on the benchmark). +The machine was not idle — load average ~4.6 — so treat these as a floor. + +An **op** is one `push` or one `pop`, so transferring an item costs two ops; +this is the unit Google Benchmark prints as `items_per_second`. + +| Benchmark (v1 MutexQueue) | Throughput | Per-op | CV | +|---|---|---|---| +| single-thread push+pop round trip | 105.1M ± 0.7M ops/s | 19.1 ns per round trip | 2.2% | +| SPSC (1 producer, 1 consumer) | 35.7M ± 0.5M ops/s | 56.0 ns | 1.4% | +| MPMC (4 producers, 4 consumers) | 21.6M ± 0.1M ops/s | 370.3 ns | 0.5% | + +The v1 story in one line: one mutex serializes everything, so **threads never +buy throughput** — the uncontended round trip moves ops ~3× faster than two +threads managing it, and going from 2 threads to 8 loses a further ~40% +(1.65× slower). That is the baseline v2's lock-free ring has to beat. + +Two caveats on reading these numbers. First, v1 notifies its condition +variables on every op, even when no thread is waiting: a variant that keeps +waiter counts and skips the no-op notify measures ~20% faster on MPMC and +~26–37% faster on SPSC (the uncontended round trip is unchanged). This +baseline is deliberately the *simple* single-mutex design, not the fastest +one — worth remembering before crediting v2 with the whole gap. Second, the +numbers above are ops/s; halve them for items transferred per second +(SPSC ≈ 17.9M items/s, MPMC ≈ 10.8M). + +_v2 → v3 to follow._ diff --git a/bench/.clang-tidy b/bench/.clang-tidy index cdfd8a4..a5ac14e 100644 --- a/bench/.clang-tidy +++ b/bench/.clang-tidy @@ -6,3 +6,5 @@ Checks: '-clang-analyzer-deadcode.DeadStores' CheckOptions: # Google Benchmark's BM_PascalCase convention. readability-identifier-naming.FunctionIgnoredRegexp: '^BM_.*' + # The same loop's `auto _` placeholder (a language feature in C++26). + readability-identifier-length.IgnoredVariableNames: '^_$' diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp index c95ff5d..286f35a 100644 --- a/bench/queue_bench.cpp +++ b/bench/queue_bench.cpp @@ -6,9 +6,10 @@ // balance and the harness keeps full control of run length. Reported items/s // is total queue ops per second (a push and its pop count as two). // -// Thread spawn and the start barrier sit inside the timed region, so a -// min_time well above that (~50ms) is required for meaningful numbers: -// Run: ./queue_bench --benchmark_min_time=1s --benchmark_repetitions=10 +// Threads synchronize on Google Benchmark's start barrier before timing +// resumes, so spawn cost is outside the timed region. The registrations below +// set MinTime so a bare run is already long enough to be meaningful; add +// repetitions for a spread: ./queue_bench --benchmark_repetitions=10 #include @@ -68,12 +69,17 @@ constexpr int kSpscThreads = 2; constexpr int kMpmcThreads = 8; static_assert(kSpscThreads % 2 == 0 && kMpmcThreads % 2 == 0, "producer/consumer pairing needs an even thread count"); +// MinTime: contended runs need a second of samples to settle; the ctest smoke +// run overrides it with --benchmark_min_time=1x. +constexpr double kMinTimeSeconds = 1.0; + BENCHMARK(BM_MutexQueueThroughput) ->Setup(setup_queue) ->Teardown(teardown_queue) ->Threads(kSpscThreads) ->Threads(kMpmcThreads) ->UseRealTime() + ->MinTime(kMinTimeSeconds) ->Name("MutexQueue/throughput"); // Uncontended single-thread round trip: the queue's raw locked cost. @@ -88,6 +94,8 @@ void BM_MutexQueuePushPopSingleThread(benchmark::State& state) { // threaded benchmark so the rates compare directly. state.SetItemsProcessed(state.iterations() * 2); } -BENCHMARK(BM_MutexQueuePushPopSingleThread)->Name("MutexQueue/single_thread_roundtrip"); +BENCHMARK(BM_MutexQueuePushPopSingleThread) + ->MinTime(kMinTimeSeconds) + ->Name("MutexQueue/single_thread_roundtrip"); } // namespace diff --git a/include/cq/mutex_queue.hpp b/include/cq/mutex_queue.hpp index 8cc4f9e..cf5a3a5 100644 --- a/include/cq/mutex_queue.hpp +++ b/include/cq/mutex_queue.hpp @@ -11,8 +11,6 @@ namespace cq { /// v1 baseline: bounded FIFO ring guarded by a single std::mutex, with /// not_full / not_empty condition variables and close() shutdown semantics. /// -/// - push()/pop() block; try_push()/try_pop() never block. -/// /// Thread-safety: after construction, all member functions may be called /// concurrently from any number of producer and consumer threads. closed() /// and size() return advisory snapshots — drive control flow off the diff --git a/tests/mutex_queue_test.cpp b/tests/mutex_queue_test.cpp index 0367ab4..a5521c0 100644 --- a/tests/mutex_queue_test.cpp +++ b/tests/mutex_queue_test.cpp @@ -75,8 +75,8 @@ TEST(MutexQueue, SupportsMoveOnlyTypes) { TEST(MutexQueue, PopBlocksUntilPush) { MutexQueue q(1); int out = 0; - EXPECT_TRUE(run_blocked([&] { return q.pop(out); }, // - [&] { EXPECT_TRUE(q.push(7)); })); + EXPECT_TRUE(testing::run_blocked([&] { return q.pop(out); }, // + [&] { EXPECT_TRUE(q.push(7)); })); EXPECT_EQ(out, 7); } @@ -84,11 +84,11 @@ TEST(MutexQueue, PushBlocksUntilPopWhenFull) { MutexQueue q(1); ASSERT_TRUE(q.push(1)); int out = 0; - EXPECT_TRUE(run_blocked([&] { return q.push(2); }, - [&] { - EXPECT_TRUE(q.pop(out)); - EXPECT_EQ(out, 1); - })); + EXPECT_TRUE(testing::run_blocked([&] { return q.push(2); }, + [&] { + EXPECT_TRUE(q.pop(out)); + EXPECT_EQ(out, 1); + })); ASSERT_TRUE(q.pop(out)); EXPECT_EQ(out, 2); } @@ -119,15 +119,15 @@ TEST(MutexQueue, PopDrainsRemainingItemsAfterClose) { TEST(MutexQueue, CloseWakesBlockedPop) { MutexQueue q(1); int out = 0; - EXPECT_FALSE(run_blocked([&] { return q.pop(out); }, // - [&] { q.close(); })); + EXPECT_FALSE(testing::run_blocked([&] { return q.pop(out); }, // + [&] { q.close(); })); } TEST(MutexQueue, CloseWakesBlockedPush) { MutexQueue q(1); ASSERT_TRUE(q.push(1)); - EXPECT_FALSE(run_blocked([&] { return q.push(2); }, // - [&] { q.close(); })); + EXPECT_FALSE(testing::run_blocked([&] { return q.push(2); }, // + [&] { q.close(); })); } } // namespace diff --git a/tests/queue_test_util.hpp b/tests/queue_test_util.hpp index 5bbf850..14273bf 100644 --- a/tests/queue_test_util.hpp +++ b/tests/queue_test_util.hpp @@ -11,7 +11,7 @@ #include -namespace cq { +namespace cq::testing { // Long enough for a spawned thread to reach its blocking call; the tests stay // correct (just less interesting) if it ever proves too short. @@ -28,24 +28,16 @@ constexpr auto kSettleTime = std::chrono::milliseconds(5); } // Launches count threads, each running body(thread_index). Hold the returned -// vector: dropping it joins every thread immediately. +// vector: destroying it (scope exit, or clear()) joins every thread. [[nodiscard]] std::vector spawn_threads(int count, auto body) { std::vector threads; threads.reserve(static_cast(count)); for (int i = 0; i < count; ++i) { - threads.emplace_back([body, i] { body(i); }); + threads.emplace_back(body, i); } return threads; } -// inline: unlike the auto-parameter helpers above this is not a template, and -// the header is included from more than one TU. -inline void join_all(std::vector& threads) { - for (auto& t : threads) { - t.join(); - } -} - -} // namespace cq +} // namespace cq::testing #endif // CQ_TESTS_QUEUE_TEST_UTIL_HPP_ diff --git a/tests/stress_test.cpp b/tests/stress_test.cpp index 9c44fe9..700f8f3 100644 --- a/tests/stress_test.cpp +++ b/tests/stress_test.cpp @@ -23,7 +23,7 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { MutexQueue q(kQueueCapacity); - auto producers = spawn_threads(kProducers, [&q](int p) { + auto producers = testing::spawn_threads(kProducers, [&q](int p) { // Assert only on failure: a per-item ASSERT is measurable under TSan. for (int i = 0; i < kItemsPerProducer; ++i) { const auto value = @@ -37,7 +37,7 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { std::atomic consumed_sum{0}; std::atomic consumed_count{0}; - auto consumers = spawn_threads(kConsumers, [&](int /*c*/) { + auto consumers = testing::spawn_threads(kConsumers, [&](int /*c*/) { std::uint64_t local_sum = 0; std::uint64_t local_count = 0; std::uint64_t value = 0; @@ -49,9 +49,9 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { consumed_count.fetch_add(local_count, std::memory_order_relaxed); }); - join_all(producers); - q.close(); // all items in; wake the consumers so they drain and exit - join_all(consumers); + producers.clear(); // joins every producer: all items are in + q.close(); // wake the consumers so they drain and exit + consumers.clear(); // joins every consumer: all items are out // Sum of 1..kTotalItems: each producer p pushes the contiguous block // [p*kItems+1, (p+1)*kItems]. From f48e831fb432ec8a74d41a7826c910aeabe9db95 Mon Sep 17 00:00:00 2001 From: Debra Date: Mon, 17 Aug 2026 07:37:58 +0800 Subject: [PATCH 8/9] =?UTF-8?q?refactor:=20round-5=20review=20cleanups=20?= =?UTF-8?q?=E2=80=94=20fix=20what=20round=204=20got=20wrong?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All four items are defects in the previous round's own changes: - tests: rename cq::testing -> cq::test_util. Inside namespace cq, the old name shadowed GoogleTest's ::testing, so any future fixture written as ': public testing::Test' (or TestWithParam, or a gmock matcher import) would fail to compile; gtest's macros only survive it by expanding to fully-qualified ::testing - README: the Per-op column mixed denominators — per round trip on row 1, per-thread on rows 2-3 — so it read as a 6.6x MPMC penalty, silently contradicting the 1.65x in the prose two lines below. Now 1/throughput throughout (9.5 / 28.0 / 46.3 ns), stated above the table - bench: MinTime() on the registration beats --benchmark_min_time=s (ComputeMinTime prefers a non-zero registered value), so that flag is accepted and ignored; only the Nx form overrides. Comment says so - CMake: CQ_BUILD_BENCHMARKS defaults to ON, so the policy block fired on the README's own TSan command and could never detect an explicit -D as the comment claimed; set() also shadowed the cache entry. Replaced with cmake_dependent_option so the cache reports what was built Co-Authored-By: Claude Fable 5 --- CMakeLists.txt | 18 ++++++++++-------- README.md | 11 +++++++---- bench/queue_bench.cpp | 6 ++++-- tests/mutex_queue_test.cpp | 22 +++++++++++----------- tests/queue_test_util.hpp | 4 ++-- tests/stress_test.cpp | 4 ++-- 6 files changed, 36 insertions(+), 29 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index b30e177..aa66c56 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -7,7 +7,16 @@ project(concurrent_queue option(ENABLE_TSAN "Build with ThreadSanitizer" OFF) option(CQ_BUILD_TESTS "Build unit and stress tests" ON) -option(CQ_BUILD_BENCHMARKS "Build Google Benchmark targets" ON) + +# Timings from a sanitized build are meaningless, so benchmarks are forced off +# under TSan. Expressed as a dependent option so the cache reports what was +# actually built (a plain set() would leave ON showing in ccmake/cmake-gui). +include(CMakeDependentOption) +cmake_dependent_option(CQ_BUILD_BENCHMARKS "Build Google Benchmark targets" ON + "NOT ENABLE_TSAN" OFF) +if(ENABLE_TSAN) + message(STATUS "ENABLE_TSAN is on: benchmarks disabled (timings under a sanitizer are meaningless)") +endif() # Every build tree gets a compile_commands.json for clangd/clang-tidy. set(CMAKE_EXPORT_COMPILE_COMMANDS ON) @@ -41,13 +50,6 @@ endif() include(FetchContent) enable_testing() -# Sanitizer builds are test-only: benchmark numbers from an instrumented build -# are meaningless. Say so rather than silently producing no benchmark targets. -if(ENABLE_TSAN AND CQ_BUILD_BENCHMARKS) - message(STATUS "ENABLE_TSAN is on: disabling benchmarks (timings under TSan are meaningless)") - set(CQ_BUILD_BENCHMARKS OFF) -endif() - if(CQ_BUILD_TESTS) FetchContent_Declare( googletest diff --git a/README.md b/README.md index 0139997..f32a925 100644 --- a/README.md +++ b/README.md @@ -62,11 +62,14 @@ The machine was not idle — load average ~4.6 — so treat these as a floor. An **op** is one `push` or one `pop`, so transferring an item costs two ops; this is the unit Google Benchmark prints as `items_per_second`. -| Benchmark (v1 MutexQueue) | Throughput | Per-op | CV | +Per-op figures below are `1 / throughput` — the aggregate cost of one op across +the whole queue, not per-thread latency. + +| Benchmark (v1 MutexQueue) | Throughput | Per op | CV | |---|---|---|---| -| single-thread push+pop round trip | 105.1M ± 0.7M ops/s | 19.1 ns per round trip | 2.2% | -| SPSC (1 producer, 1 consumer) | 35.7M ± 0.5M ops/s | 56.0 ns | 1.4% | -| MPMC (4 producers, 4 consumers) | 21.6M ± 0.1M ops/s | 370.3 ns | 0.5% | +| single-thread push+pop round trip | 105.1M ± 0.7M ops/s | 9.5 ns (19.1 ns per round trip) | 2.2% | +| SPSC (1 producer, 1 consumer) | 35.7M ± 0.5M ops/s | 28.0 ns | 1.4% | +| MPMC (4 producers, 4 consumers) | 21.6M ± 0.1M ops/s | 46.3 ns | 0.5% | The v1 story in one line: one mutex serializes everything, so **threads never buy throughput** — the uncontended round trip moves ops ~3× faster than two diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp index 286f35a..c67f4cb 100644 --- a/bench/queue_bench.cpp +++ b/bench/queue_bench.cpp @@ -69,8 +69,10 @@ constexpr int kSpscThreads = 2; constexpr int kMpmcThreads = 8; static_assert(kSpscThreads % 2 == 0 && kMpmcThreads % 2 == 0, "producer/consumer pairing needs an even thread count"); -// MinTime: contended runs need a second of samples to settle; the ctest smoke -// run overrides it with --benchmark_min_time=1x. +// MinTime: contended runs need a second of samples to settle. Setting it here +// beats the flag's seconds form — ComputeMinTime prefers a non-zero registered +// min_time — so --benchmark_min_time=0.2s is accepted and ignored. Only the +// iteration form overrides, which is what the ctest smoke run uses (1x). constexpr double kMinTimeSeconds = 1.0; BENCHMARK(BM_MutexQueueThroughput) diff --git a/tests/mutex_queue_test.cpp b/tests/mutex_queue_test.cpp index a5521c0..9abbbab 100644 --- a/tests/mutex_queue_test.cpp +++ b/tests/mutex_queue_test.cpp @@ -75,8 +75,8 @@ TEST(MutexQueue, SupportsMoveOnlyTypes) { TEST(MutexQueue, PopBlocksUntilPush) { MutexQueue q(1); int out = 0; - EXPECT_TRUE(testing::run_blocked([&] { return q.pop(out); }, // - [&] { EXPECT_TRUE(q.push(7)); })); + EXPECT_TRUE(test_util::run_blocked([&] { return q.pop(out); }, // + [&] { EXPECT_TRUE(q.push(7)); })); EXPECT_EQ(out, 7); } @@ -84,11 +84,11 @@ TEST(MutexQueue, PushBlocksUntilPopWhenFull) { MutexQueue q(1); ASSERT_TRUE(q.push(1)); int out = 0; - EXPECT_TRUE(testing::run_blocked([&] { return q.push(2); }, - [&] { - EXPECT_TRUE(q.pop(out)); - EXPECT_EQ(out, 1); - })); + EXPECT_TRUE(test_util::run_blocked([&] { return q.push(2); }, + [&] { + EXPECT_TRUE(q.pop(out)); + EXPECT_EQ(out, 1); + })); ASSERT_TRUE(q.pop(out)); EXPECT_EQ(out, 2); } @@ -119,15 +119,15 @@ TEST(MutexQueue, PopDrainsRemainingItemsAfterClose) { TEST(MutexQueue, CloseWakesBlockedPop) { MutexQueue q(1); int out = 0; - EXPECT_FALSE(testing::run_blocked([&] { return q.pop(out); }, // - [&] { q.close(); })); + EXPECT_FALSE(test_util::run_blocked([&] { return q.pop(out); }, // + [&] { q.close(); })); } TEST(MutexQueue, CloseWakesBlockedPush) { MutexQueue q(1); ASSERT_TRUE(q.push(1)); - EXPECT_FALSE(testing::run_blocked([&] { return q.push(2); }, // - [&] { q.close(); })); + EXPECT_FALSE(test_util::run_blocked([&] { return q.push(2); }, // + [&] { q.close(); })); } } // namespace diff --git a/tests/queue_test_util.hpp b/tests/queue_test_util.hpp index 14273bf..200790e 100644 --- a/tests/queue_test_util.hpp +++ b/tests/queue_test_util.hpp @@ -11,7 +11,7 @@ #include -namespace cq::testing { +namespace cq::test_util { // Long enough for a spawned thread to reach its blocking call; the tests stay // correct (just less interesting) if it ever proves too short. @@ -38,6 +38,6 @@ constexpr auto kSettleTime = std::chrono::milliseconds(5); return threads; } -} // namespace cq::testing +} // namespace cq::test_util #endif // CQ_TESTS_QUEUE_TEST_UTIL_HPP_ diff --git a/tests/stress_test.cpp b/tests/stress_test.cpp index 700f8f3..0563998 100644 --- a/tests/stress_test.cpp +++ b/tests/stress_test.cpp @@ -23,7 +23,7 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { MutexQueue q(kQueueCapacity); - auto producers = testing::spawn_threads(kProducers, [&q](int p) { + auto producers = test_util::spawn_threads(kProducers, [&q](int p) { // Assert only on failure: a per-item ASSERT is measurable under TSan. for (int i = 0; i < kItemsPerProducer; ++i) { const auto value = @@ -37,7 +37,7 @@ TEST(MutexQueueStress, ChecksumReconcilesAcrossProducersAndConsumers) { std::atomic consumed_sum{0}; std::atomic consumed_count{0}; - auto consumers = testing::spawn_threads(kConsumers, [&](int /*c*/) { + auto consumers = test_util::spawn_threads(kConsumers, [&](int /*c*/) { std::uint64_t local_sum = 0; std::uint64_t local_count = 0; std::uint64_t value = 0; From 87e9ffa526d1339e504005f80978a1638ffee55c Mon Sep 17 00:00:00 2001 From: Debra Date: Mon, 17 Aug 2026 07:41:56 +0800 Subject: [PATCH 9/9] docs: correct the CMake option comment and the round-trip figure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CMakeLists: the previous comment credited cmake_dependent_option with a cache-visibility benefit it does not provide. Verified: under TSan the default path writes no cache entry at all, and an explicit -DCQ_BUILD_BENCHMARKS=ON is demoted to CQ_BUILD_BENCHMARKS:INTERNAL=ON — recording ON while OFF was built. The STATUS line is the real signal - README: 19.1 ns predated the column's redefinition as 1/throughput; 1/105.1M ops/s makes the round trip 19.0 ns Co-Authored-By: Claude Fable 5 --- CMakeLists.txt | 6 ++++-- README.md | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index aa66c56..5d38c9a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -9,8 +9,10 @@ option(ENABLE_TSAN "Build with ThreadSanitizer" OFF) option(CQ_BUILD_TESTS "Build unit and stress tests" ON) # Timings from a sanitized build are meaningless, so benchmarks are forced off -# under TSan. Expressed as a dependent option so the cache reports what was -# actually built (a plain set() would leave ON showing in ccmake/cmake-gui). +# under TSan, and the user's choice is restored when TSan is turned back off. +# The option is hidden from ccmake/cmake-gui while forced (an explicitly passed +# -DCQ_BUILD_BENCHMARKS=ON is kept as an INTERNAL cache entry), so the STATUS +# line below — not the cache — is the signal that benchmarks were skipped. include(CMakeDependentOption) cmake_dependent_option(CQ_BUILD_BENCHMARKS "Build Google Benchmark targets" ON "NOT ENABLE_TSAN" OFF) diff --git a/README.md b/README.md index f32a925..953c574 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ the whole queue, not per-thread latency. | Benchmark (v1 MutexQueue) | Throughput | Per op | CV | |---|---|---|---| -| single-thread push+pop round trip | 105.1M ± 0.7M ops/s | 9.5 ns (19.1 ns per round trip) | 2.2% | +| single-thread push+pop round trip | 105.1M ± 0.7M ops/s | 9.5 ns (19.0 ns per round trip) | 2.2% | | SPSC (1 producer, 1 consumer) | 35.7M ± 0.5M ops/s | 28.0 ns | 1.4% | | MPMC (4 producers, 4 consumers) | 21.6M ± 0.1M ops/s | 46.3 ns | 0.5% |