dtourolle
7b7f631e6d
SharedResource::acquire() blocks on a condition variable whose predicate only becomes true when release() hands over ownership. No timeout, no stop condition. A node parked there was not observing stop flags, so teardown had no way to reach it: the worker never returned, the pool's join never completed, and shutdown waited on a resource nobody was going to release — which is precisely the situation when the holder is being stopped too. close() wakes every waiter and refuses further acquisitions, and the waiters leave through ResourceClosedError, which is an exception the node error path already handles rather than a new mechanism. StaticNetwork calls it on registered resources at the top of halt() and shutdown(), before stopping any node, since a node stopped while parked cannot respond to being stopped. The handover needed care in two places. A waiter woken by close() has not been given ownership, so it takes no Guard and leaves held_ exactly as it found it; and release() now skips handing over to waiters when closed, because handing ownership to a thread that is on its way out would leave held_ true with nobody holding it. reopen() is there for reuse across runs, which the persistent-pipeline work will want; teardown does not need it. Verified in both directions: without close() the waiter thread never returns and the test's join blocks; with it the waiter leaves through ResourceClosedError while the holder still has the resource. 145/145.
514 lines
23 KiB
C++
514 lines
23 KiB
C++
#pragma once
|
|
#include "channel.hpp"
|
|
#include "diagnostics.hpp"
|
|
#include "fanout.hpp"
|
|
#include "inode.hpp"
|
|
#include "port.hpp"
|
|
#include "tmp/fanout_groups.hpp"
|
|
#include "tmp/topo_sort.hpp"
|
|
|
|
#ifdef KPN_WEB_DEBUG
|
|
#include "web_debug.hpp"
|
|
#include <memory>
|
|
#endif
|
|
|
|
#include <iostream>
|
|
#include <map>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <tuple>
|
|
#include <type_traits>
|
|
#include <utility>
|
|
#include <vector>
|
|
|
|
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<typename SrcNode, std::size_t SrcIdx,
|
|
typename DstNode, std::size_t DstIdx>
|
|
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<typename SrcNode, std::size_t SrcIdx,
|
|
typename DstNode, std::size_t DstIdx>
|
|
auto edge(OutputPort<SrcNode, SrcIdx>, InputPort<DstNode, DstIdx> in)
|
|
-> Edge<SrcNode, SrcIdx, DstNode, DstIdx>; // deduction only; defined below
|
|
|
|
template<typename SrcNode, std::size_t SrcIdx,
|
|
typename DstNode, std::size_t DstIdx>
|
|
Edge<SrcNode, SrcIdx, DstNode, DstIdx>
|
|
edge(OutputPort<SrcNode, SrcIdx> out, InputPort<DstNode, DstIdx> in) {
|
|
return {out.node, in.node};
|
|
}
|
|
|
|
// ── StaticNetwork ─────────────────────────────────────────────────────────────
|
|
|
|
// node_label<NodeT>: returns Label NTTP as string_view if present, else empty.
|
|
template<typename NodeT, typename = void>
|
|
struct node_label_helper {
|
|
static constexpr std::string_view value = "";
|
|
};
|
|
template<typename NodeT>
|
|
struct node_label_helper<NodeT, std::void_t<decltype(NodeT::label())>> {
|
|
static constexpr std::string_view value = NodeT::label();
|
|
};
|
|
|
|
template<typename NodeT>
|
|
inline constexpr std::string_view node_label_v = node_label_helper<NodeT>::value;
|
|
|
|
// node_display_name<NodeT, UniqueTag>: Label if non-empty, else "node[UniqueTag]"
|
|
// Returned as std::string at runtime (called once at StaticNetwork construction).
|
|
template<typename NodeT>
|
|
std::string node_display_name() {
|
|
constexpr std::string_view lbl = node_label_v<NodeT>;
|
|
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::is_router_node; NodeT::unique_tag; }) {
|
|
return "router[" + std::to_string(NodeT::unique_tag) + "]";
|
|
} else if constexpr (requires { NodeT::is_filter_node; NodeT::unique_tag; }) {
|
|
return "filter[" + 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<typename FanoutStorage, typename TopoNodeList>
|
|
class StaticNetwork : public INode {
|
|
public:
|
|
// FanoutStorage = std::tuple<FanoutNode<T0,N0>, ...> (owned, heap-allocated)
|
|
// TopoNodeList = tmp::TypeList<NodeA, NodeB, ...> sources-first
|
|
|
|
StaticNetwork(std::unique_ptr<FanoutStorage> fanouts,
|
|
std::vector<INode*> user_nodes_topo,
|
|
std::vector<INode*> fanout_ptrs,
|
|
std::vector<std::string> user_node_names,
|
|
std::vector<std::string> fanout_node_names,
|
|
std::vector<std::unique_ptr<IChannelProbe>> channel_probes,
|
|
std::vector<std::string> channel_src_names)
|
|
: 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))
|
|
, channel_src_names_(std::move(channel_src_names))
|
|
{}
|
|
|
|
~StaticNetwork() override { stop(); }
|
|
|
|
void start() override {
|
|
stop_flag_ = false;
|
|
start_time_ = clock_t::now();
|
|
if (event_handler_) {
|
|
for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) {
|
|
auto* node = user_nodes_topo_[i];
|
|
const auto& n = user_node_names_[i];
|
|
node->set_network_overflow_callback(
|
|
[this, n](auto ts) { event_handler_(n, NodeEvent::Overflow, ts); });
|
|
node->set_network_closed_callback(
|
|
[this, n](auto ts) { event_handler_(n, NodeEvent::Closed, ts); });
|
|
}
|
|
}
|
|
if (error_handler_) {
|
|
for (auto* node : user_nodes_topo_)
|
|
node->set_network_error_callback(error_handler_);
|
|
}
|
|
// Install every node's channel callbacks before starting any of them.
|
|
// Those callbacks are std::function members on channels shared with
|
|
// neighbours; a neighbour that is already running reads them from its
|
|
// own thread, so writing one after the pipeline is live is a data race
|
|
// (ThreadSanitizer reports it on any multi-node network). Doing all the
|
|
// writes here, while nothing runs, makes them read-only thereafter.
|
|
for (auto* n : user_nodes_topo_) n->prepare();
|
|
for (auto* n : fanout_nodes_ptr_) n->prepare();
|
|
|
|
for (auto* n : user_nodes_topo_) n->start();
|
|
for (auto* n : fanout_nodes_ptr_) n->start();
|
|
#ifdef KPN_WEB_DEBUG
|
|
if (web_server_enabled_) {
|
|
web_server_ = std::make_unique<web_debug::WebDebugServer>(
|
|
web_debug_port_,
|
|
[this]() {
|
|
auto s = collect_snapshots();
|
|
return web_debug::to_json(s.nodes, s.channels, s.resources, s.elapsed_s, s.pools);
|
|
});
|
|
web_server_->start();
|
|
std::cerr << "[kpn] web debug UI: http://localhost:" << web_debug_port_ << "\n";
|
|
}
|
|
#endif
|
|
}
|
|
|
|
void stop() override { halt(); }
|
|
|
|
void halt() override {
|
|
stop_flag_ = true;
|
|
#ifdef KPN_WEB_DEBUG
|
|
if (web_server_) web_server_->stop();
|
|
#endif
|
|
// Release anything parked on a shared resource first. A node blocked in
|
|
// acquire() is not watching stop flags, so stopping it would wait on a
|
|
// handover that may never come — its holder is being stopped too.
|
|
for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); }
|
|
|
|
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();
|
|
}
|
|
|
|
// shutdown(): graceful drain in topological order (sources first).
|
|
// Stops source nodes, polls channels until empty, then stops each downstream layer.
|
|
void shutdown() override {
|
|
stop_flag_ = true;
|
|
#ifdef KPN_WEB_DEBUG
|
|
if (web_server_) web_server_->stop();
|
|
#endif
|
|
for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); }
|
|
// user_nodes_topo_ is already in sources-first order.
|
|
// Stop each node and drain its output channels before moving on.
|
|
for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) {
|
|
user_nodes_topo_[i]->stop();
|
|
drain_outputs_of(user_node_names_[i]);
|
|
}
|
|
for (auto* n : fanout_nodes_ptr_) n->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};
|
|
}
|
|
|
|
using EventHandler =
|
|
std::function<void(std::string_view node_name, NodeEvent,
|
|
std::chrono::steady_clock::time_point)>;
|
|
|
|
void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
|
|
|
|
/// How long shutdown() waits for one node's outputs to drain before giving
|
|
/// up on them and stopping the next layer anyway.
|
|
void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
|
|
|
|
/// Application-level error listener. Receives the exception any node's
|
|
/// function throws, after that node's own handler (if any) declined it.
|
|
/// Return true to skip the failed invocation and keep the node running,
|
|
/// false to let it stop. Without a listener the exception is discarded
|
|
/// and only a Closed event survives, which reports that a node stopped
|
|
/// but not why.
|
|
void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
|
|
|
|
#ifdef KPN_WEB_DEBUG
|
|
void set_web_debug_port(uint16_t port) { web_debug_port_ = port; }
|
|
// Called by DebugHub::register_network() so the hub owns the debug server.
|
|
void disable_web_server() { web_server_enabled_ = false; }
|
|
#endif
|
|
|
|
// Returns a snapshot of this network's nodes and channels for the DebugHub.
|
|
NetworkSnapshot network_snapshot() const {
|
|
auto s = collect_snapshots();
|
|
return {"", std::move(s.nodes), std::move(s.channels), s.elapsed_s};
|
|
}
|
|
|
|
// Register a shared resource so it appears in diagnostics and the debug UI.
|
|
// The probe must outlive this network (typically the resource is on the same stack).
|
|
void register_resource(const std::string& name, IResourceProbe* probe) {
|
|
resource_probes_.emplace_back(name, probe);
|
|
}
|
|
|
|
// Register a thread pool so it appears in diagnostics and the debug UI.
|
|
// The probe must outlive this network (typically the pool is on the same stack/shared_ptr).
|
|
void register_pool(const std::string& name, IPoolProbe* probe) {
|
|
pool_probes_.emplace_back(name, probe);
|
|
}
|
|
|
|
// 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<NodeSnapshot> nodes;
|
|
std::vector<ChannelSnapshot> channels;
|
|
std::vector<ResourceSnapshot> resources;
|
|
std::vector<PoolSnapshot> pools;
|
|
double elapsed_s;
|
|
};
|
|
|
|
Snapshots collect_snapshots() const {
|
|
double elapsed_s = std::chrono::duration<double>(
|
|
clock_t::now() - start_time_).count();
|
|
|
|
std::vector<NodeSnapshot> 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<ChannelSnapshot> channels;
|
|
for (auto& probe : channel_probes_)
|
|
channels.push_back(probe->snapshot());
|
|
|
|
std::vector<ResourceSnapshot> resources;
|
|
for (auto& [name, probe] : resource_probes_)
|
|
resources.push_back(probe->snapshot(name));
|
|
|
|
std::vector<PoolSnapshot> pools;
|
|
for (auto& [name, probe] : pool_probes_)
|
|
pools.push_back(probe->snapshot(name));
|
|
|
|
return {std::move(nodes), std::move(channels), std::move(resources), std::move(pools), elapsed_s};
|
|
}
|
|
|
|
/// Wait for the channels fed by `src` to empty, or give up.
|
|
///
|
|
/// This was an unbounded `while (anything anywhere is non-empty)` poll over
|
|
/// *every* channel in the graph, which made shutdown() wait for the whole
|
|
/// network to be idle before stopping each successive layer, and wait
|
|
/// forever if anything downstream was wedged — turning a graceful shutdown
|
|
/// into the hang it exists to avoid.
|
|
///
|
|
/// Two bounds, because they fail differently. The deadline covers a
|
|
/// consumer that has stopped consuming: fill never changes and no amount of
|
|
/// waiting helps. The no-progress counter covers a consumer that is merely
|
|
/// slow — it keeps waiting as long as the queue is shrinking, so a slow
|
|
/// drain is not cut short just for exceeding a fixed time.
|
|
///
|
|
/// Giving up is reported rather than silent: undrained data at this point
|
|
/// means values are about to be discarded by the stop that follows.
|
|
void drain_outputs_of(const std::string& src) const {
|
|
const auto deadline = clock_t::now() + drain_timeout_;
|
|
std::size_t last_fill = static_cast<std::size_t>(-1);
|
|
int stalls = 0;
|
|
|
|
for (;;) {
|
|
std::size_t fill = 0;
|
|
for (std::size_t i = 0; i < channel_probes_.size(); ++i)
|
|
if (channel_src_names_[i] == src)
|
|
fill += channel_probes_[i]->snapshot().current_fill;
|
|
|
|
if (fill == 0) return;
|
|
if (fill >= last_fill) { if (++stalls > 100) break; }
|
|
else { stalls = 0; }
|
|
last_fill = fill;
|
|
|
|
if (clock_t::now() >= deadline) break;
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
|
}
|
|
|
|
std::size_t left = 0;
|
|
for (std::size_t i = 0; i < channel_probes_.size(); ++i)
|
|
if (channel_src_names_[i] == src)
|
|
left += channel_probes_[i]->snapshot().current_fill;
|
|
if (left)
|
|
std::cerr << "[kpn] shutdown: '" << src << "' still has " << left
|
|
<< " queued item(s) its consumer did not take; "
|
|
"they are discarded\n";
|
|
}
|
|
|
|
std::string name_;
|
|
bool stop_flag_{false};
|
|
std::unique_ptr<FanoutStorage> fanouts_;
|
|
std::vector<INode*> user_nodes_topo_;
|
|
std::vector<INode*> fanout_nodes_ptr_;
|
|
std::vector<std::string> user_node_names_;
|
|
std::vector<std::string> fanout_node_names_;
|
|
std::vector<std::unique_ptr<IChannelProbe>> channel_probes_;
|
|
/// Display name of the node feeding each probe, parallel to channel_probes_.
|
|
/// shutdown() drains a node's own outputs, so it has to know which they are.
|
|
std::vector<std::string> channel_src_names_;
|
|
std::chrono::milliseconds drain_timeout_{5000};
|
|
std::vector<std::pair<std::string, IResourceProbe*>> resource_probes_;
|
|
std::vector<std::pair<std::string, IPoolProbe*>> pool_probes_;
|
|
EventHandler event_handler_;
|
|
NodeErrorHandler error_handler_;
|
|
clock_t::time_point start_time_;
|
|
#ifdef KPN_WEB_DEBUG
|
|
uint16_t web_debug_port_{9090};
|
|
bool web_server_enabled_{true};
|
|
std::unique_ptr<web_debug::WebDebugServer> web_server_;
|
|
#endif
|
|
};
|
|
|
|
// ── make_network ──────────────────────────────────────────────────────────────
|
|
|
|
template<typename... Edges>
|
|
auto make_network(Edges&&... edges) {
|
|
// 1. Expand edges — detect fan-outs, splice FanoutNodes
|
|
using FanoutSto = tmp::fanout_storage_t<std::decay_t<Edges>...>;
|
|
using ExpandedEdges = tmp::expanded_edges_t<std::decay_t<Edges>...>;
|
|
|
|
// 2. Duplicate-tag check — fires before cycle check for a cleaner error message
|
|
using UserNodes = typename tmp::all_node_types<tmp::TypeList<std::decay_t<Edges>...>>::type;
|
|
static_assert(!tmp::has_duplicate_tags_v<std::decay_t<Edges>...>,
|
|
"make_network: two nodes have the same (Func, UniqueTag) — they are "
|
|
"indistinguishable as graph vertices. Add a UniqueTag: "
|
|
"make_node<func, \"label\", 1>(capacity)");
|
|
|
|
// 3. Cycle check
|
|
using Topo = tmp::topo_sort<ExpandedEdges>;
|
|
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<FanoutSto>();
|
|
|
|
// 5. Wire all expanded SimpleEdges.
|
|
// find_node<NodeT>: searches fanout storage then user edge pack, returns NodeT*.
|
|
// Uses if constexpr in a fold so mismatched types never reach assignment.
|
|
auto find_node = [&]<typename NodeT>() -> NodeT* {
|
|
NodeT* ptr = nullptr;
|
|
std::apply([&](auto&... fn) {
|
|
([&](auto& node) {
|
|
if constexpr (std::is_same_v<std::decay_t<decltype(node)>, NodeT>)
|
|
if (!ptr) ptr = &node;
|
|
}(fn), ...);
|
|
}, *fanout_storage);
|
|
if (!ptr) {
|
|
([&](auto& e) {
|
|
if (!ptr) {
|
|
if constexpr (std::is_same_v<std::decay_t<decltype(e.src)>, NodeT>)
|
|
ptr = &e.src;
|
|
else if constexpr (std::is_same_v<std::decay_t<decltype(e.dst)>, NodeT>)
|
|
ptr = &e.dst;
|
|
}
|
|
}(edges), ...);
|
|
}
|
|
return ptr;
|
|
};
|
|
|
|
// 5. Collect user node pointers + display names in *topological* order.
|
|
//
|
|
// Topo is computed above for the cycle check and used to be discarded,
|
|
// while this vector was filled in edge-declaration order — and then named
|
|
// user_nodes_topo_ and relied upon as if it were sorted. halt() stops in
|
|
// its reverse, and shutdown() walks it forwards stopping each node and
|
|
// draining its outputs before the next, which is only a graceful drain if
|
|
// the order really is sources-first. It held for every network in the tree
|
|
// because edges happen to be declared in pipeline order, and would have
|
|
// broken silently for one that was not.
|
|
//
|
|
// Fanout nodes appear in Topo too; they are skipped here because they are
|
|
// owned separately, in fanout_storage.
|
|
std::vector<INode*> user_node_ptrs;
|
|
std::vector<std::string> user_node_names;
|
|
[&]<typename... Ns>(tmp::TypeList<Ns...>) {
|
|
([&]<typename NodeT>() {
|
|
if constexpr (!requires { NodeT::is_fanout_node; }) {
|
|
if (auto* p = find_node.template operator()<NodeT>()) {
|
|
auto* n = static_cast<INode*>(p);
|
|
if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), n)
|
|
== user_node_ptrs.end()) {
|
|
auto nm = node_display_name<NodeT>();
|
|
user_node_ptrs.push_back(n);
|
|
user_node_names.push_back(nm);
|
|
n->set_name(nm);
|
|
}
|
|
}
|
|
}
|
|
}.template operator()<Ns>(), ...);
|
|
}(typename Topo::topo{});
|
|
|
|
// 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<std::size_t, std::string> fanout_src_name;
|
|
[&]<typename... SEs>(tmp::TypeList<SEs...>) {
|
|
([&]<typename SE>(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<SrcNode>());
|
|
}(SEs{}), ...);
|
|
}(ExpandedEdges{});
|
|
|
|
// Helper: display name for any node type, resolving fanouts to "src_fanout".
|
|
auto node_name = [&]<typename NodeT>() -> 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<NodeT>();
|
|
} else {
|
|
return node_display_name<NodeT>();
|
|
}
|
|
};
|
|
|
|
std::vector<std::unique_ptr<IChannelProbe>> channel_probes;
|
|
std::vector<std::string> channel_src_names;
|
|
|
|
auto wire_one = [&]<typename SE>(SE) {
|
|
using SrcNode = typename SE::src_node_t;
|
|
using DstNode = typename SE::dst_node_t;
|
|
using out_t = std::tuple_element_t<SE::src_idx, typename SrcNode::return_tuple>;
|
|
constexpr std::size_t SrcIdx = SE::src_idx;
|
|
constexpr std::size_t DstIdx = SE::dst_idx;
|
|
|
|
auto* src = find_node.template operator()<SrcNode>();
|
|
auto* dst = find_node.template operator()<DstNode>();
|
|
if (src && dst) {
|
|
auto& ch = dst->template input_channel<DstIdx>();
|
|
src->template set_output_channel<SrcIdx>(&ch);
|
|
std::string ch_name = node_name.template operator()<SrcNode>() + ":" + std::to_string(SrcIdx)
|
|
+ " \xe2\x86\x92 " // UTF-8 →
|
|
+ node_name.template operator()<DstNode>() + ":" + std::to_string(DstIdx);
|
|
channel_probes.push_back(std::make_unique<ChannelProbe<out_t>>(ch, ch_name));
|
|
channel_src_names.push_back(node_name.template operator()<SrcNode>());
|
|
}
|
|
};
|
|
|
|
[&]<typename... SEs>(tmp::TypeList<SEs...>) {
|
|
(wire_one(SEs{}), ...);
|
|
}(ExpandedEdges{});
|
|
|
|
// 6. Collect fanout node pointers and names
|
|
std::vector<INode*> fanout_ptrs;
|
|
std::vector<std::string> fanout_node_names;
|
|
std::apply([&](auto&... fn) {
|
|
([&](auto& node) {
|
|
using NodeT = std::decay_t<decltype(node)>;
|
|
auto fname = node_name.template operator()<NodeT>();
|
|
static_cast<INode&>(node).set_name(fname);
|
|
fanout_ptrs.push_back(static_cast<INode*>(&node));
|
|
fanout_node_names.push_back(fname);
|
|
}(fn), ...);
|
|
}, *fanout_storage);
|
|
|
|
// 7. Construct and return the StaticNetwork
|
|
using Net = StaticNetwork<FanoutSto, typename Topo::topo>;
|
|
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),
|
|
std::move(channel_src_names));
|
|
}
|
|
|
|
} // namespace kpn
|