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;