dtourolle
c9aa246322
139bfbb made the channel refuse a sentinel offered while one was still
pending — correct, and the reason is a data race: overwriting wrote
eof_value_ while the consumer could be moving the previous one out of it,
which ThreadSanitizer reports as a torn refcount and a heap-use-after-free.
That part stands.
What was wrong was the accounting. The refusal recorded a drop, and PoolNode
reported it through the overflow event callback, on the theory that two
control tokens on one channel means the stream ended twice and is a caller
protocol error.
It is not. A source that has reached the end of its input keeps being polled
and keeps returning EOF — that is the normal steady state, not an error — so
the token is re-offered on every firing. Refusing a re-offer loses nothing:
the pending token carries the same meaning and is already on its way.
Found by running it. scene-actor-extraction on a 14 s clip reported
[main] ERROR: frames were dropped (channel overflow):
frame_source: 2
scene_annotate: 1
[main] The output would describe footage that was never analysed.
Refusing to report success.
and exited 2, on a run where nothing had been dropped and every frame was
analysed. frame_source emits EOF once and then returns it forever
(frame_source_node.hpp:76), so the count grows with however many times the
source is polled after the end. The pipeline's own loss detector — which
exists because a dropped frame silently corrupts the output — was being
tripped by a clean run, which is the one thing a loss detector must not do.
So SlotBusy now records nothing and reports nothing. The cost, stated
plainly: a genuinely distinct second token would also be refused silently,
and the channel cannot tell a re-offer from a distinct token. Re-offering is
the case that actually occurs; the delivery guarantee that matters — the
first token arrives — holds either way.
The test asserting the old behaviour is inverted rather than deleted, and now
also checks that the token which was accepted is the one delivered. 148/148.
1318 lines
62 KiB
C++
1318 lines
62 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>{});
|
|
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)) {
|
|
// Not parked and not reported. Parking would spin against a slot
|
|
// only the consumer can free; reporting would cry data loss on the
|
|
// normal steady state, since a source at end of input keeps being
|
|
// polled and keeps returning EOF, so the token is re-offered on
|
|
// every firing. Refusing a re-offer loses nothing — the pending
|
|
// token says the same thing. See Channel::try_push_sentinel.
|
|
ch->try_push_sentinel(val);
|
|
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.
|
|
// Closed counts as "stop trying", not as delivered: the value is gone
|
|
// and the channel has recorded the drop. Only Full means park and retry.
|
|
return ch->try_push(val) != Channel<std::tuple_element_t<I, return_tuple>>
|
|
::PushResult::Full;
|
|
}
|
|
|
|
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)) {
|
|
// Not parked and not reported. Parking would spin against a slot
|
|
// only the consumer can free; reporting would cry data loss on the
|
|
// normal steady state, since a source at end of input keeps being
|
|
// polled and keeps returning EOF, so the token is re-offered on
|
|
// every firing. Refusing a re-offer loses nothing — the pending
|
|
// token says the same thing. See Channel::try_push_sentinel.
|
|
ch->try_push_sentinel(val);
|
|
return true;
|
|
}
|
|
// See the note on the typed overload above: park rather than block.
|
|
// Closed counts as "stop trying", not as delivered: the value is gone
|
|
// and the channel has recorded the drop. Only Full means park and retry.
|
|
return ch->try_push(val) != Channel<std::tuple_element_t<I, return_tuple>>
|
|
::PushResult::Full;
|
|
}
|
|
|
|
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
|