dtourolle/KPN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a5c016833ddf4e045700bbd8d4bb7eba893e71fe
KPN/include/kpn/pool_node.hpp
T
dtourolle a5c016833d fix: install channel callbacks before any node runs 
ThreadSanitizer reported ten data races on a plain multi-node network, all
the same one:

  Read  in Channel::try_push          -> push_callback_()      (worker thread)
  Write in Channel::set_push_callback -> push_callback_ = ...  (main thread)

A node's push and space callbacks are std::function members living on
channels it shares with its neighbours. register_callbacks() wrote them from
inside start(), and a network starts its nodes one at a time — so by the
time node N is being started, nodes 1..N-1 are already running and pushing
into N's input channel, reading the very std::function that start() is
assigning. Concurrent read and write of a std::function is a data race on
its vtable pointer and buffer, not a benign one.

This is the cause of the symptom a8cfe73 patched. That commit found nodes
missing their startup wake because "enable_inputs() opens the channel
several statements before register_callbacks() installs the push callback",
and fixed it by re-asking the question with on_input_ready(). The gap it
described is this race: the callback is not merely late, it is being written
while another thread reads it.

INode gains prepare(), which installs callbacks and starts nothing. Networks
call it on every node before starting any of them, so every write happens
while the pipeline is idle and the callbacks are read-only once it is live.
start() calls prepare() itself when a node is used standalone, and prepare()
is idempotent so both paths are safe. The flag is never cleared: the
callbacks capture `this` and stay valid across a restart, so re-registering
them would only add a pointless write to a live channel.

a8cfe73's on_input_ready() stays, and is still needed — a network starts
nodes one at a time, so an upstream node can still push into this one
between its prepare() and its start(), where on_input_ready() returns early
on stop_flag_ and the empty->non-empty edge is spent. It is now a
level-triggered check against a benign ordering rather than cover for a race.

Verified with -DKPN_SANITIZER=thread: ten races before, none of these after,
across the unit suite and the contended channel stress suite. One unrelated
race remains, on overlapping fire_once invocations; it is pre-existing and
is fixed separately.
2026-08-05 13:39:26 +02:00

1151 lines
53 KiB
C++
Raw Blame History

