diff --git a/agent/docker-compose.yml b/agent/docker-compose.yml index 5032669..79538a4 100644 --- a/agent/docker-compose.yml +++ b/agent/docker-compose.yml @@ -34,6 +34,10 @@ services: # Resume the worker automatically on container start (survive a reboot / # crash-restart while you're away). Set to 0 to require a manual Start. AUTO_START: ${AUTO_START:-1} + # Autoscale the worker count (throughput hill-climb that finds the sweet + # spot + backs off under VRAM pressure). On by default; toggle live in the + # control UI. Set to 0 to start in manual mode. + AUTO_SCALE: ${AUTO_SCALE:-1} # Crop embedder (SigLIP concept bag): float16 keeps VRAM low on a shared # desktop GPU; the model itself is announced by the server. SIGLIP_DTYPE: ${SIGLIP_DTYPE:-float16} diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index 9252c2f..5ca989c 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -50,6 +50,13 @@ async def concurrency(request: Request): return JSONResponse(worker.status()) +@app.post("/auto") +async def auto(request: Request): + body = await request.json() + worker.set_auto(bool(body.get("value", True))) + return JSONResponse(worker.status()) + + @app.get("/status") def status(): s = worker.status() @@ -83,13 +90,14 @@ _PAGE = """
+ workers - (more = overlap I/O, fill the GPU) max 8 + auto-tuning to fill the GPU · max 8
stopped
state @@ -113,6 +121,10 @@ _PAGE = """ await fetch('/concurrency',{method:'POST',headers:{'Content-Type':'application/json'}, body:JSON.stringify({value:v})});refresh() } + async function setauto(on){ + await fetch('/auto',{method:'POST',headers:{'Content-Type':'application/json'}, + body:JSON.stringify({value:on})});refresh() + } async function refresh(){ const s=await (await fetch('/status')).json() CAP=s.max_concurrency||8; capn.textContent=CAP @@ -121,6 +133,11 @@ _PAGE = """ // Running but the queue read failed → curator is unreachable; show we're // riding it out rather than erroring. banner.style.display=(s.state==='running' && !s.queue)?'block':'none' + // Auto on → the dial reflects the auto-chosen count (read-only); off → manual. + if(document.activeElement!==autochk) autochk.checked=!!s.auto + conc.disabled=!!s.auto; conc.style.opacity=s.auto?0.6:1 + conchint.textContent=s.auto?('auto-tuning to fill the GPU · max '+CAP):('manual · max '+CAP) + capn.textContent=CAP if(document.activeElement!==conc) conc.value=s.concurrency conc.max=CAP cfg.textContent=s.configured?'set':'MISSING' diff --git a/agent/fc_agent/config.py b/agent/fc_agent/config.py index 630fc3f..5ab7a56 100644 --- a/agent/fc_agent/config.py +++ b/agent/fc_agent/config.py @@ -18,6 +18,7 @@ class Config: # the server announces in the lease) auto_start: bool # start the worker pool on boot (so a container restart # resumes processing without anyone clicking Start) + auto_scale: bool # autoscale the worker count (throughput hill-climb) # Crop PROPOSERS (extra YOLO detectors that say where to crop). Each weight # spec is an ultralytics name | http(s) URL | "hf_repo::file" ("" = off). person_weights: str # general COCO person detector (Western/realistic figs) @@ -43,6 +44,7 @@ class Config: embed_dtype=os.environ.get("SIGLIP_DTYPE", "float16"), embed_model_override=os.environ.get("EMBED_MODEL_NAME", ""), auto_start=os.environ.get("AUTO_START", "").lower() in ("1", "true", "yes"), + auto_scale=os.environ.get("AUTO_SCALE", "true").lower() in ("1", "true", "yes"), person_weights=os.environ.get("PERSON_WEIGHTS", "yolo11n.pt"), person_conf=float(os.environ.get("PERSON_CONF", "0.35")), anatomy_weights=os.environ.get("ANATOMY_WEIGHTS", ""), diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 2b7e412..e8b21f4 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -48,6 +48,16 @@ MAX_CONCURRENCY = 32 DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384" DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384" +# Autoscaler (when Auto is on): a throughput hill-climb that finds the worker +# count on its own — grows while jobs/sec keeps rising and VRAM stays under +# budget, backs off when a step stops helping or memory gets tight, then settles +# and periodically re-probes (the workload's GPU/IO balance shifts). +CONTROL_INTERVAL = 6.0 # seconds between control decisions +VRAM_HI = 0.90 # back off above this fraction of VRAM used +UTIL_HI = 96 # GPU util% considered saturated +TPUT_MARGIN = 0.10 # a step up must beat the baseline by this to "help" +REPROBE_TICKS = 5 # ticks to hold after settling before re-probing up + class _Slot: """One worker loop. `inflight` = jobs leased but not yet processed, so a @@ -66,6 +76,9 @@ class Worker: self._lock = threading.Lock() self._running = False self._target = max(1, min(MAX_CONCURRENCY, cfg.concurrency)) + self._auto = bool(cfg.auto_scale) # autoscale worker count + self._ctrl_stop = threading.Event() + self._ctrl_thread: threading.Thread | None = None self._slots: list[_Slot] = [] self.processed = 0 self.errors = 0 @@ -85,20 +98,43 @@ class Worker: with self._lock: self._running = True self._reconcile_locked() + # (Re)start the autoscaler control loop. + if self._ctrl_thread is None or not self._ctrl_thread.is_alive(): + self._ctrl_stop.clear() + self._ctrl_thread = threading.Thread(target=self._control_loop, daemon=True) + self._ctrl_thread.start() def stop(self): + self._ctrl_stop.set() with self._lock: self._running = False slots, self._slots = self._slots, [] for s in slots: s.stop.set() # each slot releases its inflight on exit - def set_concurrency(self, n: int): + def set_auto(self, on: bool): with self._lock: + self._auto = bool(on) + + def set_concurrency(self, n: int): + # A manual set is an override → leave Auto. + with self._lock: + self._auto = False self._target = max(1, min(MAX_CONCURRENCY, int(n))) if self._running: self._reconcile_locked() + def _apply_step(self, delta: int) -> bool: + """Nudge the target by delta (bounded). Returns True if it changed.""" + with self._lock: + new = max(1, min(MAX_CONCURRENCY, self._target + delta)) + if new == self._target: + return False + self._target = new + if self._running: + self._reconcile_locked() + return True + def _reconcile_locked(self): while len(self._slots) < self._target: slot = _Slot() @@ -113,6 +149,7 @@ class Worker: "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, @@ -171,6 +208,58 @@ class Worker: a pool-shrink doesn't hang for a full backoff window.""" slot.stop.wait(timeout=seconds) + # --- autoscaler -------------------------------------------------------- + def _control_loop(self): + """Throughput hill-climb (Auto mode): grow the pool while jobs/sec keeps + improving and VRAM stays under budget; revert a step that doesn't help; + back off under memory pressure; settle, then periodically re-probe.""" + import time as _t + + from . import gpu as gpumod + + prev_p, prev_t = self.processed, _t.monotonic() + base_tput = None # throughput baseline the current climb is judged against + 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): + if not (self._running and self._auto): + prev_p, prev_t = self.processed, _t.monotonic() + base_tput, last_dir, cooldown = None, 0, 0 + continue + now = _t.monotonic() + dt = max(1e-3, now - prev_t) + tput = (self.processed - prev_p) / dt + prev_p, prev_t = self.processed, now + + g = gpumod.read_gpu() or {} + mt = g.get("mem_total_mb") or 0 + vram = (g.get("mem_used_mb", 0) / mt) if mt else 0.0 + util = g.get("util_pct", 0) or 0 + + if vram >= VRAM_HI: # memory pressure → always shrink + self._apply_step(-1) + base_tput, last_dir, cooldown = None, 0, 2 + continue + if cooldown > 0: + 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 + def _ensure_embedder(self, model_name: str): if self._embedder is not None: return self._embedder