From f0f031782d987da0f6d52c692a213976a479d52a Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 30 Jun 2026 20:20:14 -0400 Subject: [PATCH] fix(agent): unfreeze status view + smoothed throughput-aware autoscaler + log pane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Operator: the status tiles (state/active/processed) and the Start/Stop buttons freeze while the GPU meters stay live. Root cause: /status made an INLINE blocking curator call (queue_status) on every poll, and with curator buried under a 112k-job backlog that call stalled — freezing the whole status refresh (the GPU bars survived because /gpu is a lock-free local read). Made worse by the old util-band autoscaler, which grew workers toward the 32 cap forever because util plateaus ~50% on this IO-bound load and never hit the 70 grow threshold — piling load onto curator and the agent process. - /status is now a pure in-memory read: worker.status() is lock-free, and the curator queue snapshot is refreshed by a background poller (never inline). - Autoscaler replaced with a smoothed, throughput-aware climb that SETTLES: samples util every 2s and EWMA-smooths it (raw util swings 0↔99), then every ~24s grows by one only while each grow keeps lifting smoothed jobs/s; when a grow stops helping it backs off one and holds, re-probing occasionally. No runaway, no flopping. - GPU util bar now shows a smoothed value: the agent's own EWMA (util_smooth, exposed on /gpu) when running, else smoothed client-side — so it glides instead of bouncing 0↔99. - act() aborts a slow Start/Stop POST after 8s so the buttons can't stick; the now-always-fast /status refresh recovers state regardless. - Log pane: bound the page to the viewport (height:100vh) so the Logs card scrolls INTERNALLY instead of overflowing off-screen; cap the ring buffer at 400 lines. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa --- agent/fc_agent/app.py | 41 ++++++---- agent/fc_agent/logbuf.py | 2 +- agent/fc_agent/worker.py | 172 +++++++++++++++++++++++++++------------ 3 files changed, 149 insertions(+), 66 deletions(-) diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index 55a9b81..87be87b 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -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 = """ *{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 = """ 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 = """ // 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 = """ } // 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{ diff --git a/agent/fc_agent/logbuf.py b/agent/fc_agent/logbuf.py index 465d129..2de3488 100644 --- a/agent/fc_agent/logbuf.py +++ b/agent/fc_agent/logbuf.py @@ -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): diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 0790331..2169ae7 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -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: