feat: add MutexQueue v1 with tests and benchmarks

.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

.github/workflows/ci.yml

@@ -30,9 +30,7 @@ jobs:
       - name: Build
         run: cmake --build build-tsan
       - name: Test
-        # --no-tests=error arms itself once tests/ lands; until then the
-        # scaffolding is allowed to no-op.
-        run: ctest --test-dir build-tsan --output-on-failure --no-tests="$([ -d tests ] && echo error || echo ignore)"
+        run: ctest --test-dir build-tsan --output-on-failure --no-tests=error
 
   bench-build:
     name: Benchmarks (Release, smoke-run)
@@ -44,7 +42,7 @@ jobs:
       - name: Build
         run: cmake --build build-rel
       - name: Smoke-run benchmarks
-        run: ctest --test-dir build-rel --output-on-failure --no-tests="$([ -d bench ] && echo error || echo ignore)"
+        run: ctest --test-dir build-rel --output-on-failure --no-tests=error
 
   lint:
     name: clang-format & clang-tidy

CMakeLists.txt

@@ -7,16 +7,30 @@ 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, 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)
+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)
 
-# Header-only library; consumers inherit C++20 via the compile feature.
+# 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)
 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
@@ -38,9 +52,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 +64,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)
   FetchContent_Declare(
     benchmark
     URL https://github.com/google/benchmark/archive/refs/tags/v1.9.1.tar.gz

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,34 @@ 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`.
+
+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 | 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% |
+
+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._

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:
+  # 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: '^_$'

bench/CMakeLists.txt

@@ -0,0 +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.
+add_test(NAME bench_smoke COMMAND queue_bench --benchmark_min_time=1x)

bench/queue_bench.cpp

@@ -0,0 +1,103 @@
+// 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, 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 total queue ops per second (a push and its pop count as two).
+//
+// 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 <cq/mutex_queue.hpp>
+
+#include <cstddef>
+#include <cstdint>
+#include <cstdlib>
+#include <memory>
+
+#include <benchmark/benchmark.h>
+
+namespace {
+
+constexpr std::size_t kCapacity = 1024;
+
+// Created/destroyed by the Setup/Teardown hooks below, which run once per
+// repetition outside the threaded region.
+std::unique_ptr<cq::MutexQueue<std::uint64_t>> shared_queue;
+
+void setup_queue(const benchmark::State& /*state*/) {
+  shared_queue = std::make_unique<cq::MutexQueue<std::uint64_t>>(kCapacity);
+  // Half-full start: neither side begins blocked on a condition variable, so
+  // the measurement starts in steady state.
+  bool prefilled = true;
+  for (std::uint64_t i = 0; i < kCapacity / 2; ++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) {
+  const bool is_producer = state.thread_index() < state.threads() / 2;
+  auto& queue = *shared_queue;
+
+  if (is_producer) {
+    std::uint64_t item = 0;
+    for (auto _ : state) {
+      benchmark::DoNotOptimize(queue.push(item++));
+    }
+  } else {
+    std::uint64_t value = 0;
+    for (auto _ : state) {
+      benchmark::DoNotOptimize(queue.pop(value));
+    }
+  }
+  // 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");
+// 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)
+    ->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.
+void BM_MutexQueuePushPopSingleThread(benchmark::State& state) {
+  cq::MutexQueue<std::uint64_t> queue(kCapacity);
+  std::uint64_t value = 0;
+  for (auto _ : state) {
+    benchmark::DoNotOptimize(queue.push(1));
+    benchmark::DoNotOptimize(queue.pop(value));
+  }
+  // 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)
+    ->MinTime(kMinTimeSeconds)
+    ->Name("MutexQueue/single_thread_roundtrip");
+
+}  // namespace

include/cq/mutex_queue.hpp

@@ -0,0 +1,99 @@
+#ifndef CQ_MUTEX_QUEUE_HPP_
+#define CQ_MUTEX_QUEUE_HPP_
+
+#include <condition_variable>
+#include <cstddef>
+#include <mutex>
+#include <vector>
+
+namespace cq {
+
+/// v1 baseline: bounded FIFO ring guarded by a single std::mutex, with
+/// not_full / not_empty condition variables and close() shutdown semantics.
+///
+/// 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
+/// 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
+/// 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.
+template <typename T>
+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: 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;
+  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).
+  [[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.
+  [[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.
+  [[nodiscard]] bool pop(T& out);
+
+  /// Dequeues into out without blocking.
+  /// @param[out] out Receives the dequeued element on success; untouched on
+  ///   failure.
+  /// @return false if the queue is empty.
+  [[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 (advisory snapshot).
+  [[nodiscard]] bool closed() const;
+
+  /// @return Current number of queued elements (advisory snapshot).
+  [[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<T> 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_