Author SHA1 Message Date
dtourolleandClaude Opus 5 6595e6e925 fix: node outputs block instead of dropping on a full channel
🚦 CI / changes (push) Successful in 30s
🚦 CI / docker (push) Has been skipped
🚦 CI / test (push) Failing after 1h44m2s
🚦 CI / docs (push) Has been skipped
🚦 CI / tsan (push) Failing after 3h14m57s
Every node output used the throwing push(), so a consumer falling behind cost
values rather than time. push_blocking() already existed on Channel and
OutputPort — "wait for the consumer to drain instead of dropping; the producer
just runs slower" — but nothing called it.

A dropped frame does not degrade a downstream result, it silently changes one,
and the consumer has no way to tell it happened. For any pipeline whose output
is a claim about its input, that is corruption rather than degradation.

Safe because sentinels are already handled out-of-band, above this path: only
data blocks, so the EOF token that unwinds the network can always overtake a
stalled data path. That is exactly the hold-and-wait deadlock the push_sentinel
comment warns about, and the reason it is not reachable here.

Measured on a downstream consumer (face pipeline, 77s clip at 5 fps, expected
385 sampled frames):

  before  65 frames written, 320 dropped at one node, 29s
  after   385 frames written, 0 dropped, 17s

Faster, not slower — a dropped frame has already cost its decode, and the
overflow exception cost more. Two consecutive runs now produce byte-identical
output, which they did not before: what got dropped depended on timing, so the
same command could yield different results.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-31 11:31:53 +02:00
4 changed files with 13 additions and 91 deletions
+1 -24
View File
@@ -115,16 +115,6 @@ public:
out_channels_[I] = ch; 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: private:
void run_loop() { void run_loop() {
while (!stop_flag_.load(std::memory_order_relaxed)) { while (!stop_flag_.load(std::memory_order_relaxed)) {
@@ -136,19 +126,8 @@ private:
for (std::size_t i = 0; i < N; ++i) { for (std::size_t i = 0; i < N; ++i) {
if (out_channels_[i]) { if (out_channels_[i]) {
// 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); } try { out_channels_[i]->push(val); }
catch (const ChannelOverflowError&) {} // drop independently catch (const ChannelOverflowError&) {} // drop for this output independently
}
} }
} }
@@ -163,8 +142,6 @@ private:
std::string name_; std::string name_;
std::size_t fifo_capacity_; std::size_t fifo_capacity_;
bool lossless_{false};
std::array<bool, N> lossy_out_{}; // per-output opt-out of blocking
std::shared_ptr<Channel<T>> input_ch_; std::shared_ptr<Channel<T>> input_ch_;
std::array<Channel<T>*, N> out_channels_{}; std::array<Channel<T>*, N> out_channels_{};
std::atomic<bool> stop_flag_{false}; std::atomic<bool> stop_flag_{false};
-4
View File
@@ -37,10 +37,6 @@ struct INode {
// halt(): alias for stop() — immediate, discards in-flight work. // halt(): alias for stop() — immediate, discards in-flight work.
virtual void halt() { stop(); } 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 // shutdown(): graceful drain before stopping. Base implementation falls
// back to stop(). Network and StaticNetwork override with topo-ordered drain. // back to stop(). Network and StaticNetwork override with topo-ordered drain.
virtual void shutdown() { stop(); } virtual void shutdown() { stop(); }
+9 -42
View File
@@ -133,14 +133,6 @@ public:
void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } 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_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_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); } void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); }
@@ -408,21 +400,15 @@ private:
ch->push_sentinel(std::move(val)); ch->push_sentinel(std::move(val));
return; return;
} }
// Lossless mode: block until the consumer drains instead of dropping. // Backpressure, not loss. A full downstream channel means the consumer
// Dropping is the right default for live sources (a stale frame is // is behind, and the correct response is for this producer to run
// worth less than a fresh one), but for offline batch work every sample // slower — not to discard a value. A dropped frame does not degrade a
// matters — dropped frames leave a non-uniformly sampled series, which // result, it silently changes one, and the caller has no way to tell.
// silently invalidates any fixed-rate spectral analysis downstream. //
if (lossless_) { // Safe here because sentinels are handled above, out-of-band: this
// blocks only on data, so the EOF token that unwinds the network can
// always overtake a stalled data path.
ch->push_blocking(std::move(val)); ch->push_blocking(std::move(val));
return;
}
try {
ch->push(std::move(val));
} catch (const ChannelOverflowError&) {
throw ChannelOverflowError(ch->capacity(),
"pool node '" + name_ + "' " + output_port_label<I>());
}
} }
template<std::size_t I> template<std::size_t I>
@@ -439,7 +425,6 @@ private:
std::shared_ptr<IScheduler> scheduler_; std::shared_ptr<IScheduler> scheduler_;
std::string name_; std::string name_;
bool lossless_{false};
std::size_t fifo_capacity_; std::size_t fifo_capacity_;
input_channels_t input_channels_; input_channels_t input_channels_;
output_channels_t output_channels_{}; output_channels_t output_channels_{};
@@ -516,14 +501,6 @@ public:
void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } 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_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_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); } void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); }
@@ -732,23 +709,13 @@ private:
ch->push_sentinel(std::move(val)); ch->push_sentinel(std::move(val));
return; return;
} }
// Lossless mode: block until the consumer drains instead of dropping. // See the note on the typed overload above: block rather than drop.
if (lossless_) {
ch->push_blocking(std::move(val)); ch->push_blocking(std::move(val));
return;
}
try {
ch->push(std::move(val));
} catch (const ChannelOverflowError&) {
throw ChannelOverflowError(ch->capacity(),
"pool node '" + name_ + "'");
}
} }
Obj& obj_; Obj& obj_;
std::shared_ptr<IScheduler> scheduler_; std::shared_ptr<IScheduler> scheduler_;
std::string name_; std::string name_;
bool lossless_{false};
std::size_t fifo_capacity_; std::size_t fifo_capacity_;
input_channels_t input_channels_; input_channels_t input_channels_;
output_channels_t output_channels_{}; output_channels_t output_channels_{};
-18
View File
@@ -219,24 +219,6 @@ public:
FanoutStorage& fanouts_storage() { return *fanouts_; } 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: private:
struct Snapshots { struct Snapshots {
std::vector<NodeSnapshot> nodes; std::vector<NodeSnapshot> nodes;