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
dtourolle 75b34f31bb Merge pull request 'feat: persistent-pipeline reuse — push_blocking, node introspection, stateful wrapper' (#2) from feature/persistent-pipeline-reuse into master
🚦 CI / changes (push) Successful in 4s
🚦 CI / docker (push) Has been skipped
🚦 CI / test (push) Successful in 4m39s
🚦 CI / tsan (push) Successful in 2m57s
🚦 CI / docs (push) Has been skipped
Reviewed-on: #2
2026-07-19 16:11:27 +00:00
+11 -12
View File
@@ -400,12 +400,15 @@ private:
ch->push_sentinel(std::move(val));
return;
}
try {
ch->push(std::move(val));
} catch (const ChannelOverflowError&) {
throw ChannelOverflowError(ch->capacity(),
"pool node '" + name_ + "' " + output_port_label<I>());
}
// Backpressure, not loss. A full downstream channel means the consumer
// is behind, and the correct response is for this producer to run
// slower — not to discard a value. A dropped frame does not degrade a
// result, it silently changes one, and the caller has no way to tell.
//
// 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));
}
template<std::size_t I>
@@ -706,12 +709,8 @@ private:
ch->push_sentinel(std::move(val));
return;
}
try {
ch->push(std::move(val));
} catch (const ChannelOverflowError&) {
throw ChannelOverflowError(ch->capacity(),
"pool node '" + name_ + "'");
}
// See the note on the typed overload above: block rather than drop.
ch->push_blocking(std::move(val));
}
Obj& obj_;