#pragma once #include "channel.hpp" #include "diagnostics.hpp" #include "fixed_string.hpp" #include "port.hpp" #include "traits.hpp" #include #include #include #include #include #include #include #include #include #include namespace kpn { // ── INode — type-erased interface for Network / watchdog ───────────────────── struct INode { virtual ~INode() = default; virtual void start() = 0; virtual void stop() = 0; virtual bool running() const = 0; virtual const NodeStats& stats() const = 0; virtual NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const = 0; virtual void set_name(std::string name) = 0; }; // ── Node ───────────────────────────────────────────────────────────────────── // // Template parameters: // Func — the wrapped function (auto NTTP, deduced as a function pointer) // InputNames — optional kpn::in<"a","b"> tag type (at most one) // OutputNames — optional kpn::out<"x","y"> tag type (at most one) template, typename OutputTag = out<>, fixed_string Label = "", std::size_t UniqueTag = 0> class Node; // Specialisation that unpacks the in<>/out<> tag packs template class Node, out, Label, UniqueTag> : public INode { public: using F = decltype(Func); using args_tuple = args_t; using return_raw = return_t; using return_tuple = normalised_return_t; // Identity accessors — used by StaticNetwork for diagnostics and type-level uniqueness 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; static constexpr std::size_t output_count = std::tuple_size_v; static_assert( sizeof...(InNames) == 0 || sizeof...(InNames) == input_count, "make_node: number of input names must match function arity, or provide none" ); static_assert( sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count, "make_node: number of output names must match return tuple size, or provide none" ); explicit Node(std::size_t fifo_capacity = 5) : fifo_capacity_(fifo_capacity) { init_input_channels(std::make_index_sequence{}); } ~Node() override { stop(); } // ── INode ───────────────────────────────────────────────────────────────── void start() override { enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); thread_ = std::jthread([this](std::stop_token) { run_loop(); }); } void stop() override { stop_flag_.store(true, std::memory_order_relaxed); // Disable all input channels: drops queued items and unblocks waiting pop() disable_inputs(std::make_index_sequence{}); if (thread_.joinable()) thread_.request_stop(), thread_.join(); } bool running() const override { return thread_.joinable() && !stop_flag_.load(std::memory_order_relaxed); } void set_name(std::string name) override { name_ = std::move(name); } 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 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, }; } // ── Port access — by index ──────────────────────────────────────────────── template InputPort input() { static_assert(I < input_count, "input index out of range"); return {*this}; } template OutputPort output() { static_assert(I < output_count, "output index out of range"); return {*this}; } // ── Port access — by name ───────────────────────────────────────────────── template auto input() { constexpr std::size_t idx = index_of(); static_assert(idx != npos, "unknown input port name"); return input(); } template auto output() { constexpr std::size_t idx = index_of(); static_assert(idx != npos, "unknown output port name"); return output(); } // ── Internal channel accessors (used by Network at connect time) ────────── template Channel>& input_channel() { return *std::get(input_channels_); } // Replace the owned input channel with an externally provided one. // Used by VariantNodeWrapper to share a Channel with a VariantChannel adapter. template void set_input_channel( std::shared_ptr>> ch) { std::get(input_channels_) = std::move(ch); } template void set_output_channel( Channel>* ch) { std::get(output_channels_) = ch; } private: // ── Channel storage ─────────────────────────────────────────────────────── // Input channels — shared ownership so VariantChannel adapters can share them template void init_input_channels(std::index_sequence) { ((std::get(input_channels_) = std::make_shared>>(fifo_capacity_)), ...); } template void enable_inputs(std::index_sequence) { (std::get(input_channels_)->enable(), ...); } template void disable_inputs(std::index_sequence) { (std::get(input_channels_)->disable(), ...); } template static auto make_input_channel_tuple(std::index_sequence) -> std::tuple>>...>; using input_channels_t = decltype(make_input_channel_tuple( std::make_index_sequence{})); // Output channels — non-owning pointers, set at connect time template static auto make_output_channel_tuple(std::index_sequence) -> std::tuple>*...>; using output_channels_t = decltype(make_output_channel_tuple( std::make_index_sequence{})); // ── run_loop ────────────────────────────────────────────────────────────── void run_loop() { while (!stop_flag_.load(std::memory_order_relaxed)) { try { auto t0 = clock_t::now(); auto args = pop_inputs(std::make_index_sequence{}); auto t1 = clock_t::now(); auto cpu0 = NodeStats::cpu_now(); if constexpr (std::is_void_v) { std::apply(Func, args); } else { auto result = std::apply(Func, args); push_outputs(normalise(std::move(result)), std::make_index_sequence{}); } auto cpu1 = NodeStats::cpu_now(); auto t2 = clock_t::now(); stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); } catch (const ChannelClosedError&) { break; } catch (const ChannelOverflowError& e) { std::cerr << "[kpn] overflow: " << e.what() << "\n"; } } } // Pop all inputs into a tuple of argument values template args_tuple pop_inputs(std::index_sequence) { return {std::get(input_channels_)->pop()...}; } // Normalise return value to tuple (handles void and single-value returns) template static return_tuple normalise(R&& r) { if constexpr (is_tuple_v) return std::move(r); else return std::make_tuple(std::move(r)); } static return_tuple normalise_void() { return {}; } // Push each output element to its connected channel (if connected) template void push_outputs(return_tuple&& result, std::index_sequence) { (push_one(std::get(std::move(result))), ...); } template void push_one(std::tuple_element_t&& val) { auto* ch = std::get(output_channels_); if (!ch) return; try { ch->push(std::move(val)); } catch (const ChannelOverflowError&) { throw ChannelOverflowError(ch->capacity(), "node '" + name_ + "' " + output_port_label()); } } template static std::string output_port_label() { if constexpr (sizeof...(OutNames) > 0) { constexpr std::array names{OutNames.view()...}; return std::string("output['") + std::string(names[I]) + "']"; } else { return "output[" + std::to_string(I) + "]"; } } // ── State ───────────────────────────────────────────────────────────────── std::string name_; std::size_t fifo_capacity_; input_channels_t input_channels_; output_channels_t output_channels_{}; std::atomic stop_flag_{false}; std::jthread thread_; NodeStats stats_; }; // ── ObjectNode — wraps a callable object (functor / class with operator()) ──── // // Use this when the node needs state initialised in a constructor. // The object must outlive the ObjectNode (stored by reference). // // Usage: // MyFunctor obj(...); // auto node = make_node(obj, in<"x">{}, out<"y">{}, capacity); template, typename OutputTag = out<>, fixed_string Label = "", std::size_t UniqueTag = 0> class ObjectNode; template class ObjectNode, out, Label, UniqueTag> : public INode { public: using F = decltype(&Obj::operator()); using args_tuple = args_t; using return_raw = return_t; using return_tuple = normalised_return_t; 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; static constexpr std::size_t output_count = std::tuple_size_v; static_assert( sizeof...(InNames) == 0 || sizeof...(InNames) == input_count, "make_node: number of input names must match operator() arity, or provide none" ); static_assert( sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count, "make_node: number of output names must match return tuple size, or provide none" ); explicit ObjectNode(Obj& obj, std::size_t fifo_capacity = 5) : obj_(obj), fifo_capacity_(fifo_capacity) { init_input_channels(std::make_index_sequence{}); } ~ObjectNode() override { stop(); } // ── INode ───────────────────────────────────────────────────────────────── void start() override { enable_inputs(std::make_index_sequence{}); stop_flag_.store(false, std::memory_order_relaxed); thread_ = std::jthread([this](std::stop_token) { run_loop(); }); } void stop() override { stop_flag_.store(true, std::memory_order_relaxed); disable_inputs(std::make_index_sequence{}); if (thread_.joinable()) thread_.request_stop(), thread_.join(); } bool running() const override { return thread_.joinable() && !stop_flag_.load(std::memory_order_relaxed); } void set_name(std::string name) override { name_ = std::move(name); } 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 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, }; } // ── Port access ─────────────────────────────────────────────────────────── template InputPort input() { static_assert(I < input_count, "input index out of range"); return {*this}; } template OutputPort output() { static_assert(I < output_count, "output index out of range"); return {*this}; } template auto input() { constexpr std::size_t idx = index_of(); static_assert(idx != npos, "unknown input port name"); return input(); } template auto output() { constexpr std::size_t idx = index_of(); static_assert(idx != npos, "unknown output port name"); return output(); } template Channel>& input_channel() { return *std::get(input_channels_); } template void set_input_channel( std::shared_ptr>> ch) { std::get(input_channels_) = std::move(ch); } template void set_output_channel(Channel>* ch) { std::get(output_channels_) = ch; } private: template void init_input_channels(std::index_sequence) { ((std::get(input_channels_) = std::make_shared>>(fifo_capacity_)), ...); } template void enable_inputs(std::index_sequence) { (std::get(input_channels_)->enable(), ...); } template void disable_inputs(std::index_sequence) { (std::get(input_channels_)->disable(), ...); } template static auto make_input_channel_tuple(std::index_sequence) -> std::tuple>>...>; using input_channels_t = decltype(make_input_channel_tuple( std::make_index_sequence{})); template static auto make_output_channel_tuple(std::index_sequence) -> std::tuple>*...>; using output_channels_t = decltype(make_output_channel_tuple( std::make_index_sequence{})); void run_loop() { while (!stop_flag_.load(std::memory_order_relaxed)) { try { auto t0 = clock_t::now(); auto args = pop_inputs(std::make_index_sequence{}); auto t1 = clock_t::now(); auto cpu0 = NodeStats::cpu_now(); if constexpr (std::is_void_v) { std::apply([this](auto&&... a) { obj_(std::forward(a)...); }, args); } else { auto result = std::apply([this](auto&&... a) { return obj_(std::forward(a)...); }, args); push_outputs(normalise(std::move(result)), std::make_index_sequence{}); } auto cpu1 = NodeStats::cpu_now(); auto t2 = clock_t::now(); stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); } catch (const ChannelClosedError&) { break; } catch (const ChannelOverflowError& e) { std::cerr << "[kpn] overflow: " << e.what() << "\n"; } } } template args_tuple pop_inputs(std::index_sequence) { return {std::get(input_channels_)->pop()...}; } template static return_tuple normalise(R&& r) { if constexpr (is_tuple_v) return std::move(r); else return std::make_tuple(std::move(r)); } template void push_outputs(return_tuple&& result, std::index_sequence) { (push_one(std::get(std::move(result))), ...); } template void push_one(std::tuple_element_t&& val) { auto* ch = std::get(output_channels_); if (!ch) return; try { ch->push(std::move(val)); } catch (const ChannelOverflowError&) { throw ChannelOverflowError(ch->capacity(), "node '" + name_ + "' " + output_port_label()); } } template static std::string output_port_label() { if constexpr (sizeof...(OutNames) > 0) { constexpr std::array names{OutNames.view()...}; return std::string("output['") + std::string(names[I]) + "']"; } else { return "output[" + std::to_string(I) + "]"; } } Obj& obj_; std::string name_; std::size_t fifo_capacity_; input_channels_t input_channels_; output_channels_t output_channels_{}; std::atomic stop_flag_{false}; std::jthread thread_; NodeStats stats_; }; // ── make_node overloads for callable objects ────────────────────────────────── template auto make_node(Obj& obj, std::size_t fifo_capacity = 5) { return ObjectNode, out<>>(obj, fifo_capacity); } template auto make_node(Obj& obj, in, std::size_t fifo_capacity = 5) { return ObjectNode, out<>>(obj, fifo_capacity); } template auto make_node(Obj& obj, out, std::size_t fifo_capacity = 5) { return ObjectNode, out>(obj, fifo_capacity); } template auto make_node(Obj& obj, in, out, std::size_t fifo_capacity = 5) { return ObjectNode, out>(obj, fifo_capacity); } // ── make_node factory (NTTP) ────────────────────────────────────────────────── // // Usage: // make_node(capacity) // make_node(capacity) // make_node(capacity) // UniqueTag=1 // make_node(in<"a","b">{}, capacity) // make_node(in<"a","b">{}, out<"x">{}, capacity) // No port names template auto make_node(std::size_t fifo_capacity = 5) { return Node, out<>, Label, UniqueTag>(fifo_capacity); } // in<> only template auto make_node(in, std::size_t fifo_capacity = 5) { return Node, out<>, Label, UniqueTag>(fifo_capacity); } // out<> only template auto make_node(out, std::size_t fifo_capacity = 5) { return Node, out, Label, UniqueTag>(fifo_capacity); } // in<> and out<> template auto make_node(in, out, std::size_t fifo_capacity = 5) { return Node, out, Label, UniqueTag>(fifo_capacity); } } // namespace kpn