diff --git a/include/kpn/diagnostics.hpp b/include/kpn/diagnostics.hpp index f5080b1..2bc1e4c 100644 --- a/include/kpn/diagnostics.hpp +++ b/include/kpn/diagnostics.hpp @@ -51,6 +51,14 @@ struct NodeStats { std::atomic max_exec_us{0}; std::atomic total_blocked_us{0}; + // Cumulative wall time inside fire_once, summed over every invocation. + // The EMA above cannot be turned into a total: it is exponentially + // weighted, so frames * ema_exec_us tracks the tail of the run rather than + // the whole of it, and on a workload whose per-frame cost varies (a face + // detector on a film: crowd scenes then empty landscapes) the two differ by + // a lot. Answering "how much time went into this node" needs a real sum. + std::atomic total_exec_us{0}; + // Thread CPU time — actual CPU consumed by this node's thread, // measured via CLOCK_THREAD_CPUTIME_ID. Excludes time sleeping or // blocked on mutexes/channels. Sampled once per frame. @@ -89,6 +97,7 @@ struct NodeStats { frames_processed.fetch_add(1, std::memory_order_relaxed); int64_t us = static_cast(exec_time.count() * 1000.0); + total_exec_us.fetch_add(us, std::memory_order_relaxed); uint64_t n = frames_processed.load(std::memory_order_relaxed); int64_t prev = ema_exec_us.load(std::memory_order_relaxed); @@ -147,6 +156,24 @@ struct NodeSnapshot { double total_cpu_ms; // cumulative CPU time consumed by this node's thread double cpu_util_pct; // exec_ms / (exec_ms + blocked_ms) * 100 double queue_wait_ms{0}; // PoolNode: cumulative time spent in pool queue + + // Live scheduling state, for observing the AR-004 invariant "a node never + // sleeps with a wake outstanding". The invariant was previously asserted in + // comments but invisible at runtime, so a lost wake could only be found in a + // debugger — and this bug does not reproduce under one (it needs full speed). + // Two atomic loads at snapshot time, nothing on the hot path. + // + // Read them together with the node's channel fill: + // queued=0, wake=1 -> wake recorded and never consumed + // queued=0, wake=0, input full -> wake never generated at all + // queued=1 while nothing running -> submitted but never scheduled + bool queued{false}; + bool wake_pending{false}; + // Cumulative wall time inside fire_once. Unlike ema_exec_ms this is a true + // sum, so it is the field to use for "share of the run spent in this node". + // Note it still includes time parked pushing into a full output channel; + // total_cpu_ms is the part that backpressure cannot inflate. + double total_exec_ms{0}; }; // ── Pool statistics + snapshot ──────────────────────────────────────────────── diff --git a/include/kpn/fanout.hpp b/include/kpn/fanout.hpp index 180b602..1d6bf9e 100644 --- a/include/kpn/fanout.hpp +++ b/include/kpn/fanout.hpp @@ -7,8 +7,10 @@ #include #include +#include #include #include +#include #include #include #include @@ -78,7 +80,9 @@ public: blocked_ms, elapsed_s > 0 ? frames / elapsed_s : 0.0, stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0, - total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0}; + total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0, + 0.0, // queue_wait_ms — fanout is not pool-scheduled + stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0}; } // ── Port access ─────────────────────────────────────────────────────────── @@ -116,6 +120,76 @@ public: } private: + // Deliver `val` to every connected output, losslessly. + // + // Previously a full output cost the value: push() threw and the exception was + // swallowed per output. A dropped item does not degrade a downstream result, + // it silently changes one, and the consumer cannot tell it happened — so the + // fanout waits instead, and the producer upstream runs slower. + // + // Unlike a pool node, a fanout owns a private thread, so waiting here costs + // no scheduler worker and needs no space-callback park; a bounded retry is + // enough. `stop_flag_` is re-checked every pass so teardown cannot hang on a + // full output regardless of the order the network stops its nodes in. + // + // Outputs are retried independently, so a full output never delays delivery + // to one with room. Note what that does *not* buy: the next input is not + // popped until every output has accepted the current item, so one branch can + // never run ahead of another by more than the slower branch's buffering. + // + // **That bound is a precondition on any topology where the branches rejoin.** + // If a consumer on branch B blocks waiting for something branch A computes, + // B's buffering must exceed the lead A needs, or the two wedge — B waiting on + // A, A starved because the fanout is holding an item B will not take. Making + // the fanout lossless is what puts that precondition on the topology; while + // it dropped, the question could not arise. + // + // `parked` receives the time spent waiting on a full output, which the caller + // charges to blocked rather than exec. + // + // Returns false if stopped with the value undelivered. + bool deliver(const T& val, duration_t& parked) { + std::array, N> pending; + std::size_t outstanding = 0; + for (std::size_t i = 0; i < N; ++i) + if (out_channels_[i]) { pending[i].emplace(val); ++outstanding; } + + bool first_pass = true; + auto park_from = clock_t::now(); + + for (;;) { + for (std::size_t i = 0; i < N; ++i) { + if (!pending[i]) continue; + if (out_channels_[i]->try_push(*pending[i])) { + pending[i].reset(); + --outstanding; + } + } + if (first_pass) { park_from = clock_t::now(); first_pass = false; } + + if (outstanding == 0) { + 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 per + // outstanding output, purely so the channel's own stats record + // the loss (drop if it is disabled, overflow if it is merely + // full). The whole 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. + for (std::size_t i = 0; i < N; ++i) { + if (!pending[i]) continue; + try { out_channels_[i]->push(std::move(*pending[i])); } + catch (const ChannelOverflowError&) {} + } + parked = duration_t(clock_t::now() - park_from); + return false; + } + std::this_thread::sleep_for(std::chrono::microseconds(50)); + } + } + void run_loop() { while (!stop_flag_.load(std::memory_order_relaxed)) { try { @@ -124,16 +198,17 @@ private: auto t1 = clock_t::now(); auto cpu0 = NodeStats::cpu_now(); - for (std::size_t i = 0; i < N; ++i) { - if (out_channels_[i]) { - try { out_channels_[i]->push(val); } - catch (const ChannelOverflowError&) {} // drop for this output independently - } - } + duration_t parked{0}; + const bool delivered = deliver(val, parked); auto cpu1 = NodeStats::cpu_now(); auto t2 = clock_t::now(); - stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); + // Time spent waiting on a full output is *blocked*, not exec: a + // parked fanout is idle, and charging it to exec would report the + // node as busy exactly when it is the one being held up. + 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/interrupt_node.hpp b/include/kpn/interrupt_node.hpp index 6e58c24..0b8c830 100644 --- a/include/kpn/interrupt_node.hpp +++ b/include/kpn/interrupt_node.hpp @@ -104,6 +104,7 @@ public: stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0, total_ms > 0 ? 100.0 : 0.0, qwait_ms, + stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0, }; } diff --git a/include/kpn/main_thread_node.hpp b/include/kpn/main_thread_node.hpp index 47ecb18..17ff222 100644 --- a/include/kpn/main_thread_node.hpp +++ b/include/kpn/main_thread_node.hpp @@ -90,6 +90,8 @@ public: elapsed_s > 0 ? frames / elapsed_s : 0.0, stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0, total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0, + 0.0, // queue_wait_ms — main-thread node is not pool-scheduled + stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0, }; } diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 2d12cbe..e9d8d1a 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -116,6 +116,23 @@ public: register_callbacks(std::make_index_sequence{}); if constexpr (input_count == 0) try_submit(0.5f); + else + // Never start with a wake already outstanding — the startup case of + // the invariant 9c5ce5f established for the running pipeline. + // + // enable_inputs() opens the channel several statements before + // register_callbacks() installs the push callback, and the network + // starts nodes sources-first, so an upstream node is already firing + // into this one during that gap. A push landing there is accepted by + // the ring but wakes nobody: Channel::push only invokes the callback + // on the empty→non-empty transition, and at that instant the + // callback is still null. Every later push sees a non-empty ring and + // stays silent, so the node is never submitted — the pipeline reads + // as wedged from the first frame, with no item ever delivered. + // + // on_input_ready() is the level-triggered form of the same question, + // so asking it once here converts the missed edge into a state check. + on_input_ready(); } void stop() override { @@ -156,6 +173,9 @@ public: stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0, total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0, qwait_ms, + stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0, + queued_.load(std::memory_order_relaxed), + wake_pending_.load(std::memory_order_relaxed), }; } @@ -686,6 +706,9 @@ public: register_callbacks(std::make_index_sequence{}); if constexpr (input_count == 0) try_submit(0.5f); + else + // Never start with a wake already outstanding — see PoolNode::start(). + on_input_ready(); } void stop() override { @@ -720,6 +743,9 @@ public: stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0, total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0, qwait_ms, + stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0, + queued_.load(std::memory_order_relaxed), + wake_pending_.load(std::memory_order_relaxed), }; } diff --git a/include/kpn/web_debug.hpp b/include/kpn/web_debug.hpp index cd325af..27748be 100644 --- a/include/kpn/web_debug.hpp +++ b/include/kpn/web_debug.hpp @@ -65,6 +65,11 @@ static std::string to_json(const std::vector& nodes, << ",\"fps\":" << n.throughput_fps << ",\"total_cpu_ms\":" << n.total_cpu_ms << ",\"cpu_util_pct\":" << n.cpu_util_pct + // Scheduling state — lets a WEDGED pipeline be interrogated over HTTP + // without a debugger, which matters because the lost-wake bug does not + // reproduce under one. See NodeSnapshot for how to read the pair. + << ",\"queued\":" << (n.queued ? "true" : "false") + << ",\"wake_pending\":" << (n.wake_pending ? "true" : "false") << "}"; } o << "],\"edges\":["; diff --git a/tests/test_backpressure_deadlock.cpp b/tests/test_backpressure_deadlock.cpp index 17835ec..c1ee551 100644 --- a/tests/test_backpressure_deadlock.cpp +++ b/tests/test_backpressure_deadlock.cpp @@ -169,3 +169,156 @@ TEST_CASE("a saturated chain never stalls", "[backpressure][deadlock]") { // Guard against the test passing because nothing ever ran. CHECK(last > 1000); } + +// Regression: a node must not start with a wake already outstanding. +// +// 9c5ce5f established the invariant for the running pipeline — a node never +// sleeps with a wake it dropped. start() broke the same invariant before the +// pipeline was even running: +// +// enable_inputs(...); // channel goes live here +// stop_flag_.store(false); +// queued_.store(false); +// register_callbacks(...); // push callback installed here +// +// StaticNetwork starts nodes sources-first, so an upstream node is already +// firing into this one during that gap. A push landing there is accepted by the +// ring but wakes nobody: Channel::push invokes push_callback_ only on the +// empty→non-empty transition, and at that instant the callback is null. Every +// later push sees a non-empty ring and stays silent. The node is never +// submitted, and since a sink has no outputs there is no space callback to +// rescue it either. +// +// The signature is distinctive: **zero** items delivered, not a stall partway. +// The chain reads as wedged from the first frame. Under `ctest -j4` on a loaded +// machine it reproduced 7 times in 24, and never once in 10 unloaded runs — +// contention widens the window between those two statements. That is almost +// certainly the "rare hang, ~1 run in 20 at a 300 s timeout" 28e0667 recorded as +// known-incomplete. +// +// This test needs no contention: it constructs the state the race leaves behind +// directly, by enabling the input and pushing before start() is ever called. +namespace { + +int passthrough(int x) { return x; } + +} // namespace + +TEST_CASE("a node started with data already queued still fires", + "[backpressure][startup]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = kpn::make_pool_node(pool, 8); + kpn::Channel out_ch(8); + node.set_output_channel<0>(&out_ch); + + // The missed edge: the channel is live and already holds a value, but no + // callback was installed when it arrived, so the wake has been and gone. + node.input_channel<0>().enable(); + node.input_channel<0>().push(21); + + node.start(); + + // Bounded wait — a plain pop() would hang rather than fail on a regression. + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (out_ch.size() == 0 && std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + + const bool delivered = out_ch.size() > 0; + const int got = delivered ? out_ch.pop() : -1; + + node.stop(); + pool->stop(); + + INFO("value delivered: " << got); + REQUIRE(delivered); + CHECK(got == 21); +} + +// Regression: a fanout absorbs an unequal pair of consumers by slowing, not by +// dropping. +// +// 6595e6e made node outputs lossless and 28e0667 stopped them parking a worker, +// but FanoutNode was in neither: it kept `catch (ChannelOverflowError&) {}` per +// output, so whichever branch fell behind lost items — silently, and by an +// amount that depended on timing. Two runs of the same input could therefore +// disagree, which is fatal for a fixture the rest of the suite is scored +// against. +// +// The two assertions are the two halves of the requirement: +// - no gaps: the slow branch receives *every* item, not most of them; +// - bounded lead: the fast branch is throttled to the slow one rather than +// racing ahead over a drain that is quietly discarding the difference. +// +// Either alone would pass on a broken implementation. A fanout that pushed only +// to the slow branch has no gaps; one that dropped everything for the slow +// branch keeps a bounded lead by never letting it fall behind. +namespace { + +// Records the sequence it sees, so a dropped item shows up as a gap rather than +// merely as a smaller total. +struct SeqCheck { + std::atomic* next_expected; + std::atomic* saw_gap; + int delay_us{0}; + + void record(int v) const { + if (delay_us) + std::this_thread::sleep_for(std::chrono::microseconds(delay_us)); + const int want = next_expected->load(std::memory_order_relaxed); + if (v != want) saw_gap->store(true, std::memory_order_relaxed); + else next_expected->store(want + 1, std::memory_order_relaxed); + } +}; + +struct FastBranch : SeqCheck { + static constexpr std::string_view label() { return "fast_branch"; } + void operator()(int v) { record(v); } +}; + +struct SlowBranch : SeqCheck { + static constexpr std::string_view label() { return "slow_branch"; } + void operator()(int v) { record(v); } +}; + +} // namespace + +TEST_CASE("a fanout absorbs an unequal pair by slowing, not dropping", + "[backpressure][fanout]") { + std::atomic fast_next{0}, slow_next{0}; + std::atomic fast_gap{false}, slow_gap{false}; + + FreeRun p_fn; + FastBranch fast_fn{{&fast_next, &fast_gap, 0}}; + SlowBranch slow_fn{{&slow_next, &slow_gap, 500}}; // 0.5 ms/item + + // Small channels so the slow branch saturates in the first few milliseconds + // and stays saturated for the whole run. + kpn::ObjectNode, kpn::out<"v">, "free_run", 0> p (p_fn, 8); + kpn::ObjectNode, kpn::out<>, "fast", 0> fa(fast_fn, 8); + kpn::ObjectNode, kpn::out<>, "slow", 0> sl(slow_fn, 8); + + // Two edges from one output port: make_network auto-inserts FanoutNode. + auto net = kpn::make_network( + kpn::edge(p.output<"v">(), fa.input<"fast">()), + kpn::edge(p.output<"v">(), sl.input<"slow">()) + ); + net.start(); + std::this_thread::sleep_for(std::chrono::seconds(1)); + net.stop(); + + const int fast_seen = fast_next.load(std::memory_order_relaxed); + const int slow_seen = slow_next.load(std::memory_order_relaxed); + + INFO("fast branch " << fast_seen << " items, slow branch " << slow_seen); + CHECK_FALSE(fast_gap.load(std::memory_order_relaxed)); + CHECK_FALSE(slow_gap.load(std::memory_order_relaxed)); + // Guard against passing because nothing ran: 1 s at 0.5 ms/item is ~2000. + CHECK(slow_seen > 200); + // The lead is bounded by the buffering between the two — the fanout's own + // input, the two output channels, and one item in each node's hand. A + // dropping fanout has no such bound: the fast branch runs at full speed and + // the difference is the loss. + CHECK(fast_seen - slow_seen < 200); +}