Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6595e6e925 |
+1
-24
@@ -115,16 +115,6 @@ 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)) {
|
||||
@@ -136,19 +126,8 @@ private:
|
||||
|
||||
for (std::size_t i = 0; i < N; ++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); }
|
||||
catch (const ChannelOverflowError&) {} // drop independently
|
||||
}
|
||||
catch (const ChannelOverflowError&) {} // drop for this output independently
|
||||
}
|
||||
}
|
||||
|
||||
@@ -163,8 +142,6 @@ private:
|
||||
|
||||
std::string name_;
|
||||
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::array<Channel<T>*, N> out_channels_{};
|
||||
std::atomic<bool> stop_flag_{false};
|
||||
|
||||
@@ -37,10 +37,6 @@ 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(); }
|
||||
|
||||
@@ -133,14 +133,6 @@ 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); }
|
||||
@@ -408,21 +400,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_) {
|
||||
// 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));
|
||||
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>
|
||||
@@ -439,7 +425,6 @@ private:
|
||||
|
||||
std::shared_ptr<IScheduler> scheduler_;
|
||||
std::string name_;
|
||||
bool lossless_{false};
|
||||
std::size_t fifo_capacity_;
|
||||
input_channels_t input_channels_;
|
||||
output_channels_t output_channels_{};
|
||||
@@ -516,14 +501,6 @@ 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); }
|
||||
@@ -732,23 +709,13 @@ private:
|
||||
ch->push_sentinel(std::move(val));
|
||||
return;
|
||||
}
|
||||
// Lossless mode: block until the consumer drains instead of dropping.
|
||||
if (lossless_) {
|
||||
// See the note on the typed overload above: block rather than drop.
|
||||
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_;
|
||||
std::shared_ptr<IScheduler> scheduler_;
|
||||
std::string name_;
|
||||
bool lossless_{false};
|
||||
std::size_t fifo_capacity_;
|
||||
input_channels_t input_channels_;
|
||||
output_channels_t output_channels_{};
|
||||
|
||||
@@ -219,24 +219,6 @@ 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<NodeSnapshot> nodes;
|
||||
|
||||
Reference in New Issue
Block a user