diff --git a/.gitignore b/.gitignore index 09931b4..174ad4c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,6 @@ # Build output build/ - +build_debug/ # Python __pycache__/ *.py[cod] diff --git a/examples/14_debug_hub/main.cpp b/examples/14_debug_hub/main.cpp new file mode 100644 index 0000000..56eb1d4 --- /dev/null +++ b/examples/14_debug_hub/main.cpp @@ -0,0 +1,159 @@ +// Example 14 — DebugHub with Shared Resource Token +// +// Two independent KPN networks compete for one shared inference resource +// (simulating a small GPU or single-session ONNX runtime). The DebugHub +// serves a single web UI at http://localhost:9090 with: +// +// [All Networks] — resource utilisation cards + cross-network node table +// [detect] — force-directed graph for the detection pipeline +// [classify] — force-directed graph for the classification pipeline +// +// Topology: +// +// detect pipeline: +// [source_detect] ──> [run_detect] ──> [sink_detect] +// +// classify pipeline: +// [source_classify] ──> [run_classify] ──> [sink_classify] +// +// Both [run_detect] and [run_classify] call gpu.acquire() before touching the +// simulated device. The priority-based token awards the next slot to the +// waiter that is more likely to make useful progress (higher priority score). +// +// Build: cmake -DKPN_WEB_DEBUG=ON .. && cmake --build . +// Run: ./14_debug_hub +// UI: http://localhost:9090 + +#ifdef KPN_WEB_DEBUG + +#include + +#include +#include +#include +#include + +using namespace kpn; +using namespace std::chrono_literals; + +// ── Simulated inference device ──────────────────────────────────────────────── +// +// Represents any exclusive, serialised accelerator: GPU session, ONNX runtime, +// hardware encoder, etc. Only one caller can hold it at a time. + +struct GPU { + // Detection model: fast, 8 ms per frame. + int detect(int frame_id) { + std::this_thread::sleep_for(8ms); + return frame_id * 2; // synthetic "score" + } + + // Classification model: heavier, 14 ms per frame. + int classify(int frame_id) { + std::this_thread::sleep_for(14ms); + return frame_id % 10; // synthetic "label" + } +}; + +// Global pointer so free-function nodes can reach the resource. +// In production code, capture by reference inside an ObjectNode functor instead. +static SharedResource* g_gpu = nullptr; + +// ── Detection pipeline ──────────────────────────────────────────────────────── + +static int source_detect() { + static std::atomic id{0}; + std::this_thread::sleep_for(25ms); // ~40 fps source rate + return id.fetch_add(1, std::memory_order_relaxed); +} + +static int run_detect(int frame_id) { + // Higher priority: detection is latency-critical. + auto guard = g_gpu->acquire([] { return 0.7f; }); + return guard->detect(frame_id); +} + +static std::atomic detect_out{0}; +static void sink_detect(int) { + detect_out.fetch_add(1, std::memory_order_relaxed); +} + +// ── Classification pipeline ─────────────────────────────────────────────────── + +static int source_classify() { + static std::atomic id{0}; + std::this_thread::sleep_for(40ms); // ~25 fps source rate + return id.fetch_add(1, std::memory_order_relaxed); +} + +static int run_classify(int frame_id) { + // Lower priority: classification is best-effort. + auto guard = g_gpu->acquire([] { return 0.3f; }); + return guard->classify(frame_id); +} + +static std::atomic classify_out{0}; +static void sink_classify(int) { + classify_out.fetch_add(1, std::memory_order_relaxed); +} + +// ── main ────────────────────────────────────────────────────────────────────── + +int main() { + SharedResource gpu; + g_gpu = &gpu; + + // ── Detection network ───────────────────────────────────────────────────── + auto src_det = make_node(4); + auto inf_det = make_node(4); + auto snk_det = make_node(4); + + auto net_detect = make_network( + edge(src_det.output<0>(), inf_det.input<0>()), + edge(inf_det.output<0>(), snk_det.input<0>()) + ); + + // ── Classification network ──────────────────────────────────────────────── + auto src_cls = make_node(4); + auto inf_cls = make_node(4); + auto snk_cls = make_node(4); + + auto net_classify = make_network( + edge(src_cls.output<0>(), inf_cls.input<0>()), + edge(inf_cls.output<0>(), snk_cls.input<0>()) + ); + + // ── Hub — one debug server for both networks + the shared resource ───────── + DebugHub hub(9090); + hub.register_network("detect", net_detect); + hub.register_network("classify", net_classify); + hub.register_resource("gpu", &gpu); + + net_detect.start(); + net_classify.start(); + hub.start(); + + std::cout << "Running — open http://localhost:9090\n" + << "Tabs: [All Networks] [detect] [classify]\n" + << "Press Enter to stop.\n"; + std::cin.get(); + + net_detect.stop(); + net_classify.stop(); + + std::cout << "\nResults:\n" + << " detect: " << detect_out.load() << " frames\n" + << " classify: " << classify_out.load() << " frames\n"; + return 0; +} + +#else // no KPN_WEB_DEBUG + +#include +int main() { + std::cerr << "This example requires KPN_WEB_DEBUG.\n" + << "Rebuild with: cmake -DKPN_WEB_DEBUG=ON ..\n"; + return 1; +} + +#endif diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index c25e732..b5cebc7 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -15,6 +15,10 @@ kpn_example(10_static_hello_pipeline) kpn_example(11_static_fanout) if(KPN_WEB_DEBUG) kpn_target_enable_web_debug(06_watchdog) + + add_executable(14_debug_hub 14_debug_hub/main.cpp) + target_link_libraries(14_debug_hub PRIVATE kpn) + kpn_target_enable_web_debug(14_debug_hub) endif() # 07 and 08 require the Python bindings — only add if built if(KPN_BUILD_PYTHON) diff --git a/include/kpn/debug_hub.hpp b/include/kpn/debug_hub.hpp new file mode 100644 index 0000000..e279098 --- /dev/null +++ b/include/kpn/debug_hub.hpp @@ -0,0 +1,456 @@ +#pragma once +// Only active when KPN_WEB_DEBUG is defined. + +#ifdef KPN_WEB_DEBUG +#include "diagnostics.hpp" +#include "web_debug.hpp" + +#include +#include +#include +#include +#include +#include + +namespace kpn { + +// ── Hub HTML ────────────────────────────────────────────────────────────────── +// Multi-tab UI: one tab per registered network (force-directed graph) + +// an "All Networks" tab showing shared resource cards and a cross-network +// node table. + +static const char* HUB_HTML = R"html( + + + +KPN++ Debug Hub + + + +
+

