test: add ThreadSanitizer verification for lock-free Channel<T>
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 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,288 @@
|
||||
// Contended stress tests for the lock-free SPSC Channel<T>.
|
||||
//
|
||||
// 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<T> is SPSC: exactly one producer thread and one consumer thread per
|
||||
// channel. Every scenario below honours that contract.
|
||||
|
||||
#include <catch2/catch_test_macros.hpp>
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <kpn/channel.hpp>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
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<int> 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<int> 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<long>(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<int> ch(/*capacity=*/4, /*spin_count=*/8);
|
||||
std::atomic<bool> threw{false};
|
||||
std::atomic<bool> 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<int> ch(/*capacity=*/8, /*spin_count=*/8);
|
||||
std::atomic<bool> 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<int> ch(/*capacity=*/4, /*spin_count=*/4);
|
||||
std::atomic<int> 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<int> 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<bool> 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<int> 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<bool> 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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user