From 399ee4cf9b6d900860404723ddda05315d4d092b Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Sat, 4 Jul 2026 20:40:35 +0200 Subject: [PATCH] test: add ThreadSanitizer verification for lock-free Channel Add a contended SPSC stress suite (tests/test_channel_stress.cpp) that actually exercises the ring's memory-ordering pairing and spin/futex/ lost-wakeup logic, plus the CMake and CI plumbing to run it under TSan: - KPN_SANITIZER cache var + kpn_sanitizer_flags() helper (no-op when unset) - kpn_tests_stress executable, labelled "stress" for CTest - reusable tsan.yaml workflow (gcc:14 builder image, already ships libtsan) - ci.yaml gains a tsan job on the same code/dockerfile triggers as test Co-Authored-By: Claude Opus 4.8 --- .gitea/workflows/ci.yaml | 7 + .gitea/workflows/tsan.yaml | 67 ++++++++ CMakeLists.txt | 24 +++ tests/CMakeLists.txt | 31 +++- tests/test_channel_stress.cpp | 288 ++++++++++++++++++++++++++++++++++ 5 files changed, 416 insertions(+), 1 deletion(-) create mode 100644 .gitea/workflows/tsan.yaml create mode 100644 tests/test_channel_stress.cpp diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 3373cec..e962340 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -73,6 +73,13 @@ jobs: if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }} uses: ./.gitea/workflows/test.yaml + # ThreadSanitizer run for the lock-free Channel. Same trigger conditions as + # test (code or image changed); runs in parallel with test. + tsan: + needs: [changes, docker] + if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }} + uses: ./.gitea/workflows/tsan.yaml + docs: needs: [changes, docker] if: ${{ !failure() && !cancelled() && github.ref == 'refs/heads/master' && (needs.changes.outputs.docs == 'true' || needs.changes.outputs.dockerfile == 'true') }} diff --git a/.gitea/workflows/tsan.yaml b/.gitea/workflows/tsan.yaml new file mode 100644 index 0000000..53f81ad --- /dev/null +++ b/.gitea/workflows/tsan.yaml @@ -0,0 +1,67 @@ +name: '๐Ÿงต ThreadSanitizer' + +# Reusable workflow: builds the channel stress suite with ThreadSanitizer and +# runs it. This is the dynamic half of verifying the lock-free SPSC Channel +# (the static half is the CDSChecker model-check harness in verify/). +# +# Triggering and path filtering are owned by ci.yaml (the orchestrator), which +# calls this only when code changed. workflow_dispatch is kept for manual runs. +# +# Runs in the prebuilt builder image (gcc:14), which already ships libtsan โ€” no +# package installs at job time. +on: + workflow_call: + workflow_dispatch: + +jobs: + tsan: + runs-on: linux/amd64 + container: + image: gitea.tourolle.paris/dtourolle/kpnpp-builder:latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + with: + path: tsan-${{ github.run_id }} + + - name: Cache FetchContent dependencies + uses: actions/cache@v3 + with: + path: ~/.cmake/fetchcontent + key: cmake-fetchcontent-${{ hashFiles('**/CMakeLists.txt') }} + restore-keys: cmake-fetchcontent- + + - name: Configure (TSan) + working-directory: tsan-${{ github.run_id }} + run: | + cmake -S . -B build \ + -G Ninja \ + -DCMAKE_BUILD_TYPE=Debug \ + -DKPN_SANITIZER=thread \ + -DKPN_BUILD_TESTS=ON \ + -DKPN_BUILD_EXAMPLES=OFF \ + -DKPN_BUILD_PYTHON=OFF \ + -DFETCHCONTENT_BASE_DIR=$HOME/.cmake/fetchcontent + + - name: Build (TSan) + working-directory: tsan-${{ github.run_id }} + run: cmake --build build --parallel --target kpn_tests kpn_tests_stress + + - name: Run stress suite under TSan + working-directory: tsan-${{ github.run_id }} + # halt_on_error=1 makes the first detected race fail the job; the report + # (with both stacks) is printed to the log. second_deadlock_stack gives + # the full picture for lock-order issues. + env: + TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1" + run: ./build/tests/kpn_tests_stress + + - name: Run unit tests under TSan + working-directory: tsan-${{ github.run_id }} + env: + TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1" + run: ./build/tests/kpn_tests + + - name: Cleanup + if: always() + run: rm -rf tsan-${{ github.run_id }} diff --git a/CMakeLists.txt b/CMakeLists.txt index dd147b5..63037fd 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -10,6 +10,30 @@ option(KPN_BUILD_PYTHON "Build Python bindings (requires nanobind)" ON) option(KPN_BUILD_EXAMPLES "Build examples" ON) option(KPN_WEB_DEBUG "Enable web debug UI (cpp-httplib)" OFF) +# Sanitizer build. Empty = off. Accepts "thread", "address", "undefined", +# or a combination like "address,undefined". Applied to all kpn targets via +# the kpn_sanitizer_flags() helper below. +# +# The lock-free SPSC Channel (include/kpn/channel.hpp) has hand-reasoned +# acquire/release ordering; -DKPN_SANITIZER=thread + the channel stress test +# (tests/test_channel_stress.cpp) is the dynamic half of verifying it. The +# static half is the CDSChecker model-check harness (see verify/). +set(KPN_SANITIZER "" CACHE STRING + "Build with sanitizer: thread | address | undefined | (empty = off)") + +# Translate KPN_SANITIZER into compile/link flags. No-op when empty. +function(kpn_sanitizer_flags out_var) + if(KPN_SANITIZER) + set(${out_var} + -fsanitize=${KPN_SANITIZER} + -fno-omit-frame-pointer + -g + PARENT_SCOPE) + else() + set(${out_var} "" PARENT_SCOPE) + endif() +endfunction() + # โ”€โ”€ Core library (header-only) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ add_library(kpn INTERFACE) target_include_directories(kpn INTERFACE diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a8f7583..fda6711 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -43,6 +43,35 @@ target_link_libraries(kpn_tests PRIVATE GTest::gtest ) +# โ”€โ”€ Channel stress suite (separate executable) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +# Contended SPSC tests for the lock-free Channel. Kept out of kpn_tests +# because each case runs many reps / tens of thousands of items and is slow. +# Most valuable under -DKPN_SANITIZER=thread, but correct (and run) without it. +add_executable(kpn_tests_stress test_channel_stress.cpp) +target_link_libraries(kpn_tests_stress PRIVATE kpn Catch2::Catch2WithMain) + +# โ”€โ”€ Sanitizer flags โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +# kpn_sanitizer_flags() is defined in the top-level CMakeLists and is a no-op +# unless -DKPN_SANITIZER=... is set. Sanitizer must be on both compile and link. +kpn_sanitizer_flags(_kpn_san) +if(_kpn_san) + foreach(_t kpn_tests kpn_tests_stress) + target_compile_options(${_t} PRIVATE ${_kpn_san}) + target_link_options(${_t} PRIVATE ${_kpn_san}) + endforeach() +endif() + include(CTest) include(Catch) -catch_discover_tests(kpn_tests) + +# DISCOVERY_MODE PRE_TEST defers test enumeration to `ctest` run time. The +# default (POST_BUILD) runs each test binary during the build to list its +# cases โ€” which fails a sanitizer build: a TSan/ASan binary needs a fixed +# address-space layout and aborts on startup ("unexpected memory mapping") +# under the container's ASLR, breaking the build before any test runs. The +# tsan.yaml job invokes the binaries directly (not via ctest), so deferring +# discovery costs nothing there and keeps `ctest` working for normal builds. +catch_discover_tests(kpn_tests DISCOVERY_MODE PRE_TEST) +# Register the stress suite under its own label so CI can run / time it +# separately from the fast unit tests. +catch_discover_tests(kpn_tests_stress DISCOVERY_MODE PRE_TEST PROPERTIES LABELS "stress") diff --git a/tests/test_channel_stress.cpp b/tests/test_channel_stress.cpp new file mode 100644 index 0000000..bd113d0 --- /dev/null +++ b/tests/test_channel_stress.cpp @@ -0,0 +1,288 @@ +// Contended stress tests for the lock-free SPSC Channel. +// +// The other channel tests (test_channel.cpp) are single-threaded or use a +// single 20 ms sleep to order two threads โ€” they never actually contend on the +// ring, so they exercise neither the memory-ordering pairing nor the +// spin/futex/lost-wakeup logic in pop(). +// +// These tests are written to be run under ThreadSanitizer: +// +// cmake -B build -DKPN_SANITIZER=thread -DKPN_BUILD_EXAMPLES=OFF -DKPN_BUILD_PYTHON=OFF +// cmake --build build --target kpn_tests_tsan +// ./build/tests/kpn_tests_tsan +// +// They are also valid (and meaningful) without a sanitizer: the value/sequence +// assertions catch lost or duplicated items regardless of build flags. TSan +// adds detection of the underlying data race even on runs where the race did +// not corrupt observable state. +// +// Channel is SPSC: exactly one producer thread and one consumer thread per +// channel. Every scenario below honours that contract. + +#include +#include +#include +#include +#include +#include + +using namespace kpn; +using namespace std::chrono_literals; + +namespace { + +// Repeat each scenario enough times that rare interleavings (spin window just +// missing / just catching the next push, disable landing inside the futex +// wait) actually occur across a run. Kept modest so a TSan run stays minutes, +// not hours. +constexpr int kReps = 200; + +} // namespace + +TEST_CASE("SPSC: every pushed item is popped exactly once, in order", "[channel][stress]") { + // Small capacity forces frequent full/empty transitions, so both the + // producer's overflow-retry and the consumer's spin->futex path are hit + // many times. The producer retries on overflow rather than dropping, so + // the consumer must observe a strictly contiguous 0..N-1 sequence. + constexpr int N = 50'000; + Channel ch(/*capacity=*/4, /*spin_count=*/16); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + }); + + int expected = 0; + bool in_order = true; + for (int i = 0; i < N; ++i) { + int v = ch.pop(); + if (v != expected) in_order = false; + ++expected; + } + producer.join(); + + REQUIRE(in_order); + REQUIRE(expected == N); + REQUIRE(ch.size() == 0); +} + +TEST_CASE("SPSC: tight empty<->non-empty transitions exercise spin/futex boundary", + "[channel][stress]") { + // spin_count=0 forces every empty pop() straight into atomic::wait, so this + // hammers the lost-wakeup guard (snapshot wake_, re-check tail_, then wait). + // The producer pushes one item then waits to go empty again, maximising the + // number of empty->non-empty edges relative to item count. + constexpr int N = 20'000; + Channel ch(/*capacity=*/2, /*spin_count=*/0); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + }); + + long sum = 0; + for (int i = 0; i < N; ++i) sum += ch.pop(); + producer.join(); + + // Sum of 0..N-1 โ€” detects any lost or duplicated item. + REQUIRE(sum == static_cast(N) * (N - 1) / 2); +} + +TEST_CASE("SPSC: disable() while consumer is blocked in pop() unblocks cleanly", + "[channel][stress]") { + // The data race of record: consumer blocked in pop() (spinning or parked in + // the futex) while the owner thread calls disable(). pop() must observe the + // close and throw ChannelClosedError โ€” it must not hang and must not read + // past the ring. Repeated so disable() lands at many points in pop()'s loop. + for (int rep = 0; rep < kReps; ++rep) { + Channel ch(/*capacity=*/4, /*spin_count=*/8); + std::atomic threw{false}; + std::atomic finished{false}; + + std::thread consumer([&] { + try { + ch.pop(); // empty channel: will block + } catch (const ChannelClosedError&) { + threw.store(true, std::memory_order_relaxed); + } + finished.store(true, std::memory_order_relaxed); + }); + + // Give the consumer a chance to reach the wait, then close. + std::this_thread::sleep_for(50us); + ch.disable(); + + consumer.join(); + REQUIRE(finished.load()); + REQUIRE(threw.load()); + } +} + +TEST_CASE("SPSC: producer racing a disable() never throws and never hangs", + "[channel][stress]") { + // Mirror of the above from the producer side: push() racing disable() must + // either enqueue or silently drop, never throw ChannelClosedError and never + // wedge. Overflow is still a legal outcome (full accepting channel) and is + // tolerated here. + for (int rep = 0; rep < kReps; ++rep) { + Channel ch(/*capacity=*/8, /*spin_count=*/8); + std::atomic bad{false}; + + std::thread producer([&] { + for (int i = 0; i < 1000; ++i) { + try { ch.push(i); } + catch (const ChannelOverflowError&) { /* legal: full */ } + catch (...) { bad.store(true, std::memory_order_relaxed); break; } + } + }); + + std::this_thread::sleep_for(20us); + ch.disable(); // owner closes mid-stream + producer.join(); + + REQUIRE_FALSE(bad.load()); + } +} + +TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition", + "[channel][stress]") { + // The empty->non-empty callback ([channel.hpp] was_empty branch) is read by + // the consumer-side notification path. Run it under contention to make sure + // the was_empty detection isn't torn by a concurrent pop(). + Channel ch(/*capacity=*/4, /*spin_count=*/4); + std::atomic callbacks{0}; + ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); }); + + constexpr int N = 10'000; + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + }); + + for (int i = 0; i < N; ++i) (void)ch.pop(); + producer.join(); + + // At least one transition, at most one per item; mainly we assert the run + // completed without TSan flagging a race on push_callback_/was_empty. + REQUIRE(callbacks.load() >= 1); + REQUIRE(callbacks.load() <= N); +} + +// What the out-of-band sentinel guarantees under contention โ€” and what it does +// not. push_sentinel() publishes has_eof_ (release) after the producer's N ring +// pushes; a consumer that observes has_eof_ (acquire) therefore also observes +// every value pushed before it. What these tests assert: +// +// * Losslessness โ€” every value 0..N-1 is delivered exactly once (contiguous, +// no gaps, no duplicates) and the sentinel is delivered exactly once. This +// is the invariant that must hold on every run; a broken acquire/release +// pairing would surface as a lost/duplicated value or (under TSan) a data +// race on has_eof_/eof_value_. +// +// What they deliberately do NOT assert is that the sentinel is the *strictly +// last* item popped. pop() checks emptiness (h == t) using a tail_ snapshot +// taken at the top of its loop; the producer can push more values *and* the +// sentinel in the window before take_sentinel() runs, so the consumer may +// surface the sentinel with a few real values still queued behind it. Those +// values are not lost โ€” a consumer that keeps draining still receives them โ€” +// but "sentinel arrives dead last" is not a property the channel promises, so +// asserting it would be flaky. We track how many values trailed the sentinel +// for visibility without failing on it. + +TEST_CASE("SPSC: sentinel and all values survive contention (blocking pop)", + "[channel][stress]") { + constexpr int N = 20'000; + constexpr int SENTINEL = -1; + + for (int rep = 0; rep < kReps; ++rep) { + // Small ring + tiny spin window so the ring is frequently empty exactly + // when the sentinel is published โ€” the interleaving under test. + Channel ch(/*capacity=*/4, /*spin_count=*/8); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + ch.push_sentinel(SENTINEL); // must-deliver, never overflows/blocks + }); + + std::vector seen(N, false); + int values = 0; + int sentinels = 0; + bool duplicate = false; + // Drain until the sentinel AND all N values have been received; the + // sentinel may arrive before the last few values (see note above). + while (values < N || sentinels == 0) { + int v = ch.pop(); + if (v == SENTINEL) { ++sentinels; continue; } + if (seen[v]) duplicate = true; else seen[v] = true; + ++values; + } + producer.join(); + + REQUIRE_FALSE(duplicate); + REQUIRE(values == N); // every value delivered exactly once + REQUIRE(sentinels == 1); // sentinel delivered exactly once + REQUIRE(ch.size() == 0); + REQUIRE(ch.approx_size() == 0); + } +} + +TEST_CASE("SPSC: sentinel and all values survive contention (try_pop_now)", + "[channel][stress]") { + // The pool-node consume path is try_pop_now(), not pop(): it must surface + // the out-of-band sentinel once the ring is observed empty. The consumer + // spins with no sleeps, racing the producer at full tilt across the + // empty-ring boundary where take_sentinel() is reached. + constexpr int N = 20'000; + constexpr int SENTINEL = -1; + + for (int rep = 0; rep < kReps; ++rep) { + Channel ch(/*capacity=*/4, /*spin_count=*/0); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + ch.push_sentinel(SENTINEL); + }); + + std::vector seen(N, false); + int values = 0; + int sentinels = 0; + bool duplicate = false; + int v; + while (values < N || sentinels == 0) { + if (!ch.try_pop_now(v)) { std::this_thread::yield(); continue; } + if (v == SENTINEL) { ++sentinels; continue; } + if (seen[v]) duplicate = true; else seen[v] = true; + ++values; + } + producer.join(); + + REQUIRE_FALSE(duplicate); + REQUIRE(values == N); + REQUIRE(sentinels == 1); + // Sentinel held no ring slot; once drained the channel is fully empty. + REQUIRE(ch.size() == 0); + REQUIRE(ch.approx_size() == 0); + } +}