dtourolle/KPN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7b7f631e6d3ecfd213f61227eddb2e840d341d4c
KPN/include/kpn/pool_node.hpp
T
dtourolle 97670d8ba3 fix: stop() must not return while a firing is still running 
stop() set the flag, disabled the inputs and returned, leaving an executing
fire_once touching input_channels_, stats_ and pending_ while the caller went
on to destroy them. The comment was explicit about it: callers wanting the
guarantee should call scheduler_->drain() first. But ~PoolNode calls stop(),
and a destructor cannot ask its caller to have done that.

A node with a private pool survived by accident, because Node::stop() calls
pool->stop() and that joins the worker. A node sharing a pool — which
make_pool_node exists to create — had nothing joining it, so its own
destructor raced the firing.

stop() now waits on the submit gate, which is claimed for the whole of a
firing and released as its last act. A queued but unstarted firing also holds
it and will run, observe stop_flag_ and release, so the pool must still be
running when stop() is called; that is already the documented order and what
Node/ObjectNode do.

Two ways it declines to wait. It is bounded at five seconds, because a node
function that never returns must not convert teardown into a hang — it warns
and continues. And it returns immediately when called from the firing thread
itself, since an error handler that stops its own node would otherwise wait
for a firing that is waiting for it.

Verified in both directions: with the wait removed, stop() returns while the
node function is still sleeping and the flag it sets on the way out is still
false. 143/143.
2026-08-05 15:46:27 +02:00

1314 lines
62 KiB
C++
Raw Blame History

