diff --git a/.gitignore b/.gitignore index bee8803..aa003ec 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,8 @@ __pycache__/ .venv/ venv/ .DS_Store + +# persistent telemetry store +hyperswap.db +hyperswap.db-wal +hyperswap.db-shm diff --git a/README.md b/README.md index 02390db..67df288 100644 --- a/README.md +++ b/README.md @@ -20,17 +20,18 @@ ## 1. Feature Matrix ### โšก Bidirectional VRAM Hot-Swapping & Arbitration -* **Sub-25ms Soft-Yield**: Instantly releases Ollama VRAM allocations (`keep_alive: 0`) down to 0 MB when ComfyUI needs to run diffusion workloads without evicting weights from system RAM. -* **Auto-Purge for ComfyUI**: Automatically purges diffusion pipeline checkpoints and VRAM buffers when an image/video generation job finishes, releasing 100% of VRAM back to Ollama. -* **Real-Time ComfyUI WebSocket & Watchdog Listener**: Subscribes directly to `ws://127.0.0.1:8188/ws` and runs a 300ms watchdog loop to detect prompt queueing and node execution in real time. +* **Confirmed Soft-Yield (barrier, not fire-and-forget)**: Releases Ollama VRAM allocations (`keep_alive: 0`) down to 0 MB, then **waits on NVML until the driver has actually freed the allocation** before letting ComfyUI proceed. Posting `keep_alive: 0` only *asks* Ollama to unload; on this box the HTTP call returns in ~63 ms while the driver takes a further ~77 ms to release 14.9 GB. Returning during that window is how diffusion ends up allocating into VRAM that is still occupied. +* **Idle-Aware ComfyUI Purge**: Diffusion checkpoints are held for `COMFY_IDLE_PURGE_S` (30 s) of genuinely empty queue rather than purged 1.5 s after every prompt โ€” iterating on a workflow no longer pays a full checkpoint reload per run. An immediate purge still happens the moment Ollama actually asks for VRAM (`POST /api/request-vram`). +* **Real-Time ComfyUI WebSocket & Watchdog Listener**: Subscribes directly to `ws://127.0.0.1:8188/ws`. The WebSocket is the primary signal; a connection-pooled watchdog polls `/queue` at 1 Hz purely as a fallback, backing off to 3 s while the socket is healthy. * **Process-Level VRAM Attribution**: Live NVML process inspection attributes exact GPU memory usage across Ollama (`llama-server`), ComfyUI (`python`), and Desktop display servers (`gnome-shell`, `Xorg`). -* **Hot-Swap Transition History**: Circular buffer logs all model switch events, swap durations (in ms), tokens/sec throughput, and RAM cache hit status (`RAM Cache Hit โšก` vs `Cold Disk Load ๐Ÿ’พ`). +* **Bandwidth-Classified Transition History**: Every switch is classified by the bandwidth it actually achieved (`model size รท load duration`) rather than a fixed duration threshold: `RAM Cache Hit โšก` (โ‰ฅ5 GB/s), `Partial Cache ๐ŸŒค` (โ‰ฅ1.5 GB/s), `Cold Disk Load ๐Ÿ’พ` (below that). The previous `load_duration < 2500 ms` rule called a 12.9 GB model read at 2.9 GB/s a "cold disk load" and a 0.5 GB model read from NVMe a "cache hit". ### ๐Ÿง  64GB Host RAM Cache & Page Pre-warmer * **Zero-Latency Model Discovery**: Automatic cataloging of all local Ollama models (`/usr/share/ollama/.ollama/models`, `~/.ollama/models`) and ComfyUI model directories (`checkpoints`, `diffusion_models`, `unet`, `vae`, `clip`, `loras`, `controlnet`). -* **POSIX `fadvise` & Pinned Pre-warmer**: Pre-faults multi-gigabyte GGUFs and Safetensors into the Linux OS Page Cache so that reloading models across PCIe 4.0 x16 runs at ~31.5 GB/s (sub-second VRAM loads). -* **Granular Pre-warming Controls**: Pre-warm all discovered models in bulk or target individual models/safetensors on demand. -* **Memory Telemetry**: Real-time breakdown of Total Host RAM, Applications Memory, Active Model Page Cache, Free Memory, and Cache Residency Ratio. +* **POSIX `fadvise` & Pinned Pre-warmer**: Pre-faults multi-gigabyte GGUFs and Safetensors into the Linux OS Page Cache so that reloading models runs at page-cache speed rather than disk speed. +* **Measured Residency via `cachestat(2)`**: Residency is measured, not assumed. `cachestat(2)` gives exact cached-page counts per file. Where the kernel refuses it โ€” it only permits introspection of files you own, and Ollama's blobs are owned by uid `ollama` โ€” HyperSwap falls back to a randomised read-rate probe and labels the result as such. Files it cannot measure are reported as unmeasurable rather than guessed at. +* **Budgeted, Ranked Warming**: This box has 64 GB of RAM and >270 GB of model files; reading everything simply evicts whatever was warmed first. Files are ranked by recency/frequency (from the persisted event log) and warmed until a byte budget is spent, skipping anything already resident. `GET /api/warm-plan` previews the decision without executing it. +* **Memory Telemetry**: Real-time breakdown of Total Host RAM, Applications Memory, Active Model Page Cache, Free Memory, and measured Cache Residency Ratio. ### ๐ŸŽ›๏ธ Dynamic Overclocking & Thermal Management * **Workload-Aware Overclock Profiles**: @@ -44,11 +45,25 @@ * **Hardware Fan Control**: Switch between `auto` and `manual` PWM control (30%โ€“100%) with synchronized dual-fan actuation (`[fan:0]` and `[fan:1]`). * **Automated Lockstep Profile Switching**: AutoArbitrator automatically switches hardware profiles in lockstep with the active workload (`comfy` on generation start, `ollama` on completion). +### ๐ŸŒก๏ธ Thermal Governor (closed-loop de-escalation) +* Every overclock lever here is sticky: a profile locks clocks and pins the fans to a manual PWM, and nothing used to undo that. The governor watches the telemetry the sampler already collects (so it costs no extra NVML calls) and walks the overclock back through a four-step derate ladder when the card runs hot or reports a hardware throttle. +* **Hysteresis by design**: escalation needs 5 consecutive bad samples, recovery needs 30 consecutive good ones, with a 20 s cooldown between changes โ€” a single spike during a diffusion step will not cause profile thrash. +* **Guaranteed restore**: stock clocks, default power limit and automatic fans are restored by the server's shutdown hook *and* by a systemd `ExecStopPost=`, so a `SIGKILL` cannot leave the card with locked clocks and fans pinned at 100%. + +### ๐Ÿ”ฌ Overclock Autotune (`autotune.py`) +* Walks a clock offset upward, running a fixed decode benchmark at each step, and reports the **fastest stable** value with its measured gain over baseline. +* **Instability detection**: kernel `Xid`/`NVRM` messages via `journalctl -k`, benchmark failure, degenerate output, and a temperature ceiling. The sweep stops climbing the moment a step looks unstable. +* **Safety**: refuses to start while ComfyUI is executing, and restores the original profile in a `finally` block โ€” including on exception or cancellation. + +### ๐Ÿ—„๏ธ Persistent Telemetry Store (`telemetry_store.py`) +* Swap history used to be an in-memory `deque(maxlen=50)` that evaporated on every restart. Telemetry and events now persist to SQLite (WAL, single writer thread, batched 1 Hz inserts, automatic retention pruning) at roughly **0.4 MB per hour**. +* This is what makes the app's central question answerable: **`GET /api/analytics/profiles` compares decode throughput per overclock profile**, joined against the thermals recorded while that profile was active. + ### ๐Ÿ“Š Real-Time Web Telemetry Dashboard (`:9090`) * **Live Hardware Telemetry**: GPU utilization %, GPU temperature (ยฐC), power draw (W), fan speeds (%), and graphics/memory clock frequencies (MHz). * **Live Dual-Axis Time-Series Chart**: Real-time graphical visualization of VRAM usage (GB) and Host RAM Cache (GB) with zero frontend polling overhead. * **Interactive Control Center**: Trigger model hot-swaps, soft-yields, cache pre-warms, fan adjustments, and benchmarks directly from the web interface. -* **Server-Sent Events (SSE)**: Pushes unified 1Hz telemetry updates via `GET /api/stream`. +* **Server-Sent Events (SSE)**: A single background sampler produces one 1 Hz snapshot and fans it out to every subscriber via `GET /api/stream`. Previously each connected client independently re-ran the whole snapshot โ€” NVML, `/proc/meminfo`, an HTTP round-trip each to Ollama and ComfyUI, and a recursive walk of the ComfyUI models tree with a `stat()` per checkpoint โ€” once per second, so opening the dashboard in three tabs tripled the load on the thing it was measuring. Slow clients drop stale frames instead of stalling the sampler. ### ๐Ÿค– Model Context Protocol (MCP 2.0) Server * **12 Native Agentic Tools**: Allows AI agents (Antigravity CLI, Claude Desktop, Cursor) to manage GPU resources, trigger model hot-swaps, tune fan curves, and inspect telemetry. @@ -92,7 +107,7 @@ flowchart TD ### The Physics of Sub-Second Switching * **Host RAM as Staging**: Active LLMs and diffusion checkpoints remain resident in the 64GB Linux Page Cache. * **PCIe 4.0 x16 Hot-Swapping**: Transferring weights across PCIe 4.0 x16 achieves **~31.5 GB/s** bandwidth, reducing model loads from 30+ seconds (disk) to **under 1.5 seconds**. -* **Soft-Yielding**: Dropping Ollama's VRAM allocation via `keep_alive: 0` takes **~15ms** while preserving the weights in host RAM. +* **Soft-Yielding**: Dropping Ollama's VRAM allocation via `keep_alive: 0` preserves the weights in host RAM. Measured on this box: the HTTP request returns in **~63 ms**, and the driver finishes releasing 14.9 GB **~77 ms after that**. HyperSwap waits for the second number before handing VRAM to ComfyUI โ€” the earlier "~15 ms" figure timed the request, not the release. --- @@ -119,12 +134,36 @@ The HyperSwap server runs on port `9090` by default. Interactive OpenAPI/Swagger | Endpoint | Method | Description | | :--- | :--- | :--- | | `/api/switch-model` | `POST` | Hot-swaps the active Ollama LLM in VRAM and tracks transition timing. | -| `/api/free-vram` | `POST` | Instructs Ollama to soft-yield VRAM down to 0 MB in ~15ms while retaining RAM cache. | +| `/api/free-vram` | `POST` | Soft-yields Ollama VRAM to 0 MB and **waits for NVML to confirm the release** (`?confirm=false` to skip). Returns `request_ms`, `confirm_ms` and the GB actually freed. | | `/api/comfy-free` | `POST` | Instructs ComfyUI to purge loaded diffusion weights and VRAM cache. | -| `/api/warm-all` | `POST` | Pre-faults all installed Ollama models and ComfyUI Safetensors into the Linux page cache. | -| `/api/warm-model` | `POST` | Pre-warms a specific model or file into RAM. | +| `/api/request-vram` | `POST` | Ollama-priority path: purges ComfyUI immediately if there is not enough free VRAM. | +| `/api/warm-all` | `POST` | Warms the highest-value models into page cache within a byte budget (`budget_gb`). | +| `/api/warm-plan` | `GET` | Previews what warming would read, in what order, and what it would skip โ€” without doing it. | +| `/api/warm-model` | `POST` | Pre-warms a specific model or file into RAM (`blob_only` warms weights without touching VRAM). | +| `/api/cache/report` | `GET` | Measured page-cache residency per model file, with the measurement method used for each. | | `/api/benchmark` | `POST` | Runs an automated back-and-forth model swap benchmark and calculates average latency. | +### Analytics Endpoints (persisted) + +| Endpoint | Method | Description | +| :--- | :--- | :--- | +| `/api/analytics/profiles` | `GET` | **Decode throughput per overclock profile**, joined with the thermals recorded under it. | +| `/api/analytics/swaps` | `GET` | Aggregated swap/yield/purge latencies, cache-hit split, and per-model throughput. | +| `/api/analytics/timeseries` | `GET` | Downsampled telemetry history for charts that outlive a page refresh. | +| `/api/analytics/models` | `GET` | Recency/frequency model ranking used to prioritise the warm budget. | +| `/api/history?durable=true` | `GET` | Swap history from the persistent store rather than the in-memory ring. | +| `/api/db` | `GET` | Store location, row counts and how many hours of history are held. | + +### Governor & Autotune Endpoints + +| Endpoint | Method | Description | +| :--- | :--- | :--- | +| `/api/governor` | `GET` / `POST` | Current derate level and why; enable/disable, or clear an active derate. | +| `/api/overclock/restore` | `POST` | Drop all clock locks and offsets, restore default power limit and automatic fans. | +| `/api/autotune` | `GET` | Sweep progress, last result, and every recorded autotune step. | +| `/api/autotune/sweep` | `POST` | Walk a clock offset upward, measuring tok/s and watching for instability at each step. | +| `/api/autotune/cancel` | `POST` | Stop the current sweep after the step in flight; the profile is restored either way. | + --- ## 4. Model Context Protocol (MCP 2.0) Reference diff --git a/autotune.py b/autotune.py new file mode 100644 index 0000000..c819096 --- /dev/null +++ b/autotune.py @@ -0,0 +1,263 @@ +"""Closed-loop overclock autotuner. + +The profiles in this repo were hand-tuned and had already drifted apart from the defaults +in overclock_manager.py, with no record of which numbers were actually faster. This module +answers that empirically: it walks a clock offset upward, measures real decode throughput +at each step, watches for instability, and reports the highest setting that was both +stable and fastest. + +Safety properties: + * The original profile is always restored, including on exception or cancellation. + * A sweep refuses to start while ComfyUI is executing, so it cannot corrupt someone's + render by yanking clocks mid-graph. + * Every step is bounded by a temperature ceiling and checked for kernel Xid messages, + and the sweep stops climbing the moment a step looks unstable. +""" +import asyncio +import logging +import subprocess +import time +from typing import Any, Dict, List, Optional + +import overclock_manager +import telemetry_store +import vram_arbitrator + +logger = logging.getLogger("autotune") + +BENCH_PROMPT = ("Write a detailed technical explanation of how virtual memory paging " + "works in a modern operating system kernel.") +BENCH_TOKENS = 160 +SETTLE_S = 2.5 +TEMP_CEILING_C = 84.0 + +KNOBS = { + "mem_offset_mhz": {"default_start": 0, "default_stop": 1000, "default_step": 100}, + "core_offset_mhz": {"default_start": 0, "default_stop": 300, "default_step": 25}, +} + + +def _xid_since(since_ts: float) -> List[str]: + """Look for NVIDIA Xid errors in the kernel log โ€” the clearest instability signal.""" + try: + since = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(since_ts)) + proc = subprocess.run( + ["journalctl", "-k", "--since", since, "--no-pager", "-q"], + capture_output=True, text=True, timeout=10, + ) + return [ln.strip() for ln in proc.stdout.splitlines() + if "Xid" in ln or "NVRM:" in ln] + except Exception as e: + logger.debug(f"Xid check unavailable: {e}") + return [] + + +async def _decode_benchmark(model: str) -> Dict[str, Any]: + """One fixed decode run. Throughput here is the thing being optimised.""" + client = vram_arbitrator._client(vram_arbitrator.OLLAMA_API_BASE, 300.0) + t0 = time.perf_counter() + resp = await client.post("/api/generate", json={ + "model": model, + "prompt": BENCH_PROMPT, + "stream": False, + "keep_alive": "10m", + "options": {"num_predict": BENCH_TOKENS, "temperature": 0.0, "seed": 42}, + }) + wall_ms = round((time.perf_counter() - t0) * 1000, 2) + if resp.status_code != 200: + return {"ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:200]}", + "wall_ms": wall_ms} + data = resp.json() + eval_ms = data.get("eval_duration", 0) / 1e6 + eval_count = data.get("eval_count", 0) + text = data.get("response", "") or "" + return { + "ok": True, + "tokens_per_sec": round(eval_count / (eval_ms / 1000), 2) if eval_ms > 0 else 0.0, + "eval_count": eval_count, + "eval_ms": round(eval_ms, 2), + "prompt_eval_ms": round(data.get("prompt_eval_duration", 0) / 1e6, 2), + "wall_ms": wall_ms, + "response_chars": len(text), + # A model producing almost nothing, or pure repetition, is a corruption signal. + "degenerate": eval_count < BENCH_TOKENS * 0.5 or len(set(text.split())) < 8, + } + + +class SweepState: + def __init__(self) -> None: + self.running = False + self.cancel = False + self.current: Optional[Dict[str, Any]] = None + self.last_result: Optional[Dict[str, Any]] = None + + +state = SweepState() + + +async def sweep(knob: str = "mem_offset_mhz", + profile: str = "ollama", + model: Optional[str] = None, + start: Optional[int] = None, + stop: Optional[int] = None, + step: Optional[int] = None, + repeats: int = 1, + apply_best: bool = False) -> Dict[str, Any]: + """Sweep one clock offset and return the fastest stable value.""" + if knob not in KNOBS: + return {"success": False, "error": f"unknown knob '{knob}'; try {list(KNOBS)}"} + if state.running: + return {"success": False, "error": "a sweep is already running"} + + comfy = await vram_arbitrator.get_comfyui_live_state() + if comfy.get("executing") or comfy.get("queue_remaining"): + return {"success": False, "error": "ComfyUI is busy; refusing to change clocks mid-render"} + + if not model: + ollama = await vram_arbitrator.get_ollama_live_state() + model = ollama.get("active_model_name") + if not model: + installed = ollama.get("installed_models") or [] + if not installed: + return {"success": False, "error": "no Ollama model available to benchmark"} + model = installed[0].get("name") + + defaults = KNOBS[knob] + start = defaults["default_start"] if start is None else start + stop = defaults["default_stop"] if stop is None else stop + step = defaults["default_step"] if step is None else step + if step <= 0 or stop < start: + return {"success": False, "error": "invalid sweep range"} + + baseline_cfg = overclock_manager.load_profiles().get(profile, {}) + state.running = True + state.cancel = False + results: List[Dict[str, Any]] = [] + t_start = time.time() + + try: + # Load the model once up front so the first step does not pay the load cost. + await _decode_benchmark(model) + + value = start + while value <= stop and not state.cancel: + overclock_manager.apply_profile(profile, overrides={knob: value}) + await asyncio.sleep(SETTLE_S) + step_started = time.time() + + samples = [] + for _ in range(max(repeats, 1)): + samples.append(await _decode_benchmark(model)) + if state.cancel: + break + + gpu = vram_arbitrator.get_gpu_hardware_stats() + xids = _xid_since(step_started) + ok_samples = [s for s in samples if s.get("ok") and not s.get("degenerate")] + temp = gpu.get("temperature_c", 0) or 0 + + instability = [] + if xids: + instability.append(f"kernel Xid: {xids[0][:120]}") + if len(ok_samples) < len(samples): + instability.append("benchmark failed or produced degenerate output") + if temp >= TEMP_CEILING_C: + instability.append(f"temperature ceiling hit ({temp}ยฐC)") + + tok_s = round(max((s["tokens_per_sec"] for s in ok_samples), default=0.0), 2) + row = { + "knob": knob, + "value": value, + "profile": profile, + "model": model, + "tokens_per_sec": tok_s, + "temp_c": temp, + "power_w": gpu.get("power_w"), + "clock_sm_mhz": gpu.get("clock_graphics_mhz"), + "clock_mem_mhz": gpu.get("clock_mem_mhz"), + "throttle_reasons": gpu.get("throttle_reasons"), + "stable": not instability, + "instability": "; ".join(instability) or None, + "samples": samples, + } + results.append(row) + telemetry_store.record_autotune({ + "profile": profile, "knob": knob, + "core_offset_mhz": value if knob == "core_offset_mhz" else baseline_cfg.get("core_offset_mhz"), + "mem_offset_mhz": value if knob == "mem_offset_mhz" else baseline_cfg.get("mem_offset_mhz"), + "tokens_per_sec": tok_s, "temp_c": temp, "power_w": gpu.get("power_w"), + "stable": row["stable"], "instability": row["instability"], + "note": f"sweep {knob} {start}..{stop} step {step}", + }) + state.current = {"knob": knob, "value": value, "stop": stop, + "tokens_per_sec": tok_s, "stable": row["stable"]} + logger.info(f"autotune {knob}={value}: {tok_s} tok/s, {temp}ยฐC, " + f"stable={row['stable']} {row['instability'] or ''}") + + if not row["stable"]: + logger.warning(f"autotune stopping climb at {knob}={value}: {row['instability']}") + break + value += step + + stable = [r for r in results if r["stable"] and r["tokens_per_sec"] > 0] + best = max(stable, key=lambda r: r["tokens_per_sec"]) if stable else None + baseline = next((r for r in results if r["value"] == start), None) + gain_pct = None + if best and baseline and baseline["tokens_per_sec"] > 0: + gain_pct = round((best["tokens_per_sec"] / baseline["tokens_per_sec"] - 1) * 100, 2) + + applied = None + if apply_best and best: + overclock_manager.set_profile(profile, {knob: best["value"]}) + applied = {knob: best["value"], "profile": profile} + logger.info(f"autotune wrote {knob}={best['value']} into profile '{profile}'") + + result = { + "success": True, + "knob": knob, + "profile": profile, + "model": model, + "range": {"start": start, "stop": stop, "step": step}, + "steps_run": len(results), + "duration_s": round(time.time() - t_start, 1), + "cancelled": state.cancel, + "best": {k: best[k] for k in ("value", "tokens_per_sec", "temp_c", "clock_mem_mhz", + "clock_sm_mhz")} if best else None, + "baseline_tokens_per_sec": baseline["tokens_per_sec"] if baseline else None, + "gain_pct": gain_pct, + "applied_to_profile": applied, + "first_unstable": next(({"value": r["value"], "why": r["instability"]} + for r in results if not r["stable"]), None), + "table": [{k: r[k] for k in ("value", "tokens_per_sec", "temp_c", "power_w", + "clock_mem_mhz", "clock_sm_mhz", "stable", + "instability")} for r in results], + } + state.last_result = result + return result + finally: + # Always hand the card back exactly as we found it. + state.running = False + state.current = None + try: + overclock_manager.apply_profile(profile) + logger.info(f"autotune restored profile '{profile}'") + except Exception as e: + logger.error(f"autotune failed to restore profile, forcing stock: {e}") + overclock_manager.restore_safe("autotune restore failed") + + +def get_status() -> Dict[str, Any]: + return { + "running": state.running, + "current": state.current, + "last_result": state.last_result, + "knobs": KNOBS, + "history": telemetry_store.autotune_history(100), + } + + +def cancel() -> Dict[str, Any]: + if not state.running: + return {"cancelled": False, "reason": "no sweep running"} + state.cancel = True + return {"cancelled": True} diff --git a/overclock_manager.py b/overclock_manager.py index 5f5d3b0..1fb7368 100644 --- a/overclock_manager.py +++ b/overclock_manager.py @@ -207,14 +207,20 @@ def _apply_offsets(core_mhz: int, mem_mhz: int) -> Dict[str, Any]: } -def apply_profile(name: str) -> Dict[str, Any]: - """Apply a named overclock profile to the GPU. Returns a full result report.""" +def apply_profile(name: str, overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """Apply a named overclock profile to the GPU. Returns a full result report. + + `overrides` lets the thermal governor and the autotuner apply a modified version of a + profile (a derated offset, a probe clock) without mutating what is stored on disk. + """ global ACTIVE_PROFILE, _LAST_RESULT profiles = load_profiles() if name not in profiles: return {"success": False, "error": f"unknown profile '{name}'", "profile": name} - cfg = profiles[name] + cfg = dict(profiles[name]) + if overrides: + cfg.update(overrides) fan_mode = cfg.get("fan_mode", "auto") fan_speed = int(cfg.get("fan_speed_pct", 0)) @@ -230,15 +236,31 @@ def apply_profile(name: str) -> Dict[str, Any]: } result["gpu"] = get_gpu_state() result["fan_status"] = get_fan_status() + result["overrides"] = overrides or {} + _STATE_CACHE["value"] = None + _FAN_CACHE["value"] = None ACTIVE_PROFILE = name _LAST_RESULT = result logger.info(f"Overclock profile applied: {name} -> {json.dumps(result, default=str)}") return result -def get_gpu_state() -> Dict[str, Any]: - """Read back live GPU clocks/power/limits via nvidia-smi.""" +_STATE_CACHE: Dict[str, Any] = {"ts": 0.0, "value": None} +_FAN_CACHE: Dict[str, Any] = {"ts": 0.0, "value": None} +STATE_TTL_S = 2.0 + + +def get_gpu_state(force: bool = False) -> Dict[str, Any]: + """Read back live GPU clocks/power/limits via nvidia-smi. + + Cached for STATE_TTL_S: this forks `sudo nvidia-smi`, and the dashboard polls the + status endpoint every few seconds. NVML already covers the live 1 Hz telemetry. + """ + import time as _time + if not force and _STATE_CACHE["value"] is not None and \ + (_time.time() - _STATE_CACHE["ts"]) < STATE_TTL_S: + return _STATE_CACHE["value"] state: Dict[str, Any] = {} r = _smi( "--query-gpu=driver_version,name,memory.total,power.limit,power.max_limit,power.default_limit," @@ -257,6 +279,7 @@ def get_gpu_state() -> Dict[str, Any]: state[k] = float(parts[i]) except ValueError: state[k] = parts[i] + _STATE_CACHE.update({"ts": __import__("time").time(), "value": state}) return state @@ -298,11 +321,16 @@ def set_fan_auto() -> Dict[str, Any]: ok = r["rc"] == 0 if ok: FAN_MANUAL = False + _FAN_CACHE["value"] = None return {"success": ok, "manual": False, "fan_speed_pct": None, "detail": r.get("out") or r.get("err")} -def get_fan_status() -> Dict[str, Any]: - """Read current fan control mode + target speed.""" +def get_fan_status(force: bool = False) -> Dict[str, Any]: + """Read current fan control mode + target speed (cached; forks nvidia-settings).""" + import time as _time + if not force and _FAN_CACHE["value"] is not None and \ + (_time.time() - _FAN_CACHE["ts"]) < STATE_TTL_S: + return _FAN_CACHE["value"] global FAN_MANUAL target = None manual = FAN_MANUAL @@ -320,7 +348,34 @@ def get_fan_status() -> Dict[str, Any]: target = int(line.split("):")[-1].split(".")[0].strip()) except Exception: pass - return {"manual": manual, "mode": "manual" if manual else "auto", "target_speed_pct": target} + result = {"manual": manual, "mode": "manual" if manual else "auto", "target_speed_pct": target} + _FAN_CACHE.update({"ts": __import__("time").time(), "value": result}) + return result + + +def restore_safe(reason: str = "shutdown") -> Dict[str, Any]: + """Return the card to stock: no clock locks, no offsets, default power, automatic fans. + + This matters because every lever here is sticky. If the service dies while a profile is + applied, the GPU keeps the locked clocks and, worse, keeps the fans pinned at whatever + manual PWM was last set. Nothing was undoing that. + """ + logger.warning(f"Restoring GPU to safe stock state ({reason})") + result = { + "reason": reason, + "clock_lock": _apply_clock_lock(0, 0), + "mem_lock": _apply_mem_lock(0), + "offsets": _apply_offsets(0, 0), + "fan": set_fan_auto(), + } + # Hand the power limit back to the card's own default rather than assuming 370 W. + state = get_gpu_state() + default_w = state.get("power_default_w") + if isinstance(default_w, (int, float)) and default_w > 0: + result["power_limit"] = _apply_power_limit(int(default_w)) + global ACTIVE_PROFILE + ACTIVE_PROFILE = "stock" + return result def get_status() -> Dict[str, Any]: diff --git a/overclock_profiles.json b/overclock_profiles.json index 573a6b7..d871404 100644 --- a/overclock_profiles.json +++ b/overclock_profiles.json @@ -2,8 +2,8 @@ "ollama": { "label": "Ollama \u2014 LLM decode (memory-bandwidth bound)", "power_limit_w": 370, - "core_offset_mhz": 150, - "mem_offset_mhz": 825, + "core_offset_mhz": 35, + "mem_offset_mhz": 200, "lock_core_min": 0, "lock_core_max": 0, "lock_mem_mhz": 0, @@ -14,12 +14,12 @@ "label": "ComfyUI \u2014 diffusion (core-compute bound)", "power_limit_w": 370, "core_offset_mhz": 100, - "mem_offset_mhz": 500, + "mem_offset_mhz": 150, "lock_core_min": 2900, "lock_core_max": 3105, "lock_mem_mhz": 0, "fan_mode": "manual", - "fan_speed_pct": 75 + "fan_speed_pct": 100 }, "balanced": { "label": "Balanced \u2014 stock boost, power unlocked", diff --git a/ram_optimizer.py b/ram_optimizer.py index a8306ee..bf0bff2 100644 --- a/ram_optimizer.py +++ b/ram_optimizer.py @@ -1,16 +1,48 @@ -"""RAM Optimizer and Model Pre-warmer for High-Speed Switching.""" -import os -import glob -import time -import httpx +"""RAM Optimizer and Model Pre-warmer for High-Speed Switching. + +Two things changed here versus the naive version: + + 1. Residency is *measured*, not assumed. mincore(2) tells us exactly what fraction of + each model file is resident in the Linux page cache, so "RAM Cache Hit" stops being + a guess based on how long a load took. + 2. Warming is *budgeted*. This box has 64 GB of RAM and >33 GB of models; reading every + file top-to-bottom simply evicts whatever was warmed first. Files are now scored by + recency/frequency (from the telemetry store) and warmed until a byte budget is hit, + skipping anything already resident. +""" +import ctypes +import ctypes.util +import json import logging -from typing import Dict, List, Any +import os +import random +import time +from typing import Dict, List, Any, Optional, Tuple + +import httpx + +import telemetry_store logger = logging.getLogger("ram_optimizer") OLLAMA_API_BASE = "http://localhost:11434" COMFY_API_BASE = "http://127.0.0.1:8188" -COMFY_MODELS_DIR = "/home/drjones/ComfyUI/models" +COMFY_MODELS_DIR = os.environ.get("HYPERSWAP_COMFY_MODELS", "/home/drjones/ComfyUI/models") +OLLAMA_MODEL_DIRS = [ + "/usr/share/ollama/.ollama/models", + os.path.expanduser("~/.ollama/models"), +] + +PAGE_SIZE = os.sysconf("SC_PAGE_SIZE") +# Files bigger than this are sampled rather than fully mapped for residency. +RESIDENCY_FULL_MAP_LIMIT = 2 * 1024 ** 3 +RESIDENCY_SAMPLE_WINDOWS = 64 +RESIDENCY_WINDOW_BYTES = 16 * 1024 * 1024 +# A file at/above this residency is considered warm and is skipped by the warmer. +WARM_SKIP_THRESHOLD_PCT = 90.0 + +CATALOG_TTL_S = 30.0 + def get_detailed_meminfo() -> Dict[str, Any]: """Parse /proc/meminfo for precise page cache and RAM stats.""" @@ -25,7 +57,7 @@ def get_detailed_meminfo() -> Dict[str, Any]: info[key] = int(val) * 1024 # Convert kB to bytes except Exception as e: logger.error(f"Failed to read /proc/meminfo: {e}") - + total = info.get("MemTotal", 0) free = info.get("MemFree", 0) available = info.get("MemAvailable", 0) @@ -51,43 +83,439 @@ def get_detailed_meminfo() -> Dict[str, Any]: "cache_ratio_pct": round((cached / total * 100) if total > 0 else 0, 1), } -def warm_file_to_ram(filepath: str, chunk_size: int = 16 * 1024 * 1024) -> Dict[str, Any]: - """Pre-fault/read file into Linux OS Page Cache at maximum disk read speed.""" + +# ---------------------------------------------------------------- page residency +# +# Measuring page-cache residency turned out to be the subtle part. +# +# * cachestat(2) (Linux 6.5+) is the right tool: exact cached-page counts for an fd, +# no mmap, microseconds per call. But the kernel only permits it on files you own +# or can write -- the Ollama blobs are owned by uid `ollama`, so it returns EPERM. +# * mincore(2) does NOT fail closed for those files on this kernel: it reports every +# page as resident, which produced 128 GB of "resident" model weights on a box with +# 46 GB of page cache. It is therefore not used at all. +# +# So: cachestat where permitted, and an explicit read-throughput probe where it is not. +# Anything we cannot measure is reported as unmeasurable rather than guessed at. + +_libc = None +_SYS_cachestat = 451 # x86_64 + + +class _CachestatRange(ctypes.Structure): + _fields_ = [("off", ctypes.c_uint64), ("len", ctypes.c_uint64)] + + +class _Cachestat(ctypes.Structure): + _fields_ = [ + ("nr_cache", ctypes.c_uint64), + ("nr_dirty", ctypes.c_uint64), + ("nr_writeback", ctypes.c_uint64), + ("nr_evicted", ctypes.c_uint64), + ("nr_recently_evicted", ctypes.c_uint64), + ] + + +def _get_libc(): + global _libc + if _libc is None: + _libc = ctypes.CDLL(ctypes.util.find_library("c") or "libc.so.6", use_errno=True) + return _libc + + +def _cachestat(fd: int, offset: int, length: int) -> Optional[_Cachestat]: + """Raw cachestat(2). Returns None if the kernel refuses (EPERM/ENOSYS).""" + libc = _get_libc() + rng = _CachestatRange(offset, length) + cs = _Cachestat() + ctypes.set_errno(0) + rc = libc.syscall(ctypes.c_long(_SYS_cachestat), ctypes.c_int(fd), + ctypes.byref(rng), ctypes.byref(cs), ctypes.c_uint(0)) + if rc != 0: + return None + return cs + + +PROBE_WINDOWS = 12 +PROBE_WINDOW_BYTES = 2 * 1024 * 1024 +# Measured on this box: cold NVMe reads land around 0.35-0.5 GB/s, page-cache reads at +# 3.2-13 GB/s. 1.5 GB/s sits in the empty middle of that gap. +PROBE_CACHED_GBPS = 1.5 + + +def _throughput_probe(fd: int, size: int) -> Dict[str, Any]: + """Infer residency by timing reads of small windows spread across the file. + + Used only where cachestat is not permitted (Ollama's blobs are owned by uid `ollama`). + + Two details matter for correctness: + + * Offsets are random per call. A fixed stride made the probe self-fulfilling: the + first pass faulted its 24 MB of sample windows into the page cache, and every pass + after that re-read exactly those windows and reported 100% resident for a file that + was almost entirely cold. + * Windows that read cold are handed straight back with FADV_DONTNEED. Those pages are + pollution the probe itself created, and leaving them behind would slowly warm the + cache with data nobody asked for. + """ + windows = min(PROBE_WINDOWS, max(int(size // PROBE_WINDOW_BYTES), 1)) + if windows <= 0: + return {"resident_pct": 0.0, "windows": 0} + + max_off = max(size - PROBE_WINDOW_BYTES, 0) + offsets = sorted(random.randint(0, max_off) for _ in range(windows)) if max_off else [0] + buf = bytearray(PROBE_WINDOW_BYTES) + cached = 0 + rates = [] + for off in offsets: + length = min(PROBE_WINDOW_BYTES, size - off) + if length <= 0: + continue + view = memoryview(buf)[:length] + t0 = time.perf_counter() + os.preadv(fd, [view], off) + dt = time.perf_counter() - t0 + gbps = (length / (1024 ** 3)) / dt if dt > 0 else 0.0 + rates.append(gbps) + if gbps >= PROBE_CACHED_GBPS: + cached += 1 + else: + # We just pulled this off disk; put it back the way we found it. + try: + os.posix_fadvise(fd, off, length, os.POSIX_FADV_DONTNEED) + except Exception: + pass + n = len(rates) + return { + "resident_pct": round((cached / n * 100) if n else 0.0, 1), + "windows": n, + "median_gbps": round(sorted(rates)[n // 2], 2) if n else 0.0, + "sampled_gb": round(n * PROBE_WINDOW_BYTES / (1024 ** 3), 3), + } + + +def page_residency(filepath: str, allow_probe: bool = True) -> Dict[str, Any]: + """Measure what fraction of a file is resident in the Linux page cache.""" + try: + size = os.path.getsize(filepath) + except OSError as e: + return {"success": False, "error": str(e), "resident_pct": 0.0, "measurable": False} + if size == 0: + return {"success": True, "resident_pct": 0.0, "size_bytes": 0, "measurable": True, + "method": "empty"} + + try: + fd = os.open(filepath, os.O_RDONLY) + except OSError as e: + return {"success": False, "error": str(e), "resident_pct": 0.0, "measurable": False} + try: + cs = _cachestat(fd, 0, size) + if cs is not None: + total_pages = (size + PAGE_SIZE - 1) // PAGE_SIZE + pct = round((cs.nr_cache / total_pages * 100) if total_pages else 0.0, 1) + method, measurable = "cachestat", True + extra = {"dirty_pages": cs.nr_dirty, "evicted_pages": cs.nr_evicted} + elif allow_probe: + probe = _throughput_probe(fd, size) + pct = probe["resident_pct"] + method, measurable = "probe", True + extra = {"probe_windows": probe["windows"], "probe_median_gbps": probe.get("median_gbps")} + else: + return {"success": True, "filepath": filepath, "size_bytes": size, + "size_gb": round(size / (1024**3), 3), "resident_pct": None, + "measurable": False, "method": "unavailable", "warm": None, + "reason": "cachestat not permitted for this file (not owned by us)"} + + return { + "success": True, + "filepath": filepath, + "size_bytes": size, + "size_gb": round(size / (1024**3), 3), + "resident_pct": pct, + "resident_bytes": int(size * pct / 100.0), + "method": method, + "measurable": measurable, + "warm": pct >= WARM_SKIP_THRESHOLD_PCT, + **extra, + } + except Exception as e: + return {"success": False, "error": str(e), "resident_pct": 0.0, + "size_bytes": size, "measurable": False} + finally: + os.close(fd) + + +def residency_capability() -> Dict[str, Any]: + """Report whether exact residency is available, and how to enable it if not.""" + catalog = get_model_catalog() + blocked = [] + for f in catalog["ollama"]: + try: + fd = os.open(f["full_path"], os.O_RDONLY) + except OSError: + continue + try: + if _cachestat(fd, 0, 4096) is None: + blocked.append(f["full_path"]) + finally: + os.close(fd) + break # one probe is enough; blobs share a directory and owner + if not blocked: + return {"exact_everywhere": True} + owner = "" + try: + import pwd + owner = pwd.getpwuid(os.stat(blocked[0]).st_uid).pw_name + except Exception: + owner = str(os.stat(blocked[0]).st_uid) + return { + "exact_everywhere": False, + "method_for_blocked": "probe", + "reason": f"cachestat(2) is only permitted on files you own or can write; " + f"Ollama blobs are owned by '{owner}'", + "hint": f"exact numbers for Ollama weights need read/write access, e.g. " + f"'sudo usermod -aG {owner} $USER' plus group-write on the blobs directory", + } + + +# ---------------------------------------------------------------- catalogs + +_catalog_cache: Dict[str, Any] = {"ts": 0.0, "sig": None, "comfy": [], "ollama": []} + + +def _dir_signature(root: str) -> Tuple: + """Cheap fingerprint of a model tree: (mtime, entry count) per subdirectory.""" + sig = [] + if not os.path.isdir(root): + return tuple(sig) + for dirpath, dirnames, filenames in os.walk(root): + try: + sig.append((dirpath, os.stat(dirpath).st_mtime_ns, len(filenames))) + except OSError: + continue + return tuple(sig) + + +def find_ollama_model_files() -> List[Dict[str, Any]]: + """Map installed Ollama models to their on-disk GGUF blobs via the manifest tree. + + Knowing the blob path is what lets us warm (or measure) a specific model's weights + without pulling them into VRAM. + """ + results: List[Dict[str, Any]] = [] + seen = set() + for root in OLLAMA_MODEL_DIRS: + manifests = os.path.join(root, "manifests") + blobs = os.path.join(root, "blobs") + if not os.path.isdir(manifests): + continue + for dirpath, _, filenames in os.walk(manifests): + for tag in filenames: + manifest_path = os.path.join(dirpath, tag) + try: + with open(manifest_path) as f: + manifest = json.load(f) + except Exception: + continue + rel = os.path.relpath(dirpath, manifests) + parts = rel.split(os.sep) + # registry/namespace/name -> "name:tag", keeping non-library namespaces + name = parts[-1] if parts else rel + namespace = parts[-2] if len(parts) >= 2 else "library" + model_name = f"{name}:{tag}" if namespace == "library" else f"{namespace}/{name}:{tag}" + for layer in manifest.get("layers", []): + if layer.get("mediaType") != "application/vnd.ollama.image.model": + continue + digest = (layer.get("digest") or "").replace(":", "-") + blob_path = os.path.join(blobs, digest) + if not os.path.exists(blob_path): + continue + key = (model_name, blob_path) + if key in seen: + continue + seen.add(key) + size = layer.get("size") or os.path.getsize(blob_path) + results.append({ + "model": model_name, + "filename": digest, + "full_path": blob_path, + "size_bytes": size, + "size_gb": round(size / (1024**3), 3), + "kind": "ollama", + }) + return results + + +def find_comfy_model_files(force_refresh: bool = False) -> List[Dict[str, Any]]: + """Discover all model files under ComfyUI models (cached). + + This used to run inside the 1Hz telemetry snapshot, meaning a full recursive walk plus + a stat() of every checkpoint once per second per connected dashboard. It is now cached + behind a directory-mtime fingerprint. + """ + _refresh_catalog(force_refresh) + return _catalog_cache["comfy"] + + +def get_model_catalog(force_refresh: bool = False) -> Dict[str, Any]: + _refresh_catalog(force_refresh) + return { + "comfy": _catalog_cache["comfy"], + "ollama": _catalog_cache["ollama"], + "cached_at": _catalog_cache["ts"], + } + + +def _refresh_catalog(force: bool = False) -> None: + now = time.time() + if not force and (now - _catalog_cache["ts"]) < CATALOG_TTL_S: + return + sig = _dir_signature(COMFY_MODELS_DIR) + if not force and sig == _catalog_cache["sig"] and _catalog_cache["comfy"]: + _catalog_cache["ts"] = now + return + + extensions = (".safetensors", ".ckpt", ".pt", ".bin", ".gguf", ".sft") + results = [] + if os.path.exists(COMFY_MODELS_DIR): + for root, _, files in os.walk(COMFY_MODELS_DIR): + for file in files: + if not file.endswith(extensions): + continue + full_path = os.path.join(root, file) + try: + st = os.stat(full_path) + except OSError: + continue + results.append({ + "filename": file, + "rel_path": os.path.relpath(full_path, COMFY_MODELS_DIR), + "full_path": full_path, + "category": os.path.relpath(root, COMFY_MODELS_DIR).split(os.sep)[0], + "size_bytes": st.st_size, + "size_mb": round(st.st_size / (1024**2), 2), + "size_gb": round(st.st_size / (1024**3), 3), + "mtime": st.st_mtime, + "kind": "comfy", + }) + _catalog_cache.update({"ts": now, "sig": sig, "comfy": results, + "ollama": find_ollama_model_files()}) + + +# ---------------------------------------------------------------- residency report + +_report_cache: Dict[str, Any] = {"ts": 0.0, "report": None} +REPORT_TTL_S = 15.0 + + +def get_cache_report(include_files: bool = True, force_refresh: bool = False) -> Dict[str, Any]: + """Measured page-cache residency across the whole model catalog. + + Deduplicated by blob path: several Ollama tags routinely point at the same GGUF, and + counting each tag separately produced more "resident" bytes than the box has RAM. + """ + now = time.time() + cached = _report_cache["report"] + if cached and not force_refresh and (now - _report_cache["ts"]) < REPORT_TTL_S: + return cached if include_files else {**cached, "files": []} + + t0 = time.perf_counter() + catalog = get_model_catalog() + by_path: Dict[str, Dict[str, Any]] = {} + for f in list(catalog["ollama"]) + list(catalog["comfy"]): + path = f["full_path"] + name = f.get("model") or f.get("rel_path") or f.get("filename") + if path in by_path: + by_path[path]["aliases"].append(name) + continue + by_path[path] = {"entry": f, "name": name, "aliases": []} + + entries = [] + total_bytes = resident_bytes = 0 + for path, meta in by_path.items(): + f = meta["entry"] + res = page_residency(path) + size = f.get("size_bytes") or res.get("size_bytes") or 0 + rb = res.get("resident_bytes", 0) + total_bytes += size + resident_bytes += rb + entries.append({ + "name": meta["name"], + "aliases": meta["aliases"], + "kind": f.get("kind"), + "full_path": path, + "size_gb": round(size / (1024**3), 3), + "resident_pct": res.get("resident_pct", 0.0), + "resident_gb": round(rb / (1024**3), 3), + "warm": res.get("warm", False), + }) + entries.sort(key=lambda e: e["resident_gb"], reverse=True) + + report = { + "scan_ms": round((time.perf_counter() - t0) * 1000, 1), + "files_scanned": len(entries), + "unique_blobs": len(by_path), + "catalog_total_gb": round(total_bytes / (1024**3), 2), + "resident_total_gb": round(resident_bytes / (1024**3), 2), + "residency_pct": round((resident_bytes / total_bytes * 100) if total_bytes else 0, 1), + "warm_files": sum(1 for e in entries if e["warm"]), + "files": entries, + } + _report_cache.update({"ts": now, "report": report}) + return report if include_files else {**report, "files": []} + + +# ---------------------------------------------------------------- warming + +def warm_file_to_ram(filepath: str, chunk_size: int = 16 * 1024 * 1024, + skip_if_warm: bool = True) -> Dict[str, Any]: + """Pre-fault a file into the Linux page cache, skipping it if already resident.""" if not os.path.exists(filepath): return {"success": False, "error": f"File not found: {filepath}", "duration_ms": 0} - + + before = page_residency(filepath) + if skip_if_warm and before.get("warm"): + return { + "success": True, "filepath": filepath, "skipped": True, + "reason": "already resident", "resident_pct": before.get("resident_pct"), + "size_mb": round(before.get("size_bytes", 0) / (1024**2), 2), + "duration_ms": 0.0, "bytes_read": 0, + } + t0 = time.perf_counter() file_size = os.path.getsize(filepath) bytes_read = 0 try: with open(filepath, "rb") as f: - # Hint kernel that we will read this sequentially try: os.posix_fadvise(f.fileno(), 0, file_size, os.POSIX_FADV_WILLNEED) except Exception: pass - buf = bytearray(chunk_size) while True: n = f.readinto(buf) if not n: break bytes_read += n - + duration = time.perf_counter() - t0 - duration_ms = round(duration * 1000, 2) - speed_mb_s = round((bytes_read / (1024**2)) / duration if duration > 0 else 0, 2) + after = page_residency(filepath) return { "success": True, "filepath": filepath, + "skipped": False, "size_bytes": file_size, "size_mb": round(file_size / (1024**2), 2), "bytes_read": bytes_read, - "duration_ms": duration_ms, - "speed_mb_s": speed_mb_s, + "duration_ms": round(duration * 1000, 2), + "speed_mb_s": round((bytes_read / (1024**2)) / duration if duration > 0 else 0, 2), + "resident_pct_before": before.get("resident_pct", 0.0), + "resident_pct_after": after.get("resident_pct", 0.0), } except Exception as e: - return {"success": False, "error": str(e), "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + return {"success": False, "error": str(e), + "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + async def warm_ollama_model(model_name: str, keep_alive: str = "5m") -> Dict[str, Any]: """Warm an Ollama model into memory and measure time.""" @@ -101,76 +529,148 @@ async def warm_ollama_model(model_name: str, keep_alive: str = "5m") -> Dict[str duration = time.perf_counter() - t0 if resp.status_code == 200: data = resp.json() - return { + res = { "success": True, "model": model_name, "duration_ms": round(duration * 1000, 2), "load_duration_ms": round(data.get("load_duration", 0) / 1e6, 2), "total_duration_ms": round(data.get("total_duration", 0) / 1e6, 2), } - else: - return { - "success": False, - "model": model_name, - "error": f"HTTP {resp.status_code}: {resp.text}", - "duration_ms": round(duration * 1000, 2), - } + telemetry_store.record_event({ + "event_type": "Model Warm", "source": "warmer", "target": model_name, + "duration_ms": res["duration_ms"], "load_duration_ms": res["load_duration_ms"], + }) + return res + return { + "success": False, "model": model_name, + "error": f"HTTP {resp.status_code}: {resp.text}", + "duration_ms": round(duration * 1000, 2), + } except Exception as e: - return {"success": False, "model": model_name, "error": str(e), "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + return {"success": False, "model": model_name, "error": str(e), + "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} -def find_comfy_model_files() -> List[Dict[str, Any]]: - """Discover all model files under ComfyUI models.""" - results = [] - extensions = ("*.safetensors", "*.ckpt", "*.pt", "*.bin") - if os.path.exists(COMFY_MODELS_DIR): - for root, _, files in os.walk(COMFY_MODELS_DIR): - for file in files: - if any(file.endswith(ext.replace("*", "")) for ext in extensions): - full_path = os.path.join(root, file) - rel_path = os.path.relpath(full_path, COMFY_MODELS_DIR) - size = os.path.getsize(full_path) - results.append({ - "filename": file, - "rel_path": rel_path, - "full_path": full_path, - "size_bytes": size, - "size_mb": round(size / (1024**2), 2), - "size_gb": round(size / (1024**3), 3), - }) - return results -async def warm_all_models() -> Dict[str, Any]: - """Warm all available Ollama and ComfyUI models into Linux RAM Cache.""" - t0 = time.perf_counter() - warmed_ollama = [] - warmed_comfy = [] - - # 1. Ollama models +def warm_ollama_blob(model_name: str) -> Dict[str, Any]: + """Warm a specific Ollama model's GGUF into page cache without touching VRAM.""" + for f in find_ollama_model_files(): + if f["model"] == model_name: + res = warm_file_to_ram(f["full_path"]) + res["model"] = model_name + return res + return {"success": False, "error": f"no blob found for model '{model_name}'"} + + +def _warm_priority(days: float = 30.0) -> Dict[str, float]: + """Recency/frequency score per model name, from the persisted event log.""" try: - async with httpx.AsyncClient(timeout=10.0) as client: - tags_resp = await client.get(f"{OLLAMA_API_BASE}/api/tags") - if tags_resp.status_code == 200: - models = tags_resp.json().get("models", []) - for m in models: - name = m.get("name") - res = await warm_ollama_model(name, keep_alive="1m") - warmed_ollama.append(res) - except Exception as e: - logger.error(f"Error discovering Ollama models: {e}") - - # 2. ComfyUI models - comfy_files = find_comfy_model_files() - for f in comfy_files: - res = warm_file_to_ram(f["full_path"]) - warmed_comfy.append(res) - - total_duration_ms = round((time.perf_counter() - t0) * 1000, 2) - meminfo = get_detailed_meminfo() - + return {r["model"]: r["score"] for r in telemetry_store.model_usage_ranking(days)} + except Exception: + return {} + + +def build_warm_plan(budget_gb: Optional[float] = None) -> Dict[str, Any]: + """Decide *what* to warm, in what order, within a byte budget. + + Warming everything on a 64 GB box with 33+ GB of models just evicts the earliest + files, so we rank by usage (Ollama, from history) and recency (ComfyUI, by mtime), + then fill until the budget is spent. Already-resident files cost nothing. + """ + mem = get_detailed_meminfo() + if budget_gb is None: + # Leave headroom so warming never pushes the box into reclaim. + budget_gb = max((mem["available_bytes"] * 0.7) / (1024**3), 1.0) + budget_bytes = int(budget_gb * (1024**3)) + + catalog = get_model_catalog() + scores = _warm_priority() + now = time.time() + + candidates = [] + for f in catalog["ollama"]: + candidates.append({**f, "score": scores.get(f["model"], 0.0) + 0.5, + "name": f["model"]}) + for f in catalog["comfy"]: + age_days = max((now - f.get("mtime", now)) / 86400.0, 0.01) + candidates.append({**f, "score": scores.get(f["rel_path"], 0.0) + 1.0 / (1.0 + age_days), + "name": f["rel_path"]}) + + candidates.sort(key=lambda c: c["score"], reverse=True) + + plan, spent, skipped = [], 0, [] + seen_paths = set() + for c in candidates: + if c["full_path"] in seen_paths: + continue + seen_paths.add(c["full_path"]) + res = page_residency(c["full_path"]) + entry = { + "name": c["name"], "kind": c["kind"], "full_path": c["full_path"], + "size_gb": c.get("size_gb", 0), "score": round(c["score"], 4), + "resident_pct": res.get("resident_pct", 0.0), + } + if res.get("warm"): + entry["action"] = "already-warm" + skipped.append(entry) + continue + need = int(c.get("size_bytes", 0) * (1 - res.get("resident_pct", 0) / 100.0)) + if spent + need > budget_bytes: + entry["action"] = "over-budget" + skipped.append(entry) + continue + spent += need + entry["action"] = "warm" + entry["bytes_to_read"] = need + plan.append(entry) + + return { + "budget_gb": round(budget_gb, 2), + "planned_gb": round(spent / (1024**3), 2), + "warm_count": len(plan), + "skipped_count": len(skipped), + "plan": plan, + "skipped": skipped, + "meminfo": mem, + } + + +async def warm_all_models(budget_gb: Optional[float] = None, + include_vram_load: bool = False) -> Dict[str, Any]: + """Warm the highest-value models into the page cache within a byte budget.""" + t0 = time.perf_counter() + plan = build_warm_plan(budget_gb) + warmed = [] + for entry in plan["plan"]: + res = warm_file_to_ram(entry["full_path"]) + res["name"] = entry["name"] + res["kind"] = entry["kind"] + warmed.append(res) + # Budgets are computed up front, but the page cache is shared with the rest of + # the box; bail out if we start pushing the system into reclaim. + if get_detailed_meminfo()["available_gb"] < 4.0: + logger.warning("warm_all_models: stopping early, MemAvailable below 4 GB") + break + + if include_vram_load: + try: + async with httpx.AsyncClient(timeout=10.0) as client: + tags = await client.get(f"{OLLAMA_API_BASE}/api/tags") + if tags.status_code == 200: + top = sorted(tags.json().get("models", []), + key=lambda m: _warm_priority().get(m.get("name"), 0), + reverse=True)[:1] + for m in top: + await warm_ollama_model(m.get("name"), keep_alive="1m") + except Exception as e: + logger.debug(f"optional VRAM preload skipped: {e}") + return { "status": "completed", - "total_duration_ms": total_duration_ms, - "ollama_models_warmed": warmed_ollama, - "comfy_files_warmed": warmed_comfy, - "meminfo_after": meminfo, + "total_duration_ms": round((time.perf_counter() - t0) * 1000, 2), + "budget_gb": plan["budget_gb"], + "planned_gb": plan["planned_gb"], + "files_warmed": warmed, + "bytes_read": sum(w.get("bytes_read", 0) for w in warmed), + "skipped": plan["skipped"], + "meminfo_after": get_detailed_meminfo(), } diff --git a/server.py b/server.py index 5ebfb7f..c172792 100644 --- a/server.py +++ b/server.py @@ -1,27 +1,169 @@ """FastAPI Backend Server with SSE Real-Time Telemetry and Model Orchestration API.""" import asyncio +import contextlib import json import logging -from typing import Dict, Any, Optional, List +import time +from contextlib import asynccontextmanager +from typing import Dict, Any, Optional, List, Set + from fastapi import FastAPI, Request, HTTPException, Query from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field -import ram_optimizer -import vram_arbitrator +import autotune import overclock_manager +import ram_optimizer +import telemetry_store +import thermal_governor +import vram_arbitrator logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") logger = logging.getLogger("model_manager_server") +# The sampler makes four HTTP calls a second; at INFO, httpx narrates every one of them. +logging.getLogger("httpx").setLevel(logging.WARNING) +logging.getLogger("httpcore").setLevel(logging.WARNING) + +BASE_DIR = "/home/drjones/unified-model-manager" + + +# ========================================== +# TELEMETRY BROKER +# ========================================== + +class TelemetryBroker: + """One sampler, many subscribers. + + Every SSE client used to run its own copy of the full snapshot once per second: + NVML queries, /proc/meminfo, an HTTP round-trip each to Ollama and ComfyUI, and โ€” the + expensive one โ€” a recursive walk of the ComfyUI models tree with a stat() per + checkpoint. Opening the dashboard in three tabs tripled the load on the very thing it + was measuring. Now a single background task samples at 1 Hz and fans the snapshot out. + + The sampler is also the natural feed for the thermal governor and the persistence + layer, so neither needs to poll the GPU on its own. + """ + + def __init__(self, interval_s: float = 1.0) -> None: + self.interval_s = interval_s + self.snapshot: Dict[str, Any] = {} + self.subscribers: Set[asyncio.Queue] = set() + self.task: Optional[asyncio.Task] = None + self.running = False + self.samples = 0 + self.last_sample_ms = 0.0 + + async def start(self) -> None: + if self.running: + return + self.running = True + self.task = asyncio.create_task(self._loop()) + + async def stop(self) -> None: + self.running = False + if self.task: + self.task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self.task + + def subscribe(self) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=2) + self.subscribers.add(q) + return q + + def unsubscribe(self, q: asyncio.Queue) -> None: + self.subscribers.discard(q) + + async def _loop(self) -> None: + while self.running: + t0 = time.perf_counter() + try: + snap = await self._sample() + self.snapshot = snap + self.samples += 1 + self.last_sample_ms = round((time.perf_counter() - t0) * 1000, 2) + + # Feed the governor and the durable store from the sample we already have. + thermal_governor.governor.observe(snap.get("gpu", {}), + overclock_manager.ACTIVE_PROFILE) + telemetry_store.record_telemetry( + snap.get("gpu", {}), snap.get("ram", {}), + profile=overclock_manager.ACTIVE_PROFILE, + throttle_reasons=",".join(snap.get("gpu", {}).get("throttle_reasons") or []), + ) + + for q in list(self.subscribers): + if q.full(): + # Slow client: drop the stale frame rather than stalling the sampler. + with contextlib.suppress(asyncio.QueueEmpty): + q.get_nowait() + with contextlib.suppress(asyncio.QueueFull): + q.put_nowait(snap) + except asyncio.CancelledError: + raise + except Exception as e: + logger.error(f"telemetry sampler error: {e}") + await asyncio.sleep(max(self.interval_s - (time.perf_counter() - t0), 0.05)) + + async def _sample(self) -> Dict[str, Any]: + gpu_stats = vram_arbitrator.get_gpu_hardware_stats() + mem_stats = ram_optimizer.get_detailed_meminfo() + ollama_state, comfy_state = await asyncio.gather( + vram_arbitrator.get_ollama_live_state(), + vram_arbitrator.get_comfyui_live_state(), + ) + return { + "timestamp": time.time(), # wall clock, not the event loop's monotonic clock + "monotonic": asyncio.get_running_loop().time(), + "gpu": gpu_stats, + "ram": mem_stats, + "ollama": ollama_state, + "comfyui": comfy_state, + "arbitrator": vram_arbitrator.arbitrator.get_status(), + "governor": thermal_governor.governor.get_status(), + "overclock": {"active_profile": overclock_manager.ACTIVE_PROFILE}, + "history": vram_arbitrator.get_switch_history(), + "comfy_models_count": len(ram_optimizer.find_comfy_model_files()), + "sampler": {"samples": self.samples, "last_sample_ms": self.last_sample_ms, + "subscribers": len(self.subscribers)}, + } + + async def get(self) -> Dict[str, Any]: + """Latest snapshot, sampling on demand if the loop has not produced one yet.""" + if not self.snapshot: + self.snapshot = await self._sample() + return self.snapshot + + +broker = TelemetryBroker() + + +@asynccontextmanager +async def lifespan(app: FastAPI): + telemetry_store.start() + await broker.start() + await vram_arbitrator.arbitrator.start() + yield + await vram_arbitrator.arbitrator.stop() + await broker.stop() + # Never leave the card with locked clocks and pinned fans after we exit. + try: + overclock_manager.restore_safe("server shutdown") + except Exception as e: + logger.error(f"restore_safe on shutdown failed: {e}") + telemetry_store.stop() + + app = FastAPI( title="HyperSwap // GPU Program Swapper & Telemetry API", - version="1.0.0", + version="2.0.0", description="High-performance VRAM arbitration and 64GB RAM cache orchestrator for simultaneous Ollama and ComfyUI workloads on Linux.", docs_url="/docs", redoc_url="/redoc", + lifespan=lifespan, ) app.add_middleware( @@ -32,26 +174,24 @@ app.add_middleware( allow_headers=["*"], ) -@app.on_event("startup") -async def on_startup(): - await vram_arbitrator.arbitrator.start() - -@app.on_event("shutdown") -async def on_shutdown(): - await vram_arbitrator.arbitrator.stop() # Pydantic Request Models class SwitchRequest(BaseModel): model: str = Field(..., description="Name of the Ollama model to hot-swap to in VRAM", example="qwen3.8fast:latest") keep_alive: Optional[str] = Field("30m", description="Keep-alive duration in VRAM (e.g. 5m, 30m, 0)", example="30m") + free_comfy_first: bool = Field(False, description="Purge ComfyUI VRAM first if it is holding memory") class WarmRequest(BaseModel): - model_name: Optional[str] = Field(None, description="Ollama model name to warm into OS page cache", example="gemma4:26b") - filepath: Optional[str] = Field(None, description="Absolute file path of Safetensors/GGUF to warm into RAM", example="/home/drjones/ComfyUI/models/checkpoints/v1-5-pruned-emaonly-fp16.safetensors") + model_name: Optional[str] = Field(None, description="Ollama model name to warm", example="gemma4:26b") + filepath: Optional[str] = Field(None, description="Absolute file path of Safetensors/GGUF to warm into RAM") + blob_only: bool = Field(False, description="Warm the model's weights into page cache without loading VRAM") + +class WarmAllRequest(BaseModel): + budget_gb: Optional[float] = Field(None, description="Byte budget for warming; defaults to 70% of MemAvailable", example=24.0) class BenchmarkRequest(BaseModel): iterations: Optional[int] = Field(2, description="Number of back-and-forth switch iterations to measure", example=2) - models: Optional[List[str]] = Field(None, description="Optional pair of models to benchmark between", example=["qwen3.8fast:latest", "smtek/Qwen3.8-27B:Q2_K_XL"]) + models: Optional[List[str]] = Field(None, description="Optional pair of models to benchmark between") class OverclockApplyRequest(BaseModel): profile: str = Field(..., description="Profile name: ollama | comfy | balanced", example="ollama") @@ -64,6 +204,23 @@ class FanRequest(BaseModel): percent: Optional[int] = Field(None, description="Fan speed 30-100 when mode=manual", example=70) speed_pct: Optional[int] = Field(None, description="Alias for percent (30-100)", example=70) +class GovernorRequest(BaseModel): + enabled: Optional[bool] = Field(None, description="Enable or disable the thermal governor") + reset: bool = Field(False, description="Clear any active derate and reapply the full profile") + +class SweepRequest(BaseModel): + knob: str = Field("mem_offset_mhz", description="mem_offset_mhz | core_offset_mhz") + profile: str = Field("ollama", description="Profile to tune") + model: Optional[str] = Field(None, description="Model to benchmark with; defaults to the loaded one") + start: Optional[int] = Field(None, description="First offset value") + stop: Optional[int] = Field(None, description="Last offset value") + step: Optional[int] = Field(None, description="Offset increment") + repeats: int = Field(1, description="Benchmark runs per step") + apply_best: bool = Field(False, description="Write the winning value into the profile") + +class RequestVramRequest(BaseModel): + needed_gb: float = Field(0.0, description="How much free VRAM Ollama needs", example=12.0) + # ========================================== # REST API ENDPOINTS @@ -71,52 +228,44 @@ class FanRequest(BaseModel): @app.get("/api/stats", summary="Full System Snapshot", tags=["Telemetry"]) async def get_all_stats() -> Dict[str, Any]: - """Gather complete live snapshot of GPU hardware, host RAM, Ollama, ComfyUI, and switch history.""" - gpu_stats = vram_arbitrator.get_gpu_hardware_stats() - mem_stats = ram_optimizer.get_detailed_meminfo() - ollama_state = await vram_arbitrator.get_ollama_live_state() - comfy_state = await vram_arbitrator.get_comfyui_live_state() - history = vram_arbitrator.get_switch_history() - comfy_models = ram_optimizer.find_comfy_model_files() - arbitrator_status = vram_arbitrator.arbitrator.get_status() - - return { - "timestamp": asyncio.get_event_loop().time(), - "gpu": gpu_stats, - "ram": mem_stats, - "ollama": ollama_state, - "comfyui": comfy_state, - "arbitrator": arbitrator_status, - "history": history, - "comfy_models_count": len(comfy_models), - } + """Latest unified snapshot of GPU hardware, host RAM, Ollama, ComfyUI and swap history.""" + return await broker.get() @app.get("/api/gpu", summary="GPU Sensors and VRAM Breakdown", tags=["Telemetry"]) async def get_gpu_metrics() -> Dict[str, Any]: - """Retrieve detailed NVML sensors (utilization %, temp, power, fan, clocks, and per-process VRAM allocation).""" + """Detailed NVML sensors (utilization, temp, power, fan, clocks, throttle reasons, per-process VRAM).""" return vram_arbitrator.get_gpu_hardware_stats() @app.get("/api/memory", summary="Host RAM and Page Cache Breakdown", tags=["Telemetry"]) async def get_ram_metrics() -> Dict[str, Any]: - """Retrieve precise host 64GB DDR5 RAM breakdown, active cache size, and cache hit ratios.""" + """Precise host RAM breakdown, active cache size and cache ratios.""" return ram_optimizer.get_detailed_meminfo() @app.get("/api/stream", summary="Real-Time SSE Telemetry Stream", tags=["Telemetry"]) async def sse_telemetry_stream(request: Request): - """Server-Sent Events (SSE) streaming real-time statistics at 1Hz for dynamic dashboards.""" + """Server-Sent Events stream of the shared 1Hz snapshot.""" async def event_generator(): - while True: - if await request.is_disconnected(): - break - try: - stats = await get_all_stats() - yield f"data: {json.dumps(stats)}\n\n" - except Exception as e: - logger.error(f"SSE stream error: {e}") - yield f"data: {json.dumps({'error': str(e)})}\n\n" - await asyncio.sleep(1.0) - + q = broker.subscribe() + try: + snap = await broker.get() + yield f"data: {json.dumps(snap)}\n\n" + while True: + if await request.is_disconnected(): + break + try: + snap = await asyncio.wait_for(q.get(), timeout=15.0) + yield f"data: {json.dumps(snap)}\n\n" + except asyncio.TimeoutError: + yield ": keepalive\n\n" + except asyncio.CancelledError: + raise + except Exception as e: + logger.error(f"SSE stream error: {e}") + yield f"data: {json.dumps({'error': str(e)})}\n\n" + finally: + broker.unsubscribe(q) + return StreamingResponse( event_generator(), media_type="text/event-stream", @@ -129,80 +278,142 @@ async def sse_telemetry_stream(request: Request): @app.post("/api/switch-model", summary="Hot-Swap Ollama LLM in VRAM", tags=["Orchestration"]) async def api_switch_model(req: SwitchRequest): - """Hot-swap the active Ollama model in VRAM and measure exact load duration and token evaluation speed.""" + """Hot-swap the active Ollama model, measuring real load bandwidth and token throughput.""" + if req.free_comfy_first: + await vram_arbitrator.arbitrator.request_vram_for_ollama() res = await vram_arbitrator.switch_ollama_model(req.model, keep_alive=req.keep_alive or "30m") if not res.get("success"): raise HTTPException(status_code=500, detail=res.get("error")) return res @app.post("/api/free-vram", summary="Soft-Yield Ollama VRAM", tags=["Orchestration"]) -async def api_free_vram(): - """Instruct Ollama to instantly yield VRAM to 0MB in ~15ms while preserving model weights in the 64GB host RAM page cache.""" - return await vram_arbitrator.instant_free_ollama_vram() +async def api_free_vram(confirm: bool = Query(True, description="Wait for the driver to actually release the allocation")): + """Yield Ollama's VRAM and wait for the release to be confirmed by NVML.""" + return await vram_arbitrator.instant_free_ollama_vram(confirm=confirm) @app.post("/api/comfy-free", summary="Purge ComfyUI VRAM Cache", tags=["Orchestration"]) async def api_comfy_free(): """Purge loaded diffusion models and VRAM cache from the ComfyUI pipeline.""" return await vram_arbitrator.instant_free_comfyui_vram() -@app.post("/api/warm-all", summary="Pre-warm All Models into RAM Cache", tags=["Memory Optimization"]) -async def api_warm_all(): - """Pre-fault and read all installed Ollama GGUF models and ComfyUI Safetensors checkpoints into the Linux OS Page Cache.""" - return await ram_optimizer.warm_all_models() +@app.post("/api/request-vram", summary="Ask for VRAM on Ollama's behalf", tags=["Orchestration"]) +async def api_request_vram(req: RequestVramRequest): + """Force an immediate ComfyUI purge if there is not enough free VRAM for Ollama.""" + return await vram_arbitrator.arbitrator.request_vram_for_ollama(req.needed_gb) + +@app.post("/api/warm-all", summary="Pre-warm Models into RAM Cache", tags=["Memory Optimization"]) +async def api_warm_all(req: Optional[WarmAllRequest] = None): + """Warm the highest-value models into the page cache within a byte budget.""" + return await ram_optimizer.warm_all_models(budget_gb=req.budget_gb if req else None) + +@app.get("/api/warm-plan", summary="Preview the Warm Plan", tags=["Memory Optimization"]) +async def api_warm_plan(budget_gb: Optional[float] = Query(None, description="Override the byte budget")): + """Show what warming would read, in what order, and what it would skip โ€” without doing it.""" + return ram_optimizer.build_warm_plan(budget_gb) @app.post("/api/warm-model", summary="Pre-warm Single Model or File", tags=["Memory Optimization"]) async def api_warm_model(req: WarmRequest): - """Pre-warm a specific Ollama model or individual file path into Linux RAM cache.""" + """Pre-warm a specific Ollama model or file path into the Linux page cache.""" + if req.model_name and req.blob_only: + return ram_optimizer.warm_ollama_blob(req.model_name) if req.model_name: return await ram_optimizer.warm_ollama_model(req.model_name, keep_alive="1m") - elif req.filepath: + if req.filepath: return ram_optimizer.warm_file_to_ram(req.filepath) - else: - raise HTTPException(status_code=400, detail="model_name or filepath required") + raise HTTPException(status_code=400, detail="model_name or filepath required") + +@app.get("/api/cache/report", summary="Measured Page-Cache Residency", tags=["Memory Optimization"]) +async def api_cache_report(files: bool = Query(True), refresh: bool = Query(False)): + """Measured (not assumed) page-cache residency for every model on disk.""" + report = ram_optimizer.get_cache_report(include_files=files, force_refresh=refresh) + report["capability"] = ram_optimizer.residency_capability() + return report @app.get("/api/models", summary="List All Installed Models", tags=["Catalog"]) -async def api_get_models(): - """List all installed Ollama models and discovered ComfyUI model checkpoints/safetensors on disk with sizes and quantization levels.""" +async def api_get_models(refresh: bool = Query(False)): + """All installed Ollama models (with their on-disk blobs) and ComfyUI checkpoints.""" + catalog = ram_optimizer.get_model_catalog(force_refresh=refresh) ollama_state = await vram_arbitrator.get_ollama_live_state() - comfy_models = ram_optimizer.find_comfy_model_files() return { "ollama_models": ollama_state.get("installed_models", []), - "comfy_models": comfy_models, + "ollama_blobs": catalog["ollama"], + "comfy_models": catalog["comfy"], + "cached_at": catalog["cached_at"], } @app.get("/api/history", summary="Model Switch History Log", tags=["Analytics"]) -async def api_get_history(limit: int = Query(20, description="Max history items to return")): - """Get the recent history log of model switch events, swap durations (in ms), and RAM cache hit status.""" - history = vram_arbitrator.get_switch_history() - return history[:limit] +async def api_get_history(limit: int = Query(20, description="Max history items to return"), + durable: bool = Query(False, description="Read from the persistent store instead of the in-memory ring")): + """Recent swap events, durations, achieved bandwidth and cache status.""" + if durable: + return telemetry_store.recent_events(limit) + return vram_arbitrator.get_switch_history()[:limit] @app.post("/api/benchmark", summary="Run Latency Benchmark", tags=["Analytics"]) async def api_run_benchmark(req: BenchmarkRequest): - """Run an automated benchmark swapping between available models to measure round-trip latency and RAM cache effectiveness.""" + """Automated round-trip switch benchmark measuring latency and cache effectiveness.""" from mcp_server import run_model_switch_benchmark res_str = await run_model_switch_benchmark(iterations=req.iterations or 2) return json.loads(res_str) + +# ========================================== +# ANALYTICS (persisted) +# ========================================== + +@app.get("/api/analytics/profiles", summary="Which Overclock Profile Is Actually Faster", tags=["Analytics"]) +async def api_analytics_profiles(days: float = Query(7.0)): + """Decode throughput and thermals grouped by the profile that was active at the time.""" + return {"window_days": days, "profiles": telemetry_store.profile_comparison(days)} + +@app.get("/api/analytics/swaps", summary="Swap Statistics", tags=["Analytics"]) +async def api_analytics_swaps(days: float = Query(7.0)): + """Aggregated swap/yield/purge latencies, cache-hit split and per-model throughput.""" + return telemetry_store.swap_stats(days) + +@app.get("/api/analytics/timeseries", summary="Downsampled Telemetry History", tags=["Analytics"]) +async def api_analytics_timeseries(hours: float = Query(6.0), buckets: int = Query(240)): + """Long-range history for charts that outlive a page refresh.""" + return {"hours": hours, "points": telemetry_store.timeseries(hours, buckets)} + +@app.get("/api/analytics/models", summary="Model Usage Ranking", tags=["Analytics"]) +async def api_analytics_models(days: float = Query(30.0)): + """Recency/frequency ranking used to prioritise the RAM warm budget.""" + return {"window_days": days, "models": telemetry_store.model_usage_ranking(days)} + +@app.get("/api/db", summary="Telemetry Store Info", tags=["Analytics"]) +async def api_db_info(): + """Where the persistent store lives and how much history it holds.""" + return telemetry_store.db_info() + + # ========================================== # OVERCLOCK MANAGEMENT # ========================================== @app.get("/api/overclock", summary="Overclock Status & Profiles", tags=["Overclock"]) async def api_overclock_status(): - """Get live GPU overclock state, active profile, and all per-app profiles.""" - return overclock_manager.get_status() + """Live GPU overclock state, active profile, governor state and all per-app profiles.""" + status = overclock_manager.get_status() + status["governor"] = thermal_governor.governor.get_status() + return status @app.post("/api/overclock/apply", summary="Apply Overclock Profile", tags=["Overclock"]) async def api_overclock_apply(req: OverclockApplyRequest): - """Apply a named overclock profile (ollama | comfy | balanced) to the GPU immediately.""" + """Apply a named overclock profile (ollama | comfy | balanced) immediately.""" res = overclock_manager.apply_profile(req.profile) if not res.get("success"): raise HTTPException(status_code=400, detail=res.get("error")) return res +@app.post("/api/overclock/restore", summary="Restore Stock GPU State", tags=["Overclock"]) +async def api_overclock_restore(): + """Drop all clock locks and offsets, restore default power limit and automatic fans.""" + return overclock_manager.restore_safe("manual request") + @app.get("/api/overclock/profiles", summary="List Overclock Profiles", tags=["Overclock"]) async def api_overclock_profiles(): - """List all overclock profiles with their current settings.""" + """All overclock profiles with their current settings.""" return overclock_manager.get_profiles() @app.post("/api/overclock/profiles/{name}", summary="Update Overclock Profile", tags=["Overclock"]) @@ -216,7 +427,7 @@ async def api_overclock_update_profile(name: str, req: OverclockProfileUpdate): @app.get("/api/overclock/fan", summary="Get GPU Fan Status", tags=["Overclock"]) @app.get("/api/gpu/fan", summary="Get GPU Fan Status", tags=["Overclock"]) async def api_get_fan_status(): - """Get current GPU fan control mode and speed.""" + """Current GPU fan control mode and speed.""" return overclock_manager.get_fan_status() @app.post("/api/overclock/fan", summary="Set GPU Fan Speed", tags=["Overclock"]) @@ -228,12 +439,59 @@ async def api_set_fan(req: FanRequest): return overclock_manager.set_fan_speed(pct) return overclock_manager.set_fan_auto() + +# ========================================== +# THERMAL GOVERNOR +# ========================================== + +@app.get("/api/governor", summary="Thermal Governor State", tags=["Governor"]) +async def api_governor_status(): + """Current derate level, why it was applied, and the escalation history.""" + return thermal_governor.governor.get_status() + +@app.post("/api/governor", summary="Control the Thermal Governor", tags=["Governor"]) +async def api_governor_control(req: GovernorRequest): + """Enable/disable the governor, or clear an active derate.""" + if req.enabled is not None: + thermal_governor.governor.set_enabled(req.enabled) + if req.reset: + thermal_governor.governor.reset() + return thermal_governor.governor.get_status() + + +# ========================================== +# AUTOTUNE +# ========================================== + +@app.get("/api/autotune", summary="Autotune Status & History", tags=["Autotune"]) +async def api_autotune_status(): + """Sweep progress, the last result, and every recorded autotune step.""" + return autotune.get_status() + +@app.post("/api/autotune/sweep", summary="Run an Overclock Sweep", tags=["Autotune"]) +async def api_autotune_sweep(req: SweepRequest): + """Walk a clock offset upward, measuring tok/s and watching for instability at each step.""" + res = await autotune.sweep( + knob=req.knob, profile=req.profile, model=req.model, + start=req.start, stop=req.stop, step=req.step, + repeats=req.repeats, apply_best=req.apply_best, + ) + if not res.get("success"): + raise HTTPException(status_code=400, detail=res.get("error")) + return res + +@app.post("/api/autotune/cancel", summary="Cancel a Running Sweep", tags=["Autotune"]) +async def api_autotune_cancel(): + """Stop the current sweep after the step in flight; the profile is restored either way.""" + return autotune.cancel() + + # Mount static web UI files -app.mount("/static", StaticFiles(directory="/home/drjones/unified-model-manager/static"), name="static") +app.mount("/static", StaticFiles(directory=f"{BASE_DIR}/static"), name="static") @app.get("/", summary="Dashboard Web UI", tags=["UI"]) async def root_index(): - with open("/home/drjones/unified-model-manager/static/index.html", "r") as f: + with open(f"{BASE_DIR}/static/index.html", "r") as f: content = f.read() return HTMLResponse(content=content) diff --git a/static/app.js b/static/app.js index 686932a..b388f61 100644 --- a/static/app.js +++ b/static/app.js @@ -33,6 +33,9 @@ function initSSE() { function updateDashboard(data) { if (!data) return; + // Governor state rides along in the shared snapshot โ€” no extra polling needed. + if (data.governor) renderGovernor(data.governor); + // 1. GPU VRAM Stats const gpu = data.gpu || {}; const ram = data.ram || {}; @@ -646,3 +649,231 @@ async function saveOverclockProfile() { alert(`Error: ${err}`); } } + +// ============================================================================ +// THERMAL GOVERNOR / RESIDENCY / ANALYTICS / AUTOTUNE +// ============================================================================ + +let governorEnabled = true; + +function renderGovernor(gov) { + if (!gov) return; + governorEnabled = gov.enabled; + const levels = 3; + const el = (id) => document.getElementById(id); + if (!el('gov-label')) return; + + el('gov-label').textContent = gov.label || 'โ€”'; + el('gov-label').className = 'text-2xl font-bold ' + + (gov.level === 0 ? 'text-emerald-400' : gov.level < 3 ? 'text-amber-400' : 'text-rose-400'); + el('gov-level').textContent = `level ${gov.level} / ${levels}`; + el('gov-bar').style.width = `${(gov.level / levels) * 100}%`; + el('gov-esc').textContent = `${gov.escalate_at_c}ยฐC`; + el('gov-rec').textContent = `${gov.recover_below_c}ยฐC`; + el('gov-scale').textContent = `${Math.round((gov.offset_scale ?? 1) * 100)}%`; + el('gov-reason').textContent = gov.last_reason || 'โ€”'; + + const toggle = el('gov-toggle'); + toggle.textContent = gov.enabled ? 'Enabled' : 'Disabled'; + toggle.className = 'px-2.5 py-1 text-xs font-semibold rounded-lg border transition ' + + (gov.enabled ? 'bg-emerald-950/70 border-emerald-800 text-emerald-300 hover:bg-emerald-900' + : 'bg-slate-800 border-slate-700 text-slate-400 hover:bg-slate-700'); + + const hist = el('gov-history'); + hist.innerHTML = (gov.history || []).map(h => { + const t = new Date(h.ts * 1000).toLocaleTimeString(); + const up = h.to_level > h.from_level; + return `
+ ${t} ${up ? 'โ–ฒ' : 'โ–ผ'} L${h.to_level} ${h.label} + ${h.reason || ''}
`; + }).join('') || '
No derates recorded โ€” the card has stayed happy.
'; +} + +async function toggleGovernor() { + const res = await fetch('/api/governor', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ enabled: !governorEnabled }) + }); + renderGovernor(await res.json()); +} + +async function resetGovernor() { + const res = await fetch('/api/governor', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ reset: true }) + }); + renderGovernor(await res.json()); +} + +// ---------------------------------------------------------------- residency + +async function fetchCacheReport(refresh = false) { + const body = document.getElementById('cache-files'); + if (!body) return; + if (refresh) body.innerHTML = '
Scanningโ€ฆ
'; + try { + const res = await fetch(`/api/cache/report?refresh=${refresh ? 'true' : 'false'}`); + const d = await res.json(); + + document.getElementById('cache-total').textContent = + `${d.resident_total_gb} / ${d.catalog_total_gb} GB`; + document.getElementById('cache-meta').textContent = + `${d.residency_pct}% resident ยท ${d.warm_files}/${d.files_scanned} warm ยท ${d.scan_ms}ms`; + + const cap = d.capability || {}; + document.getElementById('cache-capability').innerHTML = cap.exact_everywhere + ? 'Exact measurement available for every model file.' + : `Ollama weights measured by read-rate probe โ€” ${cap.reason || ''}`; + + body.innerHTML = (d.files || []).slice(0, 40).map(f => { + const pct = f.resident_pct ?? 0; + const color = pct >= 90 ? 'bg-emerald-500' : pct >= 25 ? 'bg-amber-500' : 'bg-slate-700'; + const kind = f.kind === 'ollama' ? 'text-cyan-400' : 'text-fuchsia-400'; + return `
+
+ ${f.name} + ${pct}% ยท ${f.size_gb}GB +
+
+
+
+
`; + }).join(''); + } catch (e) { + body.innerHTML = `
Residency scan failed: ${e}
`; + } +} + +async function fetchWarmPlan() { + const body = document.getElementById('cache-files'); + body.innerHTML = '
Building planโ€ฆ
'; + try { + const d = await (await fetch('/api/warm-plan')).json(); + document.getElementById('cache-total').textContent = `${d.planned_gb} GB planned`; + document.getElementById('cache-meta').textContent = + `budget ${d.budget_gb} GB ยท ${d.warm_count} to warm ยท ${d.skipped_count} skipped`; + document.getElementById('cache-capability').innerHTML = + 'Ranked by recency/frequency; already-resident files cost nothing.'; + body.innerHTML = d.plan.map((p, i) => `
+ ${i + 1}. ${p.name} + ${p.size_gb}GB ยท ${p.resident_pct}% ยท s=${p.score} +
`).join('') || '
Everything worth warming is already resident.
'; + } catch (e) { + body.innerHTML = `
${e}
`; + } +} + +// ---------------------------------------------------------------- analytics + +async function fetchAnalytics() { + const el = document.getElementById('analytics-body'); + if (!el) return; + const days = document.getElementById('analytics-days').value; + try { + const d = await (await fetch(`/api/analytics/profiles?days=${days}`)).json(); + if (!d.profiles || !d.profiles.length) { + el.innerHTML = '
No swaps recorded yet in this window. Switch a model to start building the comparison.
'; + return; + } + const best = Math.max(...d.profiles.map(p => p.avg_tok_s || 0)); + el.innerHTML = ` + + + ` + + d.profiles.map(p => { + const win = (p.avg_tok_s || 0) === best; + return ` + + + + `; + }).join('') + '
profiletok/sload GB/sยฐC avgW avgSM MHzn
${win ? 'โ˜… ' : ''}${p.profile ?? 'โ€”'}${p.avg_tok_s ?? 'โ€”'}${p.avg_load_gbps ?? 'โ€”'}${p.avg_temp_c ?? 'โ€”'}${p.avg_power_w ?? 'โ€”'}${p.avg_clock_sm ?? 'โ€”'}${p.swaps}
'; + } catch (e) { + el.innerHTML = `
${e}
`; + } +} + +// ---------------------------------------------------------------- autotune + +let sweepPoll = null; + +async function runSweep() { + const btn = document.getElementById('at-run'); + const body = document.getElementById('autotune-body'); + const payload = { + knob: document.getElementById('at-knob').value, + profile: document.getElementById('at-profile').value, + start: parseInt(document.getElementById('at-start').value, 10), + stop: parseInt(document.getElementById('at-stop').value, 10), + step: parseInt(document.getElementById('at-step').value, 10), + apply_best: document.getElementById('at-apply').checked, + }; + btn.disabled = true; + btn.classList.add('opacity-50'); + body.innerHTML = '
Sweepingโ€ฆ each step loads the model and benchmarks decode. This takes a few minutes.
'; + sweepPoll = setInterval(pollSweep, 2000); + try { + const res = await fetch('/api/autotune/sweep', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(payload) + }); + const d = await res.json(); + renderSweep(res.ok ? d : { error: d.detail }); + } catch (e) { + body.innerHTML = `
${e}
`; + } finally { + clearInterval(sweepPoll); + btn.disabled = false; + btn.classList.remove('opacity-50'); + document.getElementById('autotune-state').textContent = 'idle'; + } +} + +async function pollSweep() { + try { + const d = await (await fetch('/api/autotune')).json(); + const s = document.getElementById('autotune-state'); + if (d.running && d.current) { + s.textContent = `${d.current.knob}=${d.current.value} โ†’ ${d.current.tokens_per_sec} tok/s`; + s.className = 'text-xs font-mono text-fuchsia-400'; + } + } catch (e) { /* transient */ } +} + +async function cancelSweep() { + await fetch('/api/autotune/cancel', { method: 'POST' }); + document.getElementById('autotune-state').textContent = 'cancellingโ€ฆ'; +} + +function renderSweep(d) { + const body = document.getElementById('autotune-body'); + if (d.error) { body.innerHTML = `
${d.error}
`; return; } + const rows = (d.table || []).map(r => ` + ${r.value}${r.tokens_per_sec}${r.temp_c} + ${r.power_w ?? 'โ€”'}${r.clock_mem_mhz ?? 'โ€”'} + ${r.instability || 'stable'}`).join(''); + const best = d.best + ? `
โ˜… Best stable: ${d.knob}=${d.best.value} + โ†’ ${d.best.tokens_per_sec} tok/s at ${d.best.temp_c}ยฐC + ${d.gain_pct != null ? `(${d.gain_pct > 0 ? '+' : ''}${d.gain_pct}% vs baseline)` : ''} + ${d.applied_to_profile ? 'ยท saved to profile' : ''}
` + : '
No stable step produced throughput.
'; + const unstable = d.first_unstable + ? `
Stopped climbing at ${d.first_unstable.value}: ${d.first_unstable.why}
` : ''; + body.innerHTML = best + unstable + ` + + + ${rows}
offsettok/sยฐCWmem MHzstatus
`; +} + +// ---------------------------------------------------------------- wiring + +document.addEventListener('DOMContentLoaded', () => { + fetchCacheReport(false); + fetchAnalytics(); + setInterval(fetchAnalytics, 60000); + setInterval(() => fetchCacheReport(false), 30000); + fetch('/api/autotune').then(r => r.json()).then(d => { + if (d.last_result) renderSweep(d.last_result); + }).catch(() => {}); +}); diff --git a/static/index.html b/static/index.html index 7d6e7ae..3d9fb4b 100644 --- a/static/index.html +++ b/static/index.html @@ -665,6 +665,144 @@ + +
+ + +
+
+
+
+ +
+
+

Thermal Governor

+

Walks the overclock back when the card complains

+
+
+
+ + +
+
+
+
+ full + level 0 / 3 +
+
+
+
+
+
+
Escalate
+
83ยฐC
+
+
+
Recover
+
72ยฐC
+
+
+
Offset Scale
+
100%
+
+
+

Last action: cold start

+
+
+
+ + +
+
+
+
+ +
+
+

Measured Page-Cache Residency

+

What is genuinely in RAM, not what we hope is

+
+
+
+ + +
+
+
+
+ โ€” GB + โ€” +
+
+
+
+
+ + +
+
+
+
+ +
+
+

Is the Overclock Actually Working?

+

Decode throughput per profile, from persisted history

+
+
+ +
+
Collecting dataโ€ฆ
+
+ + +
+
+
+
+ +
+
+

Overclock Autotune

+

Sweep a clock offset, measure tok/s, stop at instability

+
+
+ idle +
+
+ + + + + + +
+
+ + + Restores the profile when done, even on error. +
+
+
+ +
+ diff --git a/telemetry_store.py b/telemetry_store.py new file mode 100644 index 0000000..e846250 --- /dev/null +++ b/telemetry_store.py @@ -0,0 +1,400 @@ +"""SQLite time-series persistence for HyperSwap telemetry, swap events and autotune runs. + +Everything the arbitrator learns used to die with the process (SWITCH_HISTORY was an +in-memory deque of 50). This module keeps it on disk so we can answer the question the +whole app exists to answer: does a given overclock profile actually deliver more tok/s? + +Design notes: + * WAL mode + a single writer thread -> the 1Hz sampler never blocks the event loop. + * Telemetry rows are batched and flushed every FLUSH_INTERVAL_S. + * Retention pruning runs opportunistically, keeping the DB bounded and small. +""" +import logging +import os +import queue +import sqlite3 +import threading +import time +from typing import Any, Dict, List, Optional + +logger = logging.getLogger("telemetry_store") + +_BASE = os.path.dirname(os.path.abspath(__file__)) +DB_PATH = os.environ.get("HYPERSWAP_DB", os.path.join(_BASE, "hyperswap.db")) + +TELEMETRY_RETENTION_DAYS = float(os.environ.get("HYPERSWAP_TELEMETRY_RETENTION_DAYS", "14")) +EVENT_RETENTION_DAYS = float(os.environ.get("HYPERSWAP_EVENT_RETENTION_DAYS", "180")) +FLUSH_INTERVAL_S = 2.0 +PRUNE_INTERVAL_S = 3600.0 + +SCHEMA = """ +PRAGMA journal_mode=WAL; +PRAGMA synchronous=NORMAL; + +CREATE TABLE IF NOT EXISTS telemetry ( + ts REAL NOT NULL, + profile TEXT, + gpu_util_pct REAL, + mem_util_pct REAL, + temp_c REAL, + power_w REAL, + power_limit_w REAL, + fan_pct REAL, + clock_sm_mhz REAL, + clock_mem_mhz REAL, + vram_used_bytes INTEGER, + ollama_bytes INTEGER, + comfy_bytes INTEGER, + system_bytes INTEGER, + ram_used_bytes INTEGER, + ram_cached_bytes INTEGER, + pcie_tx_kbps INTEGER, + pcie_rx_kbps INTEGER, + throttle_reasons TEXT +); +CREATE INDEX IF NOT EXISTS idx_telemetry_ts ON telemetry(ts); + +CREATE TABLE IF NOT EXISTS events ( + ts REAL NOT NULL, + event_type TEXT, + source TEXT, + target TEXT, + profile TEXT, + duration_ms REAL, + load_duration_ms REAL, + yield_confirm_ms REAL, + tokens_per_sec REAL, + bytes_loaded INTEGER, + load_gbps REAL, + cache_status TEXT, + detail TEXT +); +CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts); +CREATE INDEX IF NOT EXISTS idx_events_type ON events(event_type); + +CREATE TABLE IF NOT EXISTS autotune_runs ( + ts REAL NOT NULL, + profile TEXT, + knob TEXT, + core_offset_mhz INTEGER, + mem_offset_mhz INTEGER, + tokens_per_sec REAL, + load_gbps REAL, + temp_c REAL, + power_w REAL, + stable INTEGER, + instability TEXT, + note TEXT +); +CREATE INDEX IF NOT EXISTS idx_autotune_ts ON autotune_runs(ts); +""" + + +class _Writer(threading.Thread): + """Single background writer: batches telemetry, commits events immediately.""" + + daemon = True + + def __init__(self) -> None: + super().__init__(name="telemetry-writer") + self.q: "queue.Queue[Optional[tuple]]" = queue.Queue(maxsize=10000) + self._stop = threading.Event() + self._last_prune = 0.0 + + def run(self) -> None: + conn = sqlite3.connect(DB_PATH) + conn.executescript(SCHEMA) + conn.commit() + pending: List[tuple] = [] + last_flush = time.time() + while not self._stop.is_set(): + try: + item = self.q.get(timeout=0.5) + except queue.Empty: + item = None + if item is not None: + kind, sql, params = item + if kind == "telemetry": + pending.append((sql, params)) + else: + try: + conn.execute(sql, params) + conn.commit() + except Exception as e: + logger.warning("event write failed: %s", e) + now = time.time() + if pending and (now - last_flush) >= FLUSH_INTERVAL_S: + try: + for sql, params in pending: + conn.execute(sql, params) + conn.commit() + except Exception as e: + logger.warning("telemetry flush failed: %s", e) + pending.clear() + last_flush = now + if now - self._last_prune > PRUNE_INTERVAL_S: + self._last_prune = now + try: + conn.execute("DELETE FROM telemetry WHERE ts < ?", + (now - TELEMETRY_RETENTION_DAYS * 86400,)) + conn.execute("DELETE FROM events WHERE ts < ?", + (now - EVENT_RETENTION_DAYS * 86400,)) + conn.commit() + except Exception as e: + logger.debug("prune failed: %s", e) + try: + for sql, params in pending: + conn.execute(sql, params) + conn.commit() + conn.close() + except Exception: + pass + + def stop(self) -> None: + self._stop.set() + + +_writer: Optional[_Writer] = None +_writer_lock = threading.Lock() + + +def start() -> None: + global _writer + with _writer_lock: + if _writer is None or not _writer.is_alive(): + _writer = _Writer() + _writer.start() + logger.info("telemetry store started at %s", DB_PATH) + + +def stop() -> None: + global _writer + with _writer_lock: + if _writer is not None: + _writer.stop() + _writer.join(timeout=3.0) + _writer = None + + +def _submit(kind: str, sql: str, params: tuple) -> None: + w = _writer + if w is None: + return + try: + w.q.put_nowait((kind, sql, params)) + except queue.Full: + logger.debug("telemetry queue full, dropping sample") + + +_TELEMETRY_SQL = """ +INSERT INTO telemetry (ts, profile, gpu_util_pct, mem_util_pct, temp_c, power_w, power_limit_w, + fan_pct, clock_sm_mhz, clock_mem_mhz, vram_used_bytes, ollama_bytes, comfy_bytes, system_bytes, + ram_used_bytes, ram_cached_bytes, pcie_tx_kbps, pcie_rx_kbps, throttle_reasons) +VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) +""" + + +def record_telemetry(gpu: Dict[str, Any], ram: Dict[str, Any], profile: Optional[str] = None, + throttle_reasons: Optional[str] = None) -> None: + """Persist a single 1Hz telemetry sample. Never raises.""" + try: + if not gpu.get("available"): + return + bd = gpu.get("breakdown", {}) + _submit("telemetry", _TELEMETRY_SQL, ( + time.time(), profile, + gpu.get("gpu_util_pct"), gpu.get("mem_util_pct"), gpu.get("temperature_c"), + gpu.get("power_w"), gpu.get("power_limit_w"), gpu.get("fan_pct"), + gpu.get("clock_graphics_mhz"), gpu.get("clock_mem_mhz"), + gpu.get("vram_used_bytes"), + int(bd.get("ollama_gb", 0) * (1024 ** 3)), + int(bd.get("comfyui_gb", 0) * (1024 ** 3)), + int(bd.get("system_gb", 0) * (1024 ** 3)), + ram.get("used_bytes"), ram.get("cached_bytes"), + gpu.get("pcie_tx_kbps"), gpu.get("pcie_rx_kbps"), + throttle_reasons, + )) + except Exception as e: + logger.debug("record_telemetry failed: %s", e) + + +_EVENT_SQL = """ +INSERT INTO events (ts, event_type, source, target, profile, duration_ms, load_duration_ms, + yield_confirm_ms, tokens_per_sec, bytes_loaded, load_gbps, cache_status, detail) +VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) +""" + + +def record_event(event: Dict[str, Any], profile: Optional[str] = None) -> None: + """Persist a swap/yield/purge event. Never raises.""" + try: + _submit("event", _EVENT_SQL, ( + event.get("ts", time.time()), + event.get("event_type"), event.get("source"), event.get("target"), + profile or event.get("profile"), + event.get("duration_ms"), event.get("load_duration_ms"), + event.get("yield_confirm_ms"), event.get("tokens_per_sec"), + event.get("bytes_loaded"), event.get("load_gbps"), + event.get("cache_status"), event.get("detail"), + )) + except Exception as e: + logger.debug("record_event failed: %s", e) + + +_AUTOTUNE_SQL = """ +INSERT INTO autotune_runs (ts, profile, knob, core_offset_mhz, mem_offset_mhz, tokens_per_sec, + load_gbps, temp_c, power_w, stable, instability, note) +VALUES (?,?,?,?,?,?,?,?,?,?,?,?) +""" + + +def record_autotune(row: Dict[str, Any]) -> None: + try: + _submit("event", _AUTOTUNE_SQL, ( + row.get("ts", time.time()), row.get("profile"), row.get("knob"), + row.get("core_offset_mhz"), row.get("mem_offset_mhz"), + row.get("tokens_per_sec"), row.get("load_gbps"), + row.get("temp_c"), row.get("power_w"), + 1 if row.get("stable") else 0, + row.get("instability"), row.get("note"), + )) + except Exception as e: + logger.debug("record_autotune failed: %s", e) + + +# ------------------------------------------------------------------ queries + +def _read_conn() -> sqlite3.Connection: + conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True, timeout=5.0) + conn.row_factory = sqlite3.Row + return conn + + +def _rows(sql: str, params: tuple = ()) -> List[Dict[str, Any]]: + if not os.path.exists(DB_PATH): + return [] + try: + with _read_conn() as conn: + return [dict(r) for r in conn.execute(sql, params).fetchall()] + except Exception as e: + logger.debug("query failed: %s", e) + return [] + + +def profile_comparison(days: float = 7.0) -> List[Dict[str, Any]]: + """The headline question: which overclock profile actually produces more tok/s? + + Joins decode throughput from switch events against thermals sampled while that + profile was active. + """ + since = time.time() - days * 86400 + perf = _rows(""" + SELECT profile, + COUNT(*) AS swaps, + AVG(tokens_per_sec) AS avg_tok_s, + MAX(tokens_per_sec) AS max_tok_s, + AVG(load_gbps) AS avg_load_gbps, + AVG(load_duration_ms) AS avg_load_ms + FROM events + WHERE ts > ? AND event_type = 'LLM Model Switch' AND tokens_per_sec > 0 + GROUP BY profile + """, (since,)) + thermals = {r["profile"]: r for r in _rows(""" + SELECT profile, + AVG(temp_c) AS avg_temp_c, + MAX(temp_c) AS max_temp_c, + AVG(power_w) AS avg_power_w, + AVG(clock_sm_mhz) AS avg_clock_sm, + AVG(clock_mem_mhz) AS avg_clock_mem, + COUNT(*) AS samples + FROM telemetry + WHERE ts > ? AND gpu_util_pct > 5 + GROUP BY profile + """, (since,))} + out = [] + for row in perf: + merged = dict(row) + merged.update(thermals.get(row["profile"], {})) + for k, v in list(merged.items()): + if isinstance(v, float): + merged[k] = round(v, 2) + out.append(merged) + out.sort(key=lambda r: r.get("avg_tok_s") or 0, reverse=True) + return out + + +def swap_stats(days: float = 7.0) -> Dict[str, Any]: + since = time.time() - days * 86400 + by_type = _rows(""" + SELECT event_type, COUNT(*) AS n, + AVG(duration_ms) AS avg_ms, MIN(duration_ms) AS min_ms, MAX(duration_ms) AS max_ms, + AVG(yield_confirm_ms) AS avg_confirm_ms + FROM events WHERE ts > ? GROUP BY event_type ORDER BY n DESC + """, (since,)) + cache = _rows(""" + SELECT cache_status, COUNT(*) AS n, AVG(load_gbps) AS avg_gbps + FROM events WHERE ts > ? AND cache_status IS NOT NULL GROUP BY cache_status + """, (since,)) + models = _rows(""" + SELECT target AS model, COUNT(*) AS loads, AVG(tokens_per_sec) AS avg_tok_s, + AVG(load_gbps) AS avg_gbps, MAX(ts) AS last_used + FROM events WHERE ts > ? AND event_type = 'LLM Model Switch' + GROUP BY target ORDER BY loads DESC LIMIT 25 + """, (since,)) + return {"by_type": by_type, "by_cache_status": cache, "by_model": models, "window_days": days} + + +def model_usage_ranking(days: float = 30.0) -> List[Dict[str, Any]]: + """Recency+frequency score per model, used to prioritise the RAM warm budget.""" + since = time.time() - days * 86400 + now = time.time() + rows = _rows(""" + SELECT target AS model, COUNT(*) AS loads, MAX(ts) AS last_used + FROM events WHERE ts > ? AND target IS NOT NULL AND event_type IN + ('LLM Model Switch','Model Warm') GROUP BY target + """, (since,)) + for r in rows: + age_h = max((now - (r["last_used"] or since)) / 3600.0, 0.01) + # frequency, decayed by recency (half-life ~24h) + r["score"] = round(r["loads"] * (0.5 ** (age_h / 24.0)) + 1.0 / age_h, 4) + r["age_hours"] = round(age_h, 2) + rows.sort(key=lambda r: r["score"], reverse=True) + return rows + + +def timeseries(hours: float = 6.0, buckets: int = 240) -> List[Dict[str, Any]]: + """Downsampled history for long-range dashboard charts.""" + since = time.time() - hours * 3600 + width = max((hours * 3600) / max(buckets, 1), 1.0) + return _rows(""" + SELECT CAST(ts / ? AS INTEGER) * ? AS bucket_ts, + AVG(gpu_util_pct) AS gpu_util_pct, AVG(temp_c) AS temp_c, + AVG(power_w) AS power_w, AVG(fan_pct) AS fan_pct, + AVG(vram_used_bytes) AS vram_used_bytes, + AVG(ollama_bytes) AS ollama_bytes, AVG(comfy_bytes) AS comfy_bytes, + AVG(ram_cached_bytes) AS ram_cached_bytes, + AVG(clock_sm_mhz) AS clock_sm_mhz, AVG(clock_mem_mhz) AS clock_mem_mhz + FROM telemetry WHERE ts > ? + GROUP BY bucket_ts ORDER BY bucket_ts + """, (width, width, since)) + + +def recent_events(limit: int = 50) -> List[Dict[str, Any]]: + return _rows("SELECT * FROM events ORDER BY ts DESC LIMIT ?", (limit,)) + + +def autotune_history(limit: int = 200) -> List[Dict[str, Any]]: + return _rows("SELECT * FROM autotune_runs ORDER BY ts DESC LIMIT ?", (limit,)) + + +def db_info() -> Dict[str, Any]: + info = {"path": DB_PATH, "exists": os.path.exists(DB_PATH)} + if info["exists"]: + info["size_mb"] = round(os.path.getsize(DB_PATH) / (1024 ** 2), 2) + for tbl in ("telemetry", "events", "autotune_runs"): + r = _rows(f"SELECT COUNT(*) AS n FROM {tbl}") + info[f"{tbl}_rows"] = r[0]["n"] if r else 0 + r = _rows("SELECT MIN(ts) AS a, MAX(ts) AS b FROM telemetry") + if r and r[0]["a"]: + info["coverage_hours"] = round((r[0]["b"] - r[0]["a"]) / 3600.0, 2) + return info diff --git a/thermal_governor.py b/thermal_governor.py new file mode 100644 index 0000000..1d8b1b6 --- /dev/null +++ b/thermal_governor.py @@ -0,0 +1,175 @@ +"""Thermal governor: walks the overclock back when the card says it is unhappy. + +Nothing in HyperSwap used to de-escalate. A profile pinned the fans to a manual PWM and +locked the clocks, and those settings stayed exactly where they were put no matter what +the card reported. This module watches the telemetry the sampler already collects (so it +costs no extra NVML calls) and derates the active profile when the GPU is hot or +throttling, then restores it once things settle. + +Hysteresis is deliberate: escalation needs HOT_SAMPLES consecutive bad samples and +recovery needs COOL_SAMPLES consecutive good ones, so a single spike during a diffusion +step does not cause profile thrash. +""" +import logging +import threading +import time +from typing import Any, Dict, List, Optional + +import overclock_manager + +logger = logging.getLogger("thermal_governor") + +# Derate ladder: each step scales the profile's clock offsets and raises the fan floor. +DERATE_LADDER = [ + {"level": 0, "offset_scale": 1.00, "fan_floor": 0, "label": "full"}, + {"level": 1, "offset_scale": 0.60, "fan_floor": 80, "label": "derated -40%"}, + {"level": 2, "offset_scale": 0.25, "fan_floor": 90, "label": "derated -75%"}, + {"level": 3, "offset_scale": 0.00, "fan_floor": 100, "label": "stock clocks, fans max"}, +] + +TEMP_ESCALATE_C = 83.0 +TEMP_RECOVER_C = 72.0 +HOT_SAMPLES = 5 # ~5 s at 1 Hz before we act +COOL_SAMPLES = 30 # ~30 s of calm before we give the clocks back +REAPPLY_COOLDOWN_S = 20.0 + +# Throttle reasons that mean the hardware is protecting itself, not just hitting a cap. +HARD_THROTTLES = {"hw_thermal_slowdown", "sw_thermal_slowdown", "hw_slowdown", + "hw_power_brake_slowdown"} + + +class ThermalGovernor: + def __init__(self) -> None: + self.enabled = True + self.level = 0 + self.hot_streak = 0 + self.cool_streak = 0 + self.last_change = 0.0 + self.last_reason = "cold start" + self.history: List[Dict[str, Any]] = [] + self._lock = threading.Lock() + self._base_profile: Optional[str] = None + + # ---------------------------------------------------------------- observation + + def observe(self, gpu: Dict[str, Any], active_profile: Optional[str] = None) -> None: + """Feed one telemetry sample. Cheap and non-blocking; actuation runs in a thread.""" + if not self.enabled or not gpu.get("available"): + return + temp = gpu.get("temperature_c") or 0 + reasons = set(gpu.get("throttle_reasons") or []) + hard = bool(reasons & HARD_THROTTLES) + + hot = temp >= TEMP_ESCALATE_C or hard + cool = temp <= TEMP_RECOVER_C and not hard + + with self._lock: + if hot: + self.hot_streak += 1 + self.cool_streak = 0 + elif cool: + self.cool_streak += 1 + self.hot_streak = 0 + else: + self.hot_streak = 0 + self.cool_streak = 0 + + now = time.time() + if now - self.last_change < REAPPLY_COOLDOWN_S: + return + + if self.hot_streak >= HOT_SAMPLES and self.level < len(DERATE_LADDER) - 1: + why = (f"{temp:.0f}ยฐC" if temp >= TEMP_ESCALATE_C else "") + \ + (f" throttling: {','.join(sorted(reasons & HARD_THROTTLES))}" if hard else "") + self._step(self.level + 1, why.strip(), active_profile, temp) + elif self.cool_streak >= COOL_SAMPLES and self.level > 0: + self._step(self.level - 1, f"stable at {temp:.0f}ยฐC", active_profile, temp) + + def _step(self, new_level: int, reason: str, active_profile: Optional[str], + temp: float) -> None: + old = self.level + self.level = new_level + self.hot_streak = self.cool_streak = 0 + self.last_change = time.time() + self.last_reason = reason + profile = active_profile or overclock_manager.ACTIVE_PROFILE + self._base_profile = profile + entry = { + "ts": time.time(), "from_level": old, "to_level": new_level, + "label": DERATE_LADDER[new_level]["label"], "reason": reason, + "temp_c": temp, "profile": profile, + } + self.history.insert(0, entry) + del self.history[50:] + direction = "escalating" if new_level > old else "recovering" + logger.warning(f"Thermal governor {direction} to level {new_level} " + f"({DERATE_LADDER[new_level]['label']}) โ€” {reason}") + threading.Thread(target=self._actuate, args=(profile,), daemon=True).start() + + def _actuate(self, profile: str) -> None: + try: + overclock_manager.apply_profile(profile, overrides=self.overrides_for(profile)) + except Exception as e: + logger.error(f"Governor failed to apply derate: {e}") + + # ---------------------------------------------------------------- overrides + + def overrides_for(self, profile: str) -> Dict[str, Any]: + """Translate the current derate level into profile overrides.""" + step = DERATE_LADDER[self.level] + if self.level == 0: + return {} + cfg = overclock_manager.load_profiles().get(profile, {}) + scale = step["offset_scale"] + overrides: Dict[str, Any] = { + "core_offset_mhz": int(cfg.get("core_offset_mhz", 0) * scale), + "mem_offset_mhz": int(cfg.get("mem_offset_mhz", 0) * scale), + } + if step["level"] >= 2: + # Stop pinning the core clock high when the card is already backing off. + overrides["lock_core_min"] = 0 + overrides["lock_core_max"] = 0 + floor = step["fan_floor"] + if floor: + overrides["fan_mode"] = "manual" + overrides["fan_speed_pct"] = max(int(cfg.get("fan_speed_pct", 0)), floor) + return overrides + + # ---------------------------------------------------------------- control + + def reset(self) -> Dict[str, Any]: + with self._lock: + self.level = 0 + self.hot_streak = self.cool_streak = 0 + self.last_change = time.time() + self.last_reason = "manual reset" + profile = self._base_profile or overclock_manager.ACTIVE_PROFILE + if profile and profile != "stock": + overclock_manager.apply_profile(profile) + return self.get_status() + + def set_enabled(self, enabled: bool) -> Dict[str, Any]: + self.enabled = bool(enabled) + if not enabled and self.level > 0: + self.reset() + return self.get_status() + + def get_status(self) -> Dict[str, Any]: + step = DERATE_LADDER[self.level] + return { + "enabled": self.enabled, + "level": self.level, + "label": step["label"], + "offset_scale": step["offset_scale"], + "fan_floor": step["fan_floor"], + "last_reason": self.last_reason, + "hot_streak": self.hot_streak, + "cool_streak": self.cool_streak, + "escalate_at_c": TEMP_ESCALATE_C, + "recover_below_c": TEMP_RECOVER_C, + "seconds_since_change": round(time.time() - self.last_change, 1) if self.last_change else None, + "history": self.history[:10], + } + + +governor = ThermalGovernor() diff --git a/vram_arbitrator.py b/vram_arbitrator.py index f1fcea2..6473fcb 100644 --- a/vram_arbitrator.py +++ b/vram_arbitrator.py @@ -11,6 +11,8 @@ import json import websockets import overclock_manager +import ram_optimizer +import telemetry_store try: import pynvml @@ -24,9 +26,116 @@ logger = logging.getLogger("vram_arbitrator") OLLAMA_API_BASE = "http://localhost:11434" COMFY_API_BASE = "http://127.0.0.1:8188" -# Circular buffer for transition events +# Circular buffer for transition events (the durable log lives in telemetry_store) SWITCH_HISTORY = deque(maxlen=50) +# Bandwidth thresholds used to classify how a model actually got into VRAM. +# PCIe 4.0 x16 tops out near 31.5 GB/s; this NVMe sustains well under 2 GB/s. +RAM_HIT_GBPS = 5.0 +PARTIAL_HIT_GBPS = 1.5 + +# How long Ollama's VRAM may take to actually drain before we stop waiting. +YIELD_CONFIRM_TIMEOUT_S = 3.0 +YIELD_CONFIRM_POLL_S = 0.02 +YIELD_RESIDUAL_BYTES = 256 * 1024 ** 2 # treat <256 MB as "released" + +# Connection-pooled clients. Re-creating an AsyncClient per call meant a fresh TCP +# handshake on every one of the watchdog's polls. +_clients: Dict[str, httpx.AsyncClient] = {} + + +def _client(base_url: str, timeout: float) -> httpx.AsyncClient: + key = f"{base_url}|{timeout}" + c = _clients.get(key) + if c is None or c.is_closed: + c = httpx.AsyncClient( + base_url=base_url, + timeout=timeout, + limits=httpx.Limits(max_keepalive_connections=4, max_connections=8), + ) + _clients[key] = c + return c + + +async def close_clients() -> None: + for c in list(_clients.values()): + try: + await c.aclose() + except Exception: + pass + _clients.clear() + + +# Bit flags from nvmlDeviceGetCurrentClocksThrottleReasons, decoded for the governor. +THROTTLE_REASONS = { + 0x0000000000000001: "gpu_idle", + 0x0000000000000002: "applications_clocks_setting", + 0x0000000000000004: "sw_power_cap", + 0x0000000000000008: "hw_slowdown", + 0x0000000000000010: "sync_boost", + 0x0000000000000020: "sw_thermal_slowdown", + 0x0000000000000040: "hw_thermal_slowdown", + 0x0000000000000080: "hw_power_brake_slowdown", + 0x0000000000000100: "display_clock_setting", +} + + +def decode_throttle_reasons(bits: int) -> List[str]: + return [name for mask, name in THROTTLE_REASONS.items() if bits & mask] + + +def get_process_vram_bytes() -> Dict[str, int]: + """Fast NVML-only VRAM attribution, used by the yield barrier's tight poll loop. + + Deliberately avoids psutil lookups: this runs every 20 ms while we wait for VRAM + to actually drain. + """ + out = {"ollama_bytes": 0, "comfyui_bytes": 0, "other_bytes": 0, "free_bytes": 0} + if not NVML_AVAILABLE: + return out + try: + handle = pynvml.nvmlDeviceGetHandleByIndex(0) + out["free_bytes"] = pynvml.nvmlDeviceGetMemoryInfo(handle).free + procs = list(pynvml.nvmlDeviceGetComputeRunningProcesses(handle)) + try: + procs += list(pynvml.nvmlDeviceGetGraphicsRunningProcesses(handle)) + except Exception: + pass + merged: Dict[int, int] = {} + for p in procs: + merged[p.pid] = max(merged.get(p.pid, 0), p.usedGpuMemory or 0) + for pid, used in merged.items(): + kind = _PID_KIND_CACHE.get(pid) + if kind is None: + kind = _classify_pid(pid) + _PID_KIND_CACHE[pid] = kind + if kind == "ollama": + out["ollama_bytes"] += used + elif kind == "comfy": + out["comfyui_bytes"] += used + else: + out["other_bytes"] += used + except Exception as e: + logger.debug(f"get_process_vram_bytes failed: {e}") + return out + + +_PID_KIND_CACHE: Dict[int, str] = {} + + +def _classify_pid(pid: int) -> str: + try: + proc = psutil.Process(pid) + pname = proc.name().lower() + cmdline = " ".join(proc.cmdline()).lower() + except Exception: + return "other" + if "ollama" in pname or "llama-server" in cmdline: + return "ollama" + if "comfy" in cmdline or "main.py" in cmdline: + return "comfy" + return "other" + def get_gpu_hardware_stats() -> Dict[str, Any]: """Retrieve comprehensive GPU hardware and process metrics via NVML.""" @@ -77,6 +186,49 @@ def get_gpu_hardware_stats() -> Dict[str, Any]: clock_graphics = 0 clock_mem = 0 + # PCIe throughput (KB/s) โ€” TX + RX. Key metric for the RAM-cache + # PCIe-speed swap thesis (assimilated from pmady/gpu-mcp-server). + pcie_tx_kbps = 0 + pcie_rx_kbps = 0 + try: + pcie_tx_kbps = pynvml.nvmlDeviceGetPcieThroughput(handle, pynvml.NVML_PCIE_UTIL_TX_BYTES) + except Exception: + pass + try: + pcie_rx_kbps = pynvml.nvmlDeviceGetPcieThroughput(handle, pynvml.NVML_PCIE_UTIL_RX_BYTES) + except Exception: + pass + + # Why the GPU is not running at full clocks โ€” the thermal governor reads this. + throttle_bits = 0 + throttle_reasons: List[str] = [] + try: + throttle_bits = pynvml.nvmlDeviceGetCurrentClocksThrottleReasons(handle) + throttle_reasons = decode_throttle_reasons(throttle_bits) + except Exception: + pass + + # Power management limit (watts) โ€” the OC ceiling. + power_limit_w = 0.0 + try: + power_limit_w = round(pynvml.nvmlDeviceGetPowerManagementLimit(handle) / 1000.0, 1) + except Exception: + pass + + # Driver + CUDA version (completeness). + driver_version = "" + cuda_version = "" + try: + dv = pynvml.nvmlSystemGetDriverVersion() + driver_version = dv.decode("utf-8") if isinstance(dv, bytes) else str(dv) + except Exception: + pass + try: + cv = pynvml.nvmlSystemGetCudaDriverVersion() + cuda_version = cv # int like 12030 == CUDA 12.3 + except Exception: + pass + # Discover processes on GPU proc_breakdown = { "ollama_bytes": 0, @@ -142,6 +294,13 @@ def get_gpu_hardware_stats() -> Dict[str, Any]: "mem_util_pct": util_rates.memory, "temperature_c": temp_c, "power_w": power_w, + "power_limit_w": power_limit_w, + "pcie_tx_kbps": pcie_tx_kbps, + "pcie_rx_kbps": pcie_rx_kbps, + "throttle_bits": throttle_bits, + "throttle_reasons": throttle_reasons, + "driver_version": driver_version, + "cuda_version": cuda_version, "fan_pct": fan_pct, "fans": fans, "num_fans": len(fans), @@ -174,25 +333,25 @@ async def get_ollama_live_state() -> Dict[str, Any]: "installed_models": [] } try: - async with httpx.AsyncClient(timeout=3.0) as client: - # Check running models (ps) - ps_resp = await client.get(f"{OLLAMA_API_BASE}/api/ps") - if ps_resp.status_code == 200: - state["online"] = True - models = ps_resp.json().get("models", []) - state["loaded_models"] = models - if models: - first = models[0] - state["active_model_name"] = first.get("name") - vram_bytes = first.get("size_vram", first.get("size", 0)) - state["active_model_vram_gb"] = round(vram_bytes / (1024**3), 2) - state["active_context"] = first.get("context_length", 0) - state["expires_at"] = first.get("expires_at") - - # Check all tags - tags_resp = await client.get(f"{OLLAMA_API_BASE}/api/tags") - if tags_resp.status_code == 200: - state["installed_models"] = tags_resp.json().get("models", []) + client = _client(OLLAMA_API_BASE, 3.0) + # Check running models (ps) + ps_resp = await client.get("/api/ps") + if ps_resp.status_code == 200: + state["online"] = True + models = ps_resp.json().get("models", []) + state["loaded_models"] = models + if models: + first = models[0] + state["active_model_name"] = first.get("name") + vram_bytes = first.get("size_vram", first.get("size", 0)) + state["active_model_vram_gb"] = round(vram_bytes / (1024**3), 2) + state["active_context"] = first.get("context_length", 0) + state["expires_at"] = first.get("expires_at") + + # Check all tags + tags_resp = await client.get("/api/tags") + if tags_resp.status_code == 200: + state["installed_models"] = tags_resp.json().get("models", []) except Exception as e: logger.debug(f"Ollama check error: {e}") @@ -211,156 +370,286 @@ async def get_comfyui_live_state() -> Dict[str, Any]: "vram_total_mb": 0, } try: - async with httpx.AsyncClient(timeout=3.0) as client: - # Check system stats - stats_resp = await client.get(f"{COMFY_API_BASE}/system_stats") - if stats_resp.status_code == 200: - state["online"] = True - data = stats_resp.json() - devices = data.get("devices", []) - if devices: - dev = devices[0] - state["vram_free_mb"] = round(dev.get("vram_free", 0) / (1024**2), 1) - state["vram_total_mb"] = round(dev.get("vram_total", 0) / (1024**2), 1) - - # Check queue - queue_resp = await client.get(f"{COMFY_API_BASE}/queue") - if queue_resp.status_code == 200: - qdata = queue_resp.json() - running = qdata.get("queue_running", []) - pending = qdata.get("queue_pending", []) - state["queue_running"] = len(running) - state["queue_remaining"] = len(pending) - state["executing"] = len(running) > 0 - if running: - state["current_prompt_id"] = running[0][1] if len(running[0]) > 1 else str(running[0]) + client = _client(COMFY_API_BASE, 3.0) + # Check system stats + stats_resp = await client.get("/system_stats") + if stats_resp.status_code == 200: + state["online"] = True + data = stats_resp.json() + devices = data.get("devices", []) + if devices: + dev = devices[0] + state["vram_free_mb"] = round(dev.get("vram_free", 0) / (1024**2), 1) + state["vram_total_mb"] = round(dev.get("vram_total", 0) / (1024**2), 1) + + # Check queue + queue_resp = await client.get("/queue") + if queue_resp.status_code == 200: + qdata = queue_resp.json() + running = qdata.get("queue_running", []) + pending = qdata.get("queue_pending", []) + state["queue_running"] = len(running) + state["queue_remaining"] = len(pending) + state["executing"] = len(running) > 0 + if running: + state["current_prompt_id"] = running[0][1] if len(running[0]) > 1 else str(running[0]) except Exception as e: logger.debug(f"ComfyUI check error: {e}") return state -async def instant_free_ollama_vram(model_name: Optional[str] = None) -> Dict[str, Any]: - """Tell Ollama to instantly yield VRAM without evicting from OS page cache.""" +async def _await_vram_release(baseline_bytes: int, + timeout_s: float = YIELD_CONFIRM_TIMEOUT_S) -> Dict[str, Any]: + """Block until Ollama's VRAM has actually drained, or we give up. + + Posting keep_alive:0 only *asks* Ollama to unload; the driver frees the allocation + some milliseconds later. Returning before that happens is how ComfyUI ends up + allocating into VRAM that is still occupied, which shows up as a CUDA OOM mid-graph. + """ + t0 = time.perf_counter() + last = baseline_bytes + while True: + snap = get_process_vram_bytes() + last = snap["ollama_bytes"] + if last <= YIELD_RESIDUAL_BYTES: + return { + "confirmed": True, + "confirm_ms": round((time.perf_counter() - t0) * 1000, 2), + "residual_bytes": last, + "free_bytes": snap["free_bytes"], + } + if (time.perf_counter() - t0) >= timeout_s: + return { + "confirmed": False, + "confirm_ms": round((time.perf_counter() - t0) * 1000, 2), + "residual_bytes": last, + "free_bytes": snap["free_bytes"], + "error": f"Ollama still holding {round(last / (1024**3), 2)} GB after {timeout_s}s", + } + await asyncio.sleep(YIELD_CONFIRM_POLL_S) + + +def _record(event: Dict[str, Any]) -> None: + """Push an event to both the in-memory ring and the durable store.""" + event.setdefault("ts", time.time()) + event.setdefault("timestamp", time.strftime("%H:%M:%S")) + SWITCH_HISTORY.appendleft(event) + telemetry_store.record_event(event, profile=overclock_manager.ACTIVE_PROFILE) + + +async def instant_free_ollama_vram(model_name: Optional[str] = None, + confirm: bool = True) -> Dict[str, Any]: + """Yield Ollama's VRAM and wait for the driver to actually release it. + + The returned duration_ms is now the real end-to-end release time, not just how long + the HTTP POST took. + """ t0 = time.perf_counter() if not model_name: ollama_state = await get_ollama_live_state() model_name = ollama_state.get("active_model_name") - + if not model_name: - return {"success": True, "message": "No active Ollama model in VRAM", "duration_ms": 0} - + return {"success": True, "message": "No active Ollama model in VRAM", + "duration_ms": 0, "confirmed": True} + + baseline = get_process_vram_bytes()["ollama_bytes"] try: - async with httpx.AsyncClient(timeout=5.0) as client: - resp = await client.post( - f"{OLLAMA_API_BASE}/api/generate", - json={"model": model_name, "keep_alive": 0}, - ) - duration_ms = round((time.perf_counter() - t0) * 1000, 2) - event = { - "timestamp": time.strftime("%H:%M:%S"), - "event_type": "Ollama VRAM Yield", - "source": model_name, - "target": "VRAM 0MB (Kept in RAM)", - "duration_ms": duration_ms, - "cache_status": "RAM-Cached", - } - SWITCH_HISTORY.appendleft(event) - return {"success": True, "model": model_name, "duration_ms": duration_ms} + client = _client(OLLAMA_API_BASE, 5.0) + await client.post("/api/generate", json={"model": model_name, "keep_alive": 0}) + request_ms = round((time.perf_counter() - t0) * 1000, 2) + + barrier = {"confirmed": None, "confirm_ms": 0.0, "residual_bytes": baseline} + if confirm: + barrier = await _await_vram_release(baseline) + + duration_ms = round((time.perf_counter() - t0) * 1000, 2) + freed_gb = round(max(baseline - barrier.get("residual_bytes", 0), 0) / (1024**3), 2) + + _record({ + "event_type": "Ollama VRAM Yield", + "source": model_name, + "target": "VRAM 0MB (Kept in RAM)", + "duration_ms": duration_ms, + "yield_confirm_ms": barrier.get("confirm_ms"), + "cache_status": "RAM-Cached" if barrier.get("confirmed") else "Yield Timeout", + "detail": barrier.get("error"), + }) + return { + "success": True, + "model": model_name, + "duration_ms": duration_ms, + "request_ms": request_ms, + "confirm_ms": barrier.get("confirm_ms"), + "confirmed": barrier.get("confirmed"), + "freed_gb": freed_gb, + "residual_gb": round(barrier.get("residual_bytes", 0) / (1024**3), 2), + "free_vram_gb": round(barrier.get("free_bytes", 0) / (1024**3), 2), + "error": barrier.get("error"), + } except Exception as e: - return {"success": False, "error": str(e), "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + return {"success": False, "error": str(e), + "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + async def instant_free_comfyui_vram() -> Dict[str, Any]: """Tell ComfyUI to purge loaded diffusion models from VRAM.""" t0 = time.perf_counter() try: - async with httpx.AsyncClient(timeout=5.0) as client: - resp = await client.post( - f"{COMFY_API_BASE}/free", - json={"unload_models": True, "free_memory": True}, - ) - duration_ms = round((time.perf_counter() - t0) * 1000, 2) - event = { - "timestamp": time.strftime("%H:%M:%S"), - "event_type": "ComfyUI VRAM Purge", - "source": "ComfyUI Pipeline", - "target": "VRAM Free", - "duration_ms": duration_ms, - "cache_status": "Cleaned", - } - SWITCH_HISTORY.appendleft(event) - return {"success": True, "duration_ms": duration_ms} + client = _client(COMFY_API_BASE, 5.0) + await client.post("/free", json={"unload_models": True, "free_memory": True}) + duration_ms = round((time.perf_counter() - t0) * 1000, 2) + snap = get_process_vram_bytes() + _record({ + "event_type": "ComfyUI VRAM Purge", + "source": "ComfyUI Pipeline", + "target": "VRAM Free", + "duration_ms": duration_ms, + "cache_status": "Cleaned", + }) + return {"success": True, "duration_ms": duration_ms, + "free_vram_gb": round(snap["free_bytes"] / (1024**3), 2)} except Exception as e: - return {"success": False, "error": str(e), "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + return {"success": False, "error": str(e), + "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + + +_MODEL_SIZE_CACHE: Dict[str, int] = {} + + +def _model_size_bytes(model_name: str) -> int: + """On-disk weight size for an Ollama model, used to turn load time into bandwidth.""" + if model_name in _MODEL_SIZE_CACHE: + return _MODEL_SIZE_CACHE[model_name] + try: + for f in ram_optimizer.find_ollama_model_files(): + _MODEL_SIZE_CACHE[f["model"]] = f["size_bytes"] + except Exception as e: + logger.debug(f"model size lookup failed: {e}") + return _MODEL_SIZE_CACHE.get(model_name, 0) + + +def classify_load(size_bytes: int, load_duration_ms: float) -> Dict[str, Any]: + """Classify how a model reached VRAM, from achieved bandwidth rather than a constant. + + The old rule was `load_duration_ms < 2500`, which called a 27B Q2_K read from NVMe a + cache hit and a small model read from RAM a cold load. Bandwidth separates them + cleanly: page cache feeds PCIe at many GB/s, this NVMe does not. + """ + if load_duration_ms <= 1.0: + return {"cache_status": "Already in VRAM", "load_gbps": None, "is_ram_hit": True} + if not size_bytes: + # No size on record โ€” fall back to the old heuristic, but say so. + return { + "cache_status": "RAM Cache Hit โšก" if load_duration_ms < 2500 else "Cold Disk Load ๐Ÿ’พ", + "load_gbps": None, + "is_ram_hit": load_duration_ms < 2500, + "detail": "size unknown, fell back to duration heuristic", + } + gbps = (size_bytes / (1024**3)) / (load_duration_ms / 1000.0) + if gbps >= RAM_HIT_GBPS: + status = "RAM Cache Hit โšก" + elif gbps >= PARTIAL_HIT_GBPS: + status = "Partial Cache ๐ŸŒค" + else: + status = "Cold Disk Load ๐Ÿ’พ" + return {"cache_status": status, "load_gbps": round(gbps, 2), "is_ram_hit": gbps >= RAM_HIT_GBPS} + async def switch_ollama_model(target_model: str, keep_alive: str = "30m") -> Dict[str, Any]: """High-speed hot-swap to target Ollama model, tracking swap metrics.""" t0 = time.perf_counter() cur_state = await get_ollama_live_state() prev_model = cur_state.get("active_model_name") or "None" - + try: - async with httpx.AsyncClient(timeout=180.0) as client: - resp = await client.post( - f"{OLLAMA_API_BASE}/api/generate", - json={"model": target_model, "prompt": "Ready check", "stream": False, "keep_alive": keep_alive}, - ) - total_duration = time.perf_counter() - t0 - total_duration_ms = round(total_duration * 1000, 2) - - if resp.status_code == 200: - data = resp.json() - load_dur_ms = round(data.get("load_duration", 0) / 1e6, 2) - eval_dur_ms = round(data.get("eval_duration", 0) / 1e6, 2) - eval_count = data.get("eval_count", 0) - tokens_per_sec = round((eval_count / (eval_dur_ms / 1000)) if eval_dur_ms > 0 else 0, 1) - - # Check if it was a RAM cache hit (load duration < 1500ms for large model indicates RAM hit) - is_ram_hit = load_dur_ms < 2500 - - event = { - "timestamp": time.strftime("%H:%M:%S"), - "event_type": "LLM Model Switch", - "source": prev_model, - "target": target_model, - "duration_ms": total_duration_ms, - "load_duration_ms": load_dur_ms, - "tokens_per_sec": tokens_per_sec, - "cache_status": "RAM Cache Hit โšก" if is_ram_hit else "Cold Disk Load ๐Ÿ’พ", - } - SWITCH_HISTORY.appendleft(event) - - return { - "success": True, - "prev_model": prev_model, - "target_model": target_model, - "total_duration_ms": total_duration_ms, - "load_duration_ms": load_dur_ms, - "tokens_per_sec": tokens_per_sec, - "is_ram_hit": is_ram_hit, - "response": data.get("response", ""), - } - else: - return {"success": False, "error": f"HTTP {resp.status_code}: {resp.text}", "duration_ms": total_duration_ms} + client = _client(OLLAMA_API_BASE, 180.0) + resp = await client.post( + "/api/generate", + json={"model": target_model, "prompt": "Ready check", "stream": False, + "keep_alive": keep_alive}, + ) + total_duration_ms = round((time.perf_counter() - t0) * 1000, 2) + + if resp.status_code == 200: + data = resp.json() + load_dur_ms = round(data.get("load_duration", 0) / 1e6, 2) + eval_dur_ms = round(data.get("eval_duration", 0) / 1e6, 2) + eval_count = data.get("eval_count", 0) + tokens_per_sec = round((eval_count / (eval_dur_ms / 1000)) if eval_dur_ms > 0 else 0, 1) + + size_bytes = _model_size_bytes(target_model) + cls = classify_load(size_bytes, load_dur_ms) + + _record({ + "event_type": "LLM Model Switch", + "source": prev_model, + "target": target_model, + "duration_ms": total_duration_ms, + "load_duration_ms": load_dur_ms, + "tokens_per_sec": tokens_per_sec, + "bytes_loaded": size_bytes, + "load_gbps": cls["load_gbps"], + "cache_status": cls["cache_status"], + "detail": cls.get("detail"), + }) + + return { + "success": True, + "prev_model": prev_model, + "target_model": target_model, + "total_duration_ms": total_duration_ms, + "load_duration_ms": load_dur_ms, + "tokens_per_sec": tokens_per_sec, + "model_size_gb": round(size_bytes / (1024**3), 2) if size_bytes else None, + "load_gbps": cls["load_gbps"], + "cache_status": cls["cache_status"], + "is_ram_hit": cls["is_ram_hit"], + "response": data.get("response", ""), + } + return {"success": False, "error": f"HTTP {resp.status_code}: {resp.text}", + "duration_ms": total_duration_ms} except Exception as e: - return {"success": False, "error": str(e), "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + return {"success": False, "error": str(e), + "duration_ms": round((time.perf_counter() - t0) * 1000, 2)} + def get_switch_history() -> List[Dict[str, Any]]: return list(SWITCH_HISTORY) class AutoArbitrator: - """Real-time bidirectional background arbitrator for seamless Ollama <-> ComfyUI hot-swapping.""" + """Real-time bidirectional background arbitrator for seamless Ollama <-> ComfyUI hot-swapping. + + Two behavioural changes worth knowing about: + + * ComfyUI's VRAM is no longer purged 1.5 s after every finished prompt. Iterating on + a workflow is the common case, and purging between runs forced a full checkpoint + reload each time. The purge now waits for COMFY_IDLE_PURGE_S of genuinely empty + queue, and happens immediately only when Ollama actually needs the VRAM. + * The watchdog no longer polls two ComfyUI endpoints every 300 ms. The WebSocket is + the primary signal; polling is a fallback that runs at 1 Hz and only hits /queue, + backing off further while the socket is healthy. + """ + + COMFY_IDLE_PURGE_S = 30.0 + WATCHDOG_INTERVAL_S = 1.0 + WATCHDOG_INTERVAL_WS_OK_S = 3.0 def __init__(self): self.running = False self.ws_task: Optional[asyncio.Task] = None self.poll_task: Optional[asyncio.Task] = None + self.idle_task: Optional[asyncio.Task] = None self.last_yield_time = 0.0 self.last_comfy_free_time = 0.0 self.connected_ws = False self.last_action = "Idle" self.comfy_was_active = False + self.comfy_idle_since: Optional[float] = None self.oc_profile = None + self.pending_purge = False + self.stats = {"yields": 0, "purges": 0, "yield_timeouts": 0, "deferred_purges": 0} async def start(self): if self.running: @@ -368,24 +657,27 @@ class AutoArbitrator: self.running = True self.ws_task = asyncio.create_task(self._ws_listener()) self.poll_task = asyncio.create_task(self._poll_watchdog()) + self.idle_task = asyncio.create_task(self._idle_purge_loop()) logger.info("AutoArbitrator background engine started (Bidirectional).") - # Apply the default (balanced) overclock profile on startup. try: - await asyncio.get_event_loop().run_in_executor(None, overclock_manager.apply_profile, "balanced") + await asyncio.get_running_loop().run_in_executor( + None, overclock_manager.apply_profile, "balanced") + self.oc_profile = "balanced" except Exception as e: logger.warning(f"Startup overclock apply failed: {e}") async def stop(self): self.running = False - if self.ws_task: - self.ws_task.cancel() - if self.poll_task: - self.poll_task.cancel() + for task in (self.ws_task, self.poll_task, self.idle_task): + if task: + task.cancel() + await close_clients() logger.info("AutoArbitrator background engine stopped.") async def trigger_comfy_priority(self, reason: str = "ComfyUI prompt detected"): - """Instantly yield Ollama VRAM to 0MB when ComfyUI needs to run diffusion models.""" + """Yield Ollama's VRAM โ€” and confirm it is gone โ€” before diffusion allocates.""" self.comfy_was_active = True + self.comfy_idle_since = None self._apply_oc_profile("comfy") now = time.time() if now - self.last_yield_time < 1.0: @@ -394,89 +686,143 @@ class AutoArbitrator: ollama_state = await get_ollama_live_state() if ollama_state.get("active_model_name"): model = ollama_state["active_model_name"] - logger.info(f"โšก ComfyUI active ({reason}) -> Auto-yielding Ollama model '{model}' from VRAM...") + logger.info(f"โšก ComfyUI active ({reason}) -> Auto-yielding Ollama model '{model}'...") self.last_yield_time = time.time() - res = await instant_free_ollama_vram(model) - dur = res.get("duration_ms", 0) - self.last_action = f"Auto-yielded '{model}' for ComfyUI ({dur}ms)" + res = await instant_free_ollama_vram(model, confirm=True) + self.stats["yields"] += 1 + if res.get("confirmed"): + self.last_action = (f"Yielded '{model}' for ComfyUI in " + f"{res.get('confirm_ms')}ms (confirmed {res.get('freed_gb')}GB free)") + else: + self.stats["yield_timeouts"] += 1 + self.last_action = (f"โš  Yield of '{model}' NOT confirmed: " + f"{res.get('residual_gb')}GB still held") + logger.warning(f"VRAM yield barrier timed out: {res.get('error')}") logger.info(f"Ollama auto-yield completed: {res}") - async def trigger_comfy_completed(self): - """Purge ComfyUI VRAM cache when generation finishes, keeping VRAM 100% free for Ollama.""" + async def trigger_comfy_completed(self, immediate: bool = False): + """Mark the end of a generation. The actual purge is deferred unless forced.""" + self.comfy_was_active = False + if self.comfy_idle_since is None: + self.comfy_idle_since = time.time() + self._apply_oc_profile("ollama") + if immediate: + await self._purge_comfy_now("Ollama needs VRAM") + else: + self.pending_purge = True + self.stats["deferred_purges"] += 1 + self.last_action = (f"ComfyUI idle โ€” holding its checkpoints for " + f"{int(self.COMFY_IDLE_PURGE_S)}s in case you iterate") + + async def _purge_comfy_now(self, reason: str): now = time.time() if now - self.last_comfy_free_time < 3.0: return self.last_comfy_free_time = now - self.comfy_was_active = False - self._apply_oc_profile("ollama") - logger.info("โšก ComfyUI finished generation -> Auto-purging ComfyUI VRAM cache for Ollama...") + self.pending_purge = False + logger.info(f"โšก Purging ComfyUI VRAM cache ({reason})...") res = await instant_free_comfyui_vram() - dur = res.get("duration_ms", 0) - self.last_action = f"Auto-purged ComfyUI VRAM ({dur}ms) - Ready for Ollama" - logger.info(f"ComfyUI auto-purge completed: {res}") + self.stats["purges"] += 1 + self.last_action = f"Purged ComfyUI VRAM ({res.get('duration_ms')}ms) โ€” {reason}" + logger.info(f"ComfyUI purge completed: {res}") + + async def _idle_purge_loop(self): + """Purge ComfyUI's VRAM only after a real idle gap, not between iterations.""" + while self.running: + try: + if self.pending_purge and self.comfy_idle_since and not self.comfy_was_active: + idle_for = time.time() - self.comfy_idle_since + if idle_for >= self.COMFY_IDLE_PURGE_S: + await self._purge_comfy_now( + f"idle {int(idle_for)}s") + except Exception as e: + logger.debug(f"idle purge loop error: {e}") + await asyncio.sleep(2.0) + + async def request_vram_for_ollama(self, needed_gb: float = 0.0) -> Dict[str, Any]: + """Called when Ollama needs VRAM now: purge ComfyUI immediately rather than waiting.""" + snap = get_process_vram_bytes() + free_gb = snap["free_bytes"] / (1024**3) + if needed_gb and free_gb >= needed_gb: + return {"purged": False, "free_gb": round(free_gb, 2), "reason": "enough free VRAM"} + if snap["comfyui_bytes"] > YIELD_RESIDUAL_BYTES: + await self._purge_comfy_now(f"Ollama requested {needed_gb or '?'}GB") + snap = get_process_vram_bytes() + return {"purged": True, "free_gb": round(snap["free_bytes"] / (1024**3), 2)} + return {"purged": False, "free_gb": round(free_gb, 2), "reason": "ComfyUI holds no VRAM"} async def _ws_listener(self): client_id = "hyperswap-arbitrator" ws_url = f"ws://127.0.0.1:8188/ws?clientId={client_id}" + backoff = 2.0 while self.running: try: async with websockets.connect(ws_url, ping_interval=10, ping_timeout=10) as ws: self.connected_ws = True + backoff = 2.0 logger.info("AutoArbitrator connected to ComfyUI WebSocket.") while self.running: msg = await ws.recv() - if isinstance(msg, str): - try: - data = json.loads(msg) - msg_type = data.get("type") - msg_data = data.get("data", {}) + if not isinstance(msg, str): + continue + try: + data = json.loads(msg) + msg_type = data.get("type") + msg_data = data.get("data", {}) - if msg_type == "status": - queue_rem = msg_data.get("status", {}).get("exec_info", {}).get("queue_remaining", 0) - if queue_rem > 0: - await self.trigger_comfy_priority(f"Queue remaining: {queue_rem}") - elif queue_rem == 0 and self.comfy_was_active: - # Prompt queue finished - await asyncio.sleep(1.5) - await self.trigger_comfy_completed() - elif msg_type in ("execution_start", "execution_cached"): - await self.trigger_comfy_priority(f"Event: {msg_type}") - elif msg_type == "executing": - node = msg_data.get("node") - if node is not None: - await self.trigger_comfy_priority(f"Executing node: {node}") - elif node is None and self.comfy_was_active: - # Finished executing graph - await asyncio.sleep(1.5) - await self.trigger_comfy_completed() - elif msg_type == "execution_success": - await asyncio.sleep(1.5) + if msg_type == "status": + queue_rem = (msg_data.get("status", {}) + .get("exec_info", {}).get("queue_remaining", 0)) + if queue_rem > 0: + await self.trigger_comfy_priority(f"Queue remaining: {queue_rem}") + elif queue_rem == 0 and self.comfy_was_active: await self.trigger_comfy_completed() - except Exception as e: - logger.debug(f"WS parse error: {e}") + elif msg_type in ("execution_start", "execution_cached"): + await self.trigger_comfy_priority(f"Event: {msg_type}") + elif msg_type == "executing": + node = msg_data.get("node") + if node is not None: + await self.trigger_comfy_priority(f"Executing node: {node}") + elif self.comfy_was_active: + await self.trigger_comfy_completed() + elif msg_type == "execution_success": + await self.trigger_comfy_completed() + elif msg_type == "execution_error": + logger.warning(f"ComfyUI execution error: {msg_data}") + await self.trigger_comfy_completed() + except Exception as e: + logger.debug(f"WS parse error: {e}") except (websockets.exceptions.ConnectionClosed, OSError, asyncio.CancelledError): self.connected_ws = False except Exception as e: self.connected_ws = False logger.debug(f"WS connection error: {e}") - await asyncio.sleep(2.0) + await asyncio.sleep(backoff) + backoff = min(backoff * 1.5, 15.0) async def _poll_watchdog(self): - """Watchdog polling /queue every 300ms for robust bidirectional arbitration.""" + """Fallback for when the WebSocket is down. One cheap /queue call, 1 Hz. + + The previous version hit /system_stats and /queue every 300 ms on fresh TCP + connections โ€” roughly 6.6 requests/second against ComfyUI, forever. + """ while self.running: + interval = self.WATCHDOG_INTERVAL_WS_OK_S if self.connected_ws else self.WATCHDOG_INTERVAL_S try: - comfy_state = await get_comfyui_live_state() - is_executing = comfy_state.get("queue_running", 0) > 0 or comfy_state.get("queue_remaining", 0) > 0 or comfy_state.get("executing", False) - if is_executing: - await self.trigger_comfy_priority("Polling detected active queue/execution") - elif self.comfy_was_active and not is_executing: - await asyncio.sleep(1.5) - await self.trigger_comfy_completed() + client = _client(COMFY_API_BASE, 3.0) + resp = await client.get("/queue") + if resp.status_code == 200: + q = resp.json() + busy = len(q.get("queue_running", [])) > 0 or len(q.get("queue_pending", [])) > 0 + if busy: + await self.trigger_comfy_priority("Watchdog saw an active queue") + elif self.comfy_was_active: + await self.trigger_comfy_completed() except Exception: pass - await asyncio.sleep(0.3) + await asyncio.sleep(interval) def _apply_oc_profile(self, profile: str): """Apply an overclock profile in a background thread; only fire on transition.""" @@ -484,18 +830,25 @@ class AutoArbitrator: return self.oc_profile = profile try: - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() loop.run_in_executor(None, overclock_manager.apply_profile, profile) logger.info(f"๐ŸŽ›๏ธ Overclock profile switched -> '{profile}'") except Exception as e: logger.warning(f"Overclock profile switch failed ({profile}): {e}") def get_status(self) -> Dict[str, Any]: + idle_for = (time.time() - self.comfy_idle_since) if self.comfy_idle_since else None return { "running": self.running, "connected_ws": self.connected_ws, "last_action": self.last_action, "mode": "Bidirectional Hot-Swap (ComfyUI <-> Ollama)", + "comfy_active": self.comfy_was_active, + "pending_purge": self.pending_purge, + "comfy_idle_s": round(idle_for, 1) if idle_for is not None else None, + "idle_purge_after_s": self.COMFY_IDLE_PURGE_S, + "oc_profile": self.oc_profile, + "counters": dict(self.stats), }