Add shared reasource tag to allow coordination of usage

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

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"

include/kpn/shared_resource.hpp

@@ -0,0 +1,190 @@
+#pragma once
+#include "diagnostics.hpp"
+
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <utility>
+#include <vector>
+
+namespace kpn {
+
+template<typename T> 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<OrtSession> 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<typename T>
+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<typename... Args>
+    explicit SharedResource(Args&&... args)
+        : resource_(std::forward<Args>(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<typename PriorityFn>
+    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<float()>(std::forward<PriorityFn>(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<std::chrono::microseconds>(
+            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<uint64_t>(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<typename In, typename Out>
+    Guard acquire_balanced(const Channel<In>& in_ch, const Channel<Out>& 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<float>(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<float()>  priority_fn;
+        clock_t::time_point     wait_start;
+        std::condition_variable cv;
+        bool                    ready{false};
+
+        Waiter(std::function<float()> 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<Waiter*>     waiters_;
+    std::atomic<uint64_t>    acq_{0};
+    std::atomic<uint64_t>    total_wait_us_{0};
+    std::atomic<uint64_t>    peak_waiters_{0};
+    std::atomic<uint64_t>    current_waiters_{0};
+};
+
+// ── Factory ───────────────────────────────────────────────────────────────────
+
+template<typename T, typename... Args>
+SharedResource<T> make_shared_resource(Args&&... args) {
+    return SharedResource<T>(std::forward<Args>(args)...);
+}
+
+} // namespace kpn

include/kpn/static_network.hpp

@@ -13,6 +13,7 @@
 #endif
 
 #include <iostream>
+#include <map>
 #include <string>
 #include <tuple>
 #include <type_traits>
@@ -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<NodeSnapshot>    nodes;
-        std::vector<ChannelSnapshot> channels;
-        double                       elapsed_s;
+        std::vector<NodeSnapshot>     nodes;
+        std::vector<ChannelSnapshot>  channels;
+        std::vector<ResourceSnapshot> 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<ResourceSnapshot> 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<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_;
     clock_t::time_point                                 start_time_;
 #ifdef KPN_WEB_DEBUG
     uint16_t                                            web_debug_port_{9090};

include/kpn/web_debug.hpp

@@ -46,6 +46,7 @@ static std::pair<std::string,std::string> parse_edge_name(const std::string& nam
 
 static std::string to_json(const std::vector<NodeSnapshot>& nodes,
                             const std::vector<ChannelSnapshot>& channels,
+                            const std::vector<ResourceSnapshot>& resources = {},
                             double elapsed_s = 0.0) {
     std::ostringstream o;
     o << std::fixed;
@@ -84,6 +85,18 @@ static std::string to_json(const std::vector<NodeSnapshot>& 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(<!DOCTYPE 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; }
 </style>
 </head>
 <body>
 <div id="header"><h1>KPN++ Web Debug</h1><span id="status">connecting...</span></div>
 <svg id="graph"></svg>
 <div id="tooltip"></div>
+<div id="resources"><h2>Shared Resources</h2><div id="res-list"></div></div>
 <script src="https://d3js.org/d3.v7.min.js"></script>
 <script>
 const nodeRadius = 30;
@@ -200,6 +225,7 @@ function init(data) {
     .text(d => `${d.ema_exec_ms.toFixed(1)}ms ${d.fps.toFixed(1)}fps`);
 
   nodeSel = nodeG;
+  renderResources(data.resources);
 
   // Tooltips
   const tip = d3.select('#tooltip');
@@ -224,7 +250,25 @@ function init(data) {
   }).on('mouseleave', () => tip.style('display','none'));
 }
 
+function renderResources(resources) {
+  const panel = document.getElementById('resources');
+  const list  = document.getElementById('res-list');
+  if (!resources || resources.length === 0) { panel.style.display = 'none'; return; }
+  panel.style.display = 'block';
+  list.innerHTML = resources.map(r => {
+    const heldCls = r.held ? 'res-held-y' : 'res-held-n';
+    const heldTxt = r.held ? 'HELD' : 'free';
+    return `<div class="res-row">
+      <span class="res-name">${r.name}</span>
+      <span class="${heldCls}">${heldTxt}</span>
+      <span>wait ${r.avg_wait_ms.toFixed(1)}ms</span>
+      <span>waiters ${r.current_waiters}/${r.peak_waiters}</span>
+    </div>`;
+  }).join('');
+}
+
 function update(data) {
+  renderResources(data.resources);
   // Update node stats in-place (preserve simulation x/y positions)
   const byId = Object.fromEntries(data.nodes.map(n => [n.id, n]));
   nodes.forEach(n => {

tests/CMakeLists.txt

@@ -32,6 +32,7 @@ add_executable(kpn_tests
     test_node.cpp
     test_network.cpp
     test_static_network.cpp
+    test_shared_resource.cpp
 )
 
 target_link_libraries(kpn_tests PRIVATE

tests/test_shared_resource.cpp

@@ -0,0 +1,239 @@
+#include <catch2/catch_test_macros.hpp>
+#include <catch2/catch_approx.hpp>
+#include <kpn/shared_resource.hpp>
+#include <kpn/channel.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <thread>
+#include <vector>
+
+using namespace kpn;
+using namespace std::chrono_literals;
+
+// ── Basic acquire / release ───────────────────────────────────────────────────
+
+TEST_CASE("acquire returns guard that accesses the resource", "[shared_resource]") {
+    SharedResource<int> res(42);
+    {
+        auto g = res.acquire();
+        REQUIRE(*g == 42);
+        *g = 99;
+    } // g released here — second acquire must not overlap in the same thread
+    {
+        auto g2 = res.acquire();
+        REQUIRE(*g2 == 99);
+    }
+}
+
+TEST_CASE("guard operator-> reaches resource members", "[shared_resource]") {
+    struct Pair { int x{1}; int y{2}; };
+    SharedResource<Pair> res;
+    auto g = res.acquire();
+    REQUIRE(g->x == 1);
+    REQUIRE(g->y == 2);
+    g->x = 10;
+    REQUIRE(g.get().x == 10);
+}
+
+TEST_CASE("guard releases on scope exit", "[shared_resource]") {
+    SharedResource<int> res(0);
+    {
+        auto g = res.acquire();
+        REQUIRE(res.snapshot("r").held);
+    }
+    // After guard destroyed, resource is free
+    REQUIRE(!res.snapshot("r").held);
+}
+
+// ── Mutual exclusion ──────────────────────────────────────────────────────────
+
+TEST_CASE("only one thread holds the resource at a time", "[shared_resource]") {
+    SharedResource<int> res(0);
+    std::atomic<int> concurrent_holders{0};
+    std::atomic<int> violations{0};
+    std::atomic<bool> go{false};
+
+    auto worker = [&] {
+        while (!go.load()) std::this_thread::yield();
+        for (int i = 0; i < 20; ++i) {
+            auto g = res.acquire();
+            int h = concurrent_holders.fetch_add(1) + 1;
+            if (h > 1) violations.fetch_add(1);
+            std::this_thread::sleep_for(100us);
+            concurrent_holders.fetch_sub(1);
+        }
+    };
+
+    std::vector<std::thread> threads;
+    for (int i = 0; i < 4; ++i) threads.emplace_back(worker);
+    go.store(true);
+    for (auto& t : threads) t.join();
+
+    REQUIRE(violations.load() == 0);
+}
+
+// ── Priority ordering ─────────────────────────────────────────────────────────
+
+TEST_CASE("higher priority waiter is served before lower priority waiter", "[shared_resource]") {
+    SharedResource<int> res(0);
+
+    // Hold the resource so threads have to queue.
+    auto holder = res.acquire();
+
+    std::vector<int> order;
+    std::mutex order_mtx;
+
+    // Launch two waiters: low priority first, then high priority.
+    std::thread low([&] {
+        auto g = res.acquire([] { return 0.1f; });
+        std::lock_guard lk(order_mtx);
+        order.push_back(1);
+    });
+    std::this_thread::sleep_for(5ms); // ensure low is queued first
+
+    std::thread high([&] {
+        auto g = res.acquire([] { return 0.9f; });
+        std::lock_guard lk(order_mtx);
+        order.push_back(2);
+    });
+    std::this_thread::sleep_for(5ms); // ensure high is also queued
+
+    // Release — high priority should win even though low arrived first.
+    { auto drop = std::move(holder); }
+
+    low.join();
+    high.join();
+
+    REQUIRE(order.size() == 2);
+    REQUIRE(order[0] == 2); // high priority served first
+    REQUIRE(order[1] == 1);
+}
+
+// ── acquire_balanced uses channel fills ───────────────────────────────────────
+
+TEST_CASE("acquire_balanced: full input + empty output gives score ~1.0", "[shared_resource]") {
+    // We test the priority function indirectly via ordering.
+    // Node A: in=full, out=empty  → score ≈ 1.0  (high)
+    // Node B: in=empty, out=full  → score ≈ 0.0  (low)
+
+    Channel<int> in_a(4);   // fill it
+    Channel<int> out_a(4);  // leave empty
+    Channel<int> in_b(4);   // leave empty
+    Channel<int> out_b(4);  // fill it
+
+    in_a.enable();  out_a.enable();
+    in_b.enable();  out_b.enable();
+
+    for (int i = 0; i < 4; ++i) { in_a.push(i); out_b.push(i); }
+
+    SharedResource<int> res(0);
+    auto holder = res.acquire(); // block others
+
+    std::vector<int> order;
+    std::mutex mtx;
+
+    // Node B (low priority) waits first
+    std::thread tb([&] {
+        auto g = res.acquire_balanced(in_b, out_b);
+        std::lock_guard lk(mtx);
+        order.push_back(2);
+    });
+    std::this_thread::sleep_for(5ms);
+
+    // Node A (high priority) waits second
+    std::thread ta([&] {
+        auto g = res.acquire_balanced(in_a, out_a);
+        std::lock_guard lk(mtx);
+        order.push_back(1);
+    });
+    std::this_thread::sleep_for(5ms);
+
+    { auto drop = std::move(holder); } // release
+
+    ta.join();
+    tb.join();
+
+    REQUIRE(order.size() == 2);
+    REQUIRE(order[0] == 1); // node A served first despite arriving second
+}
+
+// ── Statistics ────────────────────────────────────────────────────────────────
+
+TEST_CASE("stats: acquisitions counted correctly", "[shared_resource]") {
+    SharedResource<int> res(0);
+    {
+        auto g1 = res.acquire();
+    }
+    {
+        auto g2 = res.acquire();
+    }
+    REQUIRE(res.snapshot("r").acquisitions == 2);
+}
+
+TEST_CASE("stats: peak_waiters reflects maximum concurrent queue depth", "[shared_resource]") {
+    SharedResource<int> res(0);
+    auto holder = res.acquire();
+
+    std::atomic<int> ready{0};
+
+    auto waiter = [&] {
+        ready.fetch_add(1);
+        auto g = res.acquire();
+    };
+
+    std::thread t1(waiter), t2(waiter), t3(waiter);
+
+    // Wait until all three are queued
+    while (ready.load() < 3) std::this_thread::sleep_for(1ms);
+    std::this_thread::sleep_for(5ms); // give them time to block on acquire
+
+    { auto drop = std::move(holder); } // release
+
+    t1.join(); t2.join(); t3.join();
+
+    REQUIRE(res.snapshot("r").peak_waiters >= 2); // at least 2 queued simultaneously
+}
+
+TEST_CASE("stats: current_waiters returns to 0 after all served", "[shared_resource]") {
+    SharedResource<int> res(0);
+    auto holder = res.acquire();
+
+    std::thread t1([&] { auto g = res.acquire(); });
+    std::thread t2([&] { auto g = res.acquire(); });
+    std::this_thread::sleep_for(10ms);
+
+    { auto drop = std::move(holder); }
+    t1.join(); t2.join();
+
+    REQUIRE(res.snapshot("r").current_waiters == 0);
+}
+
+TEST_CASE("stats: avg_wait_ms is positive when contention occurred", "[shared_resource]") {
+    SharedResource<int> res(0);
+    {
+        auto holder = res.acquire();
+        std::thread t([&] { auto g = res.acquire(); });
+        std::this_thread::sleep_for(10ms);
+        { auto drop = std::move(holder); }
+        t.join();
+    }
+    REQUIRE(res.snapshot("r").avg_wait_ms > 0.0);
+}
+
+// ── No-arg acquire ────────────────────────────────────────────────────────────
+
+TEST_CASE("no-arg acquire works and releases correctly", "[shared_resource]") {
+    SharedResource<int> res(7);
+    auto g = res.acquire();
+    REQUIRE(*g == 7);
+    REQUIRE(res.snapshot("r").held);
+}
+
+// ── make_shared_resource factory ──────────────────────────────────────────────
+
+TEST_CASE("make_shared_resource constructs with forwarded args", "[shared_resource]") {
+    auto res = make_shared_resource<std::string>("hello");
+    auto g = res.acquire();
+    REQUIRE(*g == "hello");
+}