From 4e81752838af0ce1fc5599c9e6cc628d48b2765e Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Sat, 25 Jul 2026 23:03:56 +0200 Subject: [PATCH] feat: opt-in lossless (blocking) output for nodes and fanouts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Channels drop on overflow by default. That is the right behaviour for live sources, where a stale item is worth less than a fresh one, but it makes the library unusable for offline batch work: items vanish with no diagnostic, and any downstream analysis that assumes a fixed sample rate is silently invalid. Auto-inserted fanouts were the harder half of this. They are created inside make_network(), so user code cannot reach them to configure, and they drop per-output inside a swallowed catch — so a pipeline whose own nodes were all configured lossless could still lose items with nothing reported anywhere. In the pipeline this came from, capture's 2505 frames arrived at the detector as 774 while the overflow counter read zero. Adds: - INode::set_lossless_output(bool), defaulted to a no-op so node types with no output channels ignore it - PoolNode / PoolObjectNode: route push_one_out() through push_blocking() - FanoutNode: same, plus set_lossy_output(i) to opt a single branch back out - StaticNetwork::set_lossless(), which reaches user nodes and fanouts alike - StaticNetwork::drain(), a public wrapper over the existing private drain_all_channels(), so callers can flush in-flight work before stop() Default behaviour is unchanged; every path is off unless explicitly enabled. Blocking output is only safe when every consumer eventually drains. A branch that can stall indefinitely — a display node nobody is servicing — will apply backpressure to the whole pipeline, which is what set_lossy_output() is for. Co-Authored-By: Claude Opus 5 --- include/kpn/fanout.hpp | 27 +++++++++++++++++++++++++-- include/kpn/inode.hpp | 4 ++++ include/kpn/pool_node.hpp | 27 +++++++++++++++++++++++++++ include/kpn/static_network.hpp | 18 ++++++++++++++++++ 4 files changed, 74 insertions(+), 2 deletions(-) diff --git a/include/kpn/fanout.hpp b/include/kpn/fanout.hpp index 180b602..39bf9be 100644 --- a/include/kpn/fanout.hpp +++ b/include/kpn/fanout.hpp @@ -115,6 +115,16 @@ public: out_channels_[I] = ch; } + // Lossless fanout: block until each consumer drains rather than dropping. + void set_lossless_output(bool on) override { lossless_ = on; } + + // Opt a single output back out of blocking. Needed when one branch may + // stall indefinitely — a display tap nobody is servicing, say — since + // blocking on it would apply backpressure to every other branch too. + void set_lossy_output(std::size_t i, bool lossy = true) { + if (i < N) lossy_out_[i] = lossy; + } + private: void run_loop() { while (!stop_flag_.load(std::memory_order_relaxed)) { @@ -126,8 +136,19 @@ private: 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 + // Lossless: block until this consumer drains. Note the + // branches differ in more than blocking — the dropping + // path discards per-output independently and silently, + // so a slow consumer on one branch costs frames on that + // branch only, with no diagnostic. That is the right + // default for display taps but hides frame loss from + // analysis branches. + if (lossless_ && !lossy_out_[i]) + out_channels_[i]->push_blocking(val); + else { + try { out_channels_[i]->push(val); } + catch (const ChannelOverflowError&) {} // drop independently + } } } @@ -142,6 +163,8 @@ private: std::string name_; std::size_t fifo_capacity_; + bool lossless_{false}; + std::array lossy_out_{}; // per-output opt-out of blocking std::shared_ptr> input_ch_; std::array*, N> out_channels_{}; std::atomic stop_flag_{false}; diff --git a/include/kpn/inode.hpp b/include/kpn/inode.hpp index 56f4b89..acc23b1 100644 --- a/include/kpn/inode.hpp +++ b/include/kpn/inode.hpp @@ -37,6 +37,10 @@ struct INode { // halt(): alias for stop() — immediate, discards in-flight work. virtual void halt() { stop(); } + // Opt into lossless (blocking) output for nodes that support it. Default + // is a no-op so node types with no output channels ignore it. + virtual void set_lossless_output(bool) {} + // shutdown(): graceful drain before stopping. Base implementation falls // back to stop(). Network and StaticNetwork override with topo-ordered drain. virtual void shutdown() { stop(); } diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 4fe37f2..96eac9d 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -133,6 +133,14 @@ public: void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } + // Lossless output: block the producer until the consumer drains rather than + // dropping on a full channel. Default is drop, which suits live sources + // where a stale frame is worth less than a fresh one. Enable for offline + // batch runs, where a dropped item leaves a gap that downstream analysis + // cannot recover. Must be set before start(). + void set_lossless_output(bool on) override { lossless_ = on; } + void set_lossless(bool on) { set_lossless_output(on); } + void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); } void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); } void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); } @@ -400,6 +408,15 @@ private: ch->push_sentinel(std::move(val)); return; } + // Lossless mode: block until the consumer drains instead of dropping. + // Dropping is the right default for live sources (a stale frame is + // worth less than a fresh one), but for offline batch work every sample + // matters — dropped frames leave a non-uniformly sampled series, which + // silently invalidates any fixed-rate spectral analysis downstream. + if (lossless_) { + ch->push_blocking(std::move(val)); + return; + } try { ch->push(std::move(val)); } catch (const ChannelOverflowError&) { @@ -422,6 +439,7 @@ private: std::shared_ptr scheduler_; std::string name_; + bool lossless_{false}; std::size_t fifo_capacity_; input_channels_t input_channels_; output_channels_t output_channels_{}; @@ -498,6 +516,14 @@ public: void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } + // Lossless output: block the producer until the consumer drains rather than + // dropping on a full channel. Default is drop, which suits live sources + // where a stale frame is worth less than a fresh one. Enable for offline + // batch runs, where a dropped item leaves a gap that downstream analysis + // cannot recover. Must be set before start(). + void set_lossless_output(bool on) override { lossless_ = on; } + void set_lossless(bool on) { set_lossless_output(on); } + void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); } void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); } void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); } @@ -717,6 +743,7 @@ private: Obj& obj_; std::shared_ptr scheduler_; std::string name_; + bool lossless_{false}; std::size_t fifo_capacity_; input_channels_t input_channels_; output_channels_t output_channels_{}; diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index eeb5304..4c7fc1b 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -219,6 +219,24 @@ public: FanoutStorage& fanouts_storage() { return *fanouts_; } + // Make every node in this network push losslessly (block until the consumer + // drains) instead of dropping on a full channel. Includes the fanout nodes + // make_network() inserts automatically, which is the part user code cannot + // reach: they are unnamed, and they drop silently per-output, so a network + // whose own nodes are all lossless can still lose items at a fanout. + // + // Only safe when every consumer eventually drains. A branch that can stall + // indefinitely — a display node nobody is servicing, say — will block the + // whole pipeline through backpressure. Call before start(). + void set_lossless(bool on = true) { + for (auto* n : user_nodes_topo_) if (n) n->set_lossless_output(on); + for (auto* n : fanout_nodes_ptr_) if (n) n->set_lossless_output(on); + } + + // Block until every channel is empty. Useful before stop() so work already + // in flight completes rather than being discarded at teardown. + void drain() const { drain_all_channels(); } + private: struct Snapshots { std::vector nodes;