KPN/include/kpn/diagnostics.hpp

#pragma once
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <string>
#include <vector>
#include <time.h>   // clock_gettime, CLOCK_THREAD_CPUTIME_ID

namespace kpn {

using clock_t    = std::chrono::steady_clock;
using duration_t = std::chrono::duration<double, std::milli>; // milliseconds

// ── Per-channel statistics ────────────────────────────────────────────────────

struct ChannelStats {
    std::atomic<uint64_t> pushes{0};
    std::atomic<uint64_t> bytes_pushed{0};
    std::atomic<uint64_t> drops{0};
    std::atomic<uint64_t> overflows{0};
    std::atomic<uint64_t> pops{0};
    std::atomic<std::size_t> peak_fill{0};

    ChannelStats() = default;
    ChannelStats(const ChannelStats&) = delete;
    ChannelStats& operator=(const ChannelStats&) = delete;

    void record_push(std::size_t current_fill, std::size_t data_bytes) {
        pushes.fetch_add(1, std::memory_order_relaxed);
        bytes_pushed.fetch_add(data_bytes, std::memory_order_relaxed);
        std::size_t prev = peak_fill.load(std::memory_order_relaxed);
        while (current_fill > prev &&
               !peak_fill.compare_exchange_weak(prev, current_fill,
                   std::memory_order_relaxed, std::memory_order_relaxed))
            ;
    }
    void record_drop()     { drops.fetch_add(1,    std::memory_order_relaxed); }
    void record_overflow() { overflows.fetch_add(1, std::memory_order_relaxed); }
    void record_pop()      { pops.fetch_add(1,      std::memory_order_relaxed); }
};

// ── Per-node statistics ───────────────────────────────────────────────────────

struct NodeStats {
    std::atomic<uint64_t> frames_processed{0};

    // Wall-clock execution time EMA — warmup mean for first WARMUP_FRAMES,
    // then EMA alpha=0.1. Stored as integer microseconds for atomic updates.
    static constexpr int WARMUP_FRAMES = 5;
    std::atomic<int64_t> ema_exec_us{0};
    std::atomic<int64_t> max_exec_us{0};
    std::atomic<int64_t> total_blocked_us{0};

    // Cumulative wall time inside fire_once, summed over every invocation.
    // The EMA above cannot be turned into a total: it is exponentially
    // weighted, so frames * ema_exec_us tracks the tail of the run rather than
    // the whole of it, and on a workload whose per-frame cost varies (a face
    // detector on a film: crowd scenes then empty landscapes) the two differ by
    // a lot. Answering "how much time went into this node" needs a real sum.
    std::atomic<int64_t> total_exec_us{0};

    // Thread CPU time — actual CPU consumed by this node's thread,
    // measured via CLOCK_THREAD_CPUTIME_ID. Excludes time sleeping or
    // blocked on mutexes/channels. Sampled once per frame.
    std::atomic<int64_t> total_cpu_us{0};  // cumulative CPU µs consumed

    // Pool scheduling stats — only meaningful for PoolNode / InterruptNode.
    // exec_start_us: wall-clock µs when fire_once began; 0 when idle.
    // Used by the watchdog to detect hung nodes (elapsed > max_exec_time).
    std::atomic<int64_t> queue_wait_us{0};  // cumulative µs spent in pool queue
    std::atomic<int64_t> exec_start_us{0};  // non-zero while fire_once is running

    NodeStats() = default;
    NodeStats(const NodeStats&) = delete;
    NodeStats& operator=(const NodeStats&) = delete;

    // Call at the start of run_loop to capture thread CPU baseline.
    // Returns the raw timespec for use in record_exec.
    static struct timespec cpu_now() {
        struct timespec ts{};
        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts);
        return ts;
    }

    static int64_t timespec_us(const struct timespec& ts) {
        return static_cast<int64_t>(ts.tv_sec) * 1'000'000
             + static_cast<int64_t>(ts.tv_nsec) / 1'000;
    }

    void record_queue_wait(duration_t wait) {
        int64_t us = static_cast<int64_t>(wait.count() * 1000.0);
        if (us > 0) queue_wait_us.fetch_add(us, std::memory_order_relaxed);
    }

    void record_exec(duration_t exec_time, duration_t blocked_time,
                     const struct timespec& cpu_before, const struct timespec& cpu_after) {
        frames_processed.fetch_add(1, std::memory_order_relaxed);

        int64_t us = static_cast<int64_t>(exec_time.count() * 1000.0);
        total_exec_us.fetch_add(us, std::memory_order_relaxed);

        uint64_t n = frames_processed.load(std::memory_order_relaxed);
        int64_t prev = ema_exec_us.load(std::memory_order_relaxed);
        int64_t next = (n <= static_cast<uint64_t>(WARMUP_FRAMES))
            ? prev + (us - prev) / static_cast<int64_t>(n)
            : prev + (us - prev) / 10;
        ema_exec_us.store(next, std::memory_order_relaxed);

        int64_t cur_max = max_exec_us.load(std::memory_order_relaxed);
        if (us > cur_max)
            max_exec_us.store(us, std::memory_order_relaxed);

        int64_t blocked_us = static_cast<int64_t>(blocked_time.count() * 1000.0);
        total_blocked_us.fetch_add(blocked_us, std::memory_order_relaxed);

        int64_t cpu_delta = timespec_us(cpu_after) - timespec_us(cpu_before);
        if (cpu_delta > 0)
            total_cpu_us.fetch_add(cpu_delta, std::memory_order_relaxed);
    }
};

// ── Snapshot for reporting (copyable, taken by watchdog) ─────────────────────

struct ChannelSnapshot {
    std::string name;
    std::size_t capacity;
    std::size_t current_fill;
    std::size_t peak_fill;
    uint64_t    pushes;
    uint64_t    bytes_pushed;  // actual bytes accumulated via channel_data_size<T>
    uint64_t    drops;
    uint64_t    overflows;
    uint64_t    pops;
    std::size_t item_bytes;   // sizeof(T) — nominal struct size, not necessarily data size

    double fill_pct() const {
        return capacity ? 100.0 * current_fill / capacity : 0.0;
    }
    double peak_pct() const {
        return capacity ? 100.0 * peak_fill / capacity : 0.0;
    }
    // Bandwidth in MB/s: actual bytes transferred / elapsed seconds
    double bandwidth_mbs(double elapsed_s) const {
        if (elapsed_s <= 0.0) return 0.0;
        return static_cast<double>(bytes_pushed) / elapsed_s / 1e6;
    }
};

struct NodeSnapshot {
    std::string name;
    uint64_t    frames_processed;
    double      ema_exec_ms;
    double      max_exec_ms;
    double      total_blocked_ms;   // ThreadPerNode: time blocked in channel pop
    double      throughput_fps;
    double      total_cpu_ms;       // cumulative CPU time consumed by this node's thread
    double      cpu_util_pct;       // exec_ms / (exec_ms + blocked_ms) * 100
    double      queue_wait_ms{0};   // PoolNode: cumulative time spent in pool queue

    // Cumulative wall time inside fire_once. Unlike ema_exec_ms this is a true
    // sum, so it is the field to use for "share of the run spent in this node".
    // Note it still includes time parked pushing into a full output channel;
    // total_cpu_ms is the part that backpressure cannot inflate.
    //
    // Declared before the two bools below because every node type initialises
    // this aggregate positionally, and all of them supply total_exec_ms as the
    // element after queue_wait_ms.
    double      total_exec_ms{0};

    // Live scheduling state, for observing the AR-004 invariant "a node never
    // sleeps with a wake outstanding". The invariant was previously asserted in
    // comments but invisible at runtime, so a lost wake could only be found in a
    // debugger — and this bug does not reproduce under one (it needs full speed).
    // Two atomic loads at snapshot time, nothing on the hot path.
    //
    // Read them together with the node's channel fill:
    //   queued=0, wake=1                -> wake recorded and never consumed
    //   queued=0, wake=0, input full    -> wake never generated at all
    //   queued=1 while nothing running  -> submitted but never scheduled
    bool        queued{false};
    bool        wake_pending{false};
};

// ── Pool statistics + snapshot ────────────────────────────────────────────────

struct PoolSnapshot {
    std::string  name;
    std::size_t  thread_count;
    std::size_t  queue_depth;     // tasks waiting in the priority queue
    std::size_t  active_count;    // tasks currently executing
    uint64_t     tasks_submitted;
    uint64_t     tasks_completed;
};

struct IPoolProbe {
    virtual ~IPoolProbe() = default;
    virtual PoolSnapshot snapshot(const std::string& name) const = 0;
};

// ── Cross-network snapshot (used by DebugHub) ─────────────────────────────────

struct NetworkSnapshot {
    std::string                  name;
    std::vector<NodeSnapshot>    nodes;
    std::vector<ChannelSnapshot> channels;
    double                       elapsed_s;
};

// ── 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;

    /// Release every thread waiting for the resource, so teardown is not held
    /// up by one. A network calls this on the resources registered with it when
    /// it halts; default no-op for probes with nothing to wake.
    virtual void close() {}
};

} // namespace kpn