diff --git a/include/kpn/branch.hpp b/include/kpn/branch.hpp index fcd53ce..676a1f8 100644 --- a/include/kpn/branch.hpp +++ b/include/kpn/branch.hpp @@ -7,12 +7,70 @@ #include #include +#include #include #include #include namespace kpn { +// ── Lossless single-output delivery ─────────────────────────────────────────── +// +// Shared by RouterNode and FilterNode, which each deliver a value to exactly one +// channel. Both previously did +// +// try { ch->push(val); } catch (const ChannelOverflowError&) {} +// +// which discards the value whenever the consumer is behind. 6595e6e made node +// outputs lossless, 28e0667 stopped them parking a worker, and a8cfe73 did the +// same for FanoutNode — these two were in none of them, and were the last +// remaining users of the throwing push() on a data path. +// +// A dropped item does not degrade a downstream result, it silently changes one. +// Worse, a dropped *sentinel* wedges the pipeline outright: EOF is what tells +// every downstream node to shut down, and there is nothing after it to retry. +// A filter that passes EOF by predicate but drops it by backpressure is a +// pipeline that never terminates. +// +// So sentinels go out-of-band via push_sentinel (a dedicated slot that consumes +// no ring capacity and cannot overflow), and everything else is retried until +// taken. Like FanoutNode and unlike a pool node, these own a private thread, so +// waiting here costs no scheduler worker and needs no space-callback park. +// stop_flag_ is rechecked every pass so teardown cannot hang on a full output. +// +// `parked` receives the time spent waiting, which the caller charges to blocked +// rather than exec — a parked node is idle, and charging it to exec reports the +// node as busy exactly when it is the one being held up. +// +// Returns false if stopped with the value undelivered. +template +bool deliver_one(Channel* ch, T& val, const std::atomic& stop_flag, + duration_t& parked) { + if (is_sentinel_value(val)) { + ch->push_sentinel(std::move(val)); + return true; + } + const auto park_from = clock_t::now(); + for (;;) { + if (ch->try_push(val)) { + parked = duration_t(clock_t::now() - park_from); + return true; + } + if (stop_flag.load(std::memory_order_relaxed)) { + // Teardown with work in hand. One last throwing push, purely so the + // channel's own stats record the loss (drop if it is disabled, + // overflow if it is merely full). The point of the lossless path is + // that a loss is never invisible, and a silent return here would + // reintroduce exactly the hole this function exists to close. + try { ch->push(std::move(val)); } + catch (const ChannelOverflowError&) {} + parked = duration_t(clock_t::now() - park_from); + return false; + } + std::this_thread::sleep_for(std::chrono::microseconds(50)); + } +} + // ── RouterNode ──────────────────────────────────────────────────────────────── // // Reads one item and pushes it to exactly one of N output channels, chosen by @@ -126,15 +184,20 @@ private: auto t1 = clock_t::now(); auto cpu0 = NodeStats::cpu_now(); + // An out-of-range selector still drops by design (documented on + // the class): the item was routed nowhere, not lost to a full + // channel. Only the latter is what deliver_one exists to stop. std::size_t idx = selector_(val); - if (idx < N && out_channels_[idx]) { - try { out_channels_[idx]->push(val); } - catch (const ChannelOverflowError&) {} - } + duration_t parked{0}; + bool delivered = true; + if (idx < N && out_channels_[idx]) + delivered = deliver_one(out_channels_[idx], val, stop_flag_, parked); auto cpu1 = NodeStats::cpu_now(); auto t2 = clock_t::now(); - stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); + stats_.record_exec(duration_t(t2 - t1) - parked, + duration_t(t1 - t0) + parked, cpu0, cpu1); + if (!delivered) break; } catch (const ChannelClosedError&) { break; } @@ -261,12 +324,19 @@ private: auto t1 = clock_t::now(); auto cpu0 = NodeStats::cpu_now(); + // A value the predicate rejects is dropped by design and is not + // counted as a processed frame. One it accepts is now delivered + // losslessly — including a sentinel, which a filter typically + // passes unconditionally so downstream can shut down, and which + // the old throwing push discarded whenever the output was full. if (pred_(val) && out_ch_) { - try { out_ch_->push(val); } - catch (const ChannelOverflowError&) {} + duration_t parked{0}; + const bool delivered = deliver_one(out_ch_, val, stop_flag_, parked); auto cpu1 = NodeStats::cpu_now(); auto t2 = clock_t::now(); - stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); + stats_.record_exec(duration_t(t2 - t1) - parked, + duration_t(t1 - t0) + parked, cpu0, cpu1); + if (!delivered) break; } } catch (const ChannelClosedError&) { break; diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index a821e6f..042a35d 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -23,37 +23,8 @@ 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; -} +// Sentinel detection (has_eof_field / is_sentinel_value) lives in traits.hpp — +// every node type that forwards values needs it, not just pool-scheduled ones. // ── PoolNode ────────────────────────────────────────────────────────────────── // diff --git a/include/kpn/traits.hpp b/include/kpn/traits.hpp index 6381822..e81983e 100644 --- a/include/kpn/traits.hpp +++ b/include/kpn/traits.hpp @@ -97,4 +97,41 @@ struct repeat_tuple> { template using repeat_tuple_t = typename repeat_tuple::type; +// ── 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 message types that 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. +// +// Lives here rather than in pool_node.hpp because every node type that forwards +// values needs it, not just the pool-scheduled ones. FilterNode and RouterNode +// not having it is what let an EOF token be dropped on a full output. + +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; +} + } // namespace kpn diff --git a/tests/test_backpressure_deadlock.cpp b/tests/test_backpressure_deadlock.cpp index c1ee551..f6909d0 100644 --- a/tests/test_backpressure_deadlock.cpp +++ b/tests/test_backpressure_deadlock.cpp @@ -322,3 +322,100 @@ TEST_CASE("a fanout absorbs an unequal pair by slowing, not dropping", // the difference is the loss. CHECK(fast_seen - slow_seen < 200); } + +// Regression: a filter must not drop an EOF sentinel into a full output. +// +// RouterNode and FilterNode were the last nodes on a data path still using the +// throwing push() and swallowing the result: +// +// try { out_ch_->push(val); } catch (const ChannelOverflowError&) {} +// +// 6595e6e made node outputs lossless, 28e0667 stopped them parking a worker, +// a8cfe73 did the same for FanoutNode. These two were in none of them. +// +// For ordinary values that is the familiar silent-loss problem. For a sentinel +// it is a hang. EOF is what tells every downstream node to shut down, and +// nothing comes after it to retry — so a filter that passes EOF by predicate +// but drops it by backpressure produces a pipeline that never terminates. The +// scene-actor-extraction decimator is exactly this shape: `if (f.eof) return +// true;` in the predicate, feeding a chain whose slowest node is an ONNX +// embedder, so the output is reliably full at the moment EOF arrives. +// +// The test forces that state rather than racing for it: the sink is slow enough +// that the filter's output channel is saturated for the whole run, so EOF meets +// a full ring with certainty. +// +// Both assertions are needed. `saw_eof` alone would pass on an implementation +// that dropped every ordinary value and delivered only the sentinel; `count` +// alone would pass on the broken one, which delivers plenty of values and loses +// only the token that matters. +namespace { + +struct EofFrame { + int seq{0}; + bool eof{false}; +}; + +// EOF is emitted exactly once, as a real source does. Everything after it is a +// filler frame the predicate rejects, which keeps the node alive without +// re-offering the sentinel — a source that retried EOF would mask the bug, +// since a later attempt could find the channel drained. +struct EofSource { + static constexpr std::string_view label() { return "eof_source"; } + int n{0}; + int total{0}; + EofFrame operator()() { + if (n > total) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + return {-1, false}; // filler: dropped by the predicate + } + EofFrame f{n, n == total}; + ++n; + return f; + } +}; + +struct EofSink { + static constexpr std::string_view label() { return "eof_sink"; } + std::atomic* count; + std::atomic* saw_eof; + void operator()(EofFrame f) { + std::this_thread::sleep_for(std::chrono::microseconds(200)); + if (f.eof) saw_eof->store(true, std::memory_order_release); + else count->fetch_add(1, std::memory_order_relaxed); + } +}; + +} // namespace + +TEST_CASE("a filter delivers EOF into a saturated output", "[backpressure][filter]") { + std::atomic count{0}; + std::atomic saw_eof{false}; + + EofSource src_fn{0, 40}; + EofSink sink_fn{&count, &saw_eof}; + + // Every real frame passes the predicate, so the only thing between source + // and sink is backpressure. Small channels keep the output saturated. + auto filt = kpn::make_filter( + [](const EofFrame& f) { return f.seq >= 0; }, 4); + + kpn::ObjectNode, kpn::out<"f">, "eof_source", 0> s(src_fn, 4); + kpn::ObjectNode, kpn::out<>, "eof_sink", 0> k(sink_fn, 4); + + auto net = kpn::make_network( + kpn::edge(s.output<"f">(), filt.input<0>()), + kpn::edge(filt.output<0>(), k.input<"f">()) + ); + net.start(); + + // Generous relative to 41 frames at 200 us, and this is a liveness test: + // the broken implementation never sets saw_eof no matter how long it runs. + for (int i = 0; i < 200 && !saw_eof.load(std::memory_order_acquire); ++i) + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + net.stop(); + + INFO("values delivered: " << count.load() << " of 40"); + CHECK(saw_eof.load(std::memory_order_acquire)); + CHECK(count.load(std::memory_order_relaxed) == 40); +}