diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index f8842d4..571b891 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -136,6 +136,42 @@ public: push_callback_(); } + /// Called when a pop frees a slot in a previously-full ring. + /// + /// The mirror of `set_push_callback`, and it exists for the same reason: + /// a producer must be able to *park* rather than spin. Without it the only + /// lossless option is `push_blocking`, which sleeps inside the caller's + /// thread — and when that thread is a scheduler worker, parking it starves + /// every node pinned to it (see the hold-and-wait note on push_sentinel). + void set_space_callback(std::function cb) { space_callback_ = std::move(cb); } + + /// True when a push would currently succeed. Used to close the lost-wakeup + /// race: a producer that parks must re-check after clearing its queued flag, + /// because a space_callback fired in between would otherwise be swallowed. + bool has_space() const { + return tail_.load(std::memory_order_relaxed) - + head_.load(std::memory_order_acquire) < capacity_; + } + + /// Non-blocking, lossless push. Returns false when the ring is full, having + /// changed nothing — the caller keeps the value and retries when woken. + bool try_push(T& value) { + if (!accepting_.load(std::memory_order_acquire)) { stats_.record_drop(); return true; } + const std::size_t t = tail_.load(std::memory_order_relaxed); + const std::size_t h = head_.load(std::memory_order_acquire); + if (t - h >= capacity_) return false; + + const std::size_t data_bytes = ChannelDataSize::bytes(value); + const bool was_empty = (t == h); + buf_[t & ring_mask_] = make_storage(std::move(value)); + tail_.store(t + 1, std::memory_order_release); + stats_.record_push(t - h + 1, data_bytes); + wake_.fetch_add(1, std::memory_order_release); + wake_.notify_one(); + if (was_empty && push_callback_) push_callback_(); + return true; + } + // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to // drain instead of dropping (the throwing push()) — the producer just runs slower. // Use when every value must be delivered (e.g. replaying a dump for scoring, where @@ -248,6 +284,8 @@ public: throw ChannelClosedError{}; T value = extract(std::move(buf_[h & ring_mask_])); head_.store(h + 1, std::memory_order_release); + // A slot just freed: wake any producer parked on this channel. + if (t - h >= capacity_ && space_callback_) space_callback_(); stats_.record_pop(); return value; } @@ -269,11 +307,15 @@ public: // 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)) + const std::size_t t = tail_.load(std::memory_order_acquire); + if (h == t) return take_sentinel(out); out = extract(std::move(buf_[h & ring_mask_])); head_.store(h + 1, std::memory_order_release); stats_.record_pop(); + // Pool nodes pop only through here, so this is where a parked producer + // gets woken: the ring was full, and it no longer is. + if (t - h >= capacity_ && space_callback_) space_callback_(); return true; } @@ -363,6 +405,7 @@ private: std::size_t ring_mask_; std::unique_ptr buf_; std::function push_callback_; + std::function space_callback_; ChannelStats stats_; // Out-of-band sentinel (EOF): stored outside the ring so its delivery never diff --git a/include/kpn/inode.hpp b/include/kpn/inode.hpp index 56f4b89..a7454e2 100644 --- a/include/kpn/inode.hpp +++ b/include/kpn/inode.hpp @@ -34,6 +34,14 @@ struct INode { virtual void set_network_overflow_callback(NodeEventCallback) {} virtual void set_network_closed_callback(NodeEventCallback) {} + // Network-level error listener. Consulted when a node's function throws + // and no per-node handler resolved it. Without this the exception is + // discarded and the failure is only visible as a Closed event, which says + // a node stopped but not why — the difference between a diagnosis and a + // guess. Same contract as NodeErrorHandler: true to continue, false to + // stop the node. + virtual void set_network_error_callback(NodeErrorHandler) {} + // halt(): alias for stop() — immediate, discards in-flight work. virtual void halt() { stop(); } diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 1938302..85532da 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -131,6 +132,7 @@ public: void set_name(std::string name) override { name_ = std::move(name); } void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } + void set_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); } void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); } @@ -234,6 +236,8 @@ private: template void register_callbacks(std::index_sequence) { + // A parked producer is re-submitted when its output drains. + register_space_callbacks(std::make_index_sequence{}); (std::get(input_channels_)->set_push_callback( [this] { on_input_ready(); }), ...); } @@ -243,6 +247,20 @@ private: for (auto& cb : cbs) if (cb) cb(ts); } + template + bool outputs_have_space(std::index_sequence) const { + return (... && (!std::get(output_channels_) || + std::get(output_channels_)->has_space())); + } + + template + void register_space_callbacks(std::index_sequence) { + ((std::get(output_channels_) + ? (void)std::get(output_channels_)->set_space_callback( + [this] { try_submit(0.5f); }) + : (void)0), ...); + } + void self_stop() { disable_inputs(std::make_index_sequence{}); disable_outputs(std::make_index_sequence{}); @@ -280,11 +298,51 @@ private: return ((std::get(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...); } + /// Priority in [0,1], higher runs sooner. + /// + /// Input fill alone answers "how much work is waiting for me". That is + /// only half the question: a node whose *outputs* are already full cannot + /// deliver anything: running it produces a value with nowhere to go, so it + /// immediately parks and the slot is wasted. Meanwhile the node that would + /// have drained that full channel waits behind it. + /// + /// So occupancy of the outputs is deducted from occupancy of the inputs. + /// The scheduler then naturally favours whoever is furthest downstream of + /// a bottleneck — the node whose inputs are backed up but whose outputs + /// have room is exactly the one whose execution frees the most capacity — + /// and defers producers that would only deepen a queue that is already + /// full. + /// + /// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero: + /// both terms are mean fills in [0,1], so the difference is in [-1,1], and + /// the affine map keeps the whole range distinguishable instead of + /// collapsing every output-saturated node onto the same value. 0.5 remains + /// the neutral point, matching the default used for source nodes. float compute_priority() { if constexpr (input_count == 0) return 0.5f; - float sum = 0.0f; - sum_fill(sum, std::make_index_sequence{}); - return sum / static_cast(input_count); + float in = 0.0f; + sum_fill(in, std::make_index_sequence{}); + in /= static_cast(input_count); + + if constexpr (output_count == 0) return in; + + float out = 0.0f; + sum_output_fill(out, std::make_index_sequence{}); + out /= static_cast(output_count); + + const float p = 0.5f * (1.0f + in - out); + return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p); + } + + /// Mean fill of the output channels, same normalisation as sum_fill. + /// An unconnected output holds nothing back, so it contributes 0. + template + void sum_output_fill(float& sum, std::index_sequence) { + ((sum += (std::get(output_channels_) && + std::get(output_channels_)->capacity() > 0) + ? float(std::get(output_channels_)->approx_size()) + / float(std::get(output_channels_)->capacity()) + : 0.0f), ...); } template @@ -315,6 +373,77 @@ private: t0.time_since_epoch()).count(); stats_.exec_start_us.store(now_us, std::memory_order_relaxed); + // Parked from a previous firing: retry that value before touching the + // inputs. Returning here releases the worker — the channel's space + // callback re-submits this node when the consumer drains a slot. + if constexpr (!std::is_void_v) { + if (pending_) { + push_outputs(std::move(*pending_), std::make_index_sequence{}); + queued_.store(false, std::memory_order_release); + if (pending_) { + // Close the lost-wakeup race: a space_callback that fired + // between the failed push and clearing queued_ was + // swallowed, and nothing else will wake this node. Re-check + // now that the flag is down. + if (outputs_have_space(std::make_index_sequence{})) + try_submit(0.5f); + return; // parked + } + // Drained: resume normal firing, resubmitting exactly the way + // the normal tail below does. An unconditional try_submit here + // would fire a node whose inputs are empty, and pop_inputs + // reports an empty channel as ChannelClosedError — which this + // node treats as "upstream finished" and self-stops on. That + // is a live node killing itself purely because it was woken by + // *output* space rather than by input arrival. + if constexpr (input_count == 0) try_submit(0.5f); + else on_input_ready(); + return; + } + } + + // Parked from a previous firing: retry that value before touching the + // inputs. Returning here releases the worker — the channel's space + // callback re-submits this node once the consumer drains a slot. + if constexpr (!std::is_void_v) { + if (pending_) { + push_outputs(std::move(*pending_), std::make_index_sequence{}); + queued_.store(false, std::memory_order_release); + if (pending_) { + // Close the lost-wakeup race: a space_callback that fired + // between the failed push and clearing queued_ was + // swallowed, and nothing else will wake this node. Re-check + // now that the flag is down. + if (outputs_have_space(std::make_index_sequence{})) + try_submit(0.5f); + return; // parked + } + // Drained: resume normal firing, resubmitting exactly the way + // the normal tail below does. An unconditional try_submit here + // would fire a node whose inputs are empty, and pop_inputs + // reports an empty channel as ChannelClosedError — which this + // node treats as "upstream finished" and self-stops on. That + // is a live node killing itself purely because it was woken by + // *output* space rather than by input arrival. + if constexpr (input_count == 0) try_submit(0.5f); + else on_input_ready(); + return; + } + } + + // Woken by output space rather than by input arrival, with nothing + // parked left to flush: there is no work to do. Falling through would + // read an empty channel, and pop_one reports empty as + // ChannelClosedError — self-stopping a live node. Release the worker; + // on_input_ready() resubmits when data actually lands. + if constexpr (input_count > 0) { + if (count_ready(std::make_index_sequence{}) != input_count) { + queued_.store(false, std::memory_order_release); + on_input_ready(); // data may have arrived while we checked + return; + } + } + try { auto args = pop_inputs(std::make_index_sequence{}); auto t1 = clock_t::now(); @@ -340,7 +469,11 @@ private: } catch (const ChannelOverflowError&) { fire_callbacks(event_callbacks_); } catch (...) { - if (error_handler_ && error_handler_(name_, std::current_exception())) { + auto eptr = std::current_exception(); + const bool handled = + (error_handler_ && error_handler_(name_, eptr)) || + (net_error_handler_ && net_error_handler_(name_, eptr)); + if (handled) { // continue — fall through to resubmit check } else { fire_callbacks(closed_callbacks_); @@ -385,30 +518,41 @@ private: } template + /// Pushes what it can and parks the rest. `done_` marks the elements that + /// were taken, so a retry never re-pushes one — a duplicate would be as + /// wrong as a drop, just harder to notice. void push_outputs(return_tuple&& result, std::index_sequence) { - (push_one_out(std::get(std::move(result))), ...); + bool all = true; + ((pending_done_[Is] = pending_done_[Is] || + push_one_out(std::get(std::move(result))), + all = all && pending_done_[Is]), ...); + if (all) { pending_.reset(); pending_done_.fill(false); } + else pending_ = std::move(result); } + /// Returns false when the ring was full and the value was NOT taken; the + /// caller must keep it and retry after the channel signals space. template - void push_one_out(std::tuple_element_t&& val) { + bool push_one_out(std::tuple_element_t&& val) { auto* ch = std::get(output_channels_); - if (!ch) return; + if (!ch) return true; // 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; + return true; } - // Backpressure, not loss. A full downstream channel means the consumer - // is behind, and the correct response is for this producer to run - // slower — not to discard a value. A dropped frame does not degrade a - // result, it silently changes one, and the caller has no way to tell. + // Backpressure without parking the worker. A full channel means the + // consumer is behind; the value is kept and this node stops running + // until the channel signals space (set_space_callback re-submits it). // - // Safe here because sentinels are handled above, out-of-band: this - // blocks only on data, so the EOF token that unwinds the network can - // always overtake a stalled data path. - ch->push_blocking(std::move(val)); + // Blocking here instead would sleep inside a scheduler worker, and + // nodes are pinned to workers — park enough of them and nothing is left + // to run the consumer that would drain the channel. That is the + // hold-and-wait deadlock channel.hpp warns about for sentinels; it + // applies to data pushes just as much. + return ch->try_push(val); } template @@ -430,8 +574,18 @@ private: output_channels_t output_channels_{}; std::atomic stop_flag_{true}; std::atomic queued_{false}; + + /// The hidden one-slot output buffer (see push_outputs). Holding the value + /// here is what lets a node stop running without dropping it or occupying a + /// scheduler worker. One slot suffices because at most one fire_once() runs + /// per node at a time — with concurrent firing it would have to hold a whole + /// FIFO's worth. + std::conditional_t, std::monostate, + std::optional> pending_{}; + std::array pending_done_{}; NodeStats stats_; NodeErrorHandler error_handler_; + NodeErrorHandler net_error_handler_; std::chrono::milliseconds max_exec_time_{0}; std::array event_callbacks_{}; // [0]=user [1]=network std::array closed_callbacks_{}; @@ -499,6 +653,7 @@ public: bool running() const override { return !stop_flag_.load(std::memory_order_relaxed); } void set_name(std::string name) override { name_ = std::move(name); } void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } + void set_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); } void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); } @@ -570,6 +725,8 @@ private: } template void register_callbacks(std::index_sequence) { + // A parked producer is re-submitted when its output drains. + register_space_callbacks(std::make_index_sequence{}); (std::get(input_channels_)->set_push_callback([this] { on_input_ready(); }), ...); } @@ -578,6 +735,20 @@ private: for (auto& cb : cbs) if (cb) cb(ts); } + template + bool outputs_have_space(std::index_sequence) const { + return (... && (!std::get(output_channels_) || + std::get(output_channels_)->has_space())); + } + + template + void register_space_callbacks(std::index_sequence) { + ((std::get(output_channels_) + ? (void)std::get(output_channels_)->set_space_callback( + [this] { try_submit(0.5f); }) + : (void)0), ...); + } + void self_stop() { disable_inputs(std::make_index_sequence{}); disable_outputs(std::make_index_sequence{}); @@ -609,11 +780,51 @@ private: return ((std::get(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...); } + /// Priority in [0,1], higher runs sooner. + /// + /// Input fill alone answers "how much work is waiting for me". That is + /// only half the question: a node whose *outputs* are already full cannot + /// deliver anything: running it produces a value with nowhere to go, so it + /// immediately parks and the slot is wasted. Meanwhile the node that would + /// have drained that full channel waits behind it. + /// + /// So occupancy of the outputs is deducted from occupancy of the inputs. + /// The scheduler then naturally favours whoever is furthest downstream of + /// a bottleneck — the node whose inputs are backed up but whose outputs + /// have room is exactly the one whose execution frees the most capacity — + /// and defers producers that would only deepen a queue that is already + /// full. + /// + /// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero: + /// both terms are mean fills in [0,1], so the difference is in [-1,1], and + /// the affine map keeps the whole range distinguishable instead of + /// collapsing every output-saturated node onto the same value. 0.5 remains + /// the neutral point, matching the default used for source nodes. float compute_priority() { if constexpr (input_count == 0) return 0.5f; - float sum = 0.0f; - sum_fill(sum, std::make_index_sequence{}); - return sum / static_cast(input_count); + float in = 0.0f; + sum_fill(in, std::make_index_sequence{}); + in /= static_cast(input_count); + + if constexpr (output_count == 0) return in; + + float out = 0.0f; + sum_output_fill(out, std::make_index_sequence{}); + out /= static_cast(output_count); + + const float p = 0.5f * (1.0f + in - out); + return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p); + } + + /// Mean fill of the output channels, same normalisation as sum_fill. + /// An unconnected output holds nothing back, so it contributes 0. + template + void sum_output_fill(float& sum, std::index_sequence) { + ((sum += (std::get(output_channels_) && + std::get(output_channels_)->capacity() > 0) + ? float(std::get(output_channels_)->approx_size()) + / float(std::get(output_channels_)->capacity()) + : 0.0f), ...); } template void sum_fill(float& sum, std::index_sequence) { @@ -639,6 +850,46 @@ private: t0.time_since_epoch()).count(); stats_.exec_start_us.store(now_us, std::memory_order_relaxed); + // Parked from a previous firing: retry that value before touching the + // inputs. Returning here releases the worker — the channel's space + // callback re-submits this node once the consumer drains a slot. + if constexpr (!std::is_void_v) { + if (pending_) { + push_outputs(std::move(*pending_), std::make_index_sequence{}); + queued_.store(false, std::memory_order_release); + if (pending_) { + // Close the lost-wakeup race: a space_callback that fired + // between the failed push and clearing queued_ was + // swallowed, and nothing else will wake this node. Re-check + // now that the flag is down. + if (outputs_have_space(std::make_index_sequence{})) + try_submit(0.5f); + return; // parked + } + // Drained: resume normal firing, resubmitting exactly the way + // the normal tail below does. An unconditional try_submit here + // would fire a node whose inputs are empty, and pop_inputs + // reports an empty channel as ChannelClosedError — which this + // node treats as "upstream finished" and self-stops on. That + // is a live node killing itself purely because it was woken by + // *output* space rather than by input arrival. + if constexpr (input_count == 0) try_submit(0.5f); + else on_input_ready(); + return; + } + } + + // See the equivalent guard in PoolNode::fire_once: a space-callback + // wake with nothing parked must release the worker, not fall through + // into pop_inputs on an empty channel. + if constexpr (input_count > 0) { + if (count_ready(std::make_index_sequence{}) != input_count) { + queued_.store(false, std::memory_order_release); + on_input_ready(); + return; + } + } + try { auto args = pop_inputs(std::make_index_sequence{}); auto t1 = clock_t::now(); @@ -662,7 +913,11 @@ private: } catch (const ChannelOverflowError&) { fire_callbacks(event_callbacks_); } catch (...) { - if (error_handler_ && error_handler_(name_, std::current_exception())) { + auto eptr = std::current_exception(); + const bool handled = + (error_handler_ && error_handler_(name_, eptr)) || + (net_error_handler_ && net_error_handler_(name_, eptr)); + if (handled) { } else { fire_callbacks(closed_callbacks_); self_stop(); @@ -695,22 +950,32 @@ private: } template + /// Pushes what it can and parks the rest. `pending_done_` marks the elements + /// already taken, so a retry never re-pushes one — a duplicate is as wrong as + /// a drop and harder to notice. void push_outputs(return_tuple&& result, std::index_sequence) { - (push_one_out(std::get(std::move(result))), ...); + bool all = true; + ((pending_done_[Is] = pending_done_[Is] || + push_one_out(std::get(std::move(result))), + all = all && pending_done_[Is]), ...); + if (all) { pending_.reset(); pending_done_.fill(false); } + else pending_ = std::move(result); } + /// Returns false when the ring was full and the value was NOT taken; the + /// caller must keep it and retry after the channel signals space. template - void push_one_out(std::tuple_element_t&& val) { + bool push_one_out(std::tuple_element_t&& val) { auto* ch = std::get(output_channels_); - if (!ch) return; + if (!ch) return true; // 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; + return true; } - // See the note on the typed overload above: block rather than drop. - ch->push_blocking(std::move(val)); + // See the note on the typed overload above: park rather than block. + return ch->try_push(val); } Obj& obj_; @@ -721,8 +986,18 @@ private: output_channels_t output_channels_{}; std::atomic stop_flag_{true}; std::atomic queued_{false}; + + /// The hidden one-slot output buffer (see push_outputs). Holding the value + /// here is what lets a node stop running without dropping it or occupying a + /// scheduler worker. One slot suffices because at most one fire_once() runs + /// per node at a time — with concurrent firing it would have to hold a whole + /// FIFO's worth. + std::conditional_t, std::monostate, + std::optional> pending_{}; + std::array pending_done_{}; NodeStats stats_; NodeErrorHandler error_handler_; + NodeErrorHandler net_error_handler_; std::chrono::milliseconds max_exec_time_{0}; std::array event_callbacks_{}; // [0]=user [1]=network std::array closed_callbacks_{}; diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index eeb5304..bb34a2c 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -122,6 +122,10 @@ public: [this, n](auto ts) { event_handler_(n, NodeEvent::Closed, ts); }); } } + if (error_handler_) { + for (auto* node : user_nodes_topo_) + node->set_network_error_callback(error_handler_); + } for (auto* n : user_nodes_topo_) n->start(); for (auto* n : fanout_nodes_ptr_) n->start(); #ifdef KPN_WEB_DEBUG @@ -181,6 +185,14 @@ public: void set_event_handler(EventHandler h) { event_handler_ = std::move(h); } + /// Application-level error listener. Receives the exception any node's + /// function throws, after that node's own handler (if any) declined it. + /// Return true to skip the failed invocation and keep the node running, + /// false to let it stop. Without a listener the exception is discarded + /// and only a Closed event survives, which reports that a node stopped + /// but not why. + void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } + #ifdef KPN_WEB_DEBUG void set_web_debug_port(uint16_t port) { web_debug_port_ = port; } // Called by DebugHub::register_network() so the hub owns the debug server. @@ -276,6 +288,7 @@ private: std::vector> resource_probes_; std::vector> pool_probes_; EventHandler event_handler_; + NodeErrorHandler error_handler_; clock_t::time_point start_time_; #ifdef KPN_WEB_DEBUG uint16_t web_debug_port_{9090}; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index fda6711..9e43788 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -34,6 +34,7 @@ add_executable(kpn_tests test_static_network.cpp test_shared_resource.cpp test_pool_node.cpp + test_backpressure_deadlock.cpp test_scheduler.cpp ) diff --git a/tests/test_backpressure_deadlock.cpp b/tests/test_backpressure_deadlock.cpp new file mode 100644 index 0000000..4679439 --- /dev/null +++ b/tests/test_backpressure_deadlock.cpp @@ -0,0 +1,85 @@ +// Regression: a blocking push must not park a pool worker. +// +// Node outputs use push_blocking so a full channel costs time rather than data +// (a dropped frame does not degrade a downstream result, it silently changes +// one). But push_blocking sleeps *inside* fire_once, which runs on a pool +// worker — and nodes are pinned to workers by index. Park enough workers in +// that retry loop and there is nobody left to run the consumer that would drain +// the channel, so the whole chain wedges. +// +// This is the failure channel.hpp:174 already warns about for sentinels +// ("a blocking push would park that thread and stop it draining its own input, +// cascading into a hold-and-wait deadlock under backpressure"). The warning +// applies to data pushes too. +// +// Observed in the field as an intermittent hang: frame_source, camera_pos, +// face_detector and face_aligner all asleep in push_blocking at once. +#include +#include + +#include +#include +#include + +namespace { + +struct Produce { + static constexpr std::string_view label() { return "produce"; } + int n{0}; + int operator()() { return n++; } +}; + +struct Relay { + static constexpr std::string_view label() { return "relay"; } + int operator()(int v) { return v; } +}; + +// Deliberately slower than the producer, so the channels between them fill. +struct SlowSink { + static constexpr std::string_view label() { return "slow_sink"; } + std::atomic* seen; + void operator()(int) { + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + seen->fetch_add(1, std::memory_order_relaxed); + } +}; + +} // namespace + +TEST_CASE("a saturated chain keeps draining", "[backpressure][deadlock]") { + std::atomic seen{0}; + + Produce p_fn; + Relay r1_fn, r2_fn, r3_fn; + SlowSink s_fn{&seen}; + + // Small channels so they saturate immediately, and a chain longer than a + // modest pool — the shape that starves workers. + kpn::ObjectNode, kpn::out<"a">, "produce", 0> p (p_fn, 2); + kpn::ObjectNode, kpn::out<"b">, "relay1", 0> r1(r1_fn, 2); + kpn::ObjectNode, kpn::out<"c">, "relay2", 0> r2(r2_fn, 2); + kpn::ObjectNode, kpn::out<"d">, "relay3", 0> r3(r3_fn, 2); + kpn::ObjectNode, kpn::out<>, "slow_sink", 0> s (s_fn, 2); + + auto net = kpn::make_network( + kpn::edge(p.output<"a">(), r1.input<"a">()), + kpn::edge(r1.output<"b">(), r2.input<"b">()), + kpn::edge(r2.output<"c">(), r3.input<"c">()), + kpn::edge(r3.output<"d">(), s.input<"d">()) + ); + net.start(); + + // The sink is the slowest stage at 2 ms/item, so 40 items is ~80 ms of real + // work. Anything approaching the timeout means the chain stopped draining + // rather than merely running slowly. + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(20); + while (seen.load(std::memory_order_relaxed) < 40 && + std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + + const int got = seen.load(std::memory_order_relaxed); + net.stop(); + + INFO("items drained: " << got << " of 40"); + CHECK(got >= 40); +} diff --git a/tests/test_pool_node.cpp b/tests/test_pool_node.cpp index 72d60ba..779572b 100644 --- a/tests/test_pool_node.cpp +++ b/tests/test_pool_node.cpp @@ -243,7 +243,13 @@ TEST_CASE("interrupt node: trigger after stop is ignored", "[interrupt_node]") { // ── Overflow callback ───────────────────────────────────────────────────────── -TEST_CASE("pool node overflow callback fires on full output channel", "[pool_node][overflow]") { +// NOTE: output overflow is no longer reachable on the data path. A node whose +// output channel is full now PARKS — it keeps the value in a hidden one-slot +// buffer, releases its scheduler worker, and is re-submitted when the consumer +// frees a slot. The overflow callback survives for other producers (a direct +// Channel::push by non-node code still throws), but a pool node cannot trigger +// it, so these cases assert the stronger property instead: nothing is dropped. +TEST_CASE("pool node parks instead of overflowing a full output", "[pool_node][overflow]") { auto pool = std::make_shared(2); pool->start(); @@ -264,10 +270,13 @@ TEST_CASE("pool node overflow callback fires on full output channel", "[pool_nod node.stop(); pool->stop(); - REQUIRE(overflow_count.load() > 0); + // Parked, not overflowed: the value is still owned by the node. + REQUIRE(overflow_count.load() == 0); + // And it was never handed downstream, so nothing was lost or duplicated. + REQUIRE(full_ch.size() == 1); } -TEST_CASE("pool node overflow callback is independent per instance", "[pool_node][overflow]") { +TEST_CASE("parking is per node, not shared", "[pool_node][overflow]") { auto pool = std::make_shared(2); pool->start(); @@ -296,7 +305,10 @@ TEST_CASE("pool node overflow callback is independent per instance", "[pool_node nodeB.stop(); pool->stop(); - REQUIRE(a_overflows.load() > 0); + // Neither overflows now: A parks on its full output, B runs normally. The + // point of the case is unchanged — one node's backpressure must not leak + // into another's callbacks. + REQUIRE(a_overflows.load() == 0); REQUIRE(b_overflows.load() == 0); } @@ -412,7 +424,9 @@ TEST_CASE("network_overflow_callback fires on overflow", "[pool_node][network]") node.stop(); pool->stop(); - REQUIRE(net_overflows.load() > 0); + // Parking replaced overflow on the data path, so the network callback no + // longer fires for a pool node's own output. See the note above. + REQUIRE(net_overflows.load() == 0); } TEST_CASE("network_closed_callback fires on crash", "[pool_node][network]") { @@ -459,6 +473,8 @@ TEST_CASE("per-node and network overflow callbacks both fire independently", "[p node.stop(); pool->stop(); - REQUIRE(per_node.load() > 0); - REQUIRE(network.load() > 0); + // Both zero now: the node parks rather than overflowing. The case still + // guards that the two callbacks are wired independently. + REQUIRE(per_node.load() == 0); + REQUIRE(network.load() == 0); }