#pragma once
#include "channel.hpp"
#include "diagnostics.hpp"
#include "fixed_string.hpp"
#include "inode.hpp"
#include "port.hpp"
#include "scheduler.hpp"
#include "submit_gate.hpp"
#include "traits.hpp"
#include <array>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <functional>
#include <iostream>
#include <memory>
#include <optional>
#include <variant>
#include <stdexcept>
#include <thread>
#include <tuple>
#include <type_traits>
namespace kpn {
// Sentinel detection (has_eof_field / is_sentinel_value) lives in traits.hpp —
// every node type that forwards values needs it, not just pool-scheduled ones.
// ── 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 (see SubmitGate).
//
// 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 prepare() override {
if (prepared_) return; // idempotent: the network calls this,
prepared_ = true; // and start() calls it again if not.
register_callbacks(std::make_index_sequence<input_count>{});
}
void start() override {
prepare();
enable_inputs(std::make_index_sequence<input_count>{});
stop_flag_.store(false, std::memory_order_relaxed);
gate_.force_idle();
if constexpr (input_count == 0)
try_submit(0.5f);
else
// Never start with a wake already outstanding — the startup case of
// the invariant 9c5ce5f established for the running pipeline.
//
// The callback is installed by prepare(), before any node runs, but
// a network still starts its nodes one at a time: an upstream node
// that is already firing can push into this one between the two
// calls. The push is accepted by the ring and does invoke the
// callback, but on_input_ready() sees stop_flag_ still set and
// returns. Every later push sees a non-empty ring and stays silent
// — Channel invokes push_callback_ only on the empty->non-empty
// transition — so without this the node is never submitted and the
// pipeline reads as wedged from the first frame.
//
// on_input_ready() is the level-triggered form of the same
// question, so asking it once here converts the missed edge into a
// state check.
on_input_ready();
}
void stop() override {
stop_flag_.store(true, std::memory_order_seq_cst);
disable_inputs(std::make_index_sequence<input_count>{});
await_quiescence();
}
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_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); }
void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
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,
stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
gate_.queued(),
gate_.wake_pending(),
};
}
// ── 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...>) {
// A parked producer is re-submitted when its output drains.
register_space_callbacks(std::make_index_sequence<output_count>{});
(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);
}
template<std::size_t... Os>
bool outputs_have_space(std::index_sequence<Os...>) const {
return (... && (!std::get<Os>(output_channels_) ||
std::get<Os>(output_channels_)->has_space()));
}
template<std::size_t... Os>
void register_space_callbacks(std::index_sequence<Os...>) {
((std::get<Os>(output_channels_)
? (void)std::get<Os>(output_channels_)->set_space_callback(
[this] { try_submit(0.5f); })
: (void)0), ...);
}
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);
// force_idle, not finish_firing(): this node is stopping, and
// honouring a pending wake here would resubmit a dead node.
gate_.force_idle();
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) + ...);
}
/// Priority in [0,1], higher runs sooner.
///
/// Input fill alone answers "how much work is waiting for me". That is
/// only half the question: a node whose *outputs* are already full cannot
/// deliver anything: running it produces a value with nowhere to go, so it
/// immediately parks and the slot is wasted. Meanwhile the node that would
/// have drained that full channel waits behind it.
///
/// So occupancy of the outputs is deducted from occupancy of the inputs.
/// The scheduler then naturally favours whoever is furthest downstream of
/// a bottleneck — the node whose inputs are backed up but whose outputs
/// have room is exactly the one whose execution frees the most capacity —
/// and defers producers that would only deepen a queue that is already
/// full.
///
/// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero:
/// both terms are mean fills in [0,1], so the difference is in [-1,1], and
/// the affine map keeps the whole range distinguishable instead of
/// collapsing every output-saturated node onto the same value. 0.5 remains
/// the neutral point, matching the default used for source nodes.
float compute_priority() {
if constexpr (input_count == 0) return 0.5f;
float in = 0.0f;
sum_fill(in, std::make_index_sequence<input_count>{});
in /= static_cast<float>(input_count);
if constexpr (output_count == 0) return in;
float out = 0.0f;
sum_output_fill(out, std::make_index_sequence<output_count>{});
out /= static_cast<float>(output_count);
const float p = 0.5f * (1.0f + in - out);
return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p);
}
/// Mean fill of the output channels, same normalisation as sum_fill.
/// An unconnected output holds nothing back, so it contributes 0.
template<std::size_t... Os>
void sum_output_fill(float& sum, std::index_sequence<Os...>) {
((sum += (std::get<Os>(output_channels_) &&
std::get<Os>(output_channels_)->capacity() > 0)
? float(std::get<Os>(output_channels_)->approx_size())
/ float(std::get<Os>(output_channels_)->capacity())
: 0.0f), ...);
}
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), ...);
}
/// Submit unless a firing is already in flight. A wake that arrives while
/// one is is *recorded* against it, never dropped.
///
/// Wakes are edge-triggered: a channel fires its space callback on the
/// transition, once. A dropped one never returns, so a node could park a
/// value, release its worker, and sleep forever holding output its consumer
/// was waiting for, with every worker idle in cond_wait and nothing left to
/// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same
/// variable, so the two cannot both be true — see submit_gate.hpp.
void try_submit(float priority) {
// A stopped node must not claim the gate. The scheduler now refuses
// submissions after its pool stops, so the submit itself is safe — but
// claiming and never releasing would leave the gate held, and a restart
// would then have to clear it. start() does, but relying on that makes
// the invariant depend on a distant statement.
if (stop_flag_.load(std::memory_order_relaxed)) return;
if (gate_.claim())
scheduler_->submit([this] { fire_once(); }, priority);
}
// ── Execution ─────────────────────────────────────────────────────────────
/// Decide whether this node should run again, then release the gate — in
/// that order, always.
///
/// Releasing first is what let two firings of the same node overlap: the
/// moment the gate is free another worker may enter fire_once, while this
/// invocation is still reading pending_ and writing pending_done_. TSan
/// caught it as a race on pending_done_ between a firing submitted by the
/// old release_and_recheck and one submitted by try_submit. It also quietly
/// broke the one-slot park, which is sound only because "at most one
/// fire_once runs per node at a time" — with two, a value can be parked by
/// one firing and overwritten by the other.
///
/// Everything this reads belongs to the firing that holds the claim, so it
/// is all evaluated first and the release is the last thing the firing does.
void finish_firing() {
bool want_more = false;
float prio = 0.5f;
if (!stop_flag_.load(std::memory_order_relaxed)) {
bool parked = false;
if constexpr (!std::is_void_v<return_raw>)
parked = pending_.has_value();
if (parked) {
// Still holding output: only worth running again once the
// consumer has made room.
want_more = outputs_have_space(std::make_index_sequence<output_count>{});
} else {
if constexpr (input_count == 0) {
want_more = true; // sources always run again
} else {
want_more = count_ready(std::make_index_sequence<input_count>{})
== input_count;
if (want_more) prio = compute_priority();
}
}
}
if (gate_.release()) scheduler_->submit([this] { fire_once(); }, prio);
else if (want_more) try_submit(prio);
}
/// Block until no firing of this node is in flight or queued.
///
/// stop() used to set the flag and return, leaving an executing fire_once
/// touching input_channels_, stats_ and pending_ while the caller went on
/// to destroy them. For a node with a private pool that was survivable by
/// accident — Node::stop() calls pool->stop(), which joins — but a node
/// sharing a pool had nothing joining it at all, so ~PoolNode raced its own
/// members. The old comment said callers wanting the guarantee should call
/// scheduler_->drain() first; a destructor cannot, and the default should
/// not be a use-after-free.
///
/// The gate is exactly the right thing to wait on: it is claimed for the
/// whole of a firing and released as the last act of one. A queued but
/// unstarted firing also holds it, and will run, observe stop_flag_ and
/// release — which is why the pool must still be running when this is
/// called. That is already the documented order (stop nodes, then the
/// pool), and Node/ObjectNode do it that way.
///
/// Bounded, because a node function that never returns must not turn
/// teardown into a hang; and skipped entirely when called from the firing
/// thread itself, since an error handler that stops its own node would
/// otherwise wait for a firing that is waiting for it.
void await_quiescence() {
if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id())
return;
const auto deadline = clock_t::now() + std::chrono::seconds(5);
while (gate_.queued()) {
if (clock_t::now() >= deadline) {
std::cerr << "[kpn] stop: node '" << name_
<< "' still had work in flight after 5 s; "
"continuing without it\n";
return;
}
std::this_thread::sleep_for(std::chrono::microseconds(50));
}
}
/// Marks fire_once's thread for the duration of a firing, so await_quiescence
/// can tell a re-entrant stop() from an external one.
struct FiringMark {
std::atomic<std::thread::id>& slot;
explicit FiringMark(std::atomic<std::thread::id>& s) : slot(s) {
slot.store(std::this_thread::get_id(), std::memory_order_release);
}
~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); }
};
void fire_once() {
FiringMark mark(firing_thread_);
if (stop_flag_.load(std::memory_order_relaxed)) {
gate_.force_idle();
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);
// Parked from a previous firing: retry that value before touching the
// inputs. Returning here releases the worker — the channel's space
// callback re-submits this node when the consumer drains a slot.
if constexpr (!std::is_void_v<return_raw>) {
if (pending_) {
push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
// Whether the value went out or is still parked, finish_firing
// reads pending_ and picks the right follow-up: output space if
// still holding, input readiness if drained. Resubmitting
// unconditionally would fire a node whose inputs are empty, and
// pop_one reports an empty channel as ChannelClosedError — which
// this node treats as "upstream finished" and self-stops on.
finish_firing();
return;
}
}
// Woken by output space rather than by input arrival, with nothing
// parked left to flush: there is no work to do. Falling through would
// read an empty channel, and pop_one reports empty as
// ChannelClosedError — self-stopping a live node. Release the worker;
// on_input_ready() resubmits when data actually lands.
if constexpr (input_count > 0) {
if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
// finish_firing re-checks readiness after the work above, so
// data that landed while we looked is not missed.
finish_firing();
return;
}
}
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 ChannelEmptyError&) {
// Not an error: there was simply nothing to take. Release and wait
// to be woken again. fire_once checks readiness before it gets
// here, and this node is the sole consumer of its inputs, so this
// is unreachable today — it exists so that if the check is ever
// weakened the cost is a wasted firing rather than a dead node.
finish_firing();
return;
} catch (const ChannelClosedError&) {
fire_callbacks(closed_callbacks_);
self_stop();
return;
} catch (const ChannelOverflowError&) {
fire_callbacks(event_callbacks_);
} catch (...) {
auto eptr = std::current_exception();
const bool handled =
(error_handler_ && error_handler_(name_, eptr)) ||
(net_error_handler_ && net_error_handler_(name_, eptr));
if (handled) {
// continue — fall through to resubmit check
} else {
fire_callbacks(closed_callbacks_);
self_stop();
return;
}
}
stats_.exec_start_us.store(0, std::memory_order_relaxed);
// If the push above parked, finish_firing waits on output space rather
// than input arrival: this firing consumed its input, so an input-level
// check would not resubmit and the node would hold its value forever
// while its consumer waits for exactly that value.
finish_firing();
}
// Pop all inputs — safe because we're the sole consumer and fire_once
// is guarded by the submit gate (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)) {
// try_pop_now returns false for "nothing available", which covers
// two very different situations. A closed channel means upstream is
// finished and this node should stop. An open one means only that
// nothing is here at this instant — and treating that as closed
// kills a live node, which then disables its own inputs and outputs
// and takes the rest of the pipeline with it.
if (ch.is_accepting()) throw ChannelEmptyError{};
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>
/// Pushes what it can and parks the rest. `done_` marks the elements that
/// were taken, so a retry never re-pushes one — a duplicate would be as
/// wrong as a drop, just harder to notice.
void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
bool all = true;
((pending_done_[Is] = pending_done_[Is] ||
push_one_out<Is>(std::get<Is>(std::move(result))),
all = all && pending_done_[Is]), ...);
if (all) { pending_.reset(); pending_done_.fill(false); }
// The retry path calls this as push_outputs(std::move(*pending_), …), so
// on that path `result` *is* the parked tuple. Assigning it to itself is
// a self-move-assignment, which for std::tuple is elementwise — and
// libstdc++'s std::vector does not guard against it: it swaps its data
// into a temporary and leaves the vector empty. A value that failed to
// push twice would therefore be delivered with its payload silently
// erased, which downstream reads as a legitimately empty result rather
// than as a loss. Only store when it is not already stored.
else if (!pending_ || &result != &*pending_) pending_ = std::move(result);
}
/// Returns false when the ring was full and the value was NOT taken; the
/// caller must keep it and retry after the channel signals space.
template<std::size_t I>
bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
auto* ch = std::get<I>(output_channels_);
if (!ch) return true;
// 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)) {
// A refused sentinel is a protocol error, not backpressure, so it
// is reported rather than parked and retried — retrying would spin
// forever against a slot only the consumer can free, and there is
// no correct value to deliver second anyway. Closed is normal
// during teardown and stays quiet.
if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
::SentinelResult::SlotBusy)
fire_callbacks(event_callbacks_);
return true;
}
// Backpressure without parking the worker. A full channel means the
// consumer is behind; the value is kept and this node stops running
// until the channel signals space (set_space_callback re-submits it).
//
// Blocking here instead would sleep inside a scheduler worker, and
// nodes are pinned to workers — park enough of them and nothing is left
// to run the consumer that would drain the channel. That is the
// hold-and-wait deadlock channel.hpp warns about for sentinels; it
// applies to data pushes just as much.
return ch->try_push(val);
}
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_;
std::size_t fifo_capacity_;
input_channels_t input_channels_;
output_channels_t output_channels_{};
std::atomic<bool> stop_flag_{true};
/// Serialises firings and records wakes that arrive during one. See
/// submit_gate.hpp for why this cannot be two separate flags.
SubmitGate gate_;
/// Whether prepare() has installed the channel callbacks. Only ever touched
/// from the thread driving start()/stop(), never from a worker, and never
/// cleared: the callbacks capture `this` and stay valid across a restart, so
/// re-registering them would be a pointless write to a live channel.
bool prepared_{false};
/// Thread currently inside fire_once, or a default id when none is.
/// See await_quiescence.
std::atomic<std::thread::id> firing_thread_{};
/// The hidden one-slot output buffer (see push_outputs). Holding the value
/// here is what lets a node stop running without dropping it or occupying a
/// scheduler worker. One slot suffices because at most one fire_once() runs
/// per node at a time — with concurrent firing it would have to hold a whole
/// FIFO's worth.
std::conditional_t<std::is_void_v<return_raw>, std::monostate,
std::optional<return_tuple>> pending_{};
std::array<bool, (output_count ? output_count : 1)> pending_done_{};
NodeStats stats_;
NodeErrorHandler error_handler_;
NodeErrorHandler net_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 prepare() override {
if (prepared_) return;
prepared_ = true;
register_callbacks(std::make_index_sequence<input_count>{});
}
void start() override {
prepare();
enable_inputs(std::make_index_sequence<input_count>{});
stop_flag_.store(false, std::memory_order_relaxed);
gate_.force_idle();
if constexpr (input_count == 0)
try_submit(0.5f);
else
// Never start with a wake already outstanding — see PoolNode::start().
on_input_ready();
}
void stop() override {
stop_flag_.store(true, std::memory_order_seq_cst);
disable_inputs(std::make_index_sequence<input_count>{});
await_quiescence();
}
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_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); }
void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
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,
stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
gate_.queued(),
gate_.wake_pending(),
};
}
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...>) {
// A parked producer is re-submitted when its output drains.
register_space_callbacks(std::make_index_sequence<output_count>{});
(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);
}
template<std::size_t... Os>
bool outputs_have_space(std::index_sequence<Os...>) const {
return (... && (!std::get<Os>(output_channels_) ||
std::get<Os>(output_channels_)->has_space()));
}
template<std::size_t... Os>
void register_space_callbacks(std::index_sequence<Os...>) {
((std::get<Os>(output_channels_)
? (void)std::get<Os>(output_channels_)->set_space_callback(
[this] { try_submit(0.5f); })
: (void)0), ...);
}
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);
// force_idle, not finish_firing(): this node is stopping, and
// honouring a pending wake here would resubmit a dead node.
gate_.force_idle();
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) + ...);
}
/// Priority in [0,1], higher runs sooner.
///
/// Input fill alone answers "how much work is waiting for me". That is
/// only half the question: a node whose *outputs* are already full cannot
/// deliver anything: running it produces a value with nowhere to go, so it
/// immediately parks and the slot is wasted. Meanwhile the node that would
/// have drained that full channel waits behind it.
///
/// So occupancy of the outputs is deducted from occupancy of the inputs.
/// The scheduler then naturally favours whoever is furthest downstream of
/// a bottleneck — the node whose inputs are backed up but whose outputs
/// have room is exactly the one whose execution frees the most capacity —
/// and defers producers that would only deepen a queue that is already
/// full.
///
/// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero:
/// both terms are mean fills in [0,1], so the difference is in [-1,1], and
/// the affine map keeps the whole range distinguishable instead of
/// collapsing every output-saturated node onto the same value. 0.5 remains
/// the neutral point, matching the default used for source nodes.
float compute_priority() {
if constexpr (input_count == 0) return 0.5f;
float in = 0.0f;
sum_fill(in, std::make_index_sequence<input_count>{});
in /= static_cast<float>(input_count);
if constexpr (output_count == 0) return in;
float out = 0.0f;
sum_output_fill(out, std::make_index_sequence<output_count>{});
out /= static_cast<float>(output_count);
const float p = 0.5f * (1.0f + in - out);
return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p);
}
/// Mean fill of the output channels, same normalisation as sum_fill.
/// An unconnected output holds nothing back, so it contributes 0.
template<std::size_t... Os>
void sum_output_fill(float& sum, std::index_sequence<Os...>) {
((sum += (std::get<Os>(output_channels_) &&
std::get<Os>(output_channels_)->capacity() > 0)
? float(std::get<Os>(output_channels_)->approx_size())
/ float(std::get<Os>(output_channels_)->capacity())
: 0.0f), ...);
}
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), ...);
}
/// Submit unless a firing is already in flight. A wake that arrives while
/// one is is *recorded* against it, never dropped.
///
/// Wakes are edge-triggered: a channel fires its space callback on the
/// transition, once. A dropped one never returns, so a node could park a
/// value, release its worker, and sleep forever holding output its consumer
/// was waiting for, with every worker idle in cond_wait and nothing left to
/// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same
/// variable, so the two cannot both be true — see submit_gate.hpp.
void try_submit(float priority) {
// A stopped node must not claim the gate. The scheduler now refuses
// submissions after its pool stops, so the submit itself is safe — but
// claiming and never releasing would leave the gate held, and a restart
// would then have to clear it. start() does, but relying on that makes
// the invariant depend on a distant statement.
if (stop_flag_.load(std::memory_order_relaxed)) return;
if (gate_.claim())
scheduler_->submit([this] { fire_once(); }, priority);
}
/// Decide whether this node should run again, then release the gate — in
/// that order, always.
///
/// Releasing first is what let two firings of the same node overlap: the
/// moment the gate is free another worker may enter fire_once, while this
/// invocation is still reading pending_ and writing pending_done_. TSan
/// caught it as a race on pending_done_ between a firing submitted by the
/// old release_and_recheck and one submitted by try_submit. It also quietly
/// broke the one-slot park, which is sound only because "at most one
/// fire_once runs per node at a time" — with two, a value can be parked by
/// one firing and overwritten by the other.
///
/// Everything this reads belongs to the firing that holds the claim, so it
/// is all evaluated first and the release is the last thing the firing does.
void finish_firing() {
bool want_more = false;
float prio = 0.5f;
if (!stop_flag_.load(std::memory_order_relaxed)) {
bool parked = false;
if constexpr (!std::is_void_v<return_raw>)
parked = pending_.has_value();
if (parked) {
// Still holding output: only worth running again once the
// consumer has made room.
want_more = outputs_have_space(std::make_index_sequence<output_count>{});
} else {
if constexpr (input_count == 0) {
want_more = true; // sources always run again
} else {
want_more = count_ready(std::make_index_sequence<input_count>{})
== input_count;
if (want_more) prio = compute_priority();
}
}
}
if (gate_.release()) scheduler_->submit([this] { fire_once(); }, prio);
else if (want_more) try_submit(prio);
}
/// Block until no firing of this node is in flight or queued.
///
/// stop() used to set the flag and return, leaving an executing fire_once
/// touching input_channels_, stats_ and pending_ while the caller went on
/// to destroy them. For a node with a private pool that was survivable by
/// accident — Node::stop() calls pool->stop(), which joins — but a node
/// sharing a pool had nothing joining it at all, so ~PoolNode raced its own
/// members. The old comment said callers wanting the guarantee should call
/// scheduler_->drain() first; a destructor cannot, and the default should
/// not be a use-after-free.
///
/// The gate is exactly the right thing to wait on: it is claimed for the
/// whole of a firing and released as the last act of one. A queued but
/// unstarted firing also holds it, and will run, observe stop_flag_ and
/// release — which is why the pool must still be running when this is
/// called. That is already the documented order (stop nodes, then the
/// pool), and Node/ObjectNode do it that way.
///
/// Bounded, because a node function that never returns must not turn
/// teardown into a hang; and skipped entirely when called from the firing
/// thread itself, since an error handler that stops its own node would
/// otherwise wait for a firing that is waiting for it.
void await_quiescence() {
if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id())
return;
const auto deadline = clock_t::now() + std::chrono::seconds(5);
while (gate_.queued()) {
if (clock_t::now() >= deadline) {
std::cerr << "[kpn] stop: node '" << name_
<< "' still had work in flight after 5 s; "
"continuing without it\n";
return;
}
std::this_thread::sleep_for(std::chrono::microseconds(50));
}
}
/// Marks fire_once's thread for the duration of a firing, so await_quiescence
/// can tell a re-entrant stop() from an external one.
struct FiringMark {
std::atomic<std::thread::id>& slot;
explicit FiringMark(std::atomic<std::thread::id>& s) : slot(s) {
slot.store(std::this_thread::get_id(), std::memory_order_release);
}
~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); }
};
void fire_once() {
FiringMark mark(firing_thread_);
if (stop_flag_.load(std::memory_order_relaxed)) {
gate_.force_idle();
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);
// Parked from a previous firing: retry that value before touching the
// inputs. Returning here releases the worker — the channel's space
// callback re-submits this node once the consumer drains a slot.
if constexpr (!std::is_void_v<return_raw>) {
if (pending_) {
push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
// Whether the value went out or is still parked, finish_firing
// reads pending_ and picks the right follow-up: output space if
// still holding, input readiness if drained. Resubmitting
// unconditionally would fire a node whose inputs are empty, and
// pop_one reports an empty channel as ChannelClosedError — which
// this node treats as "upstream finished" and self-stops on.
finish_firing();
return;
}
}
// See the equivalent guard in PoolNode::fire_once: a space-callback
// wake with nothing parked must release the worker, not fall through
// into pop_inputs on an empty channel.
if constexpr (input_count > 0) {
if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
// finish_firing re-checks readiness after the work above, so
// data that landed while we looked is not missed.
finish_firing();
return;
}
}
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 ChannelEmptyError&) {
// Not an error: there was simply nothing to take. Release and wait
// to be woken again. fire_once checks readiness before it gets
// here, and this node is the sole consumer of its inputs, so this
// is unreachable today — it exists so that if the check is ever
// weakened the cost is a wasted firing rather than a dead node.
finish_firing();
return;
} catch (const ChannelClosedError&) {
fire_callbacks(closed_callbacks_);
self_stop();
return;
} catch (const ChannelOverflowError&) {
fire_callbacks(event_callbacks_);
} catch (...) {
auto eptr = std::current_exception();
const bool handled =
(error_handler_ && error_handler_(name_, eptr)) ||
(net_error_handler_ && net_error_handler_(name_, eptr));
if (handled) {
} else {
fire_callbacks(closed_callbacks_);
self_stop();
return;
}
}
stats_.exec_start_us.store(0, std::memory_order_relaxed);
// If the push above parked, finish_firing waits on output space rather
// than input arrival: this firing consumed its input, so an input-level
// check would not resubmit and the node would hold its value forever
// while its consumer waits for exactly that value.
finish_firing();
}
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)) {
// try_pop_now returns false for "nothing available", which covers
// two very different situations. A closed channel means upstream is
// finished and this node should stop. An open one means only that
// nothing is here at this instant — and treating that as closed
// kills a live node, which then disables its own inputs and outputs
// and takes the rest of the pipeline with it.
if (ch.is_accepting()) throw ChannelEmptyError{};
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>
/// Pushes what it can and parks the rest. `pending_done_` marks the elements
/// already taken, so a retry never re-pushes one — a duplicate is as wrong as
/// a drop and harder to notice.
void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
bool all = true;
((pending_done_[Is] = pending_done_[Is] ||
push_one_out<Is>(std::get<Is>(std::move(result))),
all = all && pending_done_[Is]), ...);
if (all) { pending_.reset(); pending_done_.fill(false); }
// The retry path calls this as push_outputs(std::move(*pending_), …), so
// on that path `result` *is* the parked tuple. Assigning it to itself is
// a self-move-assignment, which for std::tuple is elementwise — and
// libstdc++'s std::vector does not guard against it: it swaps its data
// into a temporary and leaves the vector empty. A value that failed to
// push twice would therefore be delivered with its payload silently
// erased, which downstream reads as a legitimately empty result rather
// than as a loss. Only store when it is not already stored.
else if (!pending_ || &result != &*pending_) pending_ = std::move(result);
}
/// Returns false when the ring was full and the value was NOT taken; the
/// caller must keep it and retry after the channel signals space.
template<std::size_t I>
bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
auto* ch = std::get<I>(output_channels_);
if (!ch) return true;
// 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)) {
// A refused sentinel is a protocol error, not backpressure, so it
// is reported rather than parked and retried — retrying would spin
// forever against a slot only the consumer can free, and there is
// no correct value to deliver second anyway. Closed is normal
// during teardown and stays quiet.
if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
::SentinelResult::SlotBusy)
fire_callbacks(event_callbacks_);
return true;
}
// See the note on the typed overload above: park rather than block.
return ch->try_push(val);
}
Obj& obj_;
std::shared_ptr<IScheduler> scheduler_;
std::string name_;
std::size_t fifo_capacity_;
input_channels_t input_channels_;
output_channels_t output_channels_{};
std::atomic<bool> stop_flag_{true};
/// Serialises firings and records wakes that arrive during one. See
/// submit_gate.hpp for why this cannot be two separate flags.
SubmitGate gate_;
/// Whether prepare() has installed the channel callbacks. Only ever touched
/// from the thread driving start()/stop(), never from a worker, and never
/// cleared: the callbacks capture `this` and stay valid across a restart, so
/// re-registering them would be a pointless write to a live channel.
bool prepared_{false};
/// Thread currently inside fire_once, or a default id when none is.
/// See await_quiescence.
std::atomic<std::thread::id> firing_thread_{};
/// The hidden one-slot output buffer (see push_outputs). Holding the value
/// here is what lets a node stop running without dropping it or occupying a
/// scheduler worker. One slot suffices because at most one fire_once() runs
/// per node at a time — with concurrent firing it would have to hold a whole
/// FIFO's worth.
std::conditional_t<std::is_void_v<return_raw>, std::monostate,
std::optional<return_tuple>> pending_{};
std::array<bool, (output_count ? output_count : 1)> pending_done_{};
NodeStats stats_;
NodeErrorHandler error_handler_;
NodeErrorHandler net_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
Reference in New Issue View Git Blame Copy Permalink