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); +}