Merge pull request 'Release: unfreeze agent status view + smoothed throughput autoscaler + log pane' (#165) from dev into main

2026-06-30 20:20:34 -04:00
parent 40be0a9323 f0f031782d
commit a28c33281a
3 changed files with 149 additions and 66 deletions
@@ -75,7 +75,11 @@ def gpu():
    # GPU meters poll this on their own fast cadence. It only reads local
    # nvidia-smi — no curator round-trip — so the util/VRAM bars stay live even
    # when /status is slow waiting on the (sometimes busy) curator queue call.
-    return JSONResponse(read_gpu() or {})
+    g = read_gpu() or {}
+    us = worker.util_smooth()
+    if us is not None:
+        g["util_smooth"] = round(us, 1)   # autoscaler's EWMA — the UI bar tracks this
+    return JSONResponse(g)


@app.get("/logs")
@@ -85,15 +89,13 @@ def logs():

@app.get("/status")
 def status():
+    # Pure in-memory read: worker.status() is lock-free and the queue snapshot is
+    # kept fresh by a background poller — NO inline curator call, so this can't
+    # stall the status view when curator is buried under a big backlog.
    s = worker.status()
    s["fc_url"] = cfg.fc_url
    s["configured"] = bool(cfg.token)
-    # GPU is served by /gpu (cached); /status stays light so Start/Stop never
-    # queue behind it.
-    try:
-        s["queue"] = worker.client.queue_status()
-    except Exception:
-        s["queue"] = None
+    s["queue"] = worker.latest_queue()
    return JSONResponse(s)


@@ -106,8 +108,8 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 *{box-sizing:border-box}
 body{font:14px/1.5 system-ui,-apple-system,Segoe UI,Roboto,sans-serif;margin:0;
  background:radial-gradient(1200px 600px at 50% -10%,#1a2029,#0f1216);color:var(--fg)}
- .wrap{max-width:820px;margin:0 auto;padding:28px 20px 28px;min-height:100vh;
-  display:flex;flex-direction:column}
+ .wrap{max-width:820px;margin:0 auto;padding:28px 20px 28px;height:100vh;
+  box-sizing:border-box;overflow:hidden;display:flex;flex-direction:column}
 header{display:flex;align-items:center;justify-content:space-between;margin-bottom:4px}
 .brand{display:flex;align-items:center;gap:10px;font-size:19px;font-weight:700;letter-spacing:.2px}
 .logo{color:var(--acc);font-size:20px}
@@ -163,7 +165,7 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  border:1px solid #5a4a17;color:#ffd98a;font-size:13px}
 .logs-h{display:flex;align-items:center;justify-content:space-between}
 .grow{flex:1;display:flex;flex-direction:column;min-height:0}
- .grow .logs{flex:1;min-height:120px}
+ .grow .logs{flex:1;min-height:0}
 .copybtn{font:600 11px system-ui;letter-spacing:.04em;text-transform:uppercase;
  background:#262c34;color:var(--fg);border:1px solid var(--bd);border-radius:7px;
  padding:5px 11px;cursor:pointer}
@@ -231,9 +233,12 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 // separate /status poll, which can lag behind the curator queue call.
 async function act(p){
  pending(p==='start'?'starting':'stopping')
-  try{ applyStatus(await (await fetch('/'+p,{method:'POST'})).json()) }
+  // Abort a slow POST after 8s so the buttons never stay stuck — the periodic
+  // /status refresh (now always fast) recovers the true state either way.
+  const ac=new AbortController(); const to=setTimeout(()=>ac.abort(),8000)
+  try{ applyStatus(await (await fetch('/'+p,{method:'POST',signal:ac.signal})).json()) }
  catch{ /* leave the periodic refresh to recover the real state */ }
-  finally{ startbtn.disabled=false; stopbtn.disabled=false }
+  finally{ clearTimeout(to); startbtn.disabled=false; stopbtn.disabled=false }
 }
 function pending(label){
  state.textContent=label; state.className='n busy'
@@ -285,13 +290,19 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 }
 // GPU meters poll their OWN endpoint on a fast cadence — kept off /status so a
 // slow curator queue call can't freeze the bars (they only stale on refresh).
+ let UAVG=null   // smoothed util for the bar (raw util swings 0↔99; show the trend)
 async function refreshGpu(){
  let g; try{ g=await (await fetch('/gpu')).json() }catch{ return }
  if(g && g.util_pct!=null){
-   const u=g.util_pct, used=g.mem_used_mb, tot=g.mem_total_mb||1
-   utillbl.textContent=u+'% · '+g.temp_c+'°C'; utilbar.style.width=u+'%'
+   // Prefer the agent's own EWMA (util_smooth) when running; otherwise smooth
+   // the raw reading here so a stopped agent's bar still glides, not jumps.
+   const raw=g.util_pct
+   UAVG = (g.util_smooth!=null) ? g.util_smooth
+         : (UAVG==null ? raw : 0.25*raw + 0.75*UAVG)
+   const used=g.mem_used_mb, tot=g.mem_total_mb||1
+   utillbl.textContent=Math.round(UAVG)+'% · '+g.temp_c+'°C'; utilbar.style.width=Math.round(UAVG)+'%'
   vramlbl.textContent=used+' / '+tot+' MB'; gpubar.style.width=Math.round(100*used/tot)+'%'
-  } else { utillbl.textContent='n/a'; vramlbl.textContent='n/a (CPU?)'; utilbar.style.width='0%'; gpubar.style.width='0%' }
+  } else { UAVG=null; utillbl.textContent='n/a'; vramlbl.textContent='n/a (CPU?)'; utilbar.style.width='0%'; gpubar.style.width='0%' }
 }
 async function refreshLogs(){
  try{
@@ -5,7 +5,7 @@ logging.Handler appends to it; the UI polls /logs."""
 import logging
 from collections import deque

-LINES: deque[str] = deque(maxlen=600)
+LINES: deque[str] = deque(maxlen=400)


 class RingHandler(logging.Handler):
@@ -11,6 +11,7 @@ orphaned work is re-picked at once rather than waiting out the lease.
 """
 import logging
 import threading
+import time

 import numpy as np
 import requests
@@ -49,19 +50,25 @@ MAX_CONCURRENCY = 32
 DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384"
 DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384"

-# Autoscaler (when Auto is on): a GPU-utilization-band controller. It grows the
-# pool while the GPU has spare capacity (util below the low mark + VRAM headroom)
-# and shrinks under saturation / memory pressure, then HOLDS while util sits in
-# the band — so the worker count stays steady instead of flopping. Util is EWMA-
-# smoothed and decisions are spaced out, so a single noisy nvidia-smi sample
-# can't move it.
-CONTROL_INTERVAL = 8.0     # seconds between samples
-DECIDE_EVERY = 3           # only act every Nth sample (~24s) — stability
-UTIL_LO = 70               # grow when smoothed util is below this (spare capacity)
-UTIL_HI = 92               # shrink when above this (saturated)
-VRAM_HI = 0.88             # shrink above this fraction of VRAM (memory pressure)
-VRAM_GROW_MAX = 0.80       # don't grow past this VRAM
-EWMA_ALPHA = 0.4           # util smoothing weight on the newest sample
+# Autoscaler (when Auto is on): a SMOOTHED, throughput-aware climb that SETTLES.
+# Raw GPU util swings wildly (a batched embed pegs it ~99%, then image decode/IO
+# drops it ~0%), so a single reading is meaningless — util is sampled often and
+# EWMA-smoothed. Each decision (spaced ~24s) grows the pool by one only while
+# doing so keeps lifting *throughput* (jobs/s, also smoothed); when a grow stops
+# helping the pool is IO/CPU/curator-bound, so it backs off one and SETTLES,
+# holding there before an occasional re-probe. This finds the worker count that
+# maximises real work — instead of flopping every cycle, or growing forever
+# because util never reaches a fixed threshold on an IO-bound load.
+CONTROL_INTERVAL = 2.0        # util sampling cadence (seconds)
+SAMPLES_PER_DECISION = 12     # decide ~every 24s (12 × 2s) on averaged signals
+UTIL_HI = 92                  # smoothed util above this = saturated → shrink
+UTIL_START = 85               # only begin a climb when smoothed util is below this
+VRAM_HI = 0.88                # shrink above this fraction of VRAM (memory pressure)
+VRAM_GROW_MAX = 0.80          # don't grow past this VRAM
+UTIL_ALPHA = 0.25             # util EWMA weight on the newest sample (smoother)
+TPUT_ALPHA = 0.5              # throughput EWMA weight
+TPUT_MARGIN = 0.08            # a grow must lift smoothed jobs/s by this to "help"
+REPROBE_TICKS = 8             # decisions to hold after settling before re-probing

 log = logging.getLogger("fc_agent.worker")

@@ -92,6 +99,13 @@ class Worker:
        self.transient = 0        # jobs handed back due to a server outage (NOT
        #                           failed) — the "waiting out curator" counter
        self._active = 0          # slots currently mid-image
+        self._util_smooth: float | None = None   # EWMA GPU util (set by control loop)
+        # Curator queue snapshot, refreshed by a background poller so the UI
+        # /status read is instant — never an inline curator HTTP call (which
+        # stalls the whole status view when curator is busy).
+        self._queue: dict | None = None
+        self._queue_thread = threading.Thread(target=self._queue_poll_loop, daemon=True)
+        self._queue_thread.start()
        # The crop embedder (SigLIP-family) is built lazily on the first job that
        # needs it, from the model the server announces — one shared instance.
        self._embedder = None
@@ -100,6 +114,23 @@ class Worker:
        self._proposers = None
        self._proposers_lock = threading.Lock()

+    def _queue_poll_loop(self):
+        """Refresh the curator queue snapshot in the background (every ~3s) so
+        /status is a pure in-memory read. Errors just leave the last snapshot
+        (or None) — the UI shows 'unreachable' without ever blocking."""
+        while True:
+            try:
+                self._queue = self.client.queue_status()
+            except Exception:
+                self._queue = None
+            time.sleep(3.0)
+
+    def latest_queue(self) -> dict | None:
+        return self._queue
+
+    def util_smooth(self) -> float | None:
+        return self._util_smooth
+
    # --- control -----------------------------------------------------------
    def start(self):
        with self._lock:
@@ -151,18 +182,20 @@ class Worker:
            self._slots.pop().stop.set()

    def status(self) -> dict:
-        with self._lock:
-            return {
-                "state": "running" if self._running else "stopped",
-                "concurrency": self._target,
-                "max_concurrency": MAX_CONCURRENCY,
-                "auto": self._auto,
-                "workers": len(self._slots),
-                "active": self._active,
-                "processed": self.processed,
-                "errors": self.errors,
-                "transient": self.transient,
-            }
+        # Lock-free on purpose: these are plain int / bool reads (atomic under the
+        # GIL) and this backs the UI poll — it must NEVER be able to block behind
+        # a worker holding _lock, or the whole status view freezes.
+        return {
+            "state": "running" if self._running else "stopped",
+            "concurrency": self._target,
+            "max_concurrency": MAX_CONCURRENCY,
+            "auto": self._auto,
+            "workers": len(self._slots),
+            "active": self._active,
+            "processed": self.processed,
+            "errors": self.errors,
+            "transient": self.transient,
+        }

    def _bump(self, *, processed=0, errors=0, active=0, transient=0):
        with self._lock:
@@ -217,20 +250,31 @@ class Worker:

    # --- autoscaler --------------------------------------------------------
    def _control_loop(self):
-        """GPU-utilization-band controller (Auto mode). Hold the worker count
-        steady while the GPU sits in a healthy util band; grow only when there's
-        clear spare capacity (smoothed util below the low mark + VRAM headroom),
-        shrink under saturation or memory pressure. Util is EWMA-smoothed and we
-        only act every DECIDE_EVERY samples, so a noisy nvidia-smi reading can't
-        make the pool flop — load stays consistent instead of probe/reverting
-        every cycle (the old hill-climb's failure mode)."""
+        """Smoothed, throughput-aware climb that settles (Auto mode). Samples GPU
+        util often and EWMA-smooths it (raw util swings 0↔99 between a batched
+        embed and the IO/decode around it, so one reading is noise). Every
+        SAMPLES_PER_DECISION ticks it makes ONE move: grow by one while each grow
+        keeps lifting smoothed throughput; when a grow stops helping (IO/CPU/
+        curator-bound) back off one and SETTLE, holding before an occasional
+        re-probe. Memory pressure and saturation always shrink immediately."""
        from . import gpu as gpumod

-        util_ewma = None     # smoothed GPU util%
-        tick = 0             # samples since the last decision
+        util_ewma: float | None = None
+        tput_ewma: float | None = None
+        prev_p, prev_t = self.processed, time.monotonic()
+        tick = 0
+        settled = False
+        grew_last = False        # did the previous decision grow the pool?
+        tput_before = 0.0        # smoothed jobs/s at the count before that grow
+        hold = 0                 # decisions left to hold while settled
        while not self._ctrl_stop.wait(CONTROL_INTERVAL):
            if not (self._running and self._auto):
-                util_ewma, tick = None, 0
+                util_ewma = tput_ewma = None
+                prev_p, prev_t = self.processed, time.monotonic()
+                tick = 0
+                settled = grew_last = False
+                hold = 0
+                self._util_smooth = None
                continue

            g = gpumod.read_gpu() or {}
@@ -238,36 +282,64 @@ class Worker:
            vram = (g.get("mem_used_mb", 0) / mt) if mt else 0.0
            util = g.get("util_pct", 0) or 0
            util_ewma = util if util_ewma is None else (
-                EWMA_ALPHA * util + (1 - EWMA_ALPHA) * util_ewma
+                UTIL_ALPHA * util + (1 - UTIL_ALPHA) * util_ewma
            )
+            self._util_smooth = util_ewma

            # Memory pressure overrides the cadence — react immediately.
            if vram >= VRAM_HI:
                if self._apply_step(-1):
-                    log.info(
-                        "autoscale: -1 → %d workers (vram %d%% — memory pressure)",
-                        self._target, round(vram * 100),
-                    )
-                tick = 0
+                    log.info("autoscale: -1 → %d workers (vram %d%% — memory pressure)",
+                             self._target, round(vram * 100))
+                tick, settled, grew_last, hold = 0, True, False, REPROBE_TICKS
                continue

            tick += 1
-            if tick < DECIDE_EVERY:                  # hold between decisions
+            if tick < SAMPLES_PER_DECISION:
                continue
            tick = 0

+            now = time.monotonic()
+            inst = (self.processed - prev_p) / max(1e-3, now - prev_t)
+            prev_p, prev_t = self.processed, now
+            tput_ewma = inst if tput_ewma is None else (
+                TPUT_ALPHA * inst + (1 - TPUT_ALPHA) * tput_ewma
+            )
+
            t0 = self._target
-            if util_ewma > UTIL_HI:                  # saturated → ease off
+            if util_ewma > UTIL_HI:                       # saturated → ease off
                self._apply_step(-1)
-            elif util_ewma < UTIL_LO and vram < VRAM_GROW_MAX:
-                self._apply_step(+1)                 # spare capacity → grow
-            # else: util is in the band → HOLD (steady load, no flopping)
+                settled, grew_last, hold = True, False, REPROBE_TICKS
+            elif settled:
+                hold -= 1
+                if hold <= 0:                             # re-probe: try one grow
+                    if util_ewma < UTIL_START and vram < VRAM_GROW_MAX:
+                        tput_before = tput_ewma
+                        grew_last = self._apply_step(+1)
+                        settled = not grew_last
+                    else:
+                        hold = REPROBE_TICKS              # still no room → keep holding
+            elif grew_last:
+                if tput_ewma > tput_before * (1 + TPUT_MARGIN):   # the grow helped
+                    tput_before = tput_ewma
+                    if util_ewma < UTIL_START and vram < VRAM_GROW_MAX:
+                        grew_last = self._apply_step(+1)
+                        settled = not grew_last
+                    else:
+                        settled, grew_last, hold = True, False, REPROBE_TICKS
+                else:                                     # overshot → back off + settle
+                    self._apply_step(-1)
+                    settled, grew_last, hold = True, False, REPROBE_TICKS
+            elif util_ewma < UTIL_START and vram < VRAM_GROW_MAX:  # start a climb
+                tput_before = tput_ewma
+                grew_last = self._apply_step(+1)
+                settled = not grew_last
+            else:
+                settled, hold = True, REPROBE_TICKS       # nothing to do → settle

            if self._target != t0:
-                log.info(
-                    "autoscale: %d→%d workers (util~%d%% · vram %d%%)",
-                    t0, self._target, round(util_ewma), round(vram * 100),
-                )
+                log.info("autoscale: %d→%d workers (util~%d%% · %.2f j/s · vram %d%%)",
+                         t0, self._target, round(util_ewma), tput_ewma, round(vram * 100))

    def _ensure_embedder(self, model_name: str):
        if self._embedder is not None: