bvandeusen/FabledCurator
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 35 Wiki Activity

Release: stable util-band autoscaler + live GPU meters #162

Merged
bvandeusen merged 1 commits from dev into main 2026-06-30 19:16:29 -04:00
Conversation 0 Commits 1 Files Changed 3
+70 -56
3 changed files with 70 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/fc_agent/app.py
+19 -6
View File
@@ -59,6 +59,14 @@ async def auto(request: Request):
return JSONResponse(worker.status()) return JSONResponse(worker.status())
@app.get("/gpu")
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 {})
@app.get("/logs") @app.get("/logs")
def logs(): def logs():
return JSONResponse({"lines": list(logbuf.LINES)}) return JSONResponse({"lines": list(logbuf.LINES)})
@@ -233,12 +241,17 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
conchint.textContent=s.auto?('auto-tuning to fill the GPU · max '+CAP):('manual · max '+CAP) conchint.textContent=s.auto?('auto-tuning to fill the GPU · max '+CAP):('manual · max '+CAP)
if(document.activeElement!==conc) conc.value=s.concurrency if(document.activeElement!==conc) conc.value=s.concurrency
conc.max=CAP conc.max=CAP
if(s.gpu){ queue.textContent=s.queue?('queue · pending '+s.queue.pending+' · in flight '+s.queue.leased+' · done '+s.queue.done+' · errored '+s.queue.error):'queue · unreachable'
const u=s.gpu.util_pct, used=s.gpu.mem_used_mb, tot=s.gpu.mem_total_mb||1 }
utillbl.textContent=u+'% · '+s.gpu.temp_c+'°C'; utilbar.style.width=u+'%' // 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).
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+'%'
vramlbl.textContent=used+' / '+tot+' MB'; gpubar.style.width=Math.round(100*used/tot)+'%' 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 { utillbl.textContent='n/a'; vramlbl.textContent='n/a (CPU?)'; utilbar.style.width='0%'; gpubar.style.width='0%' }
queue.textContent=s.queue?('queue · pending '+s.queue.pending+' · in flight '+s.queue.leased+' · done '+s.queue.done+' · errored '+s.queue.error):'queue · unreachable'
} }
async function refreshLogs(){ async function refreshLogs(){
try{ try{
@@ -255,6 +268,6 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
t.select(); try{document.execCommand('copy')}catch{}; t.remove() } t.select(); try{document.execCommand('copy')}catch{}; t.remove() }
copybtn.textContent='Copied'; setTimeout(()=>{copybtn.textContent='Copy'},1200) copybtn.textContent='Copied'; setTimeout(()=>{copybtn.textContent='Copy'},1200)
} }
refresh(); refreshLogs() refresh(); refreshGpu(); refreshLogs()
setInterval(refresh,3000); setInterval(refreshLogs,2500) setInterval(refresh,3000); setInterval(refreshGpu,1500); setInterval(refreshLogs,2500)
</script></body></html>""" </script></body></html>"""
agent/fc_agent/client.py
+3 -1
View File
@@ -93,6 +93,8 @@ class FcClient:
return r.content return r.content
def queue_status(self) -> dict: def queue_status(self) -> dict:
r = self.s.get(f"{self.base}/api/gpu/status", timeout=15) # Short timeout: this backs the UI /status poll, so a busy curator must
# not hang the page for long (the GPU meters poll /gpu separately).
r = self.s.get(f"{self.base}/api/gpu/status", timeout=5)
r.raise_for_status() r.raise_for_status()
return r.json() return r.json()
agent/fc_agent/worker.py
+48 -49
View File
@@ -49,15 +49,19 @@ MAX_CONCURRENCY = 32
DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384" DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384"
DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384" DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384"
# Autoscaler (when Auto is on): a throughput hill-climb that finds the worker # Autoscaler (when Auto is on): a GPU-utilization-band controller. It grows the
# count on its own — grows while jobs/sec keeps rising and VRAM stays under # pool while the GPU has spare capacity (util below the low mark + VRAM headroom)
# budget, backs off when a step stops helping or memory gets tight, then settles # and shrinks under saturation / memory pressure, then HOLDS while util sits in
# and periodically re-probes (the workload's GPU/IO balance shifts). # the band — so the worker count stays steady instead of flopping. Util is EWMA-
CONTROL_INTERVAL = 6.0 # seconds between control decisions # smoothed and decisions are spaced out, so a single noisy nvidia-smi sample
VRAM_HI = 0.90 # back off above this fraction of VRAM used # can't move it.
UTIL_HI = 96 # GPU util% considered saturated CONTROL_INTERVAL = 8.0 # seconds between samples
TPUT_MARGIN = 0.10 # a step up must beat the baseline by this to "help" DECIDE_EVERY = 3 # only act every Nth sample (~24s) — stability
REPROBE_TICKS = 5 # ticks to hold after settling before re-probing up 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
log = logging.getLogger("fc_agent.worker") log = logging.getLogger("fc_agent.worker")
@@ -213,61 +217,56 @@ class Worker:
# --- autoscaler -------------------------------------------------------- # --- autoscaler --------------------------------------------------------
def _control_loop(self): def _control_loop(self):
"""Throughput hill-climb (Auto mode): grow the pool while jobs/sec keeps """GPU-utilization-band controller (Auto mode). Hold the worker count
improving and VRAM stays under budget; revert a step that doesn't help; steady while the GPU sits in a healthy util band; grow only when there's
back off under memory pressure; settle, then periodically re-probe.""" clear spare capacity (smoothed util below the low mark + VRAM headroom),
import time as _t 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)."""
from . import gpu as gpumod from . import gpu as gpumod
prev_p, prev_t = self.processed, _t.monotonic() util_ewma = None # smoothed GPU util%
base_tput = None # throughput baseline the current climb is judged against tick = 0 # samples since the last decision
last_dir = 0 # direction of the last applied step (+1 / -1 / 0)
cooldown = 0 # ticks to wait (post-settle / post-backoff) before acting
while not self._ctrl_stop.wait(CONTROL_INTERVAL): while not self._ctrl_stop.wait(CONTROL_INTERVAL):
if not (self._running and self._auto): if not (self._running and self._auto):
prev_p, prev_t = self.processed, _t.monotonic() util_ewma, tick = None, 0
base_tput, last_dir, cooldown = None, 0, 0
continue continue
now = _t.monotonic()
dt = max(1e-3, now - prev_t)
tput = (self.processed - prev_p) / dt
prev_p, prev_t = self.processed, now
t0 = self._target
g = gpumod.read_gpu() or {} g = gpumod.read_gpu() or {}
mt = g.get("mem_total_mb") or 0 mt = g.get("mem_total_mb") or 0
vram = (g.get("mem_used_mb", 0) / mt) if mt else 0.0 vram = (g.get("mem_used_mb", 0) / mt) if mt else 0.0
util = g.get("util_pct", 0) or 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
)
if vram >= VRAM_HI: # memory pressure → always shrink # 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
continue
tick += 1
if tick < DECIDE_EVERY: # hold between decisions
continue
tick = 0
t0 = self._target
if util_ewma > UTIL_HI: # saturated → ease off
self._apply_step(-1) self._apply_step(-1)
base_tput, last_dir, cooldown = None, 0, 2 elif util_ewma < UTIL_LO and vram < VRAM_GROW_MAX:
continue self._apply_step(+1) # spare capacity → grow
if cooldown > 0: # else: util is in the band → HOLD (steady load, no flopping)
cooldown -= 1
continue
if base_tput is None: # establish a baseline + probe up
base_tput = tput
last_dir = 1 if self._apply_step(1) else 0
if last_dir == 0: # already at the cap
base_tput, cooldown = None, REPROBE_TICKS
continue
if last_dir > 0:
if tput > base_tput * (1 + TPUT_MARGIN) and util < UTIL_HI:
base_tput = tput # the step helped → keep climbing
if not self._apply_step(1):
base_tput, last_dir, cooldown = None, 0, REPROBE_TICKS
else: # didn't help → revert + settle
self._apply_step(-1)
base_tput, last_dir, cooldown = None, 0, REPROBE_TICKS
else:
base_tput = None # settled → re-probe next cycle
if self._target != t0: if self._target != t0:
log.info( log.info(
"autoscale: %d→%d workers (%.2f jobs/s · util %d%% · vram %d%%)", "autoscale: %d→%d workers (util~%d%% · vram %d%%)",
t0, self._target, tput, util, round(vram * 100), t0, self._target, round(util_ewma), round(vram * 100),
) )
def _ensure_embedder(self, model_name: str): def _ensure_embedder(self, model_name: str):