diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index 3f0b05d..4466fd8 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -136,6 +136,36 @@ public: push_callback_(); } + // Lossless, non-blocking delivery for a must-deliver control token (EOF). + // + // A sentinel is stored out-of-band — in a dedicated slot that does NOT + // consume ring capacity — so this can never overflow and never blocks the + // caller. That distinction is essential: each KPN node has a single worker + // thread, so a *blocking* push would park that thread and stop it draining + // its own input, cascading into a hold-and-wait deadlock under backpressure. + // Setting a flag and returning keeps the worker free to keep popping. + // + // The consumer's pop() drains the ring first, then delivers this sentinel, + // 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) { + if (!accepting_.load(std::memory_order_acquire)) { + stats_.record_drop(); + return false; + } + eof_value_ = make_storage(std::move(value)); + has_eof_.store(true, std::memory_order_release); + // Wake a consumer blocked in pop(): the sentinel is now deliverable even + // though the ring may be empty. + wake_.fetch_add(1, std::memory_order_release); + wake_.notify_one(); + if (push_callback_) push_callback_(); + return true; + } + // Blocking pop. Returns when an item is available. // Throws ChannelClosedError if the channel is disabled (regardless of fill). T pop() { @@ -148,21 +178,28 @@ public: // If empty, spin before sleeping: avoids the futex when the next item // arrives within the spin window (~4 µs at default spin_count=200 on x86). if (h == t) { + // Ring drained — deliver any pending out-of-band sentinel (EOF) + // now, so it always arrives after the data pushed before it. + { T s; if (take_sentinel(s)) return s; } + if (!accepting_.load(std::memory_order_acquire)) throw ChannelClosedError{}; - for (std::size_t s = 0; s < spin_count_; ++s) { + for (std::size_t si = 0; si < spin_count_; ++si) { spin_hint(); t = tail_.load(std::memory_order_acquire); if (t != h) break; + { T s; if (take_sentinel(s)) return s; } if (!accepting_.load(std::memory_order_relaxed)) throw ChannelClosedError{}; } if (h == t) { - // Still empty after spin — sleep until push() or disable() fires. - // Re-check tail after loading w to guard against a lost wakeup. + // Still empty after spin — sleep until push()/push_sentinel() + // or disable() fires. Re-check tail and the sentinel after + // loading w to guard against a lost wakeup. if (tail_.load(std::memory_order_acquire) != h) continue; + if (has_eof_.load(std::memory_order_acquire)) continue; wake_.wait(w, std::memory_order_relaxed); continue; } @@ -190,9 +227,12 @@ public: } // Immediate non-blocking pop. Returns false if the ring is empty. + // Once the ring is drained, delivers any pending out-of-band sentinel (EOF) + // so pool nodes — which pop only via this path — still receive the token. bool try_pop_now(T& out) { const std::size_t h = head_.load(std::memory_order_relaxed); - if (h == tail_.load(std::memory_order_acquire)) return false; + if (h == tail_.load(std::memory_order_acquire)) + return take_sentinel(out); out = extract(std::move(buf_[h & ring_mask_])); head_.store(h + 1, std::memory_order_release); stats_.record_pop(); @@ -217,12 +257,21 @@ public: push_callback_ = std::move(cb); } - // Size derived lazily from ring indices — no separate counter on the hot path. + // Ring occupancy, derived lazily from indices — no separate counter on the + // hot path. Excludes any out-of-band sentinel (that lives outside the ring). std::size_t size() const { return tail_.load(std::memory_order_relaxed) - head_.load(std::memory_order_relaxed); } - std::size_t approx_size() const { return size(); } + + // A pending out-of-band sentinel (EOF) counts as consumable work here even + // though it holds no ring slot. This is what node readiness checks call, so + // a channel carrying only a sentinel still schedules its consumer's next + // fire — without this the sentinel would never be popped and the pipeline + // would deadlock at teardown. + std::size_t approx_size() const { + return size() + (has_eof_.load(std::memory_order_acquire) ? 1u : 0u); + } std::size_t capacity() const { return capacity_; } bool is_accepting() const { return accepting_.load(std::memory_order_relaxed); } @@ -260,6 +309,17 @@ private: return *s; } + // Consume the out-of-band sentinel if one is pending. Consumer-only. + // Called only when the ring is observed empty, so the sentinel is always + // delivered after every value pushed before it. + bool take_sentinel(T& out) { + if (!has_eof_.load(std::memory_order_acquire)) return false; + out = extract(std::move(eof_value_)); + has_eof_.store(false, std::memory_order_release); + stats_.record_pop(); + return true; + } + const std::size_t capacity_; const std::size_t spin_count_; std::size_t ring_mask_; @@ -267,8 +327,16 @@ private: std::function push_callback_; ChannelStats stats_; + // Out-of-band sentinel (EOF): stored outside the ring so its delivery never + // depends on ring capacity and never blocks the producer. Written by the + // producer (push_sentinel), read+cleared by the consumer (take_sentinel); + // has_eof_ is the publish/consume handshake. + storage_type eof_value_{}; + std::atomic has_eof_{false}; + // Separate cache lines: head_ is written only by the consumer; // tail_ and wake_ are written only by the producer. + // wake_ wakes a blocked pop() on enqueue or on a pending sentinel. alignas(64) std::atomic head_{0}; alignas(64) std::atomic tail_{0}; std::atomic wake_{0}; diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index e4ceb54..4fe37f2 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -22,6 +22,38 @@ namespace kpn { +// ── Sentinel detection ──────────────────────────────────────────────────────── +// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type +// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw +// source Frame) or nested one level under a `.source` member (`v.source.eof`, +// as on the pipeline's SceneFrame/…/MatchedSceneFrame message types, which wrap +// the originating Frame). Sentinels are delivered losslessly and non-blockingly +// via Channel::push_sentinel() instead of the throwing push(), so backpressure +// can never drop the token that unblocks downstream teardown. +// +// Types with neither shape are never treated as sentinels — both traits are +// SFINAE-safe and the runtime check compiles away to `false` for them, so this +// stays a no-op for pipelines that don't use an eof convention. +template +struct has_eof_field : std::false_type {}; +template +struct has_eof_field(std::declval().eof))>> + : std::true_type {}; + +template +struct has_source_eof_field : std::false_type {}; +template +struct has_source_eof_field(std::declval().source.eof))>> + : std::true_type {}; + +template +constexpr bool is_sentinel_value(const T& v) { + if constexpr (has_eof_field::value) return static_cast(v.eof); + else if constexpr (has_source_eof_field::value) return static_cast(v.source.eof); + else return false; +} + // ── PoolNode ────────────────────────────────────────────────────────────────── // // Reactive alternative to Node<>. Instead of owning a blocked thread, the node @@ -361,6 +393,13 @@ private: void push_one_out(std::tuple_element_t&& val) { auto* ch = std::get(output_channels_); if (!ch) return; + // Sentinels (EOF) must never be dropped: a lost token wedges every + // 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)); + return; + } try { ch->push(std::move(val)); } catch (const ChannelOverflowError&) { @@ -660,6 +699,13 @@ private: void push_one_out(std::tuple_element_t&& val) { auto* ch = std::get(output_channels_); if (!ch) return; + // Sentinels (EOF) must never be dropped: a lost token wedges every + // 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)); + return; + } try { ch->push(std::move(val)); } catch (const ChannelOverflowError&) { diff --git a/tests/test_channel.cpp b/tests/test_channel.cpp index 50b11d0..d710214 100644 --- a/tests/test_channel.cpp +++ b/tests/test_channel.cpp @@ -175,3 +175,66 @@ TEST_CASE("bandwidth_mbs returns 0 when elapsed_s is zero or negative", "[channe REQUIRE(snap.bandwidth_mbs(0.0) == 0.0); REQUIRE(snap.bandwidth_mbs(-1.0) == 0.0); } + +TEST_CASE("push_sentinel never overflows even on a full channel", "[channel][sentinel]") { + Channel ch(2); + ch.push(1); + ch.push(2); // channel full — a plain push(3) would throw ChannelOverflowError + + // The sentinel is stored out-of-band, so it neither throws nor blocks the + // caller — the exact property an EOF token needs under backpressure. This + // returns immediately with the ring still full. + REQUIRE(ch.push_sentinel(99)); + REQUIRE(ch.size() == 2); // sentinel did not consume ring capacity +} + +TEST_CASE("push_sentinel is delivered after all ring data, in order", "[channel][sentinel]") { + Channel ch(4); + ch.push(1); + ch.push(2); + ch.push_sentinel(99); // enqueue EOF while data is still buffered + + // Data drains first; the sentinel arrives only once the ring is empty. + REQUIRE(ch.pop() == 1); + REQUIRE(ch.pop() == 2); + REQUIRE(ch.pop() == 99); +} + +TEST_CASE("push_sentinel wakes a blocked pop", "[channel][sentinel]") { + Channel ch(2); // empty + std::thread producer([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + ch.push_sentinel(99); // must wake a consumer parked on an empty ring + }); + REQUIRE(ch.pop() == 99); + producer.join(); +} + +TEST_CASE("approx_size counts a pending sentinel so consumers stay schedulable", + "[channel][sentinel]") { + Channel ch(4); + REQUIRE(ch.approx_size() == 0); + ch.push_sentinel(99); + // Node readiness checks call approx_size(); it must report the out-of-band + // sentinel as consumable work even though it holds no ring slot. + REQUIRE(ch.approx_size() == 1); + REQUIRE(ch.size() == 0); // ...but the ring itself is still empty + int out = 0; + REQUIRE(ch.try_pop_now(out)); + REQUIRE(out == 99); + REQUIRE(ch.approx_size() == 0); +} + +TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty", + "[channel][sentinel]") { + Channel ch(2); + ch.push(1); + ch.push_sentinel(99); + + int out = 0; + REQUIRE(ch.try_pop_now(out)); // ring data first + REQUIRE(out == 1); + REQUIRE(ch.try_pop_now(out)); // then the sentinel + REQUIRE(out == 99); + REQUIRE_FALSE(ch.try_pop_now(out)); // nothing left +}