dtourolleandClaude Opus 5
4e81752838
Channels drop on overflow by default. That is the right behaviour for live
sources, where a stale item is worth less than a fresh one, but it makes the
library unusable for offline batch work: items vanish with no diagnostic, and
any downstream analysis that assumes a fixed sample rate is silently invalid.
Auto-inserted fanouts were the harder half of this. They are created inside
make_network(), so user code cannot reach them to configure, and they drop
per-output inside a swallowed catch — so a pipeline whose own nodes were all
configured lossless could still lose items with nothing reported anywhere. In
the pipeline this came from, capture's 2505 frames arrived at the detector as
774 while the overflow counter read zero.
Adds:
- INode::set_lossless_output(bool), defaulted to a no-op so node types with
no output channels ignore it
- PoolNode / PoolObjectNode: route push_one_out() through push_blocking()
- FanoutNode: same, plus set_lossy_output(i) to opt a single branch back out
- StaticNetwork::set_lossless(), which reaches user nodes and fanouts alike
- StaticNetwork::drain(), a public wrapper over the existing private
drain_all_channels(), so callers can flush in-flight work before stop()
Default behaviour is unchanged; every path is off unless explicitly enabled.
Blocking output is only safe when every consumer eventually drains. A branch
that can stall indefinitely — a display node nobody is servicing — will apply
backpressure to the whole pipeline, which is what set_lossy_output() is for.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
448 lines
19 KiB
C++
448 lines
19 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)
|
|
: 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();
|
|
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); });
|
|
}
|
|
}
|
|
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
|
|
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
|
|
// user_nodes_topo_ is already in sources-first order.
|
|
// Stop each node and drain its output channels before moving on.
|
|
for (auto* n : user_nodes_topo_) {
|
|
n->stop();
|
|
drain_all_channels();
|
|
}
|
|
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); }
|
|
|
|
#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_; }
|
|
|
|
// Make every node in this network push losslessly (block until the consumer
|
|
// drains) instead of dropping on a full channel. Includes the fanout nodes
|
|
// make_network() inserts automatically, which is the part user code cannot
|
|
// reach: they are unnamed, and they drop silently per-output, so a network
|
|
// whose own nodes are all lossless can still lose items at a fanout.
|
|
//
|
|
// Only safe when every consumer eventually drains. A branch that can stall
|
|
// indefinitely — a display node nobody is servicing, say — will block the
|
|
// whole pipeline through backpressure. Call before start().
|
|
void set_lossless(bool on = true) {
|
|
for (auto* n : user_nodes_topo_) if (n) n->set_lossless_output(on);
|
|
for (auto* n : fanout_nodes_ptr_) if (n) n->set_lossless_output(on);
|
|
}
|
|
|
|
// Block until every channel is empty. Useful before stop() so work already
|
|
// in flight completes rather than being discarded at teardown.
|
|
void drain() const { drain_all_channels(); }
|
|
|
|
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};
|
|
}
|
|
|
|
void drain_all_channels() const {
|
|
bool any_full = true;
|
|
while (any_full) {
|
|
any_full = false;
|
|
for (auto& probe : channel_probes_) {
|
|
if (probe->snapshot().current_fill > 0) { any_full = true; break; }
|
|
}
|
|
if (any_full)
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
|
}
|
|
}
|
|
|
|
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_;
|
|
std::vector<std::pair<std::string, IResourceProbe*>> resource_probes_;
|
|
std::vector<std::pair<std::string, IPoolProbe*>> pool_probes_;
|
|
EventHandler event_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. Collect unique user node pointers + their display names, in edge-declaration order
|
|
std::vector<INode*> user_node_ptrs;
|
|
std::vector<std::string> user_node_names;
|
|
auto collect = [&](auto& e) {
|
|
using SrcT = std::decay_t<decltype(e.src)>;
|
|
using DstT = std::decay_t<decltype(e.dst)>;
|
|
auto* s = static_cast<INode*>(&e.src);
|
|
auto* d = static_cast<INode*>(&e.dst);
|
|
if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), s) == user_node_ptrs.end()) {
|
|
auto sname = node_display_name<SrcT>();
|
|
user_node_ptrs.push_back(s);
|
|
user_node_names.push_back(sname);
|
|
s->set_name(sname);
|
|
}
|
|
if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), d) == user_node_ptrs.end()) {
|
|
auto dname = node_display_name<DstT>();
|
|
user_node_ptrs.push_back(d);
|
|
user_node_names.push_back(dname);
|
|
d->set_name(dname);
|
|
}
|
|
};
|
|
(collect(edges), ...);
|
|
|
|
// 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;
|
|
};
|
|
|
|
// 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;
|
|
|
|
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));
|
|
}
|
|
};
|
|
|
|
[&]<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));
|
|
}
|
|
|
|
} // namespace kpn
|