#pragma once
#include "channel.hpp"
#include "diagnostics.hpp"
#include "fixed_string.hpp"
#include "inode.hpp"
#include "port.hpp"
#include "scheduler.hpp"
#include "submit_gate.hpp"
#include "traits.hpp"
#include <array>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <functional>
#include <iostream>
#include <memory>
#include <optional>
#include <variant>
#include <stdexcept>
#include <thread>
#include <tuple>
#include <type_traits>
namespace kpn {
// Sentinel detection (has_eof_field / is_sentinel_value) lives in traits.hpp —
// every node type that forwards values needs it, not just pool-scheduled ones.
// ── PoolNode ──────────────────────────────────────────────────────────────────
//
// Reactive alternative to Node<>. Instead of owning a blocked thread, the node
// is submitted to a shared IScheduler whenever all its input channels become
// non-empty. A single fire_once() call pops all inputs, executes the function,
// and pushes outputs. At most one fire_once() runs at a time (see SubmitGate).
//
// Source nodes (input_count == 0) submit themselves immediately on start() and
// resubmit after each fire_once().
//
// Multiple PoolNodes can share one ThreadPool for resource-bounded execution,
// or each can have a dedicated single-thread pool for serialisation.
template<auto Func,
typename InputTag = in<>,
typename OutputTag = out<>,
fixed_string Label = "",
std::size_t UniqueTag = 0>
class PoolNode;
template<auto Func, fixed_string... InNames, fixed_string... OutNames,
fixed_string Label, std::size_t UniqueTag>
class PoolNode<Func, in<InNames...>, out<OutNames...>, Label, UniqueTag> : public INode {
public:
using F = decltype(Func);
using args_tuple = args_t<F>;
using return_raw = return_t<F>;
using return_tuple = normalised_return_t<return_raw>;
static constexpr std::string_view label() { return Label.view(); }
static constexpr std::size_t unique_tag = UniqueTag;
static constexpr std::size_t input_count = arity_v<F>;
static constexpr std::size_t output_count = std::tuple_size_v<return_tuple>;
static_assert(
sizeof...(InNames) == 0 || sizeof...(InNames) == input_count,
"make_pool_node: number of input names must match function arity, or provide none"
);
static_assert(
sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count,
"make_pool_node: number of output names must match return tuple size, or provide none"
);
explicit PoolNode(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5)
: scheduler_(std::move(sched)), fifo_capacity_(fifo_capacity)
{
init_input_channels(std::make_index_sequence<input_count>{});
}
~PoolNode() override { stop(); }
// ── INode ─────────────────────────────────────────────────────────────────
void prepare() override {
if (prepared_) return; // idempotent: the network calls this,
prepared_ = true; // and start() calls it again if not.
register_callbacks(std::make_index_sequence<input_count>{});
}
void start() override {
prepare();
enable_inputs(std::make_index_sequence<input_count>{});
stop_flag_.store(false, std::memory_order_relaxed);
gate_.force_idle();
if constexpr (input_count == 0)
try_submit(0.5f);
else
// Never start with a wake already outstanding — the startup case of
// the invariant 9c5ce5f established for the running pipeline.
//
// The callback is installed by prepare(), before any node runs, but
// a network still starts its nodes one at a time: an upstream node
// that is already firing can push into this one between the two
// calls. The push is accepted by the ring and does invoke the
// callback, but on_input_ready() sees stop_flag_ still set and
// returns. Every later push sees a non-empty ring and stays silent
// — Channel invokes push_callback_ only on the empty->non-empty
// transition — so without this the node is never submitted and the
// pipeline reads as wedged from the first frame.
//
// on_input_ready() is the level-triggered form of the same
// question, so asking it once here converts the missed edge into a
// state check.
on_input_ready();
}
void stop() override {
stop_flag_.store(true, std::memory_order_seq_cst);
disable_inputs(std::make_index_sequence<input_count>{});
// 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);
// Plain store, not release_and_recheck(): 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) {
if (gate_.claim())
scheduler_->submit([this] { fire_once(); }, priority);
}
/// End this firing, honouring any wake recorded during it. Every path that
/// finishes or parks a firing must release the node through here rather
/// than touching the gate directly. When a wake was recorded the gate stays
/// claimed and is handed to the next firing, so the node is never
/// momentarily idle with work outstanding.
void release_and_recheck(float priority = 0.5f) {
if (gate_.release())
scheduler_->submit([this] { fire_once(); }, priority);
}
// ── Execution ─────────────────────────────────────────────────────────────
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>{});
release_and_recheck();
if (pending_) {
// Close the lost-wakeup race: a space_callback that fired
// between the failed push and releasing the gate was
// swallowed, and nothing else will wake this node. Re-check
// now that the flag is down.
if (outputs_have_space(std::make_index_sequence<output_count>{}))
try_submit(0.5f);
return; // parked
}
// Drained: resume normal firing, resubmitting exactly the way
// the normal tail below does. An unconditional try_submit here
// would fire a node whose inputs are empty, and pop_inputs
// reports an empty channel as ChannelClosedError — which this
// node treats as "upstream finished" and self-stops on. That
// is a live node killing itself purely because it was woken by
// *output* space rather than by input arrival.
if constexpr (input_count == 0) try_submit(0.5f);
else on_input_ready();
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) {
release_and_recheck();
on_input_ready(); // data may have arrived while we checked
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 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);
release_and_recheck();
if (stop_flag_.load(std::memory_order_relaxed)) return;
// Parked by the push above. Same situation as the retry path at the top
// of fire_once — and the same lost-wakeup race, which that path closes
// and this one did not. A space callback that fired while the gate was
// still claimed is recorded there, and the resubmit below
// cannot cover it: this firing consumed its input, so inputs are empty
// and on_input_ready() will not resubmit. The node would then hold its
// value forever while its consumer waits for exactly that value and its
// producer parks on an input channel that never drains. Re-check now
// that the flag is down.
if constexpr (!std::is_void_v<return_raw>) {
if (pending_) {
if (outputs_have_space(std::make_index_sequence<output_count>{}))
try_submit(0.5f);
return; // parked
}
}
// Source nodes always resubmit; others resubmit only if inputs are ready.
if constexpr (input_count == 0) {
try_submit(0.5f);
} else {
on_input_ready();
}
}
// Pop all inputs — safe because we're the sole consumer and fire_once
// is guarded by 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))
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)) {
ch->push_sentinel(std::move(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.
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);
// Plain store, not release_and_recheck(): 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) {
if (gate_.claim())
scheduler_->submit([this] { fire_once(); }, priority);
}
/// End this firing, honouring any wake recorded during it. Every path that
/// finishes or parks a firing must release the node through here rather
/// than touching the gate directly. When a wake was recorded the gate stays
/// claimed and is handed to the next firing, so the node is never
/// momentarily idle with work outstanding.
void release_and_recheck(float priority = 0.5f) {
if (gate_.release())
scheduler_->submit([this] { fire_once(); }, priority);
}
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>{});
release_and_recheck();
if (pending_) {
// Close the lost-wakeup race: a space_callback that fired
// between the failed push and releasing the gate was
// swallowed, and nothing else will wake this node. Re-check
// now that the flag is down.
if (outputs_have_space(std::make_index_sequence<output_count>{}))
try_submit(0.5f);
return; // parked
}
// Drained: resume normal firing, resubmitting exactly the way
// the normal tail below does. An unconditional try_submit here
// would fire a node whose inputs are empty, and pop_inputs
// reports an empty channel as ChannelClosedError — which this
// node treats as "upstream finished" and self-stops on. That
// is a live node killing itself purely because it was woken by
// *output* space rather than by input arrival.
if constexpr (input_count == 0) try_submit(0.5f);
else on_input_ready();
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) {
release_and_recheck();
on_input_ready();
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 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);
release_and_recheck();
if (stop_flag_.load(std::memory_order_relaxed)) return;
// Parked by the push above. Same situation as the retry path at the top
// of fire_once — and the same lost-wakeup race, which that path closes
// and this one did not. A space callback that fired while the gate was
// still claimed is recorded there, and the resubmit below
// cannot cover it: this firing consumed its input, so inputs are empty
// and on_input_ready() will not resubmit. The node would then hold its
// value forever while its consumer waits for exactly that value and its
// producer parks on an input channel that never drains. Re-check now
// that the flag is down.
if constexpr (!std::is_void_v<return_raw>) {
if (pending_) {
if (outputs_have_space(std::make_index_sequence<output_count>{}))
try_submit(0.5f);
return; // parked
}
}
if constexpr (input_count == 0) try_submit(0.5f);
else on_input_ready();
}
template<std::size_t... Is>
args_tuple pop_inputs(std::index_sequence<Is...>) { return {pop_one<Is>()...}; }
template<std::size_t I>
std::tuple_element_t<I, args_tuple> pop_one() {
auto& ch = *std::get<I>(input_channels_);
std::tuple_element_t<I, args_tuple> val;
if (!ch.try_pop_now(val)) throw ChannelClosedError{};
return val;
}
template<typename R = return_raw>
static return_tuple normalise(R&& r) {
if constexpr (is_tuple_v<R>) return std::move(r);
else return std::make_tuple(std::move(r));
}
template<std::size_t... Is>
/// 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)) {
ch->push_sentinel(std::move(val));
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
Reference in New Issue View Git Blame Copy Permalink