KPN++ Debug Hub

+
+ connecting… +
+
+
+
+
+ + + + +)html"; + +// ── DebugHub ────────────────────────────────────────────────────────────────── + +class DebugHub { +public: + explicit DebugHub(uint16_t port = 9090) : port_(port) {} + + ~DebugHub() { stop(); } + DebugHub(const DebugHub&) = delete; + DebugHub& operator=(const DebugHub&) = delete; + + // Register a network. Disables that network's own web server so the hub + // becomes the single debug endpoint. Call before network.start(). + template + void register_network(const std::string& name, Net& net) { + net.disable_web_server(); + networks_.push_back({name, [&net, name] { + auto s = net.network_snapshot(); + s.name = name; + return s; + }}); + } + + // Register a shared resource — appears in the "All Networks" resource panel. + void register_resource(const std::string& name, IResourceProbe* probe) { + resources_.emplace_back(name, probe); + } + + void start() { + server_ = std::make_unique( + port_, + [this] { return build_json(); }, + HUB_HTML); + server_->start(); + std::cerr << "[kpn] hub debug UI: http://localhost:" << port_ << "\n"; + } + + void stop() { if (server_) server_->stop(); } + +private: + // Serialise nodes array for one network snapshot + static void write_nodes(std::ostream& o, const NetworkSnapshot& s) { + o << "["; + for (std::size_t i = 0; i < s.nodes.size(); ++i) { + const auto& n = s.nodes[i]; + if (i) o << ','; + o << "{\"id\":\"" << web_debug::escape_json(n.name) << "\"" + << ",\"frames\":" << n.frames_processed + << ",\"ema_exec_ms\":" << n.ema_exec_ms + << ",\"max_exec_ms\":" << n.max_exec_ms + << ",\"blocked_ms\":" << n.total_blocked_ms + << ",\"fps\":" << n.throughput_fps + << ",\"total_cpu_ms\":" << n.total_cpu_ms + << ",\"cpu_util_pct\":" << n.cpu_util_pct + << "}"; + } + o << "]"; + } + + // Serialise edges array for one network snapshot + static void write_edges(std::ostream& o, const NetworkSnapshot& s) { + o << "["; + for (std::size_t i = 0; i < s.channels.size(); ++i) { + const auto& c = s.channels[i]; + if (i) o << ','; + auto [src, dst] = web_debug::parse_edge_name(c.name); + o << "{\"name\":\"" << web_debug::escape_json(c.name) << "\"" + << ",\"source\":\"" << web_debug::escape_json(src) << "\"" + << ",\"target\":\"" << web_debug::escape_json(dst) << "\"" + << ",\"capacity\":" << c.capacity + << ",\"current\":" << c.current_fill + << ",\"fill_pct\":" << c.fill_pct() + << ",\"peak_pct\":" << c.peak_pct() + << ",\"pushes\":" << c.pushes + << ",\"drops\":" << c.drops + << ",\"overflows\":" << c.overflows + << ",\"item_bytes\":" << c.item_bytes + << ",\"bw_mbs\":" << c.bandwidth_mbs(s.elapsed_s) + << "}"; + } + o << "]"; + } + + std::string build_json() const { + std::ostringstream o; + o << std::fixed; + o.precision(2); + + o << "{\"networks\":["; + for (std::size_t i = 0; i < networks_.size(); ++i) { + if (i) o << ','; + auto s = networks_[i].fn(); + o << "{\"name\":\"" << web_debug::escape_json(networks_[i].name) << "\"" + << ",\"nodes\":"; write_nodes(o, s); + o << ",\"edges\":"; write_edges(o, s); + o << "}"; + } + + o << "],\"resources\":["; + for (std::size_t i = 0; i < resources_.size(); ++i) { + if (i) o << ','; + const auto r = resources_[i].second->snapshot(resources_[i].first); + o << "{\"name\":\"" << web_debug::escape_json(r.name) << "\"" + << ",\"acquisitions\":" << r.acquisitions + << ",\"avg_wait_ms\":" << r.avg_wait_ms + << ",\"peak_waiters\":" << r.peak_waiters + << ",\"current_waiters\":" << r.current_waiters + << ",\"held\":" << (r.held ? "true" : "false") + << "}"; + } + o << "]}"; + return o.str(); + } + + struct Entry { + std::string name; + std::function fn; + }; + + uint16_t port_; + std::vector networks_; + std::vector> resources_; + std::unique_ptr server_; +}; + +} // namespace kpn +#endif // KPN_WEB_DEBUG diff --git a/include/kpn/diagnostics.hpp b/include/kpn/diagnostics.hpp index 86bda95..4fec2a0 100644 --- a/include/kpn/diagnostics.hpp +++ b/include/kpn/diagnostics.hpp @@ -134,6 +134,15 @@ struct NodeSnapshot { double cpu_util_pct; // exec_ms / (exec_ms + blocked_ms) * 100 }; +// ── Cross-network snapshot (used by DebugHub) ───────────────────────────────── + +struct NetworkSnapshot { + std::string name; + std::vector nodes; + std::vector channels; + double elapsed_s; +}; + // ── Resource statistics + snapshot ─────────────────────────────────────────── struct ResourceSnapshot { diff --git a/include/kpn/kpn.hpp b/include/kpn/kpn.hpp index e06f884..0c8cc10 100644 --- a/include/kpn/kpn.hpp +++ b/include/kpn/kpn.hpp @@ -9,5 +9,6 @@ #include "fanout.hpp" #include "shared_resource.hpp" #include "static_network.hpp" +#include "debug_hub.hpp" #include "main_thread_node.hpp" #include "network.hpp" diff --git a/include/kpn/network.hpp b/include/kpn/network.hpp index 41f6c3a..3cb0bfe 100644 --- a/include/kpn/network.hpp +++ b/include/kpn/network.hpp @@ -126,7 +126,7 @@ public: web_debug_port_, [this]() { auto s = collect_snapshots(); - return web_debug::to_json(s.nodes, s.channels, s.elapsed_s); + 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"; diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index 4c5a03a..dfa5898 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -110,14 +110,16 @@ public: 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.resources, s.elapsed_s); - }); - web_server_->start(); - std::cerr << "[kpn] web debug UI: http://localhost:" << web_debug_port_ << "\n"; + if (web_server_enabled_) { + web_server_ = std::make_unique( + web_debug_port_, + [this]() { + auto s = collect_snapshots(); + return web_debug::to_json(s.nodes, s.channels, s.resources, s.elapsed_s); + }); + web_server_->start(); + std::cerr << "[kpn] web debug UI: http://localhost:" << web_debug_port_ << "\n"; + } #endif } @@ -142,8 +144,16 @@ public: #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) { @@ -205,6 +215,7 @@ private: 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_server_; #endif }; diff --git a/include/kpn/web_debug.hpp b/include/kpn/web_debug.hpp index ae2e61c..2805969 100644 --- a/include/kpn/web_debug.hpp +++ b/include/kpn/web_debug.hpp @@ -355,12 +355,14 @@ class WebDebugServer { public: using SnapshotFn = std::function; - explicit WebDebugServer(uint16_t port, SnapshotFn fn) - : port_(port), snapshot_fn_(std::move(fn)) {} + explicit WebDebugServer(uint16_t port, SnapshotFn fn, + const char* custom_html = nullptr) + : port_(port), snapshot_fn_(std::move(fn)) + , html_(custom_html ? custom_html : HTML) {} void start() { - svr_.Get("/", [](const httplib::Request&, httplib::Response& res) { - res.set_content(HTML, "text/html"); + svr_.Get("/", [this](const httplib::Request&, httplib::Response& res) { + res.set_content(html_, "text/html"); }); svr_.Get("/api/snapshot", [this](const httplib::Request&, httplib::Response& res) { res.set_content(snapshot_fn_(), "application/json"); @@ -381,10 +383,11 @@ public: WebDebugServer& operator=(const WebDebugServer&) = delete; private: - uint16_t port_; - SnapshotFn snapshot_fn_; + uint16_t port_; + SnapshotFn snapshot_fn_; + const char* html_; httplib::Server svr_; - std::thread thread_; + std::thread thread_; }; } // namespace kpn::web_debug