dtourolle
00245f5760
The handler was stored in a member and never read. A node's exception was discarded at the node boundary and the only surviving evidence was a Closed event, which reports that a node stopped but not why — the difference between a diagnosis and a guess. StaticNetwork has always wired this; Network accepted the handler and silently dropped it, which is worse than not offering the setter, because the caller believes they have a listener. start() now delivers it to each node, exactly as StaticNetwork does. The type changes with it. It was void(name, exception_ptr), which cannot express the keep-running decision the node side needs — so it is now NodeErrorHandler, the same alias StaticNetwork uses. That is a breaking change in principle; in practice nothing in the tree called this setter, which is how it stayed dead long enough to be worth finding. Verified by the new case: the node throws, the handler receives the name and the exception, returns true, and the node goes on to process the next value. 148/148.
529 lines
22 KiB
C++
529 lines
22 KiB
C++
#pragma once
|
|
#include "diagnostics.hpp"
|
|
#include "inode.hpp"
|
|
#include "port.hpp"
|
|
|
|
#ifdef KPN_WEB_DEBUG
|
|
#include "web_debug.hpp"
|
|
#include <memory>
|
|
#endif
|
|
|
|
#include <condition_variable>
|
|
#include <functional>
|
|
#include <iomanip>
|
|
#include <mutex>
|
|
#include <iostream>
|
|
#include <map>
|
|
#include <set>
|
|
#include <sstream>
|
|
#include <stdexcept>
|
|
#include <string>
|
|
#include <string_view>
|
|
#include <thread>
|
|
#include <vector>
|
|
|
|
namespace kpn {
|
|
|
|
// ── Exceptions ────────────────────────────────────────────────────────────────
|
|
|
|
class NetworkCycleError : public std::runtime_error {
|
|
public:
|
|
NetworkCycleError() : std::runtime_error("network graph contains a directed cycle") {}
|
|
};
|
|
|
|
class NetworkBuildError : public std::runtime_error {
|
|
public:
|
|
explicit NetworkBuildError(std::string msg)
|
|
: std::runtime_error(std::move(msg)) {}
|
|
};
|
|
|
|
// ── Network ───────────────────────────────────────────────────────────────────
|
|
|
|
class Network : public INode {
|
|
public:
|
|
/// 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.
|
|
///
|
|
/// Same type as StaticNetwork's, deliberately: this used to be a void
|
|
/// signature, which could not express the keep-running decision and, more
|
|
/// to the point, was never delivered anywhere.
|
|
using ErrorHandler = NodeErrorHandler;
|
|
using DiagnosticsHandler =
|
|
std::function<void(const std::vector<NodeSnapshot>&,
|
|
const std::vector<ChannelSnapshot>&)>;
|
|
using EventHandler =
|
|
std::function<void(std::string_view node_name, NodeEvent,
|
|
std::chrono::steady_clock::time_point)>;
|
|
|
|
// ── Builder API ───────────────────────────────────────────────────────────
|
|
|
|
template<typename NodeT>
|
|
Network& add(std::string name, NodeT& node) {
|
|
if (nodes_.count(name))
|
|
throw NetworkBuildError("duplicate node name: " + name);
|
|
node.set_name(name);
|
|
nodes_.emplace(name, &node);
|
|
adj_[name];
|
|
return *this;
|
|
}
|
|
|
|
template<typename SrcNode, std::size_t SrcIdx,
|
|
typename DstNode, std::size_t DstIdx>
|
|
Network& connect(const std::string& src_name,
|
|
OutputPort<SrcNode, SrcIdx>,
|
|
const std::string& dst_name,
|
|
InputPort<DstNode, DstIdx>) {
|
|
using out_t = std::tuple_element_t<SrcIdx, typename SrcNode::return_tuple>;
|
|
using dst_in_t = std::tuple_element_t<DstIdx, typename DstNode::args_tuple>;
|
|
static_assert(std::is_same_v<out_t, dst_in_t>,
|
|
"connect: output type does not match input type");
|
|
|
|
auto* src = dynamic_cast<SrcNode*>(nodes_.at(src_name));
|
|
auto* dst = dynamic_cast<DstNode*>(nodes_.at(dst_name));
|
|
if (!src) throw NetworkBuildError("node '" + src_name + "' type mismatch");
|
|
if (!dst) throw NetworkBuildError("node '" + dst_name + "' type mismatch");
|
|
|
|
auto port_key = std::make_pair(src_name, SrcIdx);
|
|
if (connected_outputs_.count(port_key))
|
|
throw NetworkBuildError(
|
|
"connect: output port '" + src_name + "':" + std::to_string(SrcIdx)
|
|
+ " is already connected — use make_fanout<T, N> for fan-out");
|
|
connected_outputs_.insert(port_key);
|
|
|
|
auto& in_ch = dst->template input_channel<DstIdx>();
|
|
src->template set_output_channel<SrcIdx>(&in_ch);
|
|
|
|
// Register channel probe for diagnostics
|
|
std::string ch_name = src_name + ":" + std::to_string(SrcIdx)
|
|
+ " → " + dst_name + ":" + std::to_string(DstIdx);
|
|
channel_probes_.push_back(
|
|
std::make_unique<ChannelProbe<out_t>>(in_ch, ch_name));
|
|
channel_src_names_.push_back(src_name);
|
|
|
|
adj_[src_name].push_back(dst_name);
|
|
return *this;
|
|
}
|
|
|
|
template<typename NodeT, std::size_t Idx>
|
|
Network& expose_input(std::string boundary_name, InputPort<NodeT, Idx>) {
|
|
exposed_inputs_[boundary_name] = boundary_name;
|
|
return *this;
|
|
}
|
|
|
|
template<typename NodeT, std::size_t Idx>
|
|
Network& expose_output(std::string boundary_name, OutputPort<NodeT, Idx>) {
|
|
exposed_outputs_[boundary_name] = boundary_name;
|
|
return *this;
|
|
}
|
|
|
|
Network& build() {
|
|
topo_.clear();
|
|
std::map<std::string, int> color;
|
|
for (auto& [name, _] : nodes_)
|
|
if (color[name] == 0)
|
|
dfs(name, color);
|
|
if (event_handler_) {
|
|
for (auto& name : topo_) {
|
|
auto* node = nodes_.at(name);
|
|
node->set_network_overflow_callback(
|
|
[this, n = name](auto ts) {
|
|
event_handler_(n, NodeEvent::Overflow, ts);
|
|
});
|
|
node->set_network_closed_callback(
|
|
[this, n = name](auto ts) {
|
|
event_handler_(n, NodeEvent::Closed, ts);
|
|
});
|
|
}
|
|
}
|
|
return *this;
|
|
}
|
|
|
|
// ── INode ─────────────────────────────────────────────────────────────────
|
|
|
|
void start() override {
|
|
start_time_ = clock_t::now();
|
|
// Deliver the listener to the nodes. Without this the handler was
|
|
// stored and never read: a node's exception was discarded at the node
|
|
// boundary and the only surviving evidence was a Closed event, which
|
|
// says a node stopped but not why. StaticNetwork has always done this;
|
|
// Network accepted the handler and silently dropped it.
|
|
if (error_handler_)
|
|
for (auto& name : topo_)
|
|
nodes_.at(name)->set_network_error_callback(error_handler_);
|
|
// Callbacks first, everywhere, before anything runs — see INode::prepare.
|
|
for (auto& name : topo_)
|
|
nodes_.at(name)->prepare();
|
|
for (auto& name : topo_)
|
|
nodes_.at(name)->start();
|
|
start_watchdog();
|
|
#ifdef KPN_WEB_DEBUG
|
|
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.elapsed_s, s.pools);
|
|
});
|
|
web_server_->start();
|
|
std::cerr << "[kpn] web debug UI: http://localhost:" << web_debug_port_ << "\n";
|
|
#endif
|
|
}
|
|
|
|
void stop() override { halt(); }
|
|
|
|
// halt(): immediate stop — broadcasts disable to all channels and joins threads.
|
|
void halt() override {
|
|
#ifdef KPN_WEB_DEBUG
|
|
if (web_server_) web_server_->stop();
|
|
#endif
|
|
stop_watchdog();
|
|
for (auto it = topo_.rbegin(); it != topo_.rend(); ++it)
|
|
nodes_.at(*it)->stop();
|
|
}
|
|
|
|
// shutdown(): graceful drain in topological order.
|
|
// Stops source nodes first, polls until their output channels drain to zero,
|
|
// then stops the next layer, and so on.
|
|
void shutdown() override {
|
|
#ifdef KPN_WEB_DEBUG
|
|
if (web_server_) web_server_->stop();
|
|
#endif
|
|
stop_watchdog();
|
|
|
|
// Identify which nodes have no incoming edges (sources).
|
|
std::map<std::string, std::size_t> in_degree;
|
|
for (auto& [name, _] : nodes_) in_degree[name] = 0;
|
|
for (auto& [src, dsts] : adj_)
|
|
for (auto& dst : dsts) in_degree[dst]++;
|
|
|
|
// Walk topo order: stop each source layer, wait for its output channels
|
|
// to drain, then proceed to the next layer.
|
|
std::set<std::string> stopped;
|
|
for (auto& name : topo_) {
|
|
if (in_degree[name] == 0 || all_predecessors_stopped(name, stopped)) {
|
|
nodes_.at(name)->stop();
|
|
stopped.insert(name);
|
|
// Wait for output channels of this node to drain.
|
|
drain_output_channels(name);
|
|
}
|
|
}
|
|
// Stop any remaining nodes (sinks / nodes not yet stopped).
|
|
for (auto it = topo_.rbegin(); it != topo_.rend(); ++it)
|
|
if (!stopped.count(*it)) nodes_.at(*it)->stop();
|
|
}
|
|
|
|
bool running() const override { return watchdog_.joinable(); }
|
|
void set_name(std::string) override {}
|
|
|
|
const NodeStats& stats() const override {
|
|
static NodeStats dummy;
|
|
return dummy;
|
|
}
|
|
|
|
NodeSnapshot node_snapshot(const std::string& name, double) const override {
|
|
return {name, 0, 0, 0, 0, 0, 0, 0};
|
|
}
|
|
|
|
// ── Configuration ─────────────────────────────────────────────────────────
|
|
|
|
void set_watchdog_interval(std::chrono::milliseconds interval) {
|
|
watchdog_interval_ = interval;
|
|
}
|
|
|
|
/// 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; }
|
|
|
|
/// Must be called before start(); the handler is delivered to nodes there.
|
|
void set_error_handler(ErrorHandler h) { error_handler_ = std::move(h); }
|
|
void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_ = std::move(h); }
|
|
void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
|
|
|
|
void register_pool(const std::string& name, IPoolProbe* probe) {
|
|
pool_probes_.emplace_back(name, probe);
|
|
}
|
|
|
|
#ifdef KPN_WEB_DEBUG
|
|
void set_web_debug_port(uint16_t port) { web_debug_port_ = port; }
|
|
#endif
|
|
|
|
// Print a diagnostics report to a stream (default: stderr).
|
|
// Can be called at any time; thread-safe (reads atomics with relaxed ordering).
|
|
void print_diagnostics(std::ostream& os = std::cerr) const {
|
|
auto s = collect_snapshots();
|
|
os << format_report(s.nodes, s.channels, s.pools, s.elapsed_s);
|
|
}
|
|
|
|
private:
|
|
// ── Diagnostics collection ────────────────────────────────────────────────
|
|
|
|
struct Snapshots {
|
|
std::vector<NodeSnapshot> nodes;
|
|
std::vector<ChannelSnapshot> channels;
|
|
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 (auto& name : topo_)
|
|
nodes.push_back(nodes_.at(name)->node_snapshot(name, elapsed_s));
|
|
|
|
std::vector<ChannelSnapshot> channels;
|
|
for (auto& probe : channel_probes_)
|
|
channels.push_back(probe->snapshot());
|
|
|
|
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(pools), elapsed_s};
|
|
}
|
|
|
|
static std::string format_report(const std::vector<NodeSnapshot>& nodes,
|
|
const std::vector<ChannelSnapshot>& channels,
|
|
const std::vector<PoolSnapshot>& pools = {},
|
|
double elapsed_s = 0.0) {
|
|
std::ostringstream os;
|
|
os << std::fixed << std::setprecision(1);
|
|
|
|
os << "\n┌─ KPN++ Diagnostics ────────────────────────────────────────────────────────\n";
|
|
|
|
// Node table
|
|
os << "│ Nodes:\n";
|
|
os << "│ " << std::left
|
|
<< std::setw(16) << "name"
|
|
<< std::setw(10) << "frames"
|
|
<< std::setw(12) << "exec ms"
|
|
<< std::setw(12) << "max ms"
|
|
<< std::setw(14) << "blocked ms"
|
|
<< std::setw(10) << "fps"
|
|
<< std::setw(12) << "cpu ms"
|
|
<< std::setw(10) << "util%"
|
|
<< "\n│ " << std::string(92, '-') << "\n";
|
|
for (auto& n : nodes) {
|
|
os << "│ " << std::left
|
|
<< std::setw(16) << n.name
|
|
<< std::setw(10) << n.frames_processed
|
|
<< std::setw(12) << n.ema_exec_ms
|
|
<< std::setw(12) << n.max_exec_ms
|
|
<< std::setw(14) << n.total_blocked_ms
|
|
<< std::setw(10) << n.throughput_fps
|
|
<< std::setw(12) << n.total_cpu_ms
|
|
<< std::setw(10) << n.cpu_util_pct
|
|
<< "\n";
|
|
}
|
|
|
|
// Channel table
|
|
os << "│\n│ Channels:\n";
|
|
os << "│ " << std::left
|
|
<< std::setw(40) << "edge"
|
|
<< std::setw(8) << "fill%"
|
|
<< std::setw(8) << "peak%"
|
|
<< std::setw(8) << "pushes"
|
|
<< std::setw(8) << "drops"
|
|
<< std::setw(8) << "oflow"
|
|
<< std::setw(12) << "MB/s"
|
|
<< std::setw(10) << "item B"
|
|
<< "\n│ " << std::string(102, '-') << "\n";
|
|
for (auto& c : channels) {
|
|
std::string flag = c.fill_pct() >= 80.0 ? " <<<" :
|
|
c.peak_pct() >= 80.0 ? " (peak)" : "";
|
|
os << "│ " << std::left
|
|
<< std::setw(40) << c.name
|
|
<< std::setw(8) << c.fill_pct()
|
|
<< std::setw(8) << c.peak_pct()
|
|
<< std::setw(8) << c.pushes
|
|
<< std::setw(8) << c.drops
|
|
<< std::setw(8) << c.overflows
|
|
<< std::setw(12) << c.bandwidth_mbs(elapsed_s)
|
|
<< std::setw(10) << c.item_bytes
|
|
<< flag << "\n";
|
|
}
|
|
|
|
// Pool table
|
|
if (!pools.empty()) {
|
|
os << "│\n│ Thread Pools:\n";
|
|
os << "│ " << std::left
|
|
<< std::setw(16) << "name"
|
|
<< std::setw(10) << "threads"
|
|
<< std::setw(12) << "queued"
|
|
<< std::setw(12) << "active"
|
|
<< std::setw(14) << "in/s"
|
|
<< std::setw(14) << "out/s"
|
|
<< "\n│ " << std::string(78, '-') << "\n";
|
|
for (auto& p : pools) {
|
|
double in_rate = elapsed_s > 0.0 ? p.tasks_submitted / elapsed_s : 0.0;
|
|
double out_rate = elapsed_s > 0.0 ? p.tasks_completed / elapsed_s : 0.0;
|
|
os << "│ " << std::left
|
|
<< std::setw(16) << p.name
|
|
<< std::setw(10) << p.thread_count
|
|
<< std::setw(12) << p.queue_depth
|
|
<< std::setw(12) << p.active_count
|
|
<< std::setw(14) << in_rate
|
|
<< std::setw(14) << out_rate
|
|
<< "\n";
|
|
}
|
|
}
|
|
|
|
// Bottleneck hint: node with highest ema_exec_ms
|
|
if (!nodes.empty()) {
|
|
auto it = std::max_element(nodes.begin(), nodes.end(),
|
|
[](const NodeSnapshot& a, const NodeSnapshot& b) {
|
|
return a.ema_exec_ms < b.ema_exec_ms;
|
|
});
|
|
os << "│\n│ Bottleneck hint: '" << it->name
|
|
<< "' (avg exec " << it->ema_exec_ms << " ms)\n";
|
|
}
|
|
os << "└────────────────────────────────────────────────────────────────\n";
|
|
return os.str();
|
|
}
|
|
|
|
// ── Shutdown helpers ──────────────────────────────────────────────────────
|
|
|
|
bool all_predecessors_stopped(const std::string& name,
|
|
const std::set<std::string>& stopped) const {
|
|
for (auto& [src, dsts] : adj_)
|
|
for (auto& dst : dsts)
|
|
if (dst == name && !stopped.count(src)) return false;
|
|
return true;
|
|
}
|
|
|
|
/// Wait for the channels fed by `name` to empty, or give up.
|
|
///
|
|
/// This took a node name and ignored it, polling *every* channel in the
|
|
/// graph instead — so shutdown() waited for the whole network to be idle
|
|
/// before stopping each successive layer. With no deadline either, anything
|
|
/// wedged downstream turned 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, where fill never changes and
|
|
/// waiting cannot help. The no-progress counter covers one that is merely
|
|
/// slow: it keeps waiting while the queue is shrinking, so a slow drain is
|
|
/// not cut short just for taking a while.
|
|
void drain_output_channels(const std::string& name) const {
|
|
const auto deadline = clock_t::now() + drain_timeout_;
|
|
std::size_t last_fill = static_cast<std::size_t>(-1);
|
|
int stalls = 0;
|
|
|
|
auto fill_of = [&] {
|
|
std::size_t fill = 0;
|
|
for (std::size_t i = 0; i < channel_probes_.size(); ++i)
|
|
if (channel_src_names_[i] == name)
|
|
fill += channel_probes_[i]->snapshot().current_fill;
|
|
return fill;
|
|
};
|
|
|
|
for (;;) {
|
|
const std::size_t fill = fill_of();
|
|
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));
|
|
}
|
|
|
|
if (const std::size_t left = fill_of())
|
|
std::cerr << "[kpn] shutdown: '" << name << "' still has " << left
|
|
<< " queued item(s) its consumer did not take; "
|
|
"they are discarded\n";
|
|
}
|
|
|
|
// ── Cycle detection / topological sort ───────────────────────────────────
|
|
|
|
void dfs(const std::string& name, std::map<std::string, int>& color) {
|
|
color[name] = 1;
|
|
for (auto& nbr : adj_[name]) {
|
|
if (color[nbr] == 1) throw NetworkCycleError{};
|
|
if (color[nbr] == 0) dfs(nbr, color);
|
|
}
|
|
color[name] = 2;
|
|
topo_.insert(topo_.begin(), name);
|
|
}
|
|
|
|
// ── Watchdog ──────────────────────────────────────────────────────────────
|
|
|
|
void start_watchdog() {
|
|
watchdog_ = std::jthread([this](std::stop_token tok) {
|
|
// Interruptible wait, not sleep_for. request_stop() cannot wake a
|
|
// sleeping thread, so stop_watchdog()'s join blocked for up to a
|
|
// full interval — three seconds by default, and unbounded for
|
|
// anyone who set a long one to keep the periodic report quiet.
|
|
// Every teardown paid it.
|
|
std::mutex m;
|
|
std::condition_variable_any cv;
|
|
while (!tok.stop_requested()) {
|
|
{
|
|
std::unique_lock lk(m);
|
|
if (cv.wait_for(lk, tok, watchdog_interval_,
|
|
[&tok] { return tok.stop_requested(); }))
|
|
break;
|
|
}
|
|
|
|
auto s = collect_snapshots();
|
|
check_hung_nodes();
|
|
|
|
if (diag_handler_) {
|
|
diag_handler_(s.nodes, s.channels);
|
|
} else {
|
|
std::cerr << format_report(s.nodes, s.channels, s.pools, s.elapsed_s);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
void check_hung_nodes() const {
|
|
auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
clock_t::now().time_since_epoch()).count();
|
|
for (auto& [name, node] : nodes_) {
|
|
int64_t start = node->stats().exec_start_us.load(std::memory_order_relaxed);
|
|
if (start == 0) continue;
|
|
int64_t elapsed_ms = (now_us - start) / 1000;
|
|
// Warn if a node has been executing for > 5 s with no max_exec_time set,
|
|
// or if it exceeds its configured max. Threshold: 5000 ms default.
|
|
if (elapsed_ms > 5000) {
|
|
std::cerr << "[kpn] WARNING: node '" << name
|
|
<< "' has been executing for " << elapsed_ms << " ms\n";
|
|
}
|
|
}
|
|
}
|
|
|
|
void stop_watchdog() {
|
|
if (watchdog_.joinable())
|
|
watchdog_.request_stop(), watchdog_.join();
|
|
}
|
|
|
|
// ── State ─────────────────────────────────────────────────────────────────
|
|
|
|
std::map<std::string, INode*> nodes_;
|
|
std::map<std::string, std::vector<std::string>> adj_;
|
|
std::vector<std::string> topo_;
|
|
std::map<std::string, std::string> exposed_inputs_;
|
|
std::map<std::string, std::string> exposed_outputs_;
|
|
std::set<std::pair<std::string, std::size_t>> connected_outputs_;
|
|
std::vector<std::unique_ptr<IChannelProbe>> channel_probes_;
|
|
/// 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, IPoolProbe*>> pool_probes_;
|
|
ErrorHandler error_handler_;
|
|
DiagnosticsHandler diag_handler_;
|
|
EventHandler event_handler_;
|
|
std::chrono::milliseconds watchdog_interval_{3000};
|
|
std::jthread watchdog_;
|
|
clock_t::time_point start_time_;
|
|
#ifdef KPN_WEB_DEBUG
|
|
uint16_t web_debug_port_{9090};
|
|
std::unique_ptr<web_debug::WebDebugServer> web_server_;
|
|
#endif
|
|
};
|
|
|
|
} // namespace kpn
|