diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index 571b891..aa87523 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -58,6 +58,18 @@ public: ChannelClosedError() : std::runtime_error("channel closed") {} }; +// Nothing available *right now* on a channel that is still open. Distinct from +// ChannelClosedError, which means upstream is finished and never coming back. +// +// Conflating the two is expensive in one direction only: a consumer that reads +// "empty" as "closed" stops a live node permanently, and because a stopping +// node disables its own inputs and outputs, one benign empty read takes the +// rest of the pipeline with it. The reverse costs nothing. +class ChannelEmptyError : public std::runtime_error { +public: + ChannelEmptyError() : std::runtime_error("channel empty") {} +}; + // ── CPU pause hint ──────────────────────────────────────────────────────────── // Signals the CPU that this is a spin-wait loop, improving HT sibling throughput // and preventing branch-predictor thrash on x86. Falls back to a compiler barrier. diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index d228a84..81c0724 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -473,6 +473,14 @@ private: auto t2 = clock_t::now(); // blocked_time = 0 for pool nodes (we don't block waiting for inputs) stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1); + } catch (const ChannelEmptyError&) { + // Not an error: there was simply nothing to take. Release and wait + // to be woken again. fire_once checks readiness before it gets + // here, and this node is the sole consumer of its inputs, so this + // is unreachable today — it exists so that if the check is ever + // weakened the cost is a wasted firing rather than a dead node. + finish_firing(); + return; } catch (const ChannelClosedError&) { fire_callbacks(closed_callbacks_); self_stop(); @@ -512,8 +520,16 @@ private: std::tuple_element_t pop_one() { auto& ch = *std::get(input_channels_); std::tuple_element_t val; - if (!ch.try_pop_now(val)) + if (!ch.try_pop_now(val)) { + // try_pop_now returns false for "nothing available", which covers + // two very different situations. A closed channel means upstream is + // finished and this node should stop. An open one means only that + // nothing is here at this instant — and treating that as closed + // kills a live node, which then disables its own inputs and outputs + // and takes the rest of the pipeline with it. + if (ch.is_accepting()) throw ChannelEmptyError{}; throw ChannelClosedError{}; + } return val; } @@ -980,6 +996,14 @@ private: auto cpu1 = NodeStats::cpu_now(); auto t2 = clock_t::now(); stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1); + } catch (const ChannelEmptyError&) { + // Not an error: there was simply nothing to take. Release and wait + // to be woken again. fire_once checks readiness before it gets + // here, and this node is the sole consumer of its inputs, so this + // is unreachable today — it exists so that if the check is ever + // weakened the cost is a wasted firing rather than a dead node. + finish_firing(); + return; } catch (const ChannelClosedError&) { fire_callbacks(closed_callbacks_); self_stop(); @@ -1014,7 +1038,16 @@ private: std::tuple_element_t pop_one() { auto& ch = *std::get(input_channels_); std::tuple_element_t val; - if (!ch.try_pop_now(val)) throw ChannelClosedError{}; + if (!ch.try_pop_now(val)) { + // try_pop_now returns false for "nothing available", which covers + // two very different situations. A closed channel means upstream is + // finished and this node should stop. An open one means only that + // nothing is here at this instant — and treating that as closed + // kills a live node, which then disables its own inputs and outputs + // and takes the rest of the pipeline with it. + if (ch.is_accepting()) throw ChannelEmptyError{}; + throw ChannelClosedError{}; + } return val; } diff --git a/tests/test_pool_node.cpp b/tests/test_pool_node.cpp index 8902bae..0926556 100644 --- a/tests/test_pool_node.cpp +++ b/tests/test_pool_node.cpp @@ -602,3 +602,69 @@ TEST_CASE("a twice-parked value keeps its payload", "[pool_node][backpressure]") CHECK(parked.size() == 4); if (parked.size() == 4) CHECK(parked[0] == 2); } + +// Regression: a node woken with nothing to read must not stop itself. +// +// pop_one reported an empty channel the same way it reported a closed one, by +// throwing ChannelClosedError, and fire_once treats that as "upstream is +// finished" and calls self_stop(). self_stop disables the node's own inputs +// *and* outputs, so one benign empty read does not merely skip a frame — it +// kills the node and, through the disabled channels, the rest of the pipeline. +// +// A node genuinely does get woken with empty inputs: a space callback fires +// when its output drains, which has nothing to do with input arrival. fire_once +// guards against it by checking readiness before popping, and that guard is +// what this test pins. pop_one now also distinguishes the two cases, so if the +// guard is ever weakened the cost is a wasted firing rather than a dead node. +// +// The sequence below reaches the guard deliberately. The output is capacity 1 +// so that draining it signals space at all — Channel fires the space callback +// only on the full->not-full edge — and by the final pop the input is long +// since consumed, so the resulting firing has nothing to read. +namespace { + +struct CountingRelay { + static constexpr std::string_view label() { return "counting_relay"; } + std::atomic* calls; + int operator()(int v) { calls->fetch_add(1, std::memory_order_relaxed); return v; } +}; + +} // namespace + +TEST_CASE("a node woken with empty inputs does not stop itself", "[pool_node]") { + std::atomic calls{0}; + std::atomic closed{0}; + + auto pool = std::make_shared(2); + pool->start(); + + CountingRelay fn{&calls}; + auto node = make_pool_node(fn, pool, 8); + Channel out(1); + node.set_output_channel<0>(&out); + node.set_closed_callback([&](auto) { closed.fetch_add(1, std::memory_order_relaxed); }); + + out.push(99); // output full before the node runs + node.start(); + + node.input_channel<0>().push(1); // fires, cannot deliver, parks + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + REQUIRE(out.pop() == 99); // space -> retry delivers the parked value + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + REQUIRE(out.pop() == 1); // space again -> fires with empty inputs + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + // That firing had nothing to read. The node must still be alive. + CHECK(closed.load(std::memory_order_relaxed) == 0); + CHECK(node.running()); + + // And must still do its job when real input arrives. + node.input_channel<0>().push(2); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + CHECK(out.pop() == 2); + CHECK(calls.load(std::memory_order_relaxed) == 2); + + node.stop(); + pool->stop(); +}