dtourolle
abbb2d4770
ThreadPool::stop() ends with queues_.clear(), and submit() went straight to
queues_[target] with no check. A submission arriving after stop indexed an
empty vector and segfaulted.
This is not a contrived teardown ordering. A node's space callback fires from
whichever thread drained the channel, and that thread belongs to the
*consumer*; the callback it runs belongs to the *producer*. Stop the producer
first — which a sources-first shutdown does by design — and the consumer keeps
draining its backlog, firing the producer's space callback into a pool that
has already been torn down:
ThreadPool::submit
<- source node's space_callback
<- Channel<int>::try_pop_now (relay draining its input)
<- relay fire_once
The static-network shutdown case in the next commit crashed about 12 runs in
20 on this. It survived until now because halt() stops in reverse topological
order — consumers first — so the producer whose callback might fire is always
still alive. shutdown() stops sources first and does not have that protection.
Reading stopped_ without a lock would not fix it: the window between the read
and the indexing is exactly where clear() runs. submit() takes a shared lock
and stop() an exclusive one, so submissions still proceed in parallel with
each other while being serialised against teardown. stop() sets the flag under
the lock, releases it to join — a worker's task may itself call submit, and
holding the lock across the join would deadlock against that — then retakes it
to destroy the queues.
Refusals are counted rather than silent. A teardown race is expected, but a
node repeatedly trying to run after its pool is gone is worth being able to
see. try_submit also checks stop_flag_ first, so a stopped node cannot claim
the submit gate and leave it held.
Not a smart-pointer problem, for anyone reading the crash: nothing here is
owned by a raw pointer. It is std::vector::operator[] on a vector that was
emptied by another thread.
Verified in both directions: with the guard removed the new scheduler cases
segfault; with it they pass.
1213 lines
56 KiB
C++
1213 lines
56 KiB
C++
#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>{});
|
|
// 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_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);
|
|
}
|
|
|
|
void fire_once() {
|
|
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};
|
|
|
|
/// 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>{});
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
void fire_once() {
|
|
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};
|
|
|
|
/// 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
|