diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index a35651b..6ddad69 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -382,9 +382,15 @@ public: // Ring occupancy, derived lazily from indices — no separate counter on the // hot path. Excludes any out-of-band sentinel (that lives outside the ring). + // head_ is loaded first, deliberately. Both indices only ever increase, so + // reading head_ before tail_ can at worst under-report a concurrent push; + // the other order can read a head_ that has advanced past the tail_ already + // sampled, and the unsigned difference then wraps to ~2^64. A caller + // polling "is this channel empty yet" against that value never terminates. std::size_t size() const { - return tail_.load(std::memory_order_relaxed) - - head_.load(std::memory_order_relaxed); + const std::size_t h = head_.load(std::memory_order_relaxed); + const std::size_t t = tail_.load(std::memory_order_acquire); + return t - h; } // A pending out-of-band sentinel (EOF) counts as consumable work here even @@ -401,8 +407,9 @@ public: const ChannelStats& stats() const { return stats_; } ChannelSnapshot snapshot(const std::string& name) const { - const std::size_t t = tail_.load(std::memory_order_relaxed); + // head_ before tail_, for the reason given on size(). const std::size_t h = head_.load(std::memory_order_relaxed); + const std::size_t t = tail_.load(std::memory_order_acquire); return { name, capacity_, diff --git a/include/kpn/network.hpp b/include/kpn/network.hpp index 67686f3..281b9fa 100644 --- a/include/kpn/network.hpp +++ b/include/kpn/network.hpp @@ -91,6 +91,7 @@ public: + " → " + dst_name + ":" + std::to_string(DstIdx); channel_probes_.push_back( std::make_unique>(in_ch, ch_name)); + channel_src_names_.push_back(src_name); adj_[src_name].push_back(dst_name); return *this; @@ -213,6 +214,10 @@ public: watchdog_interval_ = interval; } + /// How long shutdown() waits for one node's outputs to drain before giving + /// up on them and stopping the next layer anyway. + void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; } + void set_error_handler(ErrorHandler h) { error_handler_ = std::move(h); } void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_ = std::move(h); } void set_event_handler(EventHandler h) { event_handler_ = std::move(h); } @@ -370,19 +375,46 @@ private: return true; } - void drain_output_channels(const std::string& /*name*/) const { - // Poll all channel probes until none report non-zero fill. - // A short sleep prevents busy-spin; 1 ms is fine for drain purposes. - bool any_full = true; - while (any_full) { - any_full = false; - for (auto& probe : channel_probes_) { - auto snap = probe->snapshot(); - if (snap.current_fill > 0) { any_full = true; break; } - } - if (any_full) - std::this_thread::sleep_for(std::chrono::milliseconds(1)); + /// Wait for the channels fed by `name` to empty, or give up. + /// + /// This took a node name and ignored it, polling *every* channel in the + /// graph instead — so shutdown() waited for the whole network to be idle + /// before stopping each successive layer. With no deadline either, anything + /// wedged downstream turned a graceful shutdown into the hang it exists to + /// avoid. + /// + /// Two bounds, because they fail differently. The deadline covers a + /// consumer that has stopped consuming, where fill never changes and + /// waiting cannot help. The no-progress counter covers one that is merely + /// slow: it keeps waiting while the queue is shrinking, so a slow drain is + /// not cut short just for taking a while. + void drain_output_channels(const std::string& name) const { + const auto deadline = clock_t::now() + drain_timeout_; + std::size_t last_fill = static_cast(-1); + int stalls = 0; + + auto fill_of = [&] { + std::size_t fill = 0; + for (std::size_t i = 0; i < channel_probes_.size(); ++i) + if (channel_src_names_[i] == name) + fill += channel_probes_[i]->snapshot().current_fill; + return fill; + }; + + for (;;) { + const std::size_t fill = fill_of(); + if (fill == 0) return; + if (fill >= last_fill) { if (++stalls > 100) break; } + else { stalls = 0; } + last_fill = fill; + if (clock_t::now() >= deadline) break; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); } + + if (const std::size_t left = fill_of()) + std::cerr << "[kpn] shutdown: '" << name << "' still has " << left + << " queued item(s) its consumer did not take; " + "they are discarded\n"; } // ── Cycle detection / topological sort ─────────────────────────────────── @@ -447,6 +479,10 @@ private: std::map exposed_outputs_; std::set> connected_outputs_; std::vector> channel_probes_; + /// Name of the node feeding each probe, parallel to channel_probes_. + /// shutdown() drains a node's own outputs, so it has to know which they are. + std::vector channel_src_names_; + std::chrono::milliseconds drain_timeout_{5000}; std::vector> pool_probes_; ErrorHandler error_handler_; DiagnosticsHandler diag_handler_; diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index 9c2dc32..5bdb9ac 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -98,13 +98,15 @@ public: std::vector fanout_ptrs, std::vector user_node_names, std::vector fanout_node_names, - std::vector> channel_probes) + std::vector> channel_probes, + std::vector channel_src_names) : fanouts_(std::move(fanouts)) , user_nodes_topo_(std::move(user_nodes_topo)) , fanout_nodes_ptr_(std::move(fanout_ptrs)) , user_node_names_(std::move(user_node_names)) , fanout_node_names_(std::move(fanout_node_names)) , channel_probes_(std::move(channel_probes)) + , channel_src_names_(std::move(channel_src_names)) {} ~StaticNetwork() override { stop(); } @@ -173,9 +175,9 @@ public: #endif // user_nodes_topo_ is already in sources-first order. // Stop each node and drain its output channels before moving on. - for (auto* n : user_nodes_topo_) { - n->stop(); - drain_all_channels(); + for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) { + user_nodes_topo_[i]->stop(); + drain_outputs_of(user_node_names_[i]); } for (auto* n : fanout_nodes_ptr_) n->stop(); } @@ -194,6 +196,10 @@ public: void set_event_handler(EventHandler h) { event_handler_ = std::move(h); } + /// How long shutdown() waits for one node's outputs to drain before giving + /// up on them and stopping the next layer anyway. + void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; } + /// 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, @@ -274,16 +280,50 @@ private: return {std::move(nodes), std::move(channels), std::move(resources), std::move(pools), elapsed_s}; } - void drain_all_channels() const { - bool any_full = true; - while (any_full) { - any_full = false; - for (auto& probe : channel_probes_) { - if (probe->snapshot().current_fill > 0) { any_full = true; break; } - } - if (any_full) - std::this_thread::sleep_for(std::chrono::milliseconds(1)); + /// Wait for the channels fed by `src` to empty, or give up. + /// + /// This was an unbounded `while (anything anywhere is non-empty)` poll over + /// *every* channel in the graph, which made shutdown() wait for the whole + /// network to be idle before stopping each successive layer, and wait + /// forever if anything downstream was wedged — turning a graceful shutdown + /// into the hang it exists to avoid. + /// + /// Two bounds, because they fail differently. The deadline covers a + /// consumer that has stopped consuming: fill never changes and no amount of + /// waiting helps. The no-progress counter covers a consumer that is merely + /// slow — it keeps waiting as long as the queue is shrinking, so a slow + /// drain is not cut short just for exceeding a fixed time. + /// + /// Giving up is reported rather than silent: undrained data at this point + /// means values are about to be discarded by the stop that follows. + void drain_outputs_of(const std::string& src) const { + const auto deadline = clock_t::now() + drain_timeout_; + std::size_t last_fill = static_cast(-1); + int stalls = 0; + + for (;;) { + std::size_t fill = 0; + for (std::size_t i = 0; i < channel_probes_.size(); ++i) + if (channel_src_names_[i] == src) + fill += channel_probes_[i]->snapshot().current_fill; + + if (fill == 0) return; + if (fill >= last_fill) { if (++stalls > 100) break; } + else { stalls = 0; } + last_fill = fill; + + if (clock_t::now() >= deadline) break; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); } + + std::size_t left = 0; + for (std::size_t i = 0; i < channel_probes_.size(); ++i) + if (channel_src_names_[i] == src) + left += channel_probes_[i]->snapshot().current_fill; + if (left) + std::cerr << "[kpn] shutdown: '" << src << "' still has " << left + << " queued item(s) its consumer did not take; " + "they are discarded\n"; } std::string name_; @@ -294,6 +334,10 @@ private: std::vector user_node_names_; std::vector fanout_node_names_; std::vector> channel_probes_; + /// Display name of the node feeding each probe, parallel to channel_probes_. + /// shutdown() drains a node's own outputs, so it has to know which they are. + std::vector channel_src_names_; + std::chrono::milliseconds drain_timeout_{5000}; std::vector> resource_probes_; std::vector> pool_probes_; EventHandler event_handler_; @@ -401,6 +445,7 @@ auto make_network(Edges&&... edges) { }; std::vector> channel_probes; + std::vector channel_src_names; auto wire_one = [&](SE) { using SrcNode = typename SE::src_node_t; @@ -418,6 +463,7 @@ auto make_network(Edges&&... edges) { + " \xe2\x86\x92 " // UTF-8 → + node_name.template operator()() + ":" + std::to_string(DstIdx); channel_probes.push_back(std::make_unique>(ch, ch_name)); + channel_src_names.push_back(node_name.template operator()()); } }; @@ -445,7 +491,8 @@ auto make_network(Edges&&... edges) { std::move(fanout_ptrs), std::move(user_node_names), std::move(fanout_node_names), - std::move(channel_probes)); + std::move(channel_probes), + std::move(channel_src_names)); } } // namespace kpn diff --git a/tests/test_static_network.cpp b/tests/test_static_network.cpp index 3dc2b26..4b1e780 100644 --- a/tests/test_static_network.cpp +++ b/tests/test_static_network.cpp @@ -271,3 +271,81 @@ TEST_CASE("static_network: fanout with labelled same-function consumers", "[stat REQUIRE(outB.pop() == 7); net.stop(); } + +// Regression: shutdown() must return even when a consumer stopped consuming. +// +// The drain step was an unbounded `while (anything anywhere is non-empty)` poll +// over *every* channel in the graph. Two defects in one loop: it waited for the +// whole network to be idle before stopping each successive layer rather than +// just the node it had stopped — the dynamic Network's version even took a node +// name and ignored it — and it had no deadline, so anything wedged downstream +// turned a graceful shutdown into the hang it exists to avoid. +// +// It could also fail to terminate with nothing wedged at all. current_fill came +// from a snapshot that loaded tail_ before head_; a concurrent pop between the +// two reads yields a head_ past the sampled tail_, and the unsigned difference +// wraps to ~2^64. Any poll for "is it empty yet" against that value runs +// forever. Both indices only ever increase, so loading head_ first can at worst +// under-report a push, which this loop tolerates and a wrap does not. +// +// Here the sink never takes anything, so its input cannot drain and the only +// correct outcome is to give up and say so. The bound asserted is deliberately +// loose: the point is that it terminates, not how fast. +namespace { + +struct DrainSource { + static constexpr std::string_view label() { return "drain_source"; } + int n{0}; + int operator()() { + std::this_thread::sleep_for(std::chrono::microseconds(100)); + return n++; + } +}; + +struct NeverConsumes { + static constexpr std::string_view label() { return "never_consumes"; } + std::atomic* wedged; + void operator()(int) { + // Blocks for the duration of the test: the input channel behind it + // fills and stays full. + while (!wedged->load(std::memory_order_acquire)) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } +}; + +} // namespace + +TEST_CASE("shutdown returns when a consumer has wedged", "[static_network][shutdown]") { + std::atomic release{false}; + + DrainSource src_fn; + NeverConsumes sink_fn{&release}; + + kpn::ObjectNode, kpn::out<"v">, "drain_source", 0> s(src_fn, 4); + kpn::ObjectNode, kpn::out<>, "never_consumes", 0> k(sink_fn, 4); + + auto net = kpn::make_network(kpn::edge(s.output<"v">(), k.input<"v">())); + net.set_drain_timeout(std::chrono::milliseconds(100)); + net.start(); + + // Let the channel fill and the sink jam. + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // Unjam the sink well after the drain timeout should have expired. Stopping + // a node joins its worker, so a sink blocked forever would hang the test in + // stop() rather than in the drain loop this case is about. + std::thread unjam([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(800)); + release.store(true, std::memory_order_release); + }); + + const auto t0 = std::chrono::steady_clock::now(); + net.shutdown(); + const auto elapsed = std::chrono::steady_clock::now() - t0; + + unjam.join(); + + const auto ms = std::chrono::duration_cast(elapsed).count(); + INFO("shutdown took " << ms << " ms"); + CHECK(ms < 3000); // unbounded before; one 100 ms drain timeout after +}