dtourolleandClaude Opus 5
4e81752838
Channels drop on overflow by default. That is the right behaviour for live
sources, where a stale item is worth less than a fresh one, but it makes the
library unusable for offline batch work: items vanish with no diagnostic, and
any downstream analysis that assumes a fixed sample rate is silently invalid.
Auto-inserted fanouts were the harder half of this. They are created inside
make_network(), so user code cannot reach them to configure, and they drop
per-output inside a swallowed catch — so a pipeline whose own nodes were all
configured lossless could still lose items with nothing reported anywhere. In
the pipeline this came from, capture's 2505 frames arrived at the detector as
774 while the overflow counter read zero.
Adds:
- INode::set_lossless_output(bool), defaulted to a no-op so node types with
no output channels ignore it
- PoolNode / PoolObjectNode: route push_one_out() through push_blocking()
- FanoutNode: same, plus set_lossy_output(i) to opt a single branch back out
- StaticNetwork::set_lossless(), which reaches user nodes and fanouts alike
- StaticNetwork::drain(), a public wrapper over the existing private
drain_all_channels(), so callers can flush in-flight work before stop()
Default behaviour is unchanged; every path is off unless explicitly enabled.
Blocking output is only safe when every consumer eventually drains. A branch
that can stall indefinitely — a display node nobody is servicing — will apply
backpressure to the whole pipeline, which is what set_lossy_output() is for.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
817 lines
35 KiB
C++
817 lines
35 KiB
C++
#pragma once
|
|
#include "channel.hpp"
|
|
#include "diagnostics.hpp"
|
|
#include "fixed_string.hpp"
|
|
#include "inode.hpp"
|
|
#include "port.hpp"
|
|
#include "scheduler.hpp"
|
|
#include "traits.hpp"
|
|
|
|
#include <array>
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstddef>
|
|
#include <functional>
|
|
#include <iostream>
|
|
#include <memory>
|
|
#include <optional>
|
|
#include <stdexcept>
|
|
#include <thread>
|
|
#include <tuple>
|
|
#include <type_traits>
|
|
|
|
namespace kpn {
|
|
|
|
// ── Sentinel detection ────────────────────────────────────────────────────────
|
|
// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
|
|
// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
|
|
// source Frame) or nested one level under a `.source` member (`v.source.eof`,
|
|
// as on the pipeline's SceneFrame/…/MatchedSceneFrame message types, which wrap
|
|
// the originating Frame). Sentinels are delivered losslessly and non-blockingly
|
|
// via Channel::push_sentinel() instead of the throwing push(), so backpressure
|
|
// can never drop the token that unblocks downstream teardown.
|
|
//
|
|
// Types with neither shape are never treated as sentinels — both traits are
|
|
// SFINAE-safe and the runtime check compiles away to `false` for them, so this
|
|
// stays a no-op for pipelines that don't use an eof convention.
|
|
template<typename T, typename = void>
|
|
struct has_eof_field : std::false_type {};
|
|
template<typename T>
|
|
struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
|
|
: std::true_type {};
|
|
|
|
template<typename T, typename = void>
|
|
struct has_source_eof_field : std::false_type {};
|
|
template<typename T>
|
|
struct has_source_eof_field<T,
|
|
std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
|
|
: std::true_type {};
|
|
|
|
template<typename T>
|
|
constexpr bool is_sentinel_value(const T& v) {
|
|
if constexpr (has_eof_field<T>::value) return static_cast<bool>(v.eof);
|
|
else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
|
|
else return false;
|
|
}
|
|
|
|
// ── PoolNode ──────────────────────────────────────────────────────────────────
|
|
//
|
|
// Reactive alternative to Node<>. Instead of owning a blocked thread, the node
|
|
// is submitted to a shared IScheduler whenever all its input channels become
|
|
// non-empty. A single fire_once() call pops all inputs, executes the function,
|
|
// and pushes outputs. At most one fire_once() runs at a time (queued_ flag).
|
|
//
|
|
// Source nodes (input_count == 0) submit themselves immediately on start() and
|
|
// resubmit after each fire_once().
|
|
//
|
|
// Multiple PoolNodes can share one ThreadPool for resource-bounded execution,
|
|
// or each can have a dedicated single-thread pool for serialisation.
|
|
|
|
template<auto Func,
|
|
typename InputTag = in<>,
|
|
typename OutputTag = out<>,
|
|
fixed_string Label = "",
|
|
std::size_t UniqueTag = 0>
|
|
class PoolNode;
|
|
|
|
template<auto Func, fixed_string... InNames, fixed_string... OutNames,
|
|
fixed_string Label, std::size_t UniqueTag>
|
|
class PoolNode<Func, in<InNames...>, out<OutNames...>, Label, UniqueTag> : public INode {
|
|
public:
|
|
using F = decltype(Func);
|
|
using args_tuple = args_t<F>;
|
|
using return_raw = return_t<F>;
|
|
using return_tuple = normalised_return_t<return_raw>;
|
|
|
|
static constexpr std::string_view label() { return Label.view(); }
|
|
static constexpr std::size_t unique_tag = UniqueTag;
|
|
|
|
static constexpr std::size_t input_count = arity_v<F>;
|
|
static constexpr std::size_t output_count = std::tuple_size_v<return_tuple>;
|
|
|
|
static_assert(
|
|
sizeof...(InNames) == 0 || sizeof...(InNames) == input_count,
|
|
"make_pool_node: number of input names must match function arity, or provide none"
|
|
);
|
|
static_assert(
|
|
sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count,
|
|
"make_pool_node: number of output names must match return tuple size, or provide none"
|
|
);
|
|
|
|
explicit PoolNode(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5)
|
|
: scheduler_(std::move(sched)), fifo_capacity_(fifo_capacity)
|
|
{
|
|
init_input_channels(std::make_index_sequence<input_count>{});
|
|
}
|
|
|
|
~PoolNode() override { stop(); }
|
|
|
|
// ── INode ─────────────────────────────────────────────────────────────────
|
|
|
|
void start() override {
|
|
enable_inputs(std::make_index_sequence<input_count>{});
|
|
stop_flag_.store(false, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_relaxed);
|
|
register_callbacks(std::make_index_sequence<input_count>{});
|
|
if constexpr (input_count == 0)
|
|
try_submit(0.5f);
|
|
}
|
|
|
|
void stop() override {
|
|
stop_flag_.store(true, std::memory_order_seq_cst);
|
|
disable_inputs(std::make_index_sequence<input_count>{});
|
|
// fire_once() observes stop_flag_ and will not resubmit.
|
|
// We do not wait for an in-flight fire_once() to complete here;
|
|
// callers that need that guarantee should call scheduler_->drain() first.
|
|
}
|
|
|
|
bool running() const override {
|
|
return !stop_flag_.load(std::memory_order_relaxed);
|
|
}
|
|
|
|
void set_name(std::string name) override { name_ = std::move(name); }
|
|
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); }
|
|
void set_network_closed_callback(NodeEventCallback cb) override { closed_callbacks_[1] = std::move(cb); }
|
|
|
|
const NodeStats& stats() const override { return stats_; }
|
|
|
|
NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
|
|
uint64_t frames = stats_.frames_processed.load(std::memory_order_relaxed);
|
|
double exec_ms = stats_.ema_exec_us.load(std::memory_order_relaxed) / 1000.0;
|
|
double blocked_ms = stats_.total_blocked_us.load(std::memory_order_relaxed) / 1000.0;
|
|
double qwait_ms = stats_.queue_wait_us.load(std::memory_order_relaxed) / 1000.0;
|
|
double total_ms = exec_ms + blocked_ms;
|
|
return {
|
|
name, frames, exec_ms,
|
|
stats_.max_exec_us.load(std::memory_order_relaxed) / 1000.0,
|
|
blocked_ms,
|
|
elapsed_s > 0 ? frames / elapsed_s : 0.0,
|
|
stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
|
|
total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
|
|
qwait_ms,
|
|
};
|
|
}
|
|
|
|
// ── Port access — by index ────────────────────────────────────────────────
|
|
|
|
template<std::size_t I>
|
|
InputPort<PoolNode, I> input() {
|
|
static_assert(I < input_count, "input index out of range");
|
|
return {*this};
|
|
}
|
|
|
|
template<std::size_t I>
|
|
OutputPort<PoolNode, I> output() {
|
|
static_assert(I < output_count, "output index out of range");
|
|
return {*this};
|
|
}
|
|
|
|
// ── Port access — by name ─────────────────────────────────────────────────
|
|
|
|
template<fixed_string Name>
|
|
auto input() {
|
|
constexpr std::size_t idx = index_of<Name, InNames...>();
|
|
static_assert(idx != npos, "unknown input port name");
|
|
return input<idx>();
|
|
}
|
|
|
|
template<fixed_string Name>
|
|
auto output() {
|
|
constexpr std::size_t idx = index_of<Name, OutNames...>();
|
|
static_assert(idx != npos, "unknown output port name");
|
|
return output<idx>();
|
|
}
|
|
|
|
// ── Internal channel accessors ────────────────────────────────────────────
|
|
|
|
template<std::size_t I>
|
|
Channel<std::tuple_element_t<I, args_tuple>>& input_channel() {
|
|
return *std::get<I>(input_channels_);
|
|
}
|
|
|
|
template<std::size_t I>
|
|
void set_input_channel(
|
|
std::shared_ptr<Channel<std::tuple_element_t<I, args_tuple>>> ch) {
|
|
std::get<I>(input_channels_) = std::move(ch);
|
|
}
|
|
|
|
template<std::size_t I>
|
|
void set_output_channel(
|
|
Channel<std::tuple_element_t<I, return_tuple>>* ch) {
|
|
std::get<I>(output_channels_) = ch;
|
|
}
|
|
|
|
private:
|
|
// ── Channel storage ───────────────────────────────────────────────────────
|
|
|
|
template<std::size_t... Is>
|
|
void init_input_channels(std::index_sequence<Is...>) {
|
|
((std::get<Is>(input_channels_) =
|
|
std::make_shared<Channel<std::tuple_element_t<Is, args_tuple>>>(fifo_capacity_)),
|
|
...);
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
void enable_inputs(std::index_sequence<Is...>) {
|
|
(std::get<Is>(input_channels_)->enable(), ...);
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
void disable_inputs(std::index_sequence<Is...>) {
|
|
(std::get<Is>(input_channels_)->disable(), ...);
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
void disable_outputs(std::index_sequence<Is...>) {
|
|
auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
|
|
(disable_one(std::get<Is>(output_channels_)), ...);
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
void register_callbacks(std::index_sequence<Is...>) {
|
|
(std::get<Is>(input_channels_)->set_push_callback(
|
|
[this] { on_input_ready(); }), ...);
|
|
}
|
|
|
|
static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
|
|
const auto ts = std::chrono::steady_clock::now();
|
|
for (auto& cb : cbs) if (cb) cb(ts);
|
|
}
|
|
|
|
void self_stop() {
|
|
disable_inputs(std::make_index_sequence<input_count>{});
|
|
disable_outputs(std::make_index_sequence<output_count>{});
|
|
stats_.exec_start_us.store(0, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_release);
|
|
stop_flag_.store(true, std::memory_order_relaxed);
|
|
}
|
|
|
|
template<typename Tup, std::size_t... Is>
|
|
static auto make_input_channel_tuple(std::index_sequence<Is...>)
|
|
-> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;
|
|
|
|
using input_channels_t = decltype(make_input_channel_tuple<args_tuple>(
|
|
std::make_index_sequence<input_count>{}));
|
|
|
|
template<typename Tup, std::size_t... Is>
|
|
static auto make_output_channel_tuple(std::index_sequence<Is...>)
|
|
-> std::tuple<Channel<std::tuple_element_t<Is, Tup>>*...>;
|
|
|
|
using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
|
|
std::make_index_sequence<output_count>{}));
|
|
|
|
// ── Scheduling ────────────────────────────────────────────────────────────
|
|
|
|
// Called by channel push_callbacks (on the producer's thread).
|
|
void on_input_ready() {
|
|
if (stop_flag_.load(std::memory_order_relaxed)) return;
|
|
std::size_t ready = count_ready(std::make_index_sequence<input_count>{});
|
|
if (ready == input_count)
|
|
try_submit(compute_priority());
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
std::size_t count_ready(std::index_sequence<Is...>) {
|
|
return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
|
|
}
|
|
|
|
float compute_priority() {
|
|
if constexpr (input_count == 0) return 0.5f;
|
|
float sum = 0.0f;
|
|
sum_fill(sum, std::make_index_sequence<input_count>{});
|
|
return sum / static_cast<float>(input_count);
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
void sum_fill(float& sum, std::index_sequence<Is...>) {
|
|
((sum += std::get<Is>(input_channels_)->capacity() > 0
|
|
? float(std::get<Is>(input_channels_)->approx_size())
|
|
/ float(std::get<Is>(input_channels_)->capacity())
|
|
: 0.5f), ...);
|
|
}
|
|
|
|
void try_submit(float priority) {
|
|
bool expected = false;
|
|
if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
|
|
scheduler_->submit([this] { fire_once(); }, priority);
|
|
}
|
|
|
|
// ── Execution ─────────────────────────────────────────────────────────────
|
|
|
|
void fire_once() {
|
|
if (stop_flag_.load(std::memory_order_relaxed)) {
|
|
queued_.store(false, std::memory_order_release);
|
|
return;
|
|
}
|
|
|
|
// Record queue wait time (submission → now) and mark as executing
|
|
auto t0 = clock_t::now();
|
|
int64_t now_us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
t0.time_since_epoch()).count();
|
|
stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
|
|
|
|
try {
|
|
auto args = pop_inputs(std::make_index_sequence<input_count>{});
|
|
auto t1 = clock_t::now();
|
|
stats_.record_queue_wait(duration_t(t1 - t0));
|
|
auto cpu0 = NodeStats::cpu_now();
|
|
|
|
if constexpr (std::is_void_v<return_raw>) {
|
|
std::apply(Func, args);
|
|
} else {
|
|
auto result = std::apply(Func, args);
|
|
push_outputs(normalise(std::move(result)),
|
|
std::make_index_sequence<output_count>{});
|
|
}
|
|
|
|
auto cpu1 = NodeStats::cpu_now();
|
|
auto t2 = clock_t::now();
|
|
// blocked_time = 0 for pool nodes (we don't block waiting for inputs)
|
|
stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
|
|
} catch (const ChannelClosedError&) {
|
|
fire_callbacks(closed_callbacks_);
|
|
self_stop();
|
|
return;
|
|
} catch (const ChannelOverflowError&) {
|
|
fire_callbacks(event_callbacks_);
|
|
} catch (...) {
|
|
if (error_handler_ && error_handler_(name_, std::current_exception())) {
|
|
// continue — fall through to resubmit check
|
|
} else {
|
|
fire_callbacks(closed_callbacks_);
|
|
self_stop();
|
|
return;
|
|
}
|
|
}
|
|
|
|
stats_.exec_start_us.store(0, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_release);
|
|
|
|
if (stop_flag_.load(std::memory_order_relaxed)) return;
|
|
|
|
// Source nodes always resubmit; others resubmit only if inputs are ready.
|
|
if constexpr (input_count == 0) {
|
|
try_submit(0.5f);
|
|
} else {
|
|
on_input_ready();
|
|
}
|
|
}
|
|
|
|
// Pop all inputs — safe because we're the sole consumer and fire_once
|
|
// is guarded by queued_ (only one fire_once runs at a time).
|
|
template<std::size_t... Is>
|
|
args_tuple pop_inputs(std::index_sequence<Is...>) {
|
|
return {pop_one<Is>()...};
|
|
}
|
|
|
|
template<std::size_t I>
|
|
std::tuple_element_t<I, args_tuple> pop_one() {
|
|
auto& ch = *std::get<I>(input_channels_);
|
|
std::tuple_element_t<I, args_tuple> val;
|
|
if (!ch.try_pop_now(val))
|
|
throw ChannelClosedError{};
|
|
return val;
|
|
}
|
|
|
|
template<typename R = return_raw>
|
|
static return_tuple normalise(R&& r) {
|
|
if constexpr (is_tuple_v<R>) return std::move(r);
|
|
else return std::make_tuple(std::move(r));
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
|
|
(push_one_out<Is>(std::get<Is>(std::move(result))), ...);
|
|
}
|
|
|
|
template<std::size_t I>
|
|
void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
|
|
auto* ch = std::get<I>(output_channels_);
|
|
if (!ch) return;
|
|
// Sentinels (EOF) must never be dropped: a lost token wedges every
|
|
// downstream pop() forever. Deliver them out-of-band (push_sentinel),
|
|
// which never overflows and never blocks this node's worker thread.
|
|
if (is_sentinel_value(val)) {
|
|
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&) {
|
|
throw ChannelOverflowError(ch->capacity(),
|
|
"pool node '" + name_ + "' " + output_port_label<I>());
|
|
}
|
|
}
|
|
|
|
template<std::size_t I>
|
|
static std::string output_port_label() {
|
|
if constexpr (sizeof...(OutNames) > 0) {
|
|
constexpr std::array<std::string_view, sizeof...(OutNames)> names{OutNames.view()...};
|
|
return std::string("output['") + std::string(names[I]) + "']";
|
|
} else {
|
|
return "output[" + std::to_string(I) + "]";
|
|
}
|
|
}
|
|
|
|
// ── State ─────────────────────────────────────────────────────────────────
|
|
|
|
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_{};
|
|
std::atomic<bool> stop_flag_{true};
|
|
std::atomic<bool> queued_{false};
|
|
NodeStats stats_;
|
|
NodeErrorHandler error_handler_;
|
|
std::chrono::milliseconds max_exec_time_{0};
|
|
std::array<NodeEventCallback, 2> event_callbacks_{}; // [0]=user [1]=network
|
|
std::array<NodeEventCallback, 2> closed_callbacks_{};
|
|
};
|
|
|
|
// ── PoolObjectNode ────────────────────────────────────────────────────────────
|
|
//
|
|
// Same as PoolNode but wraps a stateful callable object (functor / class with
|
|
// operator()). The object must outlive the PoolObjectNode.
|
|
|
|
template<typename Obj,
|
|
typename InputTag = in<>,
|
|
typename OutputTag = out<>,
|
|
fixed_string Label = "",
|
|
std::size_t UniqueTag = 0>
|
|
class PoolObjectNode;
|
|
|
|
template<typename Obj, fixed_string... InNames, fixed_string... OutNames,
|
|
fixed_string Label, std::size_t UniqueTag>
|
|
class PoolObjectNode<Obj, in<InNames...>, out<OutNames...>, Label, UniqueTag> : public INode {
|
|
public:
|
|
using F = decltype(&Obj::operator());
|
|
using args_tuple = args_t<F>;
|
|
using return_raw = return_t<F>;
|
|
using return_tuple = normalised_return_t<return_raw>;
|
|
|
|
static constexpr std::string_view label() { return Label.view(); }
|
|
static constexpr std::size_t unique_tag = UniqueTag;
|
|
|
|
static constexpr std::size_t input_count = arity_v<F>;
|
|
static constexpr std::size_t output_count = std::tuple_size_v<return_tuple>;
|
|
|
|
static_assert(
|
|
sizeof...(InNames) == 0 || sizeof...(InNames) == input_count,
|
|
"make_pool_node: number of input names must match operator() arity, or provide none"
|
|
);
|
|
static_assert(
|
|
sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count,
|
|
"make_pool_node: number of output names must match return tuple size, or provide none"
|
|
);
|
|
|
|
explicit PoolObjectNode(Obj& obj, std::shared_ptr<IScheduler> sched,
|
|
std::size_t fifo_capacity = 5)
|
|
: obj_(obj), scheduler_(std::move(sched)), fifo_capacity_(fifo_capacity)
|
|
{
|
|
init_input_channels(std::make_index_sequence<input_count>{});
|
|
}
|
|
|
|
~PoolObjectNode() override { stop(); }
|
|
|
|
void start() override {
|
|
enable_inputs(std::make_index_sequence<input_count>{});
|
|
stop_flag_.store(false, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_relaxed);
|
|
register_callbacks(std::make_index_sequence<input_count>{});
|
|
if constexpr (input_count == 0)
|
|
try_submit(0.5f);
|
|
}
|
|
|
|
void stop() override {
|
|
stop_flag_.store(true, std::memory_order_seq_cst);
|
|
disable_inputs(std::make_index_sequence<input_count>{});
|
|
}
|
|
|
|
bool running() const override { return !stop_flag_.load(std::memory_order_relaxed); }
|
|
void set_name(std::string name) override { name_ = std::move(name); }
|
|
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); }
|
|
void set_network_closed_callback(NodeEventCallback cb) override { closed_callbacks_[1] = std::move(cb); }
|
|
|
|
const NodeStats& stats() const override { return stats_; }
|
|
|
|
NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
|
|
uint64_t frames = stats_.frames_processed.load(std::memory_order_relaxed);
|
|
double exec_ms = stats_.ema_exec_us.load(std::memory_order_relaxed) / 1000.0;
|
|
double blocked_ms = stats_.total_blocked_us.load(std::memory_order_relaxed) / 1000.0;
|
|
double qwait_ms = stats_.queue_wait_us.load(std::memory_order_relaxed) / 1000.0;
|
|
double total_ms = exec_ms + blocked_ms;
|
|
return {
|
|
name, frames, exec_ms,
|
|
stats_.max_exec_us.load(std::memory_order_relaxed) / 1000.0,
|
|
blocked_ms,
|
|
elapsed_s > 0 ? frames / elapsed_s : 0.0,
|
|
stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
|
|
total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
|
|
qwait_ms,
|
|
};
|
|
}
|
|
|
|
template<std::size_t I> InputPort<PoolObjectNode, I> input() { return {*this}; }
|
|
template<std::size_t I> OutputPort<PoolObjectNode, I> output() { return {*this}; }
|
|
|
|
template<fixed_string Name>
|
|
auto input() {
|
|
constexpr std::size_t idx = index_of<Name, InNames...>();
|
|
static_assert(idx != npos, "unknown input port name");
|
|
return input<idx>();
|
|
}
|
|
template<fixed_string Name>
|
|
auto output() {
|
|
constexpr std::size_t idx = index_of<Name, OutNames...>();
|
|
static_assert(idx != npos, "unknown output port name");
|
|
return output<idx>();
|
|
}
|
|
|
|
template<std::size_t I>
|
|
Channel<std::tuple_element_t<I, args_tuple>>& input_channel() {
|
|
return *std::get<I>(input_channels_);
|
|
}
|
|
template<std::size_t I>
|
|
void set_input_channel(std::shared_ptr<Channel<std::tuple_element_t<I, args_tuple>>> ch) {
|
|
std::get<I>(input_channels_) = std::move(ch);
|
|
}
|
|
template<std::size_t I>
|
|
void set_output_channel(Channel<std::tuple_element_t<I, return_tuple>>* ch) {
|
|
std::get<I>(output_channels_) = ch;
|
|
}
|
|
|
|
private:
|
|
template<std::size_t... Is>
|
|
void init_input_channels(std::index_sequence<Is...>) {
|
|
((std::get<Is>(input_channels_) =
|
|
std::make_shared<Channel<std::tuple_element_t<Is, args_tuple>>>(fifo_capacity_)),
|
|
...);
|
|
}
|
|
template<std::size_t... Is> void enable_inputs(std::index_sequence<Is...>) { (std::get<Is>(input_channels_)->enable(), ...); }
|
|
template<std::size_t... Is> void disable_inputs(std::index_sequence<Is...>) { (std::get<Is>(input_channels_)->disable(), ...); }
|
|
template<std::size_t... Is>
|
|
void disable_outputs(std::index_sequence<Is...>) {
|
|
auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
|
|
(disable_one(std::get<Is>(output_channels_)), ...);
|
|
}
|
|
template<std::size_t... Is>
|
|
void register_callbacks(std::index_sequence<Is...>) {
|
|
(std::get<Is>(input_channels_)->set_push_callback([this] { on_input_ready(); }), ...);
|
|
}
|
|
|
|
static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
|
|
const auto ts = std::chrono::steady_clock::now();
|
|
for (auto& cb : cbs) if (cb) cb(ts);
|
|
}
|
|
|
|
void self_stop() {
|
|
disable_inputs(std::make_index_sequence<input_count>{});
|
|
disable_outputs(std::make_index_sequence<output_count>{});
|
|
stats_.exec_start_us.store(0, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_release);
|
|
stop_flag_.store(true, std::memory_order_relaxed);
|
|
}
|
|
|
|
template<typename Tup, std::size_t... Is>
|
|
static auto make_input_channel_tuple(std::index_sequence<Is...>)
|
|
-> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;
|
|
using input_channels_t = decltype(make_input_channel_tuple<args_tuple>(
|
|
std::make_index_sequence<input_count>{}));
|
|
|
|
template<typename Tup, std::size_t... Is>
|
|
static auto make_output_channel_tuple(std::index_sequence<Is...>)
|
|
-> std::tuple<Channel<std::tuple_element_t<Is, Tup>>*...>;
|
|
using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
|
|
std::make_index_sequence<output_count>{}));
|
|
|
|
void on_input_ready() {
|
|
if (stop_flag_.load(std::memory_order_relaxed)) return;
|
|
std::size_t ready = count_ready(std::make_index_sequence<input_count>{});
|
|
if (ready == input_count) try_submit(compute_priority());
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
std::size_t count_ready(std::index_sequence<Is...>) {
|
|
return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
|
|
}
|
|
|
|
float compute_priority() {
|
|
if constexpr (input_count == 0) return 0.5f;
|
|
float sum = 0.0f;
|
|
sum_fill(sum, std::make_index_sequence<input_count>{});
|
|
return sum / static_cast<float>(input_count);
|
|
}
|
|
template<std::size_t... Is>
|
|
void sum_fill(float& sum, std::index_sequence<Is...>) {
|
|
((sum += std::get<Is>(input_channels_)->capacity() > 0
|
|
? float(std::get<Is>(input_channels_)->approx_size())
|
|
/ float(std::get<Is>(input_channels_)->capacity())
|
|
: 0.5f), ...);
|
|
}
|
|
|
|
void try_submit(float priority) {
|
|
bool expected = false;
|
|
if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
|
|
scheduler_->submit([this] { fire_once(); }, priority);
|
|
}
|
|
|
|
void fire_once() {
|
|
if (stop_flag_.load(std::memory_order_relaxed)) {
|
|
queued_.store(false, std::memory_order_release);
|
|
return;
|
|
}
|
|
auto t0 = clock_t::now();
|
|
int64_t now_us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
t0.time_since_epoch()).count();
|
|
stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
|
|
|
|
try {
|
|
auto args = pop_inputs(std::make_index_sequence<input_count>{});
|
|
auto t1 = clock_t::now();
|
|
stats_.record_queue_wait(duration_t(t1 - t0));
|
|
auto cpu0 = NodeStats::cpu_now();
|
|
|
|
if constexpr (std::is_void_v<return_raw>) {
|
|
std::apply([this](auto&&... a) { obj_(std::forward<decltype(a)>(a)...); }, args);
|
|
} else {
|
|
auto result = std::apply([this](auto&&... a) { return obj_(std::forward<decltype(a)>(a)...); }, args);
|
|
push_outputs(normalise(std::move(result)), std::make_index_sequence<output_count>{});
|
|
}
|
|
|
|
auto cpu1 = NodeStats::cpu_now();
|
|
auto t2 = clock_t::now();
|
|
stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
|
|
} catch (const ChannelClosedError&) {
|
|
fire_callbacks(closed_callbacks_);
|
|
self_stop();
|
|
return;
|
|
} catch (const ChannelOverflowError&) {
|
|
fire_callbacks(event_callbacks_);
|
|
} catch (...) {
|
|
if (error_handler_ && error_handler_(name_, std::current_exception())) {
|
|
} else {
|
|
fire_callbacks(closed_callbacks_);
|
|
self_stop();
|
|
return;
|
|
}
|
|
}
|
|
|
|
stats_.exec_start_us.store(0, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_release);
|
|
if (stop_flag_.load(std::memory_order_relaxed)) return;
|
|
if constexpr (input_count == 0) try_submit(0.5f);
|
|
else on_input_ready();
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
args_tuple pop_inputs(std::index_sequence<Is...>) { return {pop_one<Is>()...}; }
|
|
|
|
template<std::size_t I>
|
|
std::tuple_element_t<I, args_tuple> pop_one() {
|
|
auto& ch = *std::get<I>(input_channels_);
|
|
std::tuple_element_t<I, args_tuple> val;
|
|
if (!ch.try_pop_now(val)) throw ChannelClosedError{};
|
|
return val;
|
|
}
|
|
|
|
template<typename R = return_raw>
|
|
static return_tuple normalise(R&& r) {
|
|
if constexpr (is_tuple_v<R>) return std::move(r);
|
|
else return std::make_tuple(std::move(r));
|
|
}
|
|
|
|
template<std::size_t... Is>
|
|
void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
|
|
(push_one_out<Is>(std::get<Is>(std::move(result))), ...);
|
|
}
|
|
template<std::size_t I>
|
|
void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
|
|
auto* ch = std::get<I>(output_channels_);
|
|
if (!ch) return;
|
|
// Sentinels (EOF) must never be dropped: a lost token wedges every
|
|
// downstream pop() forever. Deliver them out-of-band (push_sentinel),
|
|
// which never overflows and never blocks this node's worker thread.
|
|
if (is_sentinel_value(val)) {
|
|
ch->push_sentinel(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_{};
|
|
std::atomic<bool> stop_flag_{true};
|
|
std::atomic<bool> queued_{false};
|
|
NodeStats stats_;
|
|
NodeErrorHandler error_handler_;
|
|
std::chrono::milliseconds max_exec_time_{0};
|
|
std::array<NodeEventCallback, 2> event_callbacks_{}; // [0]=user [1]=network
|
|
std::array<NodeEventCallback, 2> closed_callbacks_{};
|
|
};
|
|
|
|
// ── make_pool_node factory (NTTP) ─────────────────────────────────────────────
|
|
|
|
template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0>
|
|
auto make_pool_node(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5) {
|
|
return PoolNode<Func, in<>, out<>, Label, UniqueTag>(std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0,
|
|
fixed_string... InNames>
|
|
auto make_pool_node(std::shared_ptr<IScheduler> sched, in<InNames...>,
|
|
std::size_t fifo_capacity = 5) {
|
|
return PoolNode<Func, in<InNames...>, out<>, Label, UniqueTag>(std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0,
|
|
fixed_string... OutNames>
|
|
auto make_pool_node(std::shared_ptr<IScheduler> sched, out<OutNames...>,
|
|
std::size_t fifo_capacity = 5) {
|
|
return PoolNode<Func, in<>, out<OutNames...>, Label, UniqueTag>(std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0,
|
|
fixed_string... InNames, fixed_string... OutNames>
|
|
auto make_pool_node(std::shared_ptr<IScheduler> sched, in<InNames...>, out<OutNames...>,
|
|
std::size_t fifo_capacity = 5) {
|
|
return PoolNode<Func, in<InNames...>, out<OutNames...>, Label, UniqueTag>(
|
|
std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
// ── make_pool_node factory (callable object) ──────────────────────────────────
|
|
|
|
template<typename Obj>
|
|
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched,
|
|
std::size_t fifo_capacity = 5) {
|
|
return PoolObjectNode<Obj, in<>, out<>>(obj, std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
template<typename Obj, fixed_string... InNames>
|
|
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched, in<InNames...>,
|
|
std::size_t fifo_capacity = 5) {
|
|
return PoolObjectNode<Obj, in<InNames...>, out<>>(obj, std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
template<typename Obj, fixed_string... OutNames>
|
|
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched, out<OutNames...>,
|
|
std::size_t fifo_capacity = 5) {
|
|
return PoolObjectNode<Obj, in<>, out<OutNames...>>(obj, std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
template<typename Obj, fixed_string... InNames, fixed_string... OutNames>
|
|
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched,
|
|
in<InNames...>, out<OutNames...>,
|
|
std::size_t fifo_capacity = 5) {
|
|
return PoolObjectNode<Obj, in<InNames...>, out<OutNames...>>(
|
|
obj, std::move(sched), fifo_capacity);
|
|
}
|
|
|
|
} // namespace kpn
|