diff --git a/include/kpn/diagnostics.hpp b/include/kpn/diagnostics.hpp index e95fe16..86bda95 100644 --- a/include/kpn/diagnostics.hpp +++ b/include/kpn/diagnostics.hpp @@ -134,4 +134,20 @@ struct NodeSnapshot { double cpu_util_pct; // exec_ms / (exec_ms + blocked_ms) * 100 }; +// ── Resource statistics + snapshot ─────────────────────────────────────────── + +struct ResourceSnapshot { + std::string name; + uint64_t acquisitions; + double avg_wait_ms; + uint64_t peak_waiters; + uint64_t current_waiters; + bool held; +}; + +struct IResourceProbe { + virtual ~IResourceProbe() = default; + virtual ResourceSnapshot snapshot(const std::string& name) const = 0; +}; + } // namespace kpn diff --git a/include/kpn/kpn.hpp b/include/kpn/kpn.hpp index 66a2376..e06f884 100644 --- a/include/kpn/kpn.hpp +++ b/include/kpn/kpn.hpp @@ -7,6 +7,7 @@ #include "port.hpp" #include "node.hpp" #include "fanout.hpp" +#include "shared_resource.hpp" #include "static_network.hpp" #include "main_thread_node.hpp" #include "network.hpp" diff --git a/include/kpn/shared_resource.hpp b/include/kpn/shared_resource.hpp new file mode 100644 index 0000000..79924c3 --- /dev/null +++ b/include/kpn/shared_resource.hpp @@ -0,0 +1,190 @@ +#pragma once +#include "diagnostics.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace kpn { + +template class Channel; // forward declaration for acquire_balanced + +// ── SharedResource ──────────────────────────────────────────────────────────── +// +// Wraps an exclusive resource (e.g. an ONNX session, a CUDA stream) and +// arbitrates concurrent access using a priority-based waiter queue. +// +// When multiple nodes compete, the one with the highest priority score wins +// the next slot. Priority is re-evaluated at release time so it reflects the +// current queue state, not the state when the node first started waiting. +// +// Starvation prevention: each waiter's effective score grows with elapsed wait +// time (aging_per_second), ensuring a low-priority node eventually gets served. +// +// Usage: +// SharedResource res(session_args...); +// +// // inside a node functor — +// auto guard = res.acquire_balanced(in_channel, out_channel); +// guard->Run(...); // guard releases automatically on scope exit + +template +class SharedResource : public IResourceProbe { +public: + // ── RAII guard ──────────────────────────────────────────────────────────── + + class Guard { + SharedResource* owner_; + explicit Guard(SharedResource* o) : owner_(o) {} + friend class SharedResource; + public: + Guard(Guard&& o) noexcept : owner_(std::exchange(o.owner_, nullptr)) {} + Guard& operator=(Guard&&) = delete; + Guard(const Guard&) = delete; + Guard& operator=(const Guard&) = delete; + ~Guard() { if (owner_) owner_->release(); } + + T& get() { return owner_->resource_; } + T* operator->() { return &owner_->resource_; } + T& operator*() { return owner_->resource_; } + }; + + // ── Construction ────────────────────────────────────────────────────────── + + template + explicit SharedResource(Args&&... args) + : resource_(std::forward(args)...) {} + + SharedResource(const SharedResource&) = delete; + SharedResource& operator=(const SharedResource&) = delete; + SharedResource(SharedResource&&) = delete; + SharedResource& operator=(SharedResource&&) = delete; + + // ── Acquire ─────────────────────────────────────────────────────────────── + + // Acquire with a callable that returns a priority in [0, 1]. + // Higher = more urgent. Called at every release to pick the best waiter. + template + Guard acquire(PriorityFn&& fn) { + std::unique_lock lock(mutex_); + if (!held_) { + held_ = true; + acq_.fetch_add(1, std::memory_order_relaxed); + return Guard(this); + } + Waiter w{std::function(std::forward(fn)), clock_t::now()}; + waiters_.push_back(&w); + update_peak(waiters_.size()); + current_waiters_.store(waiters_.size(), std::memory_order_relaxed); + + auto t0 = w.wait_start; + w.cv.wait(lock, [&w] { return w.ready; }); + + int64_t wait_us = std::chrono::duration_cast( + clock_t::now() - t0).count(); + waiters_.erase(std::find(waiters_.begin(), waiters_.end(), &w)); + current_waiters_.store(waiters_.size(), std::memory_order_relaxed); + acq_.fetch_add(1, std::memory_order_relaxed); + total_wait_us_.fetch_add(static_cast(wait_us > 0 ? wait_us : 0), + std::memory_order_relaxed); + return Guard(this); + } + + // Acquire with no priority (all waiters treated equally, order is fair-ish). + Guard acquire() { + return acquire([] { return 0.5f; }); + } + + // Acquire with priority derived from channel fill fractions: + // score = input_fill × output_headroom + // A node with a full input queue and empty output queue has the highest + // urgency — it has work to do and nowhere to stall downstream. + template + Guard acquire_balanced(const Channel& in_ch, const Channel& out_ch) { + return acquire([&in_ch, &out_ch] { + float in_fill = in_ch.capacity() ? float(in_ch.size()) / in_ch.capacity() : 0.5f; + float out_head = out_ch.capacity() ? 1.0f - float(out_ch.size()) / out_ch.capacity() : 0.5f; + return in_fill * out_head; + }); + } + + // ── IResourceProbe ──────────────────────────────────────────────────────── + + ResourceSnapshot snapshot(const std::string& name) const override { + std::lock_guard lock(mutex_); + uint64_t a = acq_.load(std::memory_order_relaxed); + uint64_t w = total_wait_us_.load(std::memory_order_relaxed); + return { + name, + a, + a > 0 ? double(w) / a / 1000.0 : 0.0, + peak_waiters_.load(std::memory_order_relaxed), + current_waiters_.load(std::memory_order_relaxed), + held_, + }; + } + +private: + void release() { + std::unique_lock lock(mutex_); + if (waiters_.empty()) { + held_ = false; + return; + } + // Re-evaluate every waiter's current priority and apply aging bonus. + auto now = clock_t::now(); + Waiter* best = nullptr; + float best_score = -1.0f; + for (Waiter* w : waiters_) { + float age_s = std::chrono::duration(now - w->wait_start).count(); + float score = w->priority_fn() + age_s * kAgingPerSecond; + if (score > best_score) { best_score = score; best = w; } + } + best->ready = true; + best->cv.notify_one(); + // held_ stays true — ownership transfers to the woken waiter. + } + + void update_peak(std::size_t n) { + uint64_t prev = peak_waiters_.load(std::memory_order_relaxed); + while (n > prev && + !peak_waiters_.compare_exchange_weak(prev, n, + std::memory_order_relaxed, std::memory_order_relaxed)) + ; + } + + struct Waiter { + std::function priority_fn; + clock_t::time_point wait_start; + std::condition_variable cv; + bool ready{false}; + + Waiter(std::function fn, clock_t::time_point t) + : priority_fn(std::move(fn)), wait_start(t) {} + }; + + static constexpr float kAgingPerSecond = 0.05f; + + T resource_; + bool held_{false}; + mutable std::mutex mutex_; + std::vector waiters_; + std::atomic acq_{0}; + std::atomic total_wait_us_{0}; + std::atomic peak_waiters_{0}; + std::atomic current_waiters_{0}; +}; + +// ── Factory ─────────────────────────────────────────────────────────────────── + +template +SharedResource make_shared_resource(Args&&... args) { + return SharedResource(std::forward(args)...); +} + +} // namespace kpn diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index 490815b..4c5a03a 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -13,6 +13,7 @@ #endif #include +#include #include #include #include @@ -113,7 +114,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.resources, s.elapsed_s); }); web_server_->start(); std::cerr << "[kpn] web debug UI: http://localhost:" << web_debug_port_ << "\n"; @@ -143,6 +144,12 @@ public: void set_web_debug_port(uint16_t port) { web_debug_port_ = port; } #endif + // 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); + } + // Print diagnostics using compile-time node labels void print_diagnostics(std::ostream& os = std::cerr) const { os << "\n┌─ KPN++ StaticNetwork diagnostics ─────────────────────────────\n"; @@ -159,9 +166,10 @@ public: private: struct Snapshots { - std::vector nodes; - std::vector channels; - double elapsed_s; + std::vector nodes; + std::vector channels; + std::vector resources; + double elapsed_s; }; Snapshots collect_snapshots() const { @@ -178,7 +186,11 @@ private: for (auto& probe : channel_probes_) channels.push_back(probe->snapshot()); - return {std::move(nodes), std::move(channels), elapsed_s}; + std::vector resources; + for (auto& [name, probe] : resource_probes_) + resources.push_back(probe->snapshot(name)); + + return {std::move(nodes), std::move(channels), std::move(resources), elapsed_s}; } std::string name_; @@ -189,6 +201,7 @@ private: std::vector user_node_names_; std::vector fanout_node_names_; std::vector> channel_probes_; + std::vector> resource_probes_; clock_t::time_point start_time_; #ifdef KPN_WEB_DEBUG uint16_t web_debug_port_{9090}; diff --git a/include/kpn/web_debug.hpp b/include/kpn/web_debug.hpp index f959c31..ae2e61c 100644 --- a/include/kpn/web_debug.hpp +++ b/include/kpn/web_debug.hpp @@ -46,6 +46,7 @@ static std::pair parse_edge_name(const std::string& nam static std::string to_json(const std::vector& nodes, const std::vector& channels, + const std::vector& resources = {}, double elapsed_s = 0.0) { std::ostringstream o; o << std::fixed; @@ -84,6 +85,18 @@ static std::string to_json(const std::vector& nodes, << ",\"bw_mbs\":" << c.bandwidth_mbs(elapsed_s) << "}"; } + o << "],\"resources\":["; + for (std::size_t i = 0; i < resources.size(); ++i) { + const auto& r = resources[i]; + if (i) o << ','; + o << "{\"name\":\"" << 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(); } @@ -112,12 +125,24 @@ static const char* HTML = R"html( border-radius: 4px; padding: 8px 12px; font-size: 11px; pointer-events: none; display: none; white-space: pre; line-height: 1.6; } + #resources { + position: absolute; bottom: 12px; right: 12px; + background: #16213e; border: 1px solid #0f3460; border-radius: 4px; + padding: 8px 12px; font-size: 10px; min-width: 220px; + display: none; + } + #resources h2 { margin: 0 0 6px; font-size: 11px; color: #e94560; } + .res-row { display: flex; justify-content: space-between; gap: 12px; margin-top: 3px; } + .res-name { color: #eee; } + .res-held-y { color: #e94560; } + .res-held-n { color: #4CAF50; }
+

Shared Resources