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/.github/workflows/ci.yml b/.github/workflows/ci.yml index 16f60b7..55995cc 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CMakeLists.txt b/CMakeLists.txt index c2bcc25..5d38c9a 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/README.md b/README.md index 45feca6..953c574 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,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._ diff --git a/bench/.clang-tidy b/bench/.clang-tidy new file mode 100644 index 0000000..a5ac14e --- /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: + # 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/CMakeLists.txt b/bench/CMakeLists.txt new file mode 100644 index 0000000..7dbb97a --- /dev/null +++ b/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) diff --git a/bench/queue_bench.cpp b/bench/queue_bench.cpp new file mode 100644 index 0000000..c67f4cb --- /dev/null +++ b/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 + +#include +#include +#include +#include + +#include + +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> 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. + 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 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 diff --git a/include/cq/mutex_queue.hpp b/include/cq/mutex_queue.hpp new file mode 100644 index 0000000..cf5a3a5 --- /dev/null +++ b/include/cq/mutex_queue.hpp @@ -0,0 +1,99 @@ +#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. +/// +/// 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 +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 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..b879a77 --- /dev/null +++ b/include/cq/mutex_queue.ipp @@ -0,0 +1,129 @@ +// 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"); + } +} + +// 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) { + { + 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_); + if (std::exchange(closed_, true)) { + return; // already closed; waiters were woken the first time + } + } + 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..5fa3960 --- /dev/null +++ b/tests/.clang-tidy @@ -0,0 +1,5 @@ +--- +InheritParentConfig: true +# 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 new file mode 100644 index 0000000..efd8fe3 --- /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 + 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/mutex_queue_test.cpp b/tests/mutex_queue_test.cpp new file mode 100644 index 0000000..9abbbab --- /dev/null +++ b/tests/mutex_queue_test.cpp @@ -0,0 +1,134 @@ +// Unit tests for cq::MutexQueue (v1: mutex + condition_variable bounded queue). + +#include + +#include +#include + +#include + +#include "queue_test_util.hpp" + +namespace cq { +namespace { + +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(test_util::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(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); +} + +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(test_util::run_blocked([&] { return q.pop(out); }, // + [&] { q.close(); })); +} + +TEST(MutexQueue, CloseWakesBlockedPush) { + MutexQueue q(1); + ASSERT_TRUE(q.push(1)); + EXPECT_FALSE(test_util::run_blocked([&] { return q.push(2); }, // + [&] { q.close(); })); +} + +} // namespace +} // namespace cq diff --git a/tests/queue_test_util.hpp b/tests/queue_test_util.hpp new file mode 100644 index 0000000..200790e --- /dev/null +++ b/tests/queue_test_util.hpp @@ -0,0 +1,43 @@ +// 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 +#include + +#include + +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. +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(); + return pending.get(); +} + +// Launches count threads, each running body(thread_index). Hold the returned +// 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); + } + return threads; +} + +} // namespace cq::test_util + +#endif // CQ_TESTS_QUEUE_TEST_UTIL_HPP_ diff --git a/tests/stress_test.cpp b/tests/stress_test.cpp new file mode 100644 index 0000000..0563998 --- /dev/null +++ b/tests/stress_test.cpp @@ -0,0 +1,65 @@ +// Checksum stress test (see README correctness policy): N producers push known +// values, consumers' totals must reconcile. Run under ThreadSanitizer. + +#include + +#include +#include + +#include + +#include "queue_test_util.hpp" + +namespace cq { +namespace { + +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 = 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 = + (static_cast(p) * kItemsPerProducer) + static_cast(i) + 1; + if (!q.push(value)) { + ADD_FAILURE() << "producer " << p << " push failed at item " << i; + break; + } + } + }); + + std::atomic consumed_sum{0}; + std::atomic consumed_count{0}; + 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; + 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); + }); + + 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]. + 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