dtourolle/KPN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5628447ea833e06eaf4ad116c14d1595f6183a69
KPN/include/kpn/pool_node.hpp
T
dtourolle 5628447ea8 fix: never self-move the parked output tuple 
push_outputs ended with

    else pending_ = std::move(result);

and the retry path calls it as push_outputs(std::move(*pending_), …), so on
that path `result` is the parked tuple itself. The assignment was a
self-move-assignment. std::tuple's is elementwise, and libstdc++'s
std::vector does not guard against self-move: _M_move_assign swaps its data
into a temporary, which is then destroyed. The vector ends up empty.

So the first park was clean — the argument there is a local temporary — and
the second erased the payload. The value was still delivered, still in
order, still counted, just empty. Downstream cannot distinguish that from a
frame on which the node genuinely found nothing, which is why it would never
surface as an error: in scene-actor-extraction it reads as "no faces in this
frame" and the run completes with a quietly wrong answer.

Scope, stated precisely because I first got it wrong: this needs a node with
*two or more* outputs. With one output the only thing that resubmits a
parked node is that output's own space callback, which by definition fires
when there is room, so the retry always succeeds and never reassigns. With
two, output A draining resubmits the node while output B is still full — the
retry skips A (already delivered, tracked in pending_done_) and fails on B,
and that is the reassignment that eats B's payload.

Every node in the scene-actor-extraction pipeline currently has exactly one
output, and the fanout is a separate class that does not use pending_, so
this is latent there rather than active. It is reachable by any multi-output
node under backpressure, which the library supports and documents.

Verified in both directions: on 6a4f45f the parked payload arrives with size
0; here it arrives intact. The test drives raw channels rather than consumer
nodes so each step is forced rather than raced, and both channels are
capacity 1 — Channel fires the space callback only on the full->not-full
edge, so a roomy channel A would never resubmit the node and the retry would
never happen at all.
2026-08-05 13:10:41 +02:00

1135 lines
52 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 "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 (queued_ flag).
//
// Source nodes (input_count == 0) submit themselves immediately on start() and
// resubmit after each fire_once().
//
// Multiple PoolNodes can share one ThreadPool for resource-bounded execution,
// or each can have a dedicated single-thread pool for serialisation.
template<auto Func,
typename InputTag = in<>,
typename OutputTag = out<>,
fixed_string Label = "",
std::size_t UniqueTag = 0>
class PoolNode;
template<auto Func, fixed_string... InNames, fixed_string... OutNames,
fixed_string Label, std::size_t UniqueTag>
class PoolNode<Func, in<InNames...>, out<OutNames...>, Label, UniqueTag> : public INode {
public:
using F = decltype(Func);
using args_tuple = args_t<F>;
using return_raw = return_t<F>;
using return_tuple = normalised_return_t<return_raw>;
static constexpr std::string_view label() { return Label.view(); }
static constexpr std::size_t unique_tag = UniqueTag;
static constexpr std::size_t input_count = arity_v<F>;
static constexpr std::size_t output_count = std::tuple_size_v<return_tuple>;
static_assert(
sizeof...(InNames) == 0 || sizeof...(InNames) == input_count,
"make_pool_node: number of input names must match function arity, or provide none"
);
static_assert(
sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count,
"make_pool_node: number of output names must match return tuple size, or provide none"
);
explicit PoolNode(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5)
: scheduler_(std::move(sched)), fifo_capacity_(fifo_capacity)
{
init_input_channels(std::make_index_sequence<input_count>{});
}
~PoolNode() override { stop(); }
// ── INode ─────────────────────────────────────────────────────────────────
void start() override {
enable_inputs(std::make_index_sequence<input_count>{});
stop_flag_.store(false, std::memory_order_relaxed);
queued_.store(false, std::memory_order_relaxed);
register_callbacks(std::make_index_sequence<input_count>{});
if constexpr (input_count == 0)
try_submit(0.5f);
else
// Never start with a wake already outstanding — the startup case of
// the invariant 9c5ce5f established for the running pipeline.
//
// enable_inputs() opens the channel several statements before
// register_callbacks() installs the push callback, and the network
// starts nodes sources-first, so an upstream node is already firing
// into this one during that gap. A push landing there is accepted by
// the ring but wakes nobody: Channel::push only invokes the callback
// on the empty→non-empty transition, and at that instant the
// callback is still null. Every later push sees a non-empty ring and
// stays silent, so the node is never submitted — the pipeline reads
// as wedged from the first frame, with no item ever delivered.
//
// 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,
queued_.load(std::memory_order_relaxed),
wake_pending_.load(std::memory_order_relaxed),
};
}
// ── 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.
queued_.store(false, std::memory_order_release);
stop_flag_.store(true, std::memory_order_relaxed);
}
template<typename Tup, std::size_t... Is>
static auto make_input_channel_tuple(std::index_sequence<Is...>)
-> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;
using input_channels_t = decltype(make_input_channel_tuple<args_tuple>(
std::make_index_sequence<input_count>{}));
template<typename Tup, std::size_t... Is>
static auto make_output_channel_tuple(std::index_sequence<Is...>)
-> std::tuple<Channel<std::tuple_element_t<Is, Tup>>*...>;
using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
std::make_index_sequence<output_count>{}));
// ── Scheduling ────────────────────────────────────────────────────────────
// Called by channel push_callbacks (on the producer's thread).
void on_input_ready() {
if (stop_flag_.load(std::memory_order_relaxed)) return;
std::size_t ready = count_ready(std::make_index_sequence<input_count>{});
if (ready == input_count)
try_submit(compute_priority());
}
template<std::size_t... Is>
std::size_t count_ready(std::index_sequence<Is...>) {
return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
}
/// 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 already queued. A wake that arrives while this node is
/// queued or running is *recorded*, never dropped.
///
/// Wakes are edge-triggered: a channel fires its space callback on the
/// transition, once. If that lands while queued_ is up, the CAS below fails
/// and — before wake_pending_ — the wake was gone. A node could then 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. Recording the drop turns the signal level-triggered: the
/// invariant is that a node never sleeps with a wake outstanding, enforced
/// by release_and_recheck() at every point that releases the node.
void try_submit(float priority) {
bool expected = false;
if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
scheduler_->submit([this] { fire_once(); }, priority);
else
wake_pending_.store(true, std::memory_order_release);
}
/// Clear queued_, then honour any wake that was dropped while it was up.
/// Every path that finishes or parks a firing must release the node through
/// here rather than storing queued_ directly.
void release_and_recheck(float priority = 0.5f) {
queued_.store(false, std::memory_order_release);
if (wake_pending_.exchange(false, std::memory_order_acq_rel))
try_submit(priority);
}
// ── Execution ─────────────────────────────────────────────────────────────
void fire_once() {
if (stop_flag_.load(std::memory_order_relaxed)) {
queued_.store(false, std::memory_order_release);
return;
}
// Record queue wait time (submission → now) and mark as executing
auto t0 = clock_t::now();
int64_t now_us = std::chrono::duration_cast<std::chrono::microseconds>(
t0.time_since_epoch()).count();
stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
// 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 clearing queued_ 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 queued_ was
// still up got swallowed by try_submit's CAS, 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 queued_ (only one fire_once runs at a time).
template<std::size_t... Is>
args_tuple pop_inputs(std::index_sequence<Is...>) {
return {pop_one<Is>()...};
}
template<std::size_t I>
std::tuple_element_t<I, args_tuple> pop_one() {
auto& ch = *std::get<I>(input_channels_);
std::tuple_element_t<I, args_tuple> val;
if (!ch.try_pop_now(val))
throw ChannelClosedError{};
return val;
}
template<typename R = return_raw>
static return_tuple normalise(R&& r) {
if constexpr (is_tuple_v<R>) return std::move(r);
else return std::make_tuple(std::move(r));
}
template<std::size_t... Is>
/// 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};
std::atomic<bool> queued_{false};
/// A wake that arrived while queued_ was up. See try_submit.
std::atomic<bool> wake_pending_{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 start() override {
enable_inputs(std::make_index_sequence<input_count>{});
stop_flag_.store(false, std::memory_order_relaxed);
queued_.store(false, std::memory_order_relaxed);
register_callbacks(std::make_index_sequence<input_count>{});
if constexpr (input_count == 0)
try_submit(0.5f);
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,
queued_.load(std::memory_order_relaxed),
wake_pending_.load(std::memory_order_relaxed),
};
}
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.
queued_.store(false, std::memory_order_release);
stop_flag_.store(true, std::memory_order_relaxed);
}
template<typename Tup, std::size_t... Is>
static auto make_input_channel_tuple(std::index_sequence<Is...>)
-> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;
using input_channels_t = decltype(make_input_channel_tuple<args_tuple>(
std::make_index_sequence<input_count>{}));
template<typename Tup, std::size_t... Is>
static auto make_output_channel_tuple(std::index_sequence<Is...>)
-> std::tuple<Channel<std::tuple_element_t<Is, Tup>>*...>;
using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
std::make_index_sequence<output_count>{}));
void on_input_ready() {
if (stop_flag_.load(std::memory_order_relaxed)) return;
std::size_t ready = count_ready(std::make_index_sequence<input_count>{});
if (ready == input_count) try_submit(compute_priority());
}
template<std::size_t... Is>
std::size_t count_ready(std::index_sequence<Is...>) {
return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
}
/// 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 already queued. A wake that arrives while this node is
/// queued or running is *recorded*, never dropped.
///
/// Wakes are edge-triggered: a channel fires its space callback on the
/// transition, once. If that lands while queued_ is up, the CAS below fails
/// and — before wake_pending_ — the wake was gone. A node could then 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. Recording the drop turns the signal level-triggered: the
/// invariant is that a node never sleeps with a wake outstanding, enforced
/// by release_and_recheck() at every point that releases the node.
void try_submit(float priority) {
bool expected = false;
if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
scheduler_->submit([this] { fire_once(); }, priority);
else
wake_pending_.store(true, std::memory_order_release);
}
/// Clear queued_, then honour any wake that was dropped while it was up.
/// Every path that finishes or parks a firing must release the node through
/// here rather than storing queued_ directly.
void release_and_recheck(float priority = 0.5f) {
queued_.store(false, std::memory_order_release);
if (wake_pending_.exchange(false, std::memory_order_acq_rel))
try_submit(priority);
}
void fire_once() {
if (stop_flag_.load(std::memory_order_relaxed)) {
queued_.store(false, std::memory_order_release);
return;
}
auto t0 = clock_t::now();
int64_t now_us = std::chrono::duration_cast<std::chrono::microseconds>(
t0.time_since_epoch()).count();
stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
// 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 clearing queued_ 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 queued_ was
// still up got swallowed by try_submit's CAS, 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};
std::atomic<bool> queued_{false};
/// A wake that arrived while queued_ was up. See try_submit.
std::atomic<bool> wake_pending_{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