diff --git a/.gitignore b/.gitignore index 483232f..d3fb3e0 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,5 @@ Thumbs.db # Claude Code local settings .claude/settings.local.json include/kpn/ort_cache/ +build-tsan/ +build-*/ diff --git a/include/kpn/branch.hpp b/include/kpn/branch.hpp index fcd53ce..81903ae 100644 --- a/include/kpn/branch.hpp +++ b/include/kpn/branch.hpp @@ -7,12 +7,78 @@ #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 (;;) { + switch (ch->try_push(val)) { + case Channel::PushResult::Taken: + parked = duration_t(clock_t::now() - park_from); + return true; + case Channel::PushResult::Closed: + // Nobody is listening any more; the channel has recorded the + // drop. Retrying would spin until teardown noticed. + parked = duration_t(clock_t::now() - park_from); + return false; + case Channel::PushResult::Full: + break; // fall through to the retry logic + } + if (stop_flag.load(std::memory_order_relaxed)) { + // Teardown with work in hand and the output still full. One last + // throwing push, purely so the channel's own stats record the + // overflow — 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 +192,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 +332,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/channel.hpp b/include/kpn/channel.hpp index 571b891..e7420ff 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. @@ -153,13 +165,26 @@ public: head_.load(std::memory_order_acquire) < capacity_; } - /// Non-blocking, lossless push. Returns false when the ring is full, having + /// Outcome of a non-blocking push. + /// + /// try_push used to return bool, and returned *true* for a closed channel — + /// so "delivered" and "discarded because nobody is listening" were the same + /// answer. Both mean "stop trying", which is why the callers were correct, + /// but neither they nor the producer's own accounting could tell a value + /// that arrived from one that was thrown away. Only the channel's drop + /// counter knew. + enum class PushResult { Taken, Full, Closed }; + + /// Non-blocking, lossless push. Returns Full 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; } + PushResult try_push(T& value) { + if (!accepting_.load(std::memory_order_acquire)) { + stats_.record_drop(); + return PushResult::Closed; + } 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; + if (t - h >= capacity_) return PushResult::Full; const std::size_t data_bytes = ChannelDataSize::bytes(value); const bool was_empty = (t == h); @@ -169,7 +194,7 @@ public: wake_.fetch_add(1, std::memory_order_release); wake_.notify_one(); if (was_empty && push_callback_) push_callback_(); - return true; + return PushResult::Taken; } // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to @@ -214,12 +239,37 @@ public: // preserving ordering (EOF arrives after all data pushed before it). // // Only the sole producer may call it (SPSC contract, same as push()). - // Returns false if the channel is already disabled (token discarded — - // teardown is in progress, so the sentinel is moot). - bool push_sentinel(T value) { + // + // The slot holds exactly one undelivered token. A second offered before the + // first is taken is refused, not queued and not overwritten: two control + // tokens on one channel means the stream ended twice, which is a caller + // protocol error rather than backpressure, and silently coalescing them + // would hide it. + /// Outcome of offering a sentinel. SlotBusy is a protocol error, not + /// backpressure: it means a second control token was offered while the + /// first was still undelivered, and a channel carries at most one. + enum class SentinelResult { Taken, Closed, SlotBusy }; + + /// Non-consuming form. `value` is left untouched unless the result is + /// Taken, so a refused token is still the caller's to report. + SentinelResult try_push_sentinel(T& value) { if (!accepting_.load(std::memory_order_acquire)) { stats_.record_drop(); - return false; + return SentinelResult::Closed; + } + // Refuse rather than overwrite. Overwriting lost the first token + // silently, and worse, wrote eof_value_ while the consumer could be + // moving the previous one out of it — a data race on the storage, which + // for a shared_ptr payload is a torn refcount rather than a stale read. + // + // Checking here is what makes the slot a correct SPSC handshake: the + // producer is the only writer of eof_value_ and the only one that sets + // has_eof_, the consumer is the only one that clears it, so observing + // false here means the consumer has finished with the storage and will + // not touch it again until this store publishes the next token. + if (has_eof_.load(std::memory_order_acquire)) { + stats_.record_drop(); + return SentinelResult::SlotBusy; } eof_value_ = make_storage(std::move(value)); has_eof_.store(true, std::memory_order_release); @@ -228,7 +278,13 @@ public: wake_.fetch_add(1, std::memory_order_release); wake_.notify_one(); if (push_callback_) push_callback_(); - return true; + return SentinelResult::Taken; + } + + /// Consuming convenience form. Returns false when the token was not stored, + /// whether because the channel is closed or because one is already pending. + bool push_sentinel(T value) { + return try_push_sentinel(value) == SentinelResult::Taken; } // Blocking pop. Returns when an item is available. @@ -339,9 +395,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 @@ -358,8 +420,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_, @@ -394,6 +457,24 @@ private: // delivered after every value pushed before it. bool take_sentinel(T& out) { if (!has_eof_.load(std::memory_order_acquire)) return false; + // Re-check emptiness *after* observing has_eof_, not before. + // + // Callers check the ring is empty and then call this, but the producer + // can push a value and publish the sentinel in the window between those + // two steps — so the sentinel would be delivered with a real value still + // queued behind it, breaking the "sentinel is strictly last" contract + // that downstream teardown depends on. a0c4bf5 closed the variant where + // the caller's emptiness check used a stale tail_ snapshot; this is the + // one where the check is fresh but simply too early. + // + // Checking here is what makes it sound: the producer publishes the + // sentinel with a release store *after* its ring pushes, so a consumer + // that has observed has_eof_ has also observed every tail_ advance + // before it. If the ring is non-empty now, those values genuinely + // precede the sentinel and must be delivered first. + if (head_.load(std::memory_order_relaxed) + != tail_.load(std::memory_order_acquire)) + return false; out = extract(std::move(eof_value_)); has_eof_.store(false, std::memory_order_release); stats_.record_pop(); diff --git a/include/kpn/diagnostics.hpp b/include/kpn/diagnostics.hpp index f5080b1..884d668 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,29 @@ 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 + + // 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. + // + // Declared before the two bools below because every node type initialises + // this aggregate positionally, and all of them supply total_exec_ms as the + // element after queue_wait_ms. + double total_exec_ms{0}; + + // 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}; }; // ── Pool statistics + snapshot ──────────────────────────────────────────────── @@ -188,6 +220,11 @@ struct ResourceSnapshot { struct IResourceProbe { virtual ~IResourceProbe() = default; virtual ResourceSnapshot snapshot(const std::string& name) const = 0; + + /// Release every thread waiting for the resource, so teardown is not held + /// up by one. A network calls this on the resources registered with it when + /// it halts; default no-op for probes with nothing to wake. + virtual void close() {} }; } // namespace kpn diff --git a/include/kpn/fanout.hpp b/include/kpn/fanout.hpp index 180b602..f9f2c46 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,79 @@ 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; + // Taken or Closed both mean "stop trying" — delivered, or gone + // with the drop recorded. Only Full is worth another pass. + if (out_channels_[i]->try_push(*pending[i]) + != Channel::PushResult::Full) { + 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 +201,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/inode.hpp b/include/kpn/inode.hpp index a7454e2..1ad9747 100644 --- a/include/kpn/inode.hpp +++ b/include/kpn/inode.hpp @@ -22,6 +22,23 @@ enum class NodeEvent { Overflow, Closed }; struct INode { virtual ~INode() = default; + + // Install channel callbacks, without starting anything. + // + // A node's push/space callbacks live in std::function members on channels + // it shares with its neighbours, and a neighbour that is already running + // reads them on its own thread. Writing one while the pipeline runs is a + // data race on the std::function — ThreadSanitizer reports it, and the + // consequence in the field was the missed startup wake a8cfe73 had to + // patch around. + // + // So a network calls prepare() on every node before it calls start() on + // any of them: all the writes happen while nothing is running, and once a + // node is live the callbacks are read-only. start() calls prepare() itself + // if it has not been called, so standalone nodes still work; it is + // idempotent, and the network relies on that. + virtual void prepare() {} + virtual void start() = 0; virtual void stop() = 0; virtual bool running() const = 0; 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/network.hpp b/include/kpn/network.hpp index 239ad88..5974e70 100644 --- a/include/kpn/network.hpp +++ b/include/kpn/network.hpp @@ -8,8 +8,10 @@ #include #endif +#include #include #include +#include #include #include #include @@ -39,8 +41,15 @@ public: class Network : public INode { public: - using ErrorHandler = - std::function; + /// 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. + /// + /// Same type as StaticNetwork's, deliberately: this used to be a void + /// signature, which could not express the keep-running decision and, more + /// to the point, was never delivered anywhere. + using ErrorHandler = NodeErrorHandler; using DiagnosticsHandler = std::function&, const std::vector&)>; @@ -91,6 +100,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; @@ -134,6 +144,17 @@ public: void start() override { start_time_ = clock_t::now(); + // Deliver the listener to the nodes. Without this the handler was + // stored and never read: a node's exception was discarded at the node + // boundary and the only surviving evidence was a Closed event, which + // says a node stopped but not why. StaticNetwork has always done this; + // Network accepted the handler and silently dropped it. + if (error_handler_) + for (auto& name : topo_) + nodes_.at(name)->set_network_error_callback(error_handler_); + // Callbacks first, everywhere, before anything runs — see INode::prepare. + for (auto& name : topo_) + nodes_.at(name)->prepare(); for (auto& name : topo_) nodes_.at(name)->start(); start_watchdog(); @@ -210,6 +231,11 @@ 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; } + + /// Must be called before start(); the handler is delivered to nodes there. 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); } @@ -367,19 +393,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 ─────────────────────────────────── @@ -398,9 +451,20 @@ private: void start_watchdog() { watchdog_ = std::jthread([this](std::stop_token tok) { + // Interruptible wait, not sleep_for. request_stop() cannot wake a + // sleeping thread, so stop_watchdog()'s join blocked for up to a + // full interval — three seconds by default, and unbounded for + // anyone who set a long one to keep the periodic report quiet. + // Every teardown paid it. + std::mutex m; + std::condition_variable_any cv; while (!tok.stop_requested()) { - std::this_thread::sleep_for(watchdog_interval_); - if (tok.stop_requested()) break; + { + std::unique_lock lk(m); + if (cv.wait_for(lk, tok, watchdog_interval_, + [&tok] { return tok.stop_requested(); })) + break; + } auto s = collect_snapshots(); check_hung_nodes(); @@ -444,6 +508,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/pool_node.hpp b/include/kpn/pool_node.hpp index 2d12cbe..e6113b5 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -5,6 +5,7 @@ #include "inode.hpp" #include "port.hpp" #include "scheduler.hpp" +#include "submit_gate.hpp" #include "traits.hpp" #include @@ -23,44 +24,15 @@ 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 ────────────────────────────────────────────────────────────────── // // Reactive alternative to Node<>. Instead of owning a blocked thread, the node // is submitted to a shared IScheduler whenever all its input channels become // non-empty. A single fire_once() call pops all inputs, executes the function, -// and pushes outputs. At most one fire_once() runs at a time (queued_ flag). +// and pushes outputs. At most one fire_once() runs at a time (see SubmitGate). // // Source nodes (input_count == 0) submit themselves immediately on start() and // resubmit after each fire_once(). @@ -109,21 +81,43 @@ public: // ── INode ───────────────────────────────────────────────────────────────── + void prepare() override { + if (prepared_) return; // idempotent: the network calls this, + prepared_ = true; // and start() calls it again if not. + register_callbacks(std::make_index_sequence{}); + } + void start() override { + prepare(); enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); - queued_.store(false, std::memory_order_relaxed); - register_callbacks(std::make_index_sequence{}); + gate_.force_idle(); 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. + // + // The callback is installed by prepare(), before any node runs, but + // a network still starts its nodes one at a time: an upstream node + // that is already firing can push into this one between the two + // calls. The push is accepted by the ring and does invoke the + // callback, but on_input_ready() sees stop_flag_ still set and + // returns. Every later push sees a non-empty ring and stays silent + // — Channel invokes push_callback_ only on the empty->non-empty + // transition — so without this the node is never submitted and the + // pipeline reads as wedged from the first frame. + // + // 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 { stop_flag_.store(true, std::memory_order_seq_cst); disable_inputs(std::make_index_sequence{}); - // fire_once() observes stop_flag_ and will not resubmit. - // We do not wait for an in-flight fire_once() to complete here; - // callers that need that guarantee should call scheduler_->drain() first. + await_quiescence(); } bool running() const override { @@ -156,6 +150,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, + gate_.queued(), + gate_.wake_pending(), }; } @@ -265,9 +262,9 @@ private: disable_inputs(std::make_index_sequence{}); disable_outputs(std::make_index_sequence{}); stats_.exec_start_us.store(0, std::memory_order_relaxed); - // Plain store, not release_and_recheck(): this node is stopping, and + // force_idle, not finish_firing(): this node is stopping, and // honouring a pending wake here would resubmit a dead node. - queued_.store(false, std::memory_order_release); + gate_.force_idle(); stop_flag_.store(true, std::memory_order_relaxed); } @@ -355,39 +352,121 @@ private: : 0.5f), ...); } - /// Submit unless already queued. A wake that arrives while this node is - /// queued or running is *recorded*, never dropped. + /// Submit unless a firing is already in flight. A wake that arrives while + /// one is is *recorded* against it, never dropped. /// /// Wakes are edge-triggered: a channel fires its space callback on the - /// transition, once. If that lands while queued_ is up, the CAS below fails - /// and — before wake_pending_ — the wake was gone. A node could then park a + /// transition, once. A dropped one never returns, so a node could park a /// value, release its worker, and sleep forever holding output its consumer /// was waiting for, with every worker idle in cond_wait and nothing left to - /// re-trigger it. Recording the drop turns the signal level-triggered: the - /// invariant is that a node never sleeps with a wake outstanding, enforced - /// by release_and_recheck() at every point that releases the node. + /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same + /// variable, so the two cannot both be true — see submit_gate.hpp. void try_submit(float priority) { - bool expected = false; - if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + // A stopped node must not claim the gate. The scheduler now refuses + // submissions after its pool stops, so the submit itself is safe — but + // claiming and never releasing would leave the gate held, and a restart + // would then have to clear it. start() does, but relying on that makes + // the invariant depend on a distant statement. + if (stop_flag_.load(std::memory_order_relaxed)) return; + if (gate_.claim()) scheduler_->submit([this] { fire_once(); }, priority); - else - wake_pending_.store(true, std::memory_order_release); - } - - /// Clear queued_, then honour any wake that was dropped while it was up. - /// Every path that finishes or parks a firing must release the node through - /// here rather than storing queued_ directly. - void release_and_recheck(float priority = 0.5f) { - queued_.store(false, std::memory_order_release); - if (wake_pending_.exchange(false, std::memory_order_acq_rel)) - try_submit(priority); } // ── Execution ───────────────────────────────────────────────────────────── + /// Decide whether this node should run again, then release the gate — in + /// that order, always. + /// + /// Releasing first is what let two firings of the same node overlap: the + /// moment the gate is free another worker may enter fire_once, while this + /// invocation is still reading pending_ and writing pending_done_. TSan + /// caught it as a race on pending_done_ between a firing submitted by the + /// old release_and_recheck and one submitted by try_submit. It also quietly + /// broke the one-slot park, which is sound only because "at most one + /// fire_once runs per node at a time" — with two, a value can be parked by + /// one firing and overwritten by the other. + /// + /// Everything this reads belongs to the firing that holds the claim, so it + /// is all evaluated first and the release is the last thing the firing does. + void finish_firing() { + bool want_more = false; + float prio = 0.5f; + + if (!stop_flag_.load(std::memory_order_relaxed)) { + bool parked = false; + if constexpr (!std::is_void_v) + parked = pending_.has_value(); + + if (parked) { + // Still holding output: only worth running again once the + // consumer has made room. + want_more = outputs_have_space(std::make_index_sequence{}); + } else { + if constexpr (input_count == 0) { + want_more = true; // sources always run again + } else { + want_more = count_ready(std::make_index_sequence{}) + == input_count; + if (want_more) prio = compute_priority(); + } + } + } + + if (gate_.release()) scheduler_->submit([this] { fire_once(); }, prio); + else if (want_more) try_submit(prio); + } + + /// Block until no firing of this node is in flight or queued. + /// + /// stop() used to set the flag and return, leaving an executing fire_once + /// touching input_channels_, stats_ and pending_ while the caller went on + /// to destroy them. For a node with a private pool that was survivable by + /// accident — Node::stop() calls pool->stop(), which joins — but a node + /// sharing a pool had nothing joining it at all, so ~PoolNode raced its own + /// members. The old comment said callers wanting the guarantee should call + /// scheduler_->drain() first; a destructor cannot, and the default should + /// not be a use-after-free. + /// + /// The gate is exactly the right thing to wait on: it is claimed for the + /// whole of a firing and released as the last act of one. A queued but + /// unstarted firing also holds it, and will run, observe stop_flag_ and + /// release — which is why the pool must still be running when this is + /// called. That is already the documented order (stop nodes, then the + /// pool), and Node/ObjectNode do it that way. + /// + /// Bounded, because a node function that never returns must not turn + /// teardown into a hang; and skipped entirely when called from the firing + /// thread itself, since an error handler that stops its own node would + /// otherwise wait for a firing that is waiting for it. + void await_quiescence() { + if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id()) + return; + const auto deadline = clock_t::now() + std::chrono::seconds(5); + while (gate_.queued()) { + if (clock_t::now() >= deadline) { + std::cerr << "[kpn] stop: node '" << name_ + << "' still had work in flight after 5 s; " + "continuing without it\n"; + return; + } + std::this_thread::sleep_for(std::chrono::microseconds(50)); + } + } + + /// Marks fire_once's thread for the duration of a firing, so await_quiescence + /// can tell a re-entrant stop() from an external one. + struct FiringMark { + std::atomic& slot; + explicit FiringMark(std::atomic& s) : slot(s) { + slot.store(std::this_thread::get_id(), std::memory_order_release); + } + ~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); } + }; + void fire_once() { + FiringMark mark(firing_thread_); if (stop_flag_.load(std::memory_order_relaxed)) { - queued_.store(false, std::memory_order_release); + gate_.force_idle(); return; } @@ -403,54 +482,13 @@ private: if constexpr (!std::is_void_v) { if (pending_) { push_outputs(std::move(*pending_), std::make_index_sequence{}); - release_and_recheck(); - 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{}); - release_and_recheck(); - 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(); + // Whether the value went out or is still parked, finish_firing + // reads pending_ and picks the right follow-up: output space if + // still holding, input readiness if drained. Resubmitting + // unconditionally would fire a node whose inputs are empty, and + // pop_one reports an empty channel as ChannelClosedError — which + // this node treats as "upstream finished" and self-stops on. + finish_firing(); return; } } @@ -462,8 +500,9 @@ private: // on_input_ready() resubmits when data actually lands. if constexpr (input_count > 0) { if (count_ready(std::make_index_sequence{}) != input_count) { - release_and_recheck(); - on_input_ready(); // data may have arrived while we checked + // finish_firing re-checks readiness after the work above, so + // data that landed while we looked is not missed. + finish_firing(); return; } } @@ -486,6 +525,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(); @@ -507,37 +554,15 @@ private: } stats_.exec_start_us.store(0, std::memory_order_relaxed); - release_and_recheck(); - - if (stop_flag_.load(std::memory_order_relaxed)) return; - - // Parked by the push above. Same situation as the retry path at the top - // of fire_once — and the same lost-wakeup race, which that path closes - // and this one did not. A space callback that fired while queued_ was - // still up got swallowed by try_submit's CAS, and the resubmit below - // cannot cover it: this firing consumed its input, so inputs are empty - // and on_input_ready() will not resubmit. The node would then hold its - // value forever while its consumer waits for exactly that value and its - // producer parks on an input channel that never drains. Re-check now - // that the flag is down. - if constexpr (!std::is_void_v) { - if (pending_) { - if (outputs_have_space(std::make_index_sequence{})) - try_submit(0.5f); - return; // parked - } - } - - // Source nodes always resubmit; others resubmit only if inputs are ready. - if constexpr (input_count == 0) { - try_submit(0.5f); - } else { - on_input_ready(); - } + // If the push above parked, finish_firing waits on output space rather + // than input arrival: this firing consumed its input, so an input-level + // check would not resubmit and the node would hold its value forever + // while its consumer waits for exactly that value. + finish_firing(); } // Pop all inputs — safe because we're the sole consumer and fire_once - // is guarded by queued_ (only one fire_once runs at a time). + // is guarded by the submit gate (only one fire_once runs at a time). template args_tuple pop_inputs(std::index_sequence) { return {pop_one()...}; @@ -547,8 +572,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; } @@ -568,7 +601,15 @@ private: 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); + // The retry path calls this as push_outputs(std::move(*pending_), …), so + // on that path `result` *is* the parked tuple. Assigning it to itself is + // a self-move-assignment, which for std::tuple is elementwise — and + // libstdc++'s std::vector does not guard against it: it swaps its data + // into a temporary and leaves the vector empty. A value that failed to + // push twice would therefore be delivered with its payload silently + // erased, which downstream reads as a legitimately empty result rather + // than as a loss. Only store when it is not already stored. + else if (!pending_ || &result != &*pending_) pending_ = std::move(result); } /// Returns false when the ring was full and the value was NOT taken; the @@ -581,7 +622,14 @@ private: // 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)); + // A refused sentinel is a protocol error, not backpressure, so it + // is reported rather than parked and retried — retrying would spin + // forever against a slot only the consumer can free, and there is + // no correct value to deliver second anyway. Closed is normal + // during teardown and stays quiet. + if (ch->try_push_sentinel(val) == Channel> + ::SentinelResult::SlotBusy) + fire_callbacks(event_callbacks_); return true; } // Backpressure without parking the worker. A full channel means the @@ -593,7 +641,10 @@ private: // 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); + // Closed counts as "stop trying", not as delivered: the value is gone + // and the channel has recorded the drop. Only Full means park and retry. + return ch->try_push(val) != Channel> + ::PushResult::Full; } template @@ -614,9 +665,17 @@ private: input_channels_t input_channels_; output_channels_t output_channels_{}; std::atomic stop_flag_{true}; - std::atomic queued_{false}; - /// A wake that arrived while queued_ was up. See try_submit. - std::atomic wake_pending_{false}; + /// Serialises firings and records wakes that arrive during one. See + /// submit_gate.hpp for why this cannot be two separate flags. + SubmitGate gate_; + /// Whether prepare() has installed the channel callbacks. Only ever touched + /// from the thread driving start()/stop(), never from a worker, and never + /// cleared: the callbacks capture `this` and stay valid across a restart, so + /// re-registering them would be a pointless write to a live channel. + bool prepared_{false}; + /// Thread currently inside fire_once, or a default id when none is. + /// See await_quiescence. + std::atomic firing_thread_{}; /// 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 @@ -679,18 +738,28 @@ public: ~PoolObjectNode() override { stop(); } + void prepare() override { + if (prepared_) return; + prepared_ = true; + register_callbacks(std::make_index_sequence{}); + } + void start() override { + prepare(); enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); - queued_.store(false, std::memory_order_relaxed); - register_callbacks(std::make_index_sequence{}); + gate_.force_idle(); 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 { stop_flag_.store(true, std::memory_order_seq_cst); disable_inputs(std::make_index_sequence{}); + await_quiescence(); } bool running() const override { return !stop_flag_.load(std::memory_order_relaxed); } @@ -720,6 +789,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, + gate_.queued(), + gate_.wake_pending(), }; } @@ -796,9 +868,9 @@ private: disable_inputs(std::make_index_sequence{}); disable_outputs(std::make_index_sequence{}); stats_.exec_start_us.store(0, std::memory_order_relaxed); - // Plain store, not release_and_recheck(): this node is stopping, and + // force_idle, not finish_firing(): this node is stopping, and // honouring a pending wake here would resubmit a dead node. - queued_.store(false, std::memory_order_release); + gate_.force_idle(); stop_flag_.store(true, std::memory_order_relaxed); } @@ -879,37 +951,119 @@ private: : 0.5f), ...); } - /// Submit unless already queued. A wake that arrives while this node is - /// queued or running is *recorded*, never dropped. + /// Submit unless a firing is already in flight. A wake that arrives while + /// one is is *recorded* against it, never dropped. /// /// Wakes are edge-triggered: a channel fires its space callback on the - /// transition, once. If that lands while queued_ is up, the CAS below fails - /// and — before wake_pending_ — the wake was gone. A node could then park a + /// transition, once. A dropped one never returns, so a node could park a /// value, release its worker, and sleep forever holding output its consumer /// was waiting for, with every worker idle in cond_wait and nothing left to - /// re-trigger it. Recording the drop turns the signal level-triggered: the - /// invariant is that a node never sleeps with a wake outstanding, enforced - /// by release_and_recheck() at every point that releases the node. + /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same + /// variable, so the two cannot both be true — see submit_gate.hpp. void try_submit(float priority) { - bool expected = false; - if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + // A stopped node must not claim the gate. The scheduler now refuses + // submissions after its pool stops, so the submit itself is safe — but + // claiming and never releasing would leave the gate held, and a restart + // would then have to clear it. start() does, but relying on that makes + // the invariant depend on a distant statement. + if (stop_flag_.load(std::memory_order_relaxed)) return; + if (gate_.claim()) scheduler_->submit([this] { fire_once(); }, priority); - else - wake_pending_.store(true, std::memory_order_release); } - /// Clear queued_, then honour any wake that was dropped while it was up. - /// Every path that finishes or parks a firing must release the node through - /// here rather than storing queued_ directly. - void release_and_recheck(float priority = 0.5f) { - queued_.store(false, std::memory_order_release); - if (wake_pending_.exchange(false, std::memory_order_acq_rel)) - try_submit(priority); + /// Decide whether this node should run again, then release the gate — in + /// that order, always. + /// + /// Releasing first is what let two firings of the same node overlap: the + /// moment the gate is free another worker may enter fire_once, while this + /// invocation is still reading pending_ and writing pending_done_. TSan + /// caught it as a race on pending_done_ between a firing submitted by the + /// old release_and_recheck and one submitted by try_submit. It also quietly + /// broke the one-slot park, which is sound only because "at most one + /// fire_once runs per node at a time" — with two, a value can be parked by + /// one firing and overwritten by the other. + /// + /// Everything this reads belongs to the firing that holds the claim, so it + /// is all evaluated first and the release is the last thing the firing does. + void finish_firing() { + bool want_more = false; + float prio = 0.5f; + + if (!stop_flag_.load(std::memory_order_relaxed)) { + bool parked = false; + if constexpr (!std::is_void_v) + parked = pending_.has_value(); + + if (parked) { + // Still holding output: only worth running again once the + // consumer has made room. + want_more = outputs_have_space(std::make_index_sequence{}); + } else { + if constexpr (input_count == 0) { + want_more = true; // sources always run again + } else { + want_more = count_ready(std::make_index_sequence{}) + == input_count; + if (want_more) prio = compute_priority(); + } + } + } + + if (gate_.release()) scheduler_->submit([this] { fire_once(); }, prio); + else if (want_more) try_submit(prio); } + /// Block until no firing of this node is in flight or queued. + /// + /// stop() used to set the flag and return, leaving an executing fire_once + /// touching input_channels_, stats_ and pending_ while the caller went on + /// to destroy them. For a node with a private pool that was survivable by + /// accident — Node::stop() calls pool->stop(), which joins — but a node + /// sharing a pool had nothing joining it at all, so ~PoolNode raced its own + /// members. The old comment said callers wanting the guarantee should call + /// scheduler_->drain() first; a destructor cannot, and the default should + /// not be a use-after-free. + /// + /// The gate is exactly the right thing to wait on: it is claimed for the + /// whole of a firing and released as the last act of one. A queued but + /// unstarted firing also holds it, and will run, observe stop_flag_ and + /// release — which is why the pool must still be running when this is + /// called. That is already the documented order (stop nodes, then the + /// pool), and Node/ObjectNode do it that way. + /// + /// Bounded, because a node function that never returns must not turn + /// teardown into a hang; and skipped entirely when called from the firing + /// thread itself, since an error handler that stops its own node would + /// otherwise wait for a firing that is waiting for it. + void await_quiescence() { + if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id()) + return; + const auto deadline = clock_t::now() + std::chrono::seconds(5); + while (gate_.queued()) { + if (clock_t::now() >= deadline) { + std::cerr << "[kpn] stop: node '" << name_ + << "' still had work in flight after 5 s; " + "continuing without it\n"; + return; + } + std::this_thread::sleep_for(std::chrono::microseconds(50)); + } + } + + /// Marks fire_once's thread for the duration of a firing, so await_quiescence + /// can tell a re-entrant stop() from an external one. + struct FiringMark { + std::atomic& slot; + explicit FiringMark(std::atomic& s) : slot(s) { + slot.store(std::this_thread::get_id(), std::memory_order_release); + } + ~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); } + }; + void fire_once() { + FiringMark mark(firing_thread_); if (stop_flag_.load(std::memory_order_relaxed)) { - queued_.store(false, std::memory_order_release); + gate_.force_idle(); return; } auto t0 = clock_t::now(); @@ -923,25 +1077,13 @@ private: if constexpr (!std::is_void_v) { if (pending_) { push_outputs(std::move(*pending_), std::make_index_sequence{}); - release_and_recheck(); - 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(); + // Whether the value went out or is still parked, finish_firing + // reads pending_ and picks the right follow-up: output space if + // still holding, input readiness if drained. Resubmitting + // unconditionally would fire a node whose inputs are empty, and + // pop_one reports an empty channel as ChannelClosedError — which + // this node treats as "upstream finished" and self-stops on. + finish_firing(); return; } } @@ -951,8 +1093,9 @@ private: // into pop_inputs on an empty channel. if constexpr (input_count > 0) { if (count_ready(std::make_index_sequence{}) != input_count) { - release_and_recheck(); - on_input_ready(); + // finish_firing re-checks readiness after the work above, so + // data that landed while we looked is not missed. + finish_firing(); return; } } @@ -973,6 +1116,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(); @@ -993,28 +1144,11 @@ private: } stats_.exec_start_us.store(0, std::memory_order_relaxed); - release_and_recheck(); - if (stop_flag_.load(std::memory_order_relaxed)) return; - - // Parked by the push above. Same situation as the retry path at the top - // of fire_once — and the same lost-wakeup race, which that path closes - // and this one did not. A space callback that fired while queued_ was - // still up got swallowed by try_submit's CAS, and the resubmit below - // cannot cover it: this firing consumed its input, so inputs are empty - // and on_input_ready() will not resubmit. The node would then hold its - // value forever while its consumer waits for exactly that value and its - // producer parks on an input channel that never drains. Re-check now - // that the flag is down. - if constexpr (!std::is_void_v) { - if (pending_) { - if (outputs_have_space(std::make_index_sequence{})) - try_submit(0.5f); - return; // parked - } - } - - if constexpr (input_count == 0) try_submit(0.5f); - else on_input_ready(); + // If the push above parked, finish_firing waits on output space rather + // than input arrival: this firing consumed its input, so an input-level + // check would not resubmit and the node would hold its value forever + // while its consumer waits for exactly that value. + finish_firing(); } template @@ -1024,7 +1158,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; } @@ -1044,7 +1187,15 @@ private: 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); + // The retry path calls this as push_outputs(std::move(*pending_), …), so + // on that path `result` *is* the parked tuple. Assigning it to itself is + // a self-move-assignment, which for std::tuple is elementwise — and + // libstdc++'s std::vector does not guard against it: it swaps its data + // into a temporary and leaves the vector empty. A value that failed to + // push twice would therefore be delivered with its payload silently + // erased, which downstream reads as a legitimately empty result rather + // than as a loss. Only store when it is not already stored. + else if (!pending_ || &result != &*pending_) 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. @@ -1056,11 +1207,21 @@ private: // 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)); + // A refused sentinel is a protocol error, not backpressure, so it + // is reported rather than parked and retried — retrying would spin + // forever against a slot only the consumer can free, and there is + // no correct value to deliver second anyway. Closed is normal + // during teardown and stays quiet. + if (ch->try_push_sentinel(val) == Channel> + ::SentinelResult::SlotBusy) + fire_callbacks(event_callbacks_); return true; } // See the note on the typed overload above: park rather than block. - return ch->try_push(val); + // Closed counts as "stop trying", not as delivered: the value is gone + // and the channel has recorded the drop. Only Full means park and retry. + return ch->try_push(val) != Channel> + ::PushResult::Full; } Obj& obj_; @@ -1070,9 +1231,17 @@ private: input_channels_t input_channels_; output_channels_t output_channels_{}; std::atomic stop_flag_{true}; - std::atomic queued_{false}; - /// A wake that arrived while queued_ was up. See try_submit. - std::atomic wake_pending_{false}; + /// Serialises firings and records wakes that arrive during one. See + /// submit_gate.hpp for why this cannot be two separate flags. + SubmitGate gate_; + /// Whether prepare() has installed the channel callbacks. Only ever touched + /// from the thread driving start()/stop(), never from a worker, and never + /// cleared: the callbacks capture `this` and stay valid across a restart, so + /// re-registering them would be a pointless write to a live channel. + bool prepared_{false}; + /// Thread currently inside fire_once, or a default id when none is. + /// See await_quiescence. + std::atomic firing_thread_{}; /// 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 diff --git a/include/kpn/scheduler.hpp b/include/kpn/scheduler.hpp index e7920bf..54bbb5b 100644 --- a/include/kpn/scheduler.hpp +++ b/include/kpn/scheduler.hpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -62,18 +63,30 @@ public: } void stop() override { - stopped_.store(true, std::memory_order_seq_cst); + // Close the pool to new work before touching anything, and do it under + // the lifecycle lock so no submit() is midway through indexing queues_. + { + std::unique_lock lk(lifecycle_mx_); + stopped_.store(true, std::memory_order_seq_cst); + } for (auto& q : queues_) { std::lock_guard lock(q->mx); std::size_t discarded = q->pq.size(); while (!q->pq.empty()) q->pq.pop(); total_.fetch_sub(discarded, std::memory_order_relaxed); + queued_.fetch_sub(discarded, std::memory_order_relaxed); } // Lock cv_mx_ before notifying so the stop signal can't be lost in the // gap between a worker's predicate check and its wait() (see submit()). { std::lock_guard lk(cv_mx_); } cv_.notify_all(); + // Join without the lock: a worker's task may call submit(), which takes + // it shared, and holding it here would deadlock against that. for (auto& t : workers_) if (t.joinable()) t.join(); + // Destroying the queues is what submit() must never race. By now + // stopped_ is published, so any submit() that acquires the lock after + // this point returns without touching them. + std::unique_lock lk(lifecycle_mx_); workers_.clear(); queues_.clear(); } @@ -86,6 +99,22 @@ public: } void submit(std::function task, float priority = 0.5f) override { + // A submission can arrive after this pool has been stopped, and did so + // by an ordinary route: a node's space callback fires from whichever + // thread drained the channel, which belongs to the *consumer*. Stop the + // producer first — as a sources-first shutdown does — and the consumer + // keeps draining its backlog, firing the producer's space callback into + // a pool whose stop() has already run queues_.clear(). submit() then + // indexed an empty vector: a segfault, reproducible about 12 runs in 20. + // + // The shared lock is what makes the check meaningful. Reading stopped_ + // alone leaves the window between the read and the indexing, which is + // precisely where stop() clears the vector. + std::shared_lock lk(lifecycle_mx_); + if (stopped_.load(std::memory_order_acquire) || queues_.empty()) { + rejected_.fetch_add(1, std::memory_order_relaxed); + return; + } std::size_t target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_; { std::lock_guard lock(queues_[target]->mx); @@ -93,6 +122,7 @@ public: {std::move(task), priority, seq_.fetch_add(1, std::memory_order_relaxed)}); } total_.fetch_add(1, std::memory_order_relaxed); + queued_.fetch_add(1, std::memory_order_relaxed); submitted_.fetch_add(1, std::memory_order_relaxed); // Synchronize with worker_loop's predicate evaluation: taking cv_mx_ // here guarantees a worker is either before its predicate check (and @@ -105,15 +135,17 @@ public: std::size_t thread_count() const { return thread_count_; } + /// Submissions dropped because the pool was stopped. See rejected_. + uint64_t rejected() const { return rejected_.load(std::memory_order_relaxed); } + // ── IPoolProbe ──────────────────────────────────────────────────────────── PoolSnapshot snapshot(const std::string& name) const override { std::size_t a = active_.load(std::memory_order_relaxed); - std::size_t t = total_.load(std::memory_order_relaxed); return { name, thread_count_, - t > a ? t - a : 0, // queued (approximate) - a, // executing + queued_.load(std::memory_order_relaxed), // queued (exact) + a, // executing submitted_.load(std::memory_order_relaxed), completed_.load(std::memory_order_relaxed), }; @@ -142,6 +174,7 @@ private: if (q.pq.empty()) return std::nullopt; auto fn = std::move(const_cast(q.pq.top()).fn); q.pq.pop(); + queued_.fetch_sub(1, std::memory_order_relaxed); return fn; } @@ -182,10 +215,13 @@ private: std::unique_lock lock(cv_mx_); cv_.wait(lock, [this] { return stopped_.load(std::memory_order_seq_cst) - || total_.load(std::memory_order_relaxed) > 0; + || queued_.load(std::memory_order_relaxed) > 0; }); + // Exit on queued_, not total_: waiting for total_ to reach zero + // meant waiting for someone else's task to finish, which this + // worker cannot help with and would spin through until it did. if (stopped_.load(std::memory_order_seq_cst) - && total_.load(std::memory_order_relaxed) == 0) + && queued_.load(std::memory_order_relaxed) == 0) return; } } @@ -194,17 +230,36 @@ private: std::vector> queues_; std::vector workers_; + /// Guards the lifetime of queues_/workers_ against a concurrent submit(). + /// Shared by submit, exclusive by stop, so submissions still run in + /// parallel with each other. + mutable std::shared_mutex lifecycle_mx_; + std::mutex cv_mx_; std::condition_variable cv_; std::mutex drain_mx_; std::condition_variable drain_cv_; std::atomic stopped_{true}; - std::atomic total_{0}; // queued + executing + std::atomic total_{0}; // queued + executing (drain() waits on this) + /// Queued only — never counts a task that is already executing. + /// + /// The wait predicate used total_, which includes running tasks, so while + /// any one task ran every *other* worker's predicate was true: wait() + /// returned instantly and the worker spun through try_pop / try_steal / + /// wait at full speed, try_lock-ing every peer queue on each pass. One slow + /// task therefore pinned every other core and contended the very mutexes + /// the working thread needed. Sleeping requires "no work is *waiting*", + /// which is this. + std::atomic queued_{0}; // waiting to run std::atomic active_{0}; // executing only (for snapshot) std::atomic next_{0}; // round-robin submit cursor std::atomic seq_{0}; // tie-break for equal-priority tasks std::atomic submitted_{0}; + /// Submissions refused because the pool was already stopped. Not an error — + /// teardown races are expected — but silence here would hide a node that + /// keeps trying to run after its pool is gone. + std::atomic rejected_{0}; std::atomic completed_{0}; }; diff --git a/include/kpn/shared_resource.hpp b/include/kpn/shared_resource.hpp index 79924c3..9919508 100644 --- a/include/kpn/shared_resource.hpp +++ b/include/kpn/shared_resource.hpp @@ -14,6 +14,18 @@ namespace kpn { template class Channel; // forward declaration for acquire_balanced +/// Thrown by a pending acquire() when the resource is closed underneath it. +/// +/// acquire() blocks on a condition variable with no timeout and no stop +/// condition, so a node parked there ignored teardown entirely: the worker +/// never returned, the pool's join never completed, and shutdown hung on a +/// resource nobody was going to release. Closing the resource turns that into +/// an exception the node's normal error path already handles. +class ResourceClosedError : public std::runtime_error { +public: + ResourceClosedError() : std::runtime_error("shared resource closed") {} +}; + // ── SharedResource ──────────────────────────────────────────────────────────── // // Wraps an exclusive resource (e.g. an ONNX session, a CUDA stream) and @@ -72,6 +84,7 @@ public: template Guard acquire(PriorityFn&& fn) { std::unique_lock lock(mutex_); + if (closed_) throw ResourceClosedError{}; if (!held_) { held_ = true; acq_.fetch_add(1, std::memory_order_relaxed); @@ -83,18 +96,46 @@ public: current_waiters_.store(waiters_.size(), std::memory_order_relaxed); auto t0 = w.wait_start; - w.cv.wait(lock, [&w] { return w.ready; }); + // Woken either by release() handing over ownership, or by close() + // giving up on the wait entirely. + w.cv.wait(lock, [&w] { return w.ready || w.closed; }); int64_t wait_us = std::chrono::duration_cast( clock_t::now() - t0).count(); waiters_.erase(std::find(waiters_.begin(), waiters_.end(), &w)); current_waiters_.store(waiters_.size(), std::memory_order_relaxed); - acq_.fetch_add(1, std::memory_order_relaxed); total_wait_us_.fetch_add(static_cast(wait_us > 0 ? wait_us : 0), std::memory_order_relaxed); + + // Closed without being handed ownership: no Guard, so nothing to + // release, and held_ is left exactly as close() found it. + if (!w.ready) throw ResourceClosedError{}; + + acq_.fetch_add(1, std::memory_order_relaxed); return Guard(this); } + /// Wake every waiter and refuse further acquisitions. + /// + /// Teardown is the whole point: a node parked in acquire() is not + /// observing stop flags, so without this the only way out is for whoever + /// holds the resource to release it — which, if that node is also being + /// stopped, may never happen. Idempotent, and safe to call from any thread. + void close() override { + std::lock_guard lock(mutex_); + closed_ = true; + for (Waiter* w : waiters_) { + w->closed = true; + w->cv.notify_one(); + } + } + + /// Reopen after a close(). For reuse across runs; not needed for teardown. + void reopen() { + std::lock_guard lock(mutex_); + closed_ = false; + } + // Acquire with no priority (all waiters treated equally, order is fair-ish). Guard acquire() { return acquire([] { return 0.5f; }); @@ -132,7 +173,10 @@ public: private: void release() { std::unique_lock lock(mutex_); - if (waiters_.empty()) { + // Hand over only to a waiter that is still waiting. A closed one is on + // its way out and will not take ownership, so treating it as the next + // holder would leave held_ true with nobody holding it. + if (closed_ || waiters_.empty()) { held_ = false; return; } @@ -162,7 +206,8 @@ private: std::function priority_fn; clock_t::time_point wait_start; std::condition_variable cv; - bool ready{false}; + bool ready{false}; // handed ownership by release() + bool closed{false}; // woken by close() instead Waiter(std::function fn, clock_t::time_point t) : priority_fn(std::move(fn)), wait_start(t) {} @@ -172,6 +217,7 @@ private: T resource_; bool held_{false}; + bool closed_{false}; mutable std::mutex mutex_; std::vector waiters_; std::atomic acq_{0}; diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index bb34a2c..4eda22e 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(); } @@ -126,6 +128,15 @@ public: for (auto* node : user_nodes_topo_) node->set_network_error_callback(error_handler_); } + // Install every node's channel callbacks before starting any of them. + // Those callbacks are std::function members on channels shared with + // neighbours; a neighbour that is already running reads them from its + // own thread, so writing one after the pipeline is live is a data race + // (ThreadSanitizer reports it on any multi-node network). Doing all the + // writes here, while nothing runs, makes them read-only thereafter. + for (auto* n : user_nodes_topo_) n->prepare(); + for (auto* n : fanout_nodes_ptr_) n->prepare(); + for (auto* n : user_nodes_topo_) n->start(); for (auto* n : fanout_nodes_ptr_) n->start(); #ifdef KPN_WEB_DEBUG @@ -149,6 +160,11 @@ public: #ifdef KPN_WEB_DEBUG if (web_server_) web_server_->stop(); #endif + // Release anything parked on a shared resource first. A node blocked in + // acquire() is not watching stop flags, so stopping it would wait on a + // handover that may never come — its holder is being stopped too. + for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); } + for (auto it = fanout_nodes_ptr_.rbegin(); it != fanout_nodes_ptr_.rend(); ++it) (*it)->stop(); for (auto it = user_nodes_topo_.rbegin(); it != user_nodes_topo_.rend(); ++it) @@ -162,11 +178,12 @@ public: #ifdef KPN_WEB_DEBUG if (web_server_) web_server_->stop(); #endif + for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); } // 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(); } @@ -185,6 +202,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, @@ -265,16 +286,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_; @@ -285,6 +340,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_; @@ -321,29 +380,6 @@ auto make_network(Edges&&... edges) { // 4. Construct owned fanout storage on the heap (FanoutNode has jthread — not moveable) auto fanout_storage = std::make_unique(); - // 5. Collect unique user node pointers + their display names, in edge-declaration order - std::vector user_node_ptrs; - std::vector user_node_names; - auto collect = [&](auto& e) { - using SrcT = std::decay_t; - using DstT = std::decay_t; - auto* s = static_cast(&e.src); - auto* d = static_cast(&e.dst); - if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), s) == user_node_ptrs.end()) { - auto sname = node_display_name(); - user_node_ptrs.push_back(s); - user_node_names.push_back(sname); - s->set_name(sname); - } - if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), d) == user_node_ptrs.end()) { - auto dname = node_display_name(); - user_node_ptrs.push_back(d); - user_node_names.push_back(dname); - d->set_name(dname); - } - }; - (collect(edges), ...); - // 5. Wire all expanded SimpleEdges. // find_node: searches fanout storage then user edge pack, returns NodeT*. // Uses if constexpr in a fold so mismatched types never reach assignment. @@ -368,6 +404,38 @@ auto make_network(Edges&&... edges) { return ptr; }; + // 5. Collect user node pointers + display names in *topological* order. + // + // Topo is computed above for the cycle check and used to be discarded, + // while this vector was filled in edge-declaration order — and then named + // user_nodes_topo_ and relied upon as if it were sorted. halt() stops in + // its reverse, and shutdown() walks it forwards stopping each node and + // draining its outputs before the next, which is only a graceful drain if + // the order really is sources-first. It held for every network in the tree + // because edges happen to be declared in pipeline order, and would have + // broken silently for one that was not. + // + // Fanout nodes appear in Topo too; they are skipped here because they are + // owned separately, in fanout_storage. + std::vector user_node_ptrs; + std::vector user_node_names; + [&](tmp::TypeList) { + ([&]() { + if constexpr (!requires { NodeT::is_fanout_node; }) { + if (auto* p = find_node.template operator()()) { + auto* n = static_cast(p); + if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), n) + == user_node_ptrs.end()) { + auto nm = node_display_name(); + user_node_ptrs.push_back(n); + user_node_names.push_back(nm); + n->set_name(nm); + } + } + } + }.template operator()(), ...); + }(typename Topo::topo{}); + // Pre-pass: build fanout_id → source display name map so fanout nodes // can be named after the node feeding them (e.g. "capture_fanout"). std::map fanout_src_name; @@ -392,6 +460,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; @@ -409,6 +478,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()()); } }; @@ -436,7 +506,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/include/kpn/submit_gate.hpp b/include/kpn/submit_gate.hpp new file mode 100644 index 0000000..85cbce8 --- /dev/null +++ b/include/kpn/submit_gate.hpp @@ -0,0 +1,101 @@ +#pragma once +#include + +namespace kpn { + +// ── SubmitGate ──────────────────────────────────────────────────────────────── +// +// Decides, for one node, whether a wake must turn into a scheduler submission. +// Exactly one firing of a node may be in flight at a time, and a wake that +// arrives while one is already in flight must not be lost — it has to be +// honoured when that firing finishes, or the node sleeps holding work. +// +// 9c5ce5f wrote this as two independent atomics: queued_ said a firing was in +// flight, wake_pending_ recorded a wake that arrived during one. That cannot be +// made correct, because the release side has to read and write both, and a wake +// can land between the two operations: +// +// producer (try_submit) worker (release_and_recheck) +// ------------------------ ---------------------------- +// CAS reads queued_ == true, fails +// queued_.store(false) +// wake_pending_.exchange(false) -> false +// wake_pending_.store(true) +// +// End state: queued_ false, wake_pending_ true, nothing running and nothing +// scheduled. The node sleeps with a wake outstanding, which is precisely the +// invariant that commit set out to establish. It is not a memory-ordering +// subtlety — the interleaving above holds under seq_cst. +// +// It survived because every caller happened to follow release_and_recheck() +// with a level re-check (on_input_ready(), or outputs_have_space() on the +// parked path), which rediscovers the state a lost wake would have signalled. +// That is a property of the call sites, not of the mechanism, and any new early +// return that forgets the re-check turns it back into a hang. +// +// One atomic with three states makes the race unrepresentable: "idle" and "wake +// outstanding" are the same variable, so no interleaving can produce both. +// +// Idle nothing in flight +// Queued a firing is in flight or queued; no wake since it was claimed +// QueuedWake a firing is in flight or queued, and a wake arrived meanwhile +// +class SubmitGate { +public: + /// Register a wake. Returns true when the caller must submit the node; + /// false when a firing is already in flight and the wake has been recorded + /// against it instead. + bool claim() noexcept { + int cur = state_.load(std::memory_order_acquire); + for (;;) { + if (cur == kIdle) { + if (state_.compare_exchange_weak(cur, kQueued, + std::memory_order_acq_rel, std::memory_order_acquire)) + return true; + } else if (cur == kQueued) { + if (state_.compare_exchange_weak(cur, kQueuedWake, + std::memory_order_acq_rel, std::memory_order_acquire)) + return false; + } else { + return false; // a wake is already recorded + } + } + } + + /// End the in-flight firing. Returns true when a wake arrived during it and + /// the caller must submit again — in which case the gate stays claimed, so + /// the node is handed straight from one firing to the next and is never + /// momentarily idle with work outstanding. Returns false when the node is + /// now idle. + bool release() noexcept { + int cur = state_.load(std::memory_order_acquire); + for (;;) { + if (cur == kQueuedWake) { + if (state_.compare_exchange_weak(cur, kQueued, + std::memory_order_acq_rel, std::memory_order_acquire)) + return true; + } else { + // kQueued, or kIdle if a stop already forced the gate down. + if (state_.compare_exchange_weak(cur, kIdle, + std::memory_order_acq_rel, std::memory_order_acquire)) + return false; + } + } + } + + /// Drop the claim and any recorded wake. For stop paths only: honouring a + /// wake there would resubmit a dead node. + void force_idle() noexcept { state_.store(kIdle, std::memory_order_release); } + + bool queued() const noexcept { return state_.load(std::memory_order_relaxed) != kIdle; } + bool wake_pending() const noexcept { return state_.load(std::memory_order_relaxed) == kQueuedWake; } + +private: + static constexpr int kIdle = 0; + static constexpr int kQueued = 1; + static constexpr int kQueuedWake = 2; + + std::atomic state_{kIdle}; +}; + +} // namespace kpn 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/include/kpn/variant_node.hpp b/include/kpn/variant_node.hpp index 9a98599..26320ca 100644 --- a/include/kpn/variant_node.hpp +++ b/include/kpn/variant_node.hpp @@ -162,6 +162,7 @@ public: // ── INode ───────────────────────────────────────────────────────────────── + void prepare() override { node_.prepare(); } void start() override { node_.start(); } void stop() override { node_.stop(); } bool running() const override { return node_.running(); } 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/CMakeLists.txt b/tests/CMakeLists.txt index 9e43788..f567afc 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -36,6 +36,7 @@ add_executable(kpn_tests test_pool_node.cpp test_backpressure_deadlock.cpp test_scheduler.cpp + test_submit_gate.cpp ) target_link_libraries(kpn_tests PRIVATE diff --git a/tests/test_backpressure_deadlock.cpp b/tests/test_backpressure_deadlock.cpp index 17835ec..f6909d0 100644 --- a/tests/test_backpressure_deadlock.cpp +++ b/tests/test_backpressure_deadlock.cpp @@ -169,3 +169,253 @@ 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); +} + +// 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); +} diff --git a/tests/test_channel.cpp b/tests/test_channel.cpp index d710214..7be801e 100644 --- a/tests/test_channel.cpp +++ b/tests/test_channel.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -238,3 +239,87 @@ TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty", REQUIRE(out == 99); REQUIRE_FALSE(ch.try_pop_now(out)); // nothing left } + +// Regression: the sentinel slot holds one token and refuses a second. +// +// push_sentinel used to write eof_value_ unconditionally. Offering a second +// token before the first was taken therefore did two wrong things at once: it +// lost the first silently — and a lost EOF wedges every downstream pop forever +// — and it wrote the storage while the consumer could be moving the previous +// value out of it. For the shared_ptr storage that non-trivial types use, that +// is a torn refcount, not merely a stale read. +// +// Refusing is correct rather than queueing: two control tokens on one channel +// means the stream ended twice, which is a caller protocol error. Coalescing +// them would hide it, and there is no second value that could sensibly follow +// the end of a stream. +TEST_CASE("a second sentinel is refused, not swallowed", "[channel][sentinel]") { + Channel ch(4); + + REQUIRE(ch.push_sentinel(1)); + // Slot occupied: the first token is still undelivered. + REQUIRE_FALSE(ch.push_sentinel(2)); + + // The first survives intact — the overwrite is what used to lose it. + int out = 0; + REQUIRE(ch.try_pop_now(out)); + CHECK(out == 1); + + // And the slot is reusable once drained. + REQUIRE(ch.push_sentinel(3)); + REQUIRE(ch.try_pop_now(out)); + CHECK(out == 3); +} + +TEST_CASE("a refused sentinel is counted as a drop", "[channel][sentinel]") { + // Visibility matters more here than for a dropped value: the refusal means + // a control token went nowhere, and the only alternative to a counter is + // for it to vanish. + Channel ch(4); + REQUIRE(ch.push_sentinel(1)); + const auto before = ch.stats().drops.load(); + REQUIRE_FALSE(ch.push_sentinel(2)); + CHECK(ch.stats().drops.load() == before + 1); +} + +TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][sentinel]") { + // The non-consuming form exists so a refused token is still the caller's to + // report. The consuming push_sentinel cannot offer that, since the value is + // already moved into its parameter. + Channel ch(4); + std::string first = "eof-1", second = "eof-2"; + + REQUIRE(ch.try_push_sentinel(first) == Channel::SentinelResult::Taken); + REQUIRE(ch.try_push_sentinel(second) == Channel::SentinelResult::SlotBusy); + CHECK(second == "eof-2"); // not moved from + + ch.disable(); + std::string third = "eof-3"; + CHECK(ch.try_push_sentinel(third) == Channel::SentinelResult::Closed); + CHECK(third == "eof-3"); +} + +// Regression: try_push must distinguish delivered from discarded. +// +// It returned bool, and returned *true* for a closed channel — so "the value +// arrived" and "the value was thrown away because nobody is listening" were the +// same answer. Every caller was nonetheless correct, because both cases mean +// "stop trying"; but nothing above the channel could tell the two apart, and a +// producer counting successful pushes counted discards among them. Only the +// channel's own drop counter knew, and only if someone read the diagnostics. +TEST_CASE("try_push distinguishes taken, full and closed", "[channel]") { + Channel ch(2); + int v = 1; + + CHECK(ch.try_push(v) == Channel::PushResult::Taken); + CHECK(ch.try_push(v) == Channel::PushResult::Taken); + // Ring is full: the value is untouched and the caller keeps it. + CHECK(ch.try_push(v) == Channel::PushResult::Full); + CHECK(v == 1); + + ch.disable(); + const auto drops_before = ch.stats().drops.load(); + CHECK(ch.try_push(v) == Channel::PushResult::Closed); + // Discarded, and recorded as such rather than reported as a delivery. + CHECK(ch.stats().drops.load() == drops_before + 1); +} diff --git a/tests/test_channel_stress.cpp b/tests/test_channel_stress.cpp index eb58704..9f851e3 100644 --- a/tests/test_channel_stress.cpp +++ b/tests/test_channel_stress.cpp @@ -19,6 +19,7 @@ // Channel is SPSC: exactly one producer thread and one consumer thread per // channel. Every scenario below honours that contract. +#include #include #include #include @@ -279,3 +280,51 @@ TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)", REQUIRE(ch.approx_size() == 0); } } + +// Contended: a producer offering sentinels while the consumer takes them. +// +// The old push_sentinel wrote eof_value_ with no regard for whether the +// consumer was reading it, so a second offer racing a take was a data race on +// the storage — for the shared_ptr form used by non-trivial types, on the +// refcount. Under TSan the old code reports it; the handshake added alongside +// this test makes the producer's write conditional on observing the slot free, +// which is what serialises the two. +// +// Payload is a std::string so the storage is the shared_ptr path rather than +// the trivially-copyable one, and each token carries its own identity so a torn +// value shows up as a mismatch rather than as a plausible-looking result. +TEST_CASE("SPSC: offering sentinels concurrently with takes is race-free", + "[channel][stress][sentinel]") { + constexpr int kRounds = 20000; + Channel ch(4); + + std::atomic taken{0}; + std::atomic torn{false}; + std::atomic done{false}; + + std::thread consumer([&] { + std::string out; + while (!done.load(std::memory_order_acquire) || ch.approx_size() > 0) { + if (ch.try_pop_now(out)) { + if (out.rfind("eof-", 0) != 0) torn.store(true, std::memory_order_relaxed); + taken.fetch_add(1, std::memory_order_relaxed); + } + } + }); + + int accepted = 0; + for (int i = 0; i < kRounds; ++i) { + std::string tok = "eof-" + std::to_string(i); + if (ch.try_push_sentinel(tok) == Channel::SentinelResult::Taken) + ++accepted; + } + done.store(true, std::memory_order_release); + consumer.join(); + + INFO("accepted " << accepted << " taken " << taken.load()); + CHECK_FALSE(torn.load(std::memory_order_relaxed)); + // Every accepted token must be delivered: the slot is refused while full, + // so acceptance and delivery are one-to-one. + CHECK(taken.load(std::memory_order_relaxed) == accepted); + CHECK(accepted > 0); +} diff --git a/tests/test_network.cpp b/tests/test_network.cpp index 9d75f64..11d91c4 100644 --- a/tests/test_network.cpp +++ b/tests/test_network.cpp @@ -1,6 +1,10 @@ #include #include +#include #include +#include +#include +#include #include using namespace kpn; @@ -54,3 +58,93 @@ TEST_CASE("stop disables input channels — producer push is silently dropped", in_ch.push(99); REQUIRE(in_ch.size() == 0); } + +// Regression: Network::set_error_handler must actually deliver the handler. +// +// The handler was stored in a member and never read. A node's exception was +// discarded at the node boundary and the only surviving evidence was a Closed +// event, which reports that a node stopped but not why — the difference between +// a diagnosis and a guess. StaticNetwork has always wired this; Network +// accepted the handler and silently dropped it, which is worse than not +// offering the setter at all. +// +// The type changed with the fix. It was void(name, exception_ptr), which cannot +// express the keep-running decision the node side needs, so it is now +// NodeErrorHandler like StaticNetwork's. +namespace { + +static int throwing_stage(int x) { + if (x == 42) throw std::runtime_error("boom"); + return x; +} + +} // namespace + +TEST_CASE("network error handler receives the node's exception", "[network]") { + auto src = kpn::make_node(kpn::in<"v">{}, kpn::out<"w">{}, 8); + kpn::Channel out(8); + src.set_output_channel<0>(&out); + + kpn::Network net; + net.add("stage", src).build(); + + std::atomic calls{0}; + std::string seen_name; + std::string seen_what; + std::mutex mx; + + net.set_error_handler([&](std::string_view name, std::exception_ptr ep) { + std::lock_guard lk(mx); + seen_name = std::string(name); + try { if (ep) std::rethrow_exception(ep); } + catch (const std::exception& e) { seen_what = e.what(); } + calls.fetch_add(1, std::memory_order_relaxed); + return true; // handled: keep the node running + }); + + net.set_watchdog_interval(std::chrono::hours(1)); // keep the report quiet + net.start(); + src.input_channel<0>().push(42); // throws + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + src.input_channel<0>().push(7); // must still be running + const int passed = out.pop(); + net.stop(); + + CHECK(calls.load(std::memory_order_relaxed) == 1); + CHECK(seen_name == "stage"); + CHECK(seen_what == "boom"); + CHECK(passed == 7); +} + +// Regression: stopping a network must not wait for the watchdog's next tick. +// +// The watchdog looped on std::this_thread::sleep_for(watchdog_interval_), and +// request_stop() cannot wake a sleeping thread — so stop_watchdog()'s join +// blocked until the current sleep expired. Every teardown paid up to a full +// interval, three seconds by default, and a caller who set a long one to keep +// the periodic report quiet got a stop() that looked like a hang. That is how +// this was found: the error-handler case above set an hour. +TEST_CASE("stopping a network does not wait for the watchdog interval", "[network]") { + auto node = kpn::make_node(kpn::in<"v">{}, kpn::out<"w">{}, 4); + kpn::Channel out(4); + node.set_output_channel<0>(&out); + + kpn::Network net; + net.add("inc", node).build(); + net.set_watchdog_interval(std::chrono::hours(1)); + net.start(); + + // Let the watchdog actually reach its wait. Without this the test races it: + // stop_watchdog() runs before the thread has entered the loop, the token is + // already set when it does, and it exits without ever waiting — which passes + // against the bug as well as the fix. + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + const auto t0 = std::chrono::steady_clock::now(); + net.stop(); + const auto ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0).count(); + + INFO("stop took " << ms << " ms"); + CHECK(ms < 2000); +} diff --git a/tests/test_pool_node.cpp b/tests/test_pool_node.cpp index 779572b..9ec8b34 100644 --- a/tests/test_pool_node.cpp +++ b/tests/test_pool_node.cpp @@ -478,3 +478,245 @@ TEST_CASE("per-node and network overflow callbacks both fire independently", "[p REQUIRE(per_node.load() == 0); REQUIRE(network.load() == 0); } + +// Regression: NodeSnapshot's fields must line up with what nodes initialise. +// +// The snapshot is an aggregate that every node type fills positionally, and +// a8cfe73 appended queued/wake_pending/total_exec_ms to it in an order no call +// site used: each node supplies total_exec_ms as the element straight after +// queue_wait_ms, but the struct declared the two bools there. So the exec total +// landed in `queued`, `queued` landed in `wake_pending`, and `wake_pending` +// landed in total_exec_ms. The compiler said so (-Wnarrowing, bool to double, +// once per node instantiation) and the build carried on. +// +// It matters more than a cosmetic mix-up: these three fields exist to diagnose a +// wedge, and a wedged pipeline reported total_exec_ms as 0 or 1 and `queued` as +// "did this node ever run". Reading them would have pointed at the wrong node. +// +// Asserted against ema_exec_ms because that field is independently computed and +// was already correct: a true sum over several frames cannot be below the +// exponentially-weighted average of the same samples. +TEST_CASE("node snapshot fields line up with the values nodes supply", + "[pool_node][diagnostics]") { + auto pool = std::make_shared(1); + pool->start(); + + auto node = make_pool_node(pool, 64); + Channel out(64); + node.set_output_channel<0>(&out); + node.start(); + + for (int i = 0; i < 8; ++i) node.input_channel<0>().push(i); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + auto snap = node.node_snapshot("n", 1.0); + node.stop(); + pool->stop(); + + INFO("frames=" << snap.frames_processed + << " ema=" << snap.ema_exec_ms + << " total=" << snap.total_exec_ms); + REQUIRE(snap.frames_processed == 8); + // The mis-ordered aggregate put wake_pending here, so this was 0.0 or 1.0. + CHECK(snap.total_exec_ms >= snap.ema_exec_ms); + // ...and the exec total here, which is non-zero, so `queued` read true for + // any node that had ever run — including one asleep with nothing to do. + CHECK_FALSE(snap.queued); + CHECK_FALSE(snap.wake_pending); +} + +// Regression: a value parked twice must keep its payload. +// +// push_outputs ends with +// +// else pending_ = std::move(result); +// +// and the retry path calls it as push_outputs(std::move(*pending_), …), so on +// that path `result` is the parked tuple itself. The assignment was therefore a +// self-move-assignment. std::tuple's is elementwise, and libstdc++'s +// std::vector does not guard against self-move: it swaps its data into a +// temporary and leaves the vector empty. So the first park was clean (the +// argument is a local temporary) and the second erased the payload. +// +// The value was still delivered, still in order, still counted — just empty. +// Downstream cannot distinguish that from a frame on which the node genuinely +// found nothing, which is why it never surfaced as an error: in +// scene-actor-extraction it reads as "no faces in this frame" and the run +// completes with a quietly wrong answer. +// +// Reaching it needs *two* outputs. With one, the only thing that resubmits a +// parked node is that output's own space callback, which by definition fires +// when there is room — so the retry always succeeds and never reassigns. With +// two, output A draining resubmits the node while output B is still full: the +// retry skips A (already delivered, tracked in pending_done_) and fails on B, +// and that is the reassignment that eats B's payload. +// +// Driven through raw channels rather than consumer nodes so each step is +// forced rather than raced: B is pre-filled and stays full for exactly as long +// as the test wants it to. +namespace { + +struct TwoPayloads { + static constexpr std::string_view label() { return "two_payloads"; } + std::tuple, std::vector> operator()() { + return {std::vector(4, 1), std::vector(4, 2)}; + } +}; + +} // namespace + +TEST_CASE("a twice-parked value keeps its payload", "[pool_node][backpressure]") { + auto pool = std::make_shared(2); + pool->start(); + + TwoPayloads fn; + auto node = make_pool_node(fn, pool); + + // Both capacity 1. A must be *full* for its pop to signal space at all — + // Channel fires the space callback only on the full->not-full edge, so a + // roomy A would never resubmit the node and the retry would never happen. + Channel> out_a(1), out_b(1); + node.set_output_channel<0>(&out_a); + node.set_output_channel<1>(&out_b); + + // B is full before the node ever runs, so the very first firing parks. + out_b.push(std::vector(4, 99)); + + node.start(); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + // Draining A resubmits the node while B is still full: this is the retry + // that reassigned the tuple to itself. + (void)out_a.pop(); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + // Now let B through and collect what the node had been holding for it. + (void)out_b.pop(); // the pre-fill + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + std::vector parked = out_b.pop(); // the value parked across two tries + + node.stop(); + pool->stop(); + + INFO("parked payload size " << parked.size()); + 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(); +} + +// Regression: stop() must not return while a firing is still running. +// +// stop() set the flag, disabled the inputs and returned, leaving an executing +// fire_once touching input_channels_, stats_ and pending_ while the caller went +// on to destroy them. The old comment was explicit that callers wanting the +// guarantee should call scheduler_->drain() first — but ~PoolNode calls stop(), +// and a destructor cannot ask its caller to have done that. +// +// A node with a private pool survived by accident: Node::stop() calls +// pool->stop(), which joins the worker. A node sharing a pool, which +// make_pool_node exists to create, had nothing joining it at all, so its own +// destructor raced the firing. +// +// Asserted through an observable side effect rather than by trying to catch the +// use-after-free: if stop() returns before the node function has finished, the +// flag it sets on the way out is still false. +namespace { + +struct SlowFiring { + static constexpr std::string_view label() { return "slow_firing"; } + std::atomic* entered; + std::atomic* finished; + void operator()(int) { + entered->store(true, std::memory_order_release); + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + finished->store(true, std::memory_order_release); + } +}; + +} // namespace + +TEST_CASE("stop waits for a firing already in flight", "[pool_node]") { + std::atomic entered{false}, finished{false}; + + auto pool = std::make_shared(2); + pool->start(); + + SlowFiring fn{&entered, &finished}; + auto node = make_pool_node(fn, pool, 4); + node.start(); + node.input_channel<0>().push(1); + + // Stop only once the node is demonstrably inside its function. + while (!entered.load(std::memory_order_acquire)) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + + node.stop(); + CHECK(finished.load(std::memory_order_acquire)); + + pool->stop(); +} diff --git a/tests/test_scheduler.cpp b/tests/test_scheduler.cpp index 18f3fdd..9af89b9 100644 --- a/tests/test_scheduler.cpp +++ b/tests/test_scheduler.cpp @@ -227,3 +227,116 @@ TEST_CASE("work stealing: tasks complete with more threads than initial queue ta REQUIRE(counter.load() == 4); pool.stop(); } + +// Regression: submitting to a stopped pool must be a no-op, not a segfault. +// +// stop() ends with queues_.clear(), and submit() went straight to +// queues_[target] with no check — so a submission arriving after stop indexed +// an empty vector. +// +// This is not a contrived teardown ordering; it happens on a normal path. A +// node's space callback fires from whichever thread drained the channel, and +// that thread belongs to the *consumer*. Stop the producer first — which a +// sources-first shutdown does by design — and the consumer keeps draining its +// backlog, firing the producer's space callback into a pool that has already +// been torn down. Before this fix the static-network shutdown case crashed +// about 12 runs in 20. +// +// Checking stopped_ without the lock would not be enough: the window between +// reading the flag and indexing the vector is exactly where clear() runs. +TEST_CASE("submitting to a stopped pool is refused, not fatal", "[scheduler]") { + ThreadPool pool(2); + pool.start(); + pool.stop(); + + std::atomic ran{0}; + for (int i = 0; i < 10; ++i) + pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); }); + + CHECK(ran.load(std::memory_order_relaxed) == 0); + CHECK(pool.rejected() == 10); +} + +TEST_CASE("submitting while the pool stops does not crash", "[scheduler]") { + // The racing form of the case above: a producer thread submitting + // continuously while stop() runs underneath it. Nothing is asserted about + // how many tasks run — the point is that every submission either enqueues + // or is refused, and none touches a destroyed queue. + for (int rep = 0; rep < 20; ++rep) { + ThreadPool pool(4); + pool.start(); + + std::atomic go{false}; + std::atomic ran{0}; + std::thread submitter([&] { + while (!go.load(std::memory_order_acquire)) {} + for (int i = 0; i < 2000; ++i) + pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); }); + }); + + go.store(true, std::memory_order_release); + std::this_thread::sleep_for(std::chrono::microseconds(200)); + pool.stop(); + submitter.join(); + + // Everything submitted was either executed or refused; nothing vanished + // into a queue that no longer existed. + CHECK(pool.rejected() + pool.snapshot("p").tasks_completed <= 2000); + } +} + +// Regression: idle workers must sleep while another worker is busy. +// +// The wait predicate was `stopped_ || total_ > 0`, and total_ counts queued +// *plus executing*. So while any one task ran, every other worker's predicate +// was true: wait() returned immediately and the worker spun through try_pop, +// try_steal and back to wait at full speed — try_lock-ing every peer queue on +// each pass. One slow task pinned every other core and contended the very +// mutexes the working thread needed to make progress. +// +// That is the shape of this pipeline's load exactly: a handful of nodes whose +// work is tens of milliseconds of ONNX inference. It was latent only because +// each node currently owns a private single-thread pool, where there is no +// idle peer to spin. Any use of a shared pool — which make_pool_node exists +// for — hits it immediately. +// +// Measured as CPU time rather than wall time, because the bug does not make +// anything slower to finish; it makes seven cores burn while one works. A +// sleeping task consumes no CPU, so with workers correctly asleep the whole +// pool should account for almost none. +TEST_CASE("idle workers do not spin while one task runs", "[scheduler]") { + auto cpu_ms = [] { + struct timespec ts{}; + clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &ts); + return ts.tv_sec * 1000.0 + ts.tv_nsec / 1e6; + }; + + constexpr int kThreads = 8; + constexpr int kWorkMs = 300; + + ThreadPool pool(kThreads); + pool.start(); + std::this_thread::sleep_for(20ms); // let workers reach the wait + + const double before = cpu_ms(); + + // One long task plus a trivial one per remaining worker. The trivial ones + // matter: a worker that has never been woken stays blocked in wait() and + // never re-evaluates the predicate, so the spin only appears once a worker + // *finishes* something and re-enters the loop while a peer is still busy. + // Submitting only the long task does not reproduce it. + pool.submit([&] { std::this_thread::sleep_for(std::chrono::milliseconds(kWorkMs)); }); + for (int i = 0; i < kThreads - 1; ++i) pool.submit([] {}); + + pool.drain(); + const double used = cpu_ms() - before; + pool.stop(); + + // Measured on this tree: 1991 ms of CPU with the total_ predicate against + // 0.4 ms with queued_, and 19205 voluntary context switches against 10 — + // roughly (kThreads - 1) cores burned for the duration of one sleeping + // task. The threshold sits far from both so the case is not sensitive to + // how loaded the machine is. + INFO("cpu " << used << " ms over " << kWorkMs << " ms of sleeping work"); + CHECK(used < kWorkMs); +} diff --git a/tests/test_shared_resource.cpp b/tests/test_shared_resource.cpp index 7468f11..9e8abf6 100644 --- a/tests/test_shared_resource.cpp +++ b/tests/test_shared_resource.cpp @@ -237,3 +237,51 @@ TEST_CASE("make_shared_resource constructs with forwarded args", "[shared_resour auto g = res.acquire(); REQUIRE(*g == "hello"); } + +// Regression: a waiter must be releasable, or teardown waits on it forever. +// +// acquire() blocks on a condition variable whose predicate only becomes true +// when release() hands over ownership. There was no timeout and no stop +// condition, so a node parked there ignored teardown entirely: its worker never +// returned, the pool's join never completed, and shutdown hung waiting for a +// resource nobody was going to release — which is exactly the case when the +// holder is being stopped too. +// +// close() turns that into an exception the node's existing error path already +// handles, and networks now call it on registered resources before stopping any +// node, for the same reason. +TEST_CASE("closing a shared resource releases its waiters", "[shared_resource]") { + SharedResource res(42); + + auto holder = res.acquire(); // resource is now held + + std::atomic threw{false}, returned{false}; + std::thread waiter([&] { + try { + auto g = res.acquire(); // blocks: someone else holds it + (void)g; + } catch (const ResourceClosedError&) { + threw.store(true, std::memory_order_release); + } + returned.store(true, std::memory_order_release); + }); + + // Let it park, then tear down without ever releasing the holder. + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + REQUIRE_FALSE(returned.load(std::memory_order_acquire)); + + res.close(); + waiter.join(); + + CHECK(threw.load(std::memory_order_acquire)); +} + +TEST_CASE("acquiring a closed resource fails immediately", "[shared_resource]") { + SharedResource res(7); + res.close(); + CHECK_THROWS_AS(res.acquire(), ResourceClosedError); + + // Reusable across runs once reopened. + res.reopen(); + CHECK_NOTHROW(res.acquire()); +} diff --git a/tests/test_static_network.cpp b/tests/test_static_network.cpp index 3dc2b26..188de84 100644 --- a/tests/test_static_network.cpp +++ b/tests/test_static_network.cpp @@ -271,3 +271,164 @@ 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 +} + +// Regression: node order must come from the topological sort, not from the +// order the edges happened to be written in. +// +// make_network computes Topo for the cycle check and then dropped it, filling +// the node vector in edge-declaration order — and named it user_nodes_topo_. +// halt() stops in its reverse, and shutdown() walks it forwards stopping each +// node and draining its outputs before moving to the next, which is a graceful +// drain only if the order really is sources-first. +// +// Every network in this tree declares edges in pipeline order, so the two +// coincided and nothing failed. This case declares them backwards, which is +// legal and which make_network otherwise accepts silently. +// +// Asserted through shutdown() rather than by reading the order back, because +// the order is private and the ordering is not the point — what it buys is. +// A sources-first shutdown lets the values already in flight reach the sink; +// stopping the sink first strands them, and the drain step then has nobody +// left to take them. +namespace { + +struct OrderSource { + static constexpr std::string_view label() { return "order_source"; } + std::atomic* made; + int operator()() { + std::this_thread::sleep_for(std::chrono::microseconds(20)); + return made->fetch_add(1, std::memory_order_relaxed); + } +}; + +// Deliberately slower than the source, so a deep backlog builds up in its input +// channel. That backlog is what a sources-first shutdown preserves and a +// sink-first one throws away, and it needs to be big enough that the difference +// cannot be mistaken for one value in flight. +struct OrderRelay { + static constexpr std::string_view label() { return "order_relay"; } + int operator()(int v) { + std::this_thread::sleep_for(std::chrono::microseconds(300)); + return v; + } +}; + +struct OrderSink { + static constexpr std::string_view label() { return "order_sink"; } + std::atomic* seen; + void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); } +}; + +} // namespace + +TEST_CASE("edges declared out of order still start and stop sources-first", + "[static_network][shutdown]") { + std::atomic seen{0}, made{0}; + + OrderSource src_fn{&made}; + OrderRelay relay_fn; + OrderSink sink_fn{&seen}; + + kpn::ObjectNode, kpn::out<"v">, "order_source", 0> s(src_fn, 8); + kpn::ObjectNode, kpn::out<"w">, "order_relay", 0> r(relay_fn, 64); + kpn::ObjectNode, kpn::out<>, "order_sink", 0> k(sink_fn, 64); + + // Sink edge first, source edge last — the reverse of pipeline order. + auto net = kpn::make_network( + kpn::edge(r.output<"w">(), k.input<"w">()), + kpn::edge(s.output<"v">(), r.input<"v">()) + ); + net.start(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + const int before = seen.load(std::memory_order_relaxed); + REQUIRE(before > 0); // the pipeline ran at all + + net.shutdown(); + + // Sources stop first and each layer drains before the next stops, so the + // backlog queued in front of the relay still reaches the sink. Stopping in + // declaration order stops the relay first and discards all of it. + const int after = seen.load(std::memory_order_relaxed); + INFO("made " << made.load() << ", delivered " << before + << " before shutdown, " << after << " after"); + CHECK(after - before >= 20); +} diff --git a/tests/test_submit_gate.cpp b/tests/test_submit_gate.cpp new file mode 100644 index 0000000..b84ac90 --- /dev/null +++ b/tests/test_submit_gate.cpp @@ -0,0 +1,173 @@ +// Regression: a node must never end up idle with a wake outstanding. +// +// 9c5ce5f established that invariant and implemented it as two independent +// atomics — queued_ for "a firing is in flight", wake_pending_ for "a wake +// arrived during one". Two variables cannot express it, because the release +// side has to read and write both and a wake can land in between: +// +// producer (try_submit) worker (release_and_recheck) +// ------------------------ ---------------------------- +// CAS reads queued_ == true, fails +// queued_.store(false) +// wake_pending_.exchange(false) -> false +// wake_pending_.store(true) +// +// queued_ false, wake_pending_ true, nothing running and nothing scheduled. +// Not a memory-ordering subtlety: the interleaving holds under seq_cst. +// +// LegacyGate below is that protocol verbatim, with a hook between the failed +// CAS and the wake_pending_ store so the interleaving can be forced rather than +// waited for. That makes the loss deterministic and the test non-flaky, and it +// keeps the defect on record now that the code implementing it is gone. +// +// A note on what is NOT tested here, because it would be misleading to imply +// otherwise: there is no black-box, node-level test that fails before this fix +// and passes after. Every call site of release_and_recheck() happens to follow +// it with a level re-check — on_input_ready(), or outputs_have_space() on the +// parked path — which rediscovers the state a lost wake would have signalled. +// That masking is a property of the call sites, not of the mechanism, and the +// point of the fix is that a future early return that forgets the re-check no +// longer reintroduces a hang. The value is structural, so the tests are +// structural: the state machine is pinned by contract, and the defect it +// replaces is pinned by demonstration. +#include +#include + +#include +#include +#include + +using namespace kpn; + +namespace { + +// The pre-fix protocol, with a seam at the point where the race lives. +class LegacyGate { +public: + std::function before_recording_wake; + + bool claim() noexcept { + bool expected = false; + if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + return true; + if (before_recording_wake) before_recording_wake(); + wake_pending_.store(true, std::memory_order_release); + return false; + } + bool release() noexcept { + queued_.store(false, std::memory_order_release); + if (wake_pending_.exchange(false, std::memory_order_acq_rel)) { + bool expected = false; + if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + return true; + } + return false; + } + bool queued() const noexcept { return queued_.load(std::memory_order_relaxed); } + bool wake_pending() const noexcept { return wake_pending_.load(std::memory_order_relaxed); } + +private: + std::atomic queued_{false}; + std::atomic wake_pending_{false}; +}; + +} // namespace + +TEST_CASE("the two-atomic gate loses a wake, deterministically", "[submit_gate]") { + LegacyGate gate; + bool resubmitted = true; + + REQUIRE(gate.claim()); // a firing is now in flight + + // Force the interleaving: the firing completes in the window between the + // second wake's failed CAS and its record of that wake. + gate.before_recording_wake = [&] { resubmitted = gate.release(); }; + + const bool submitted = gate.claim(); + + // The wake was neither submitted by the producer nor honoured by the + // release. Nothing is scheduled, and nothing else will re-trigger it. + CHECK_FALSE(submitted); + CHECK_FALSE(resubmitted); + CHECK_FALSE(gate.queued()); + CHECK(gate.wake_pending()); // recorded, and never to be consumed +} + +TEST_CASE("submit gate: a wake during a firing is honoured", "[submit_gate]") { + SubmitGate gate; + + REQUIRE(gate.claim()); // idle -> queued, caller submits + REQUIRE(gate.queued()); + REQUIRE_FALSE(gate.wake_pending()); + + REQUIRE_FALSE(gate.claim()); // second wake is recorded, not submitted + REQUIRE(gate.wake_pending()); + + REQUIRE(gate.release()); // and honoured when the firing ends + // The gate stays claimed across the handover, so the node is never + // momentarily idle while a submission for it is in flight. This is the + // state the legacy gate could not represent. + REQUIRE(gate.queued()); + REQUIRE_FALSE(gate.wake_pending()); + + REQUIRE_FALSE(gate.release()); // no further wake: now idle + REQUIRE_FALSE(gate.queued()); +} + +TEST_CASE("submit gate: repeated wakes collapse to one resubmission", "[submit_gate]") { + // Collapsing is deliberate. A firing consumes one item and its caller then + // re-checks the input level, so the gate only has to guarantee that at + // least one more firing follows a wake, not one per wake. + SubmitGate gate; + REQUIRE(gate.claim()); + for (int i = 0; i < 10; ++i) REQUIRE_FALSE(gate.claim()); + REQUIRE(gate.release()); + REQUIRE_FALSE(gate.release()); +} + +TEST_CASE("submit gate: force_idle drops a recorded wake", "[submit_gate]") { + // Stop paths use this deliberately — honouring a wake there would resubmit + // a node that has already been told to stop. + SubmitGate gate; + REQUIRE(gate.claim()); + REQUIRE_FALSE(gate.claim()); + REQUIRE(gate.wake_pending()); + + gate.force_idle(); + REQUIRE_FALSE(gate.queued()); + REQUIRE_FALSE(gate.wake_pending()); + REQUIRE(gate.claim()); // and the gate is reusable afterwards +} + +TEST_CASE("submit gate: concurrent claim and release stay consistent", "[submit_gate]") { + // Not a lost-wake test — see the header note. This is a TSan target and a + // check that the CAS loops always terminate and always leave the gate in a + // reachable state: exactly one party may hold the claim at a time, so the + // count of claims granted must equal the count of releases that ended idle. + SubmitGate gate; + std::atomic granted{0}, ended_idle{0}; + std::atomic stop{false}; + + std::thread waker([&] { + while (!stop.load(std::memory_order_relaxed)) + if (gate.claim()) granted.fetch_add(1, std::memory_order_relaxed); + }); + std::thread worker([&] { + while (!stop.load(std::memory_order_relaxed)) + if (gate.queued() && !gate.release()) + ended_idle.fetch_add(1, std::memory_order_relaxed); + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + stop.store(true, std::memory_order_relaxed); + waker.join(); + worker.join(); + + // Drain whatever claim is outstanding so the two counts can be compared. + while (gate.queued()) + if (!gate.release()) ended_idle.fetch_add(1, std::memory_order_relaxed); + + INFO("granted " << granted.load() << " ended idle " << ended_idle.load()); + REQUIRE(granted.load() > 0); + CHECK(granted.load() == ended_idle.load()); +}