744 lines
31 KiB
C++
744 lines
31 KiB
C++
#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 <stdexcept>
|
|
#include <thread>
|
|
#include <tuple>
|
|
#include <type_traits>
|
|
|
|
namespace kpn {
|
|
|
|
// ── 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);
|
|
}
|
|
|
|
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_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,
|
|
};
|
|
}
|
|
|
|
// ── 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...>) {
|
|
(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);
|
|
}
|
|
|
|
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);
|
|
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) + ...);
|
|
}
|
|
|
|
float compute_priority() {
|
|
if constexpr (input_count == 0) return 0.5f;
|
|
float sum = 0.0f;
|
|
sum_fill(sum, std::make_index_sequence<input_count>{});
|
|
return sum / static_cast<float>(input_count);
|
|
}
|
|
|
|
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), ...);
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
// ── 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);
|
|
|
|
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 (...) {
|
|
if (error_handler_ && error_handler_(name_, std::current_exception())) {
|
|
// continue — fall through to resubmit check
|
|
} else {
|
|
fire_callbacks(closed_callbacks_);
|
|
self_stop();
|
|
return;
|
|
}
|
|
}
|
|
|
|
stats_.exec_start_us.store(0, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_release);
|
|
|
|
if (stop_flag_.load(std::memory_order_relaxed)) return;
|
|
|
|
// 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>
|
|
void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
|
|
(push_one_out<Is>(std::get<Is>(std::move(result))), ...);
|
|
}
|
|
|
|
template<std::size_t I>
|
|
void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
|
|
auto* ch = std::get<I>(output_channels_);
|
|
if (!ch) return;
|
|
try {
|
|
ch->push(std::move(val));
|
|
} catch (const ChannelOverflowError&) {
|
|
throw ChannelOverflowError(ch->capacity(),
|
|
"pool node '" + name_ + "' " + output_port_label<I>());
|
|
}
|
|
}
|
|
|
|
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};
|
|
NodeStats stats_;
|
|
NodeErrorHandler 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);
|
|
}
|
|
|
|
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_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,
|
|
};
|
|
}
|
|
|
|
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...>) {
|
|
(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);
|
|
}
|
|
|
|
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);
|
|
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) + ...);
|
|
}
|
|
|
|
float compute_priority() {
|
|
if constexpr (input_count == 0) return 0.5f;
|
|
float sum = 0.0f;
|
|
sum_fill(sum, std::make_index_sequence<input_count>{});
|
|
return sum / static_cast<float>(input_count);
|
|
}
|
|
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), ...);
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
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);
|
|
|
|
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 (...) {
|
|
if (error_handler_ && error_handler_(name_, std::current_exception())) {
|
|
} else {
|
|
fire_callbacks(closed_callbacks_);
|
|
self_stop();
|
|
return;
|
|
}
|
|
}
|
|
|
|
stats_.exec_start_us.store(0, std::memory_order_relaxed);
|
|
queued_.store(false, std::memory_order_release);
|
|
if (stop_flag_.load(std::memory_order_relaxed)) return;
|
|
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>
|
|
void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
|
|
(push_one_out<Is>(std::get<Is>(std::move(result))), ...);
|
|
}
|
|
template<std::size_t I>
|
|
void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
|
|
auto* ch = std::get<I>(output_channels_);
|
|
if (!ch) return;
|
|
try {
|
|
ch->push(std::move(val));
|
|
} catch (const ChannelOverflowError&) {
|
|
throw ChannelOverflowError(ch->capacity(),
|
|
"pool node '" + name_ + "'");
|
|
}
|
|
}
|
|
|
|
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};
|
|
NodeStats stats_;
|
|
NodeErrorHandler 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
|