#pragma once #include "channel.hpp" #include "diagnostics.hpp" #include "fanout.hpp" #include "node.hpp" #include "port.hpp" #include "tmp/fanout_groups.hpp" #include "tmp/topo_sort.hpp" #ifdef KPN_WEB_DEBUG #include "web_debug.hpp" #include #endif #include #include #include #include #include #include namespace kpn { // ── Edge descriptor ─────────────────────────────────────────────────────────── // // Carries compile-time type info and runtime references to the two endpoints. // Constructed by the edge() factory; consumed by make_network(). template struct Edge { using src_node_t = SrcNode; using dst_node_t = DstNode; static constexpr std::size_t src_idx = SrcIdx; static constexpr std::size_t dst_idx = DstIdx; SrcNode& src; DstNode& dst; }; template auto edge(OutputPort, InputPort in) -> Edge; // deduction only; defined below template Edge edge(OutputPort out, InputPort in) { return {out.node, in.node}; } // ── StaticNetwork ───────────────────────────────────────────────────────────── // node_label: returns Label NTTP as string_view if present, else empty. template struct node_label_helper { static constexpr std::string_view value = ""; }; template struct node_label_helper> { static constexpr std::string_view value = NodeT::label(); }; template inline constexpr std::string_view node_label_v = node_label_helper::value; // node_display_name: Label if non-empty, else "node[UniqueTag]" // Returned as std::string at runtime (called once at StaticNetwork construction). template std::string node_display_name() { constexpr std::string_view lbl = node_label_v; if constexpr (!lbl.empty()) { return std::string(lbl); } else if constexpr (requires { NodeT::is_fanout_node; NodeT::unique_tag; }) { return "fanout[" + std::to_string(NodeT::unique_tag) + "]"; } else if constexpr (requires { NodeT::unique_tag; }) { return "node[" + std::to_string(NodeT::unique_tag) + "]"; } else { return "node[?]"; } } template class StaticNetwork : public INode { public: // FanoutStorage = std::tuple, ...> (owned, heap-allocated) // TopoNodeList = tmp::TypeList sources-first StaticNetwork(std::unique_ptr fanouts, std::vector user_nodes_topo, std::vector fanout_ptrs, std::vector user_node_names, std::vector fanout_node_names, std::vector> channel_probes) : fanouts_(std::move(fanouts)) , user_nodes_topo_(std::move(user_nodes_topo)) , fanout_nodes_ptr_(std::move(fanout_ptrs)) , user_node_names_(std::move(user_node_names)) , fanout_node_names_(std::move(fanout_node_names)) , channel_probes_(std::move(channel_probes)) {} ~StaticNetwork() override { stop(); } void start() override { stop_flag_ = false; start_time_ = clock_t::now(); for (auto* n : user_nodes_topo_) n->start(); for (auto* n : fanout_nodes_ptr_) n->start(); #ifdef KPN_WEB_DEBUG web_server_ = std::make_unique( web_debug_port_, [this]() { auto s = collect_snapshots(); return web_debug::to_json(s.nodes, s.channels, s.elapsed_s); }); web_server_->start(); std::cerr << "[kpn] web debug UI: http://localhost:" << web_debug_port_ << "\n"; #endif } void stop() override { stop_flag_ = true; #ifdef KPN_WEB_DEBUG if (web_server_) web_server_->stop(); #endif for (auto it = fanout_nodes_ptr_.rbegin(); it != fanout_nodes_ptr_.rend(); ++it) (*it)->stop(); for (auto it = user_nodes_topo_.rbegin(); it != user_nodes_topo_.rend(); ++it) (*it)->stop(); } bool running() const override { return !stop_flag_; } void set_name(std::string name) override { name_ = std::move(name); } const NodeStats& stats() const override { static NodeStats dummy; return dummy; } NodeSnapshot node_snapshot(const std::string& n, double) const override { return {n, 0, 0, 0, 0, 0, 0, 0}; } #ifdef KPN_WEB_DEBUG void set_web_debug_port(uint16_t port) { web_debug_port_ = port; } #endif // Print diagnostics using compile-time node labels void print_diagnostics(std::ostream& os = std::cerr) const { os << "\n┌─ KPN++ StaticNetwork diagnostics ─────────────────────────────\n"; for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) { auto snap = user_nodes_topo_[i]->node_snapshot(user_node_names_[i], 0.0); os << "│ " << snap.name << " frames=" << snap.frames_processed << " ema=" << snap.ema_exec_ms << "ms\n"; } os << "└────────────────────────────────────────────────────────────────\n"; } FanoutStorage& fanouts_storage() { return *fanouts_; } private: struct Snapshots { std::vector nodes; std::vector channels; double elapsed_s; }; Snapshots collect_snapshots() const { double elapsed_s = std::chrono::duration( clock_t::now() - start_time_).count(); std::vector nodes; for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) nodes.push_back(user_nodes_topo_[i]->node_snapshot(user_node_names_[i], elapsed_s)); for (std::size_t i = 0; i < fanout_nodes_ptr_.size(); ++i) nodes.push_back(fanout_nodes_ptr_[i]->node_snapshot(fanout_node_names_[i], elapsed_s)); std::vector channels; for (auto& probe : channel_probes_) channels.push_back(probe->snapshot()); return {std::move(nodes), std::move(channels), elapsed_s}; } std::string name_; bool stop_flag_{false}; std::unique_ptr fanouts_; std::vector user_nodes_topo_; std::vector fanout_nodes_ptr_; std::vector user_node_names_; std::vector fanout_node_names_; std::vector> channel_probes_; clock_t::time_point start_time_; #ifdef KPN_WEB_DEBUG uint16_t web_debug_port_{9090}; std::unique_ptr web_server_; #endif }; // ── make_network ────────────────────────────────────────────────────────────── template auto make_network(Edges&&... edges) { // 1. Expand edges — detect fan-outs, splice FanoutNodes using FanoutSto = tmp::fanout_storage_t...>; using ExpandedEdges = tmp::expanded_edges_t...>; // 2. Duplicate-tag check — fires before cycle check for a cleaner error message using UserNodes = typename tmp::all_node_types...>>::type; static_assert(!tmp::has_duplicate_tags_v...>, "make_network: two nodes have the same (Func, UniqueTag) — they are " "indistinguishable as graph vertices. Add a UniqueTag: " "make_node(capacity)"); // 3. Cycle check using Topo = tmp::topo_sort; constexpr bool has_cycle = Topo::has_cycle; static_assert(!has_cycle, "make_network: graph contains a directed cycle"); // 4. Construct owned fanout storage on the heap (FanoutNode has jthread — not moveable) auto fanout_storage = std::make_unique(); // 5. Collect unique user node pointers + their display names, in edge-declaration order std::vector user_node_ptrs; std::vector user_node_names; auto collect = [&](auto& e) { using SrcT = std::decay_t; using DstT = std::decay_t; auto* s = static_cast(&e.src); auto* d = static_cast(&e.dst); if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), s) == user_node_ptrs.end()) { user_node_ptrs.push_back(s); user_node_names.push_back(node_display_name()); } if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), d) == user_node_ptrs.end()) { user_node_ptrs.push_back(d); user_node_names.push_back(node_display_name()); } }; (collect(edges), ...); // 5. Wire all expanded SimpleEdges. // find_node: searches fanout storage then user edge pack, returns NodeT*. // Uses if constexpr in a fold so mismatched types never reach assignment. auto find_node = [&]() -> NodeT* { NodeT* ptr = nullptr; std::apply([&](auto&... fn) { ([&](auto& node) { if constexpr (std::is_same_v, NodeT>) if (!ptr) ptr = &node; }(fn), ...); }, *fanout_storage); if (!ptr) { ([&](auto& e) { if (!ptr) { if constexpr (std::is_same_v, NodeT>) ptr = &e.src; else if constexpr (std::is_same_v, NodeT>) ptr = &e.dst; } }(edges), ...); } return ptr; }; // Pre-pass: build fanout_id → source display name map so fanout nodes // can be named after the node feeding them (e.g. "capture_fanout"). std::map fanout_src_name; [&](tmp::TypeList) { ([&](SE) { using DstNode = typename SE::dst_node_t; using SrcNode = typename SE::src_node_t; if constexpr (requires { DstNode::is_fanout_node; }) fanout_src_name.emplace(DstNode::unique_tag, node_display_name()); }(SEs{}), ...); }(ExpandedEdges{}); // Helper: display name for any node type, resolving fanouts to "src_fanout". auto node_name = [&]() -> std::string { if constexpr (requires { NodeT::is_fanout_node; }) { auto it = fanout_src_name.find(NodeT::unique_tag); return it != fanout_src_name.end() ? it->second + "_fanout" : node_display_name(); } else { return node_display_name(); } }; std::vector> channel_probes; auto wire_one = [&](SE) { using SrcNode = typename SE::src_node_t; using DstNode = typename SE::dst_node_t; using out_t = std::tuple_element_t; constexpr std::size_t SrcIdx = SE::src_idx; constexpr std::size_t DstIdx = SE::dst_idx; auto* src = find_node.template operator()(); auto* dst = find_node.template operator()(); if (src && dst) { auto& ch = dst->template input_channel(); src->template set_output_channel(&ch); std::string ch_name = node_name.template operator()() + ":" + std::to_string(SrcIdx) + " \xe2\x86\x92 " // UTF-8 → + node_name.template operator()() + ":" + std::to_string(DstIdx); channel_probes.push_back(std::make_unique>(ch, ch_name)); } }; [&](tmp::TypeList) { (wire_one(SEs{}), ...); }(ExpandedEdges{}); // 6. Collect fanout node pointers and names std::vector fanout_ptrs; std::vector fanout_node_names; std::apply([&](auto&... fn) { ([&](auto& node) { using NodeT = std::decay_t; fanout_ptrs.push_back(static_cast(&node)); fanout_node_names.push_back(node_name.template operator()()); }(fn), ...); }, *fanout_storage); // 7. Construct and return the StaticNetwork using Net = StaticNetwork; return Net(std::move(fanout_storage), std::move(user_node_ptrs), std::move(fanout_ptrs), std::move(user_node_names), std::move(fanout_node_names), std::move(channel_probes)); } } // namespace kpn