fix: node outputs block instead of dropping on a full channel
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>
This commit is contained in:
+11
-12
@@ -400,12 +400,15 @@ private:
|
|||||||
ch->push_sentinel(std::move(val));
|
ch->push_sentinel(std::move(val));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
try {
|
// Backpressure, not loss. A full downstream channel means the consumer
|
||||||
ch->push(std::move(val));
|
// is behind, and the correct response is for this producer to run
|
||||||
} catch (const ChannelOverflowError&) {
|
// slower — not to discard a value. A dropped frame does not degrade a
|
||||||
throw ChannelOverflowError(ch->capacity(),
|
// result, it silently changes one, and the caller has no way to tell.
|
||||||
"pool node '" + name_ + "' " + output_port_label<I>());
|
//
|
||||||
}
|
// 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>
|
template<std::size_t I>
|
||||||
@@ -706,12 +709,8 @@ private:
|
|||||||
ch->push_sentinel(std::move(val));
|
ch->push_sentinel(std::move(val));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
try {
|
// See the note on the typed overload above: block rather than drop.
|
||||||
ch->push(std::move(val));
|
ch->push_blocking(std::move(val));
|
||||||
} catch (const ChannelOverflowError&) {
|
|
||||||
throw ChannelOverflowError(ch->capacity(),
|
|
||||||
"pool node '" + name_ + "'");
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
Obj& obj_;
|
Obj& obj_;
|
||||||
|
|||||||
Reference in New Issue
Block a user