From 139bfbb7946a85a0fe1f01c1913130d796ade91e Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Wed, 5 Aug 2026 14:06:26 +0200 Subject: [PATCH] fix: the sentinel slot holds one token and refuses a second MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit push_sentinel wrote eof_value_ unconditionally. Offering a second token before the first was taken did two wrong things at once. It lost the first silently, and a lost EOF is not a lost frame — it is the token every downstream node is waiting for in order to shut down, so losing it wedges the pipeline. And it wrote the storage while the consumer could be moving the previous value out of it. I expected that to be a stale read; ThreadSanitizer shows it is worse. On the shared_ptr storage that non-trivial types use, the racing write tears the refcount, and the stress case added here reports heap-use-after-free in extract() alongside the data race. try_push_sentinel now refuses when the slot is occupied, which turns the slot into a correct SPSC handshake: the producer is the only writer of eof_value_ and the only one that sets has_eof_, the consumer is the only one that clears it, so observing it false is what licenses the write. Refusal is recorded as a drop, and PoolNode reports it through the overflow event callback, because a refused control token going unnoticed is the failure this commit exists to stop. Refusing rather than queueing is deliberate. Two control tokens on one channel means the stream ended twice, which is a caller protocol error and not backpressure; parking and retrying would spin against a slot only the consumer can free, and there is no sensible second value to deliver after the end of a stream. The non-consuming try_push_sentinel exists so a refused token is still the caller's to report — the consuming push_sentinel cannot offer that, since the value has already been moved into its parameter. Single-shot EOF is what every current caller does, so this is latent for them today. It stops being latent the moment a pipeline is reused for a second input, which is what the persistent-pipeline work in 4b6e498 sets up. Verified in both directions under -DKPN_SANITIZER=thread: the new contended case reports three data races and a heap-use-after-free against the old overwrite, and is clean with the handshake. Full suite 137/137, TSan clean across unit and stress suites. --- include/kpn/channel.hpp | 41 +++++++++++++++++++++--- include/kpn/pool_node.hpp | 18 +++++++++-- tests/test_channel.cpp | 60 +++++++++++++++++++++++++++++++++++ tests/test_channel_stress.cpp | 49 ++++++++++++++++++++++++++++ 4 files changed, 161 insertions(+), 7 deletions(-) diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index aa87523..a35651b 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -226,12 +226,37 @@ public: // preserving ordering (EOF arrives after all data pushed before it). // // Only the sole producer may call it (SPSC contract, same as push()). - // Returns false if the channel is already disabled (token discarded — - // teardown is in progress, so the sentinel is moot). - bool push_sentinel(T value) { + // + // The slot holds exactly one undelivered token. A second offered before the + // first is taken is refused, not queued and not overwritten: two control + // tokens on one channel means the stream ended twice, which is a caller + // protocol error rather than backpressure, and silently coalescing them + // would hide it. + /// Outcome of offering a sentinel. SlotBusy is a protocol error, not + /// backpressure: it means a second control token was offered while the + /// first was still undelivered, and a channel carries at most one. + enum class SentinelResult { Taken, Closed, SlotBusy }; + + /// Non-consuming form. `value` is left untouched unless the result is + /// Taken, so a refused token is still the caller's to report. + SentinelResult try_push_sentinel(T& value) { if (!accepting_.load(std::memory_order_acquire)) { stats_.record_drop(); - return false; + return SentinelResult::Closed; + } + // Refuse rather than overwrite. Overwriting lost the first token + // silently, and worse, wrote eof_value_ while the consumer could be + // moving the previous one out of it — a data race on the storage, which + // for a shared_ptr payload is a torn refcount rather than a stale read. + // + // Checking here is what makes the slot a correct SPSC handshake: the + // producer is the only writer of eof_value_ and the only one that sets + // has_eof_, the consumer is the only one that clears it, so observing + // false here means the consumer has finished with the storage and will + // not touch it again until this store publishes the next token. + if (has_eof_.load(std::memory_order_acquire)) { + stats_.record_drop(); + return SentinelResult::SlotBusy; } eof_value_ = make_storage(std::move(value)); has_eof_.store(true, std::memory_order_release); @@ -240,7 +265,13 @@ public: wake_.fetch_add(1, std::memory_order_release); wake_.notify_one(); if (push_callback_) push_callback_(); - return true; + return SentinelResult::Taken; + } + + /// Consuming convenience form. Returns false when the token was not stored, + /// whether because the channel is closed or because one is already pending. + bool push_sentinel(T value) { + return try_push_sentinel(value) == SentinelResult::Taken; } // Blocking pop. Returns when an item is available. diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 81c0724..19a32be 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -570,7 +570,14 @@ private: // downstream pop() forever. Deliver them out-of-band (push_sentinel), // which never overflows and never blocks this node's worker thread. if (is_sentinel_value(val)) { - ch->push_sentinel(std::move(val)); + // A refused sentinel is a protocol error, not backpressure, so it + // is reported rather than parked and retried — retrying would spin + // forever against a slot only the consumer can free, and there is + // no correct value to deliver second anyway. Closed is normal + // during teardown and stays quiet. + if (ch->try_push_sentinel(val) == Channel> + ::SentinelResult::SlotBusy) + fire_callbacks(event_callbacks_); return true; } // Backpressure without parking the worker. A full channel means the @@ -1087,7 +1094,14 @@ private: // downstream pop() forever. Deliver them out-of-band (push_sentinel), // which never overflows and never blocks this node's worker thread. if (is_sentinel_value(val)) { - ch->push_sentinel(std::move(val)); + // A refused sentinel is a protocol error, not backpressure, so it + // is reported rather than parked and retried — retrying would spin + // forever against a slot only the consumer can free, and there is + // no correct value to deliver second anyway. Closed is normal + // during teardown and stays quiet. + if (ch->try_push_sentinel(val) == Channel> + ::SentinelResult::SlotBusy) + fire_callbacks(event_callbacks_); return true; } // See the note on the typed overload above: park rather than block. diff --git a/tests/test_channel.cpp b/tests/test_channel.cpp index d710214..8150b6d 100644 --- a/tests/test_channel.cpp +++ b/tests/test_channel.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -238,3 +239,62 @@ TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty", REQUIRE(out == 99); REQUIRE_FALSE(ch.try_pop_now(out)); // nothing left } + +// Regression: the sentinel slot holds one token and refuses a second. +// +// push_sentinel used to write eof_value_ unconditionally. Offering a second +// token before the first was taken therefore did two wrong things at once: it +// lost the first silently — and a lost EOF wedges every downstream pop forever +// — and it wrote the storage while the consumer could be moving the previous +// value out of it. For the shared_ptr storage that non-trivial types use, that +// is a torn refcount, not merely a stale read. +// +// Refusing is correct rather than queueing: two control tokens on one channel +// means the stream ended twice, which is a caller protocol error. Coalescing +// them would hide it, and there is no second value that could sensibly follow +// the end of a stream. +TEST_CASE("a second sentinel is refused, not swallowed", "[channel][sentinel]") { + Channel ch(4); + + REQUIRE(ch.push_sentinel(1)); + // Slot occupied: the first token is still undelivered. + REQUIRE_FALSE(ch.push_sentinel(2)); + + // The first survives intact — the overwrite is what used to lose it. + int out = 0; + REQUIRE(ch.try_pop_now(out)); + CHECK(out == 1); + + // And the slot is reusable once drained. + REQUIRE(ch.push_sentinel(3)); + REQUIRE(ch.try_pop_now(out)); + CHECK(out == 3); +} + +TEST_CASE("a refused sentinel is counted as a drop", "[channel][sentinel]") { + // Visibility matters more here than for a dropped value: the refusal means + // a control token went nowhere, and the only alternative to a counter is + // for it to vanish. + Channel ch(4); + REQUIRE(ch.push_sentinel(1)); + const auto before = ch.stats().drops.load(); + REQUIRE_FALSE(ch.push_sentinel(2)); + CHECK(ch.stats().drops.load() == before + 1); +} + +TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][sentinel]") { + // The non-consuming form exists so a refused token is still the caller's to + // report. The consuming push_sentinel cannot offer that, since the value is + // already moved into its parameter. + Channel ch(4); + std::string first = "eof-1", second = "eof-2"; + + REQUIRE(ch.try_push_sentinel(first) == Channel::SentinelResult::Taken); + REQUIRE(ch.try_push_sentinel(second) == Channel::SentinelResult::SlotBusy); + CHECK(second == "eof-2"); // not moved from + + ch.disable(); + std::string third = "eof-3"; + CHECK(ch.try_push_sentinel(third) == Channel::SentinelResult::Closed); + CHECK(third == "eof-3"); +} diff --git a/tests/test_channel_stress.cpp b/tests/test_channel_stress.cpp index eb58704..9f851e3 100644 --- a/tests/test_channel_stress.cpp +++ b/tests/test_channel_stress.cpp @@ -19,6 +19,7 @@ // Channel is SPSC: exactly one producer thread and one consumer thread per // channel. Every scenario below honours that contract. +#include #include #include #include @@ -279,3 +280,51 @@ TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)", REQUIRE(ch.approx_size() == 0); } } + +// Contended: a producer offering sentinels while the consumer takes them. +// +// The old push_sentinel wrote eof_value_ with no regard for whether the +// consumer was reading it, so a second offer racing a take was a data race on +// the storage — for the shared_ptr form used by non-trivial types, on the +// refcount. Under TSan the old code reports it; the handshake added alongside +// this test makes the producer's write conditional on observing the slot free, +// which is what serialises the two. +// +// Payload is a std::string so the storage is the shared_ptr path rather than +// the trivially-copyable one, and each token carries its own identity so a torn +// value shows up as a mismatch rather than as a plausible-looking result. +TEST_CASE("SPSC: offering sentinels concurrently with takes is race-free", + "[channel][stress][sentinel]") { + constexpr int kRounds = 20000; + Channel ch(4); + + std::atomic taken{0}; + std::atomic torn{false}; + std::atomic done{false}; + + std::thread consumer([&] { + std::string out; + while (!done.load(std::memory_order_acquire) || ch.approx_size() > 0) { + if (ch.try_pop_now(out)) { + if (out.rfind("eof-", 0) != 0) torn.store(true, std::memory_order_relaxed); + taken.fetch_add(1, std::memory_order_relaxed); + } + } + }); + + int accepted = 0; + for (int i = 0; i < kRounds; ++i) { + std::string tok = "eof-" + std::to_string(i); + if (ch.try_push_sentinel(tok) == Channel::SentinelResult::Taken) + ++accepted; + } + done.store(true, std::memory_order_release); + consumer.join(); + + INFO("accepted " << accepted << " taken " << taken.load()); + CHECK_FALSE(torn.load(std::memory_order_relaxed)); + // Every accepted token must be delivered: the slot is refused while full, + // so acceptance and delivery are one-to-one. + CHECK(taken.load(std::memory_order_relaxed) == accepted); + CHECK(accepted > 0); +}