diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index 615dab2..ec2d474 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -17,7 +17,7 @@ from .worker import Worker # Bump on every agent change. The page embeds this and /status reports it; the UI # warns to reload when they differ — so a stale browser-cached page can't be # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.) -VERSION = "2026-07-01.3 · video frame dedup" +VERSION = "2026-07-01.4 · gentler downloads, failure-aware scaling" logbuf.install() cfg = Config.from_env() diff --git a/agent/fc_agent/client.py b/agent/fc_agent/client.py index f226a08..1ac9925 100644 --- a/agent/fc_agent/client.py +++ b/agent/fc_agent/client.py @@ -113,7 +113,11 @@ class FcClient: def fetch_image(self, image_url: str) -> bytes: # image_url is a server-relative path ("/images/..."). - r = self.s.get(f"{self.base}{image_url}", timeout=180) + # timeout=(connect, read): the read timeout is BETWEEN-BYTES, not total, + # so a large-but-flowing download still completes — but a stuck/dead + # connection (curator overloaded) fails in 60s instead of hanging a + # downloader for 180s and piling up concurrent stuck requests on curator. + r = self.s.get(f"{self.base}{image_url}", timeout=(10, 60)) r.raise_for_status() return r.content diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index ee223c5..3ceff61 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -61,10 +61,14 @@ def _is_transient(exc: requests.RequestException) -> bool: return resp.status_code >= 500 or resp.status_code in (401, 403, 408, 409, 429) -# Pipeline sizing. Downloaders are I/O-bound so the ceiling is generous; consumers -# are GPU-bound so a couple saturate the card. The buffer is small on purpose — -# each slot can hold many decoded video frames, so it bounds RAM, not just depth. -DL_MAX = 24 # max downloader threads +# Pipeline sizing. Downloaders are I/O-bound, but every download streams a full +# original (large videos included) THROUGH curator's single Python file-serving +# path — so the ceiling is deliberately modest: too many concurrent large-file +# GETs saturate curator's web workers + NFS and slow everything (including the +# browser). 8 keeps a fast GPU fed without stampeding curator. Consumers are +# GPU-bound so a couple saturate the card. The buffer is small on purpose — each +# slot can hold many decoded video frames, so it bounds RAM, not just depth. +DL_MAX = 8 # max downloader threads CONSUMER_MAX = 2 # max GPU consumer threads BUFFER_MAX = 12 # bounded decoded-frame buffer (backpressure + RAM cap) @@ -712,13 +716,22 @@ class Worker: (the GPU is the bottleneck → add a 2nd consumer if it has headroom and the add lifts throughput, else trim a downloader). Occupancy, util and throughput are EWMA-smoothed and decisions spaced so moves ride averaged - signals, not tick-to-tick noise. VRAM pressure sheds a consumer at once.""" + signals, not tick-to-tick noise. VRAM pressure sheds a consumer at once. + + Failure guard (critical): an empty buffer can mean the GPU is starving OR + that downloads are FAILING (curator slow/unreachable). In the latter case + adding downloaders piles more concurrent large-file requests onto a + struggling curator — a congestion collapse that slows curator (and its + browser) further and never recovers. So if transient download failures + rose since the last decision, SHRINK toward the floor instead of growing, + and let the pool ramp back up only once downloads succeed again.""" from . import gpu as gpumod occ_ewma: float | None = None util_ewma: float | None = None tput_ewma: float | None = None prev_p, prev_t = self.processed, time.monotonic() + prev_fail = self.transient tick = 0 con_grew = False # did the previous decision add a consumer? tput_before = 0.0 # smoothed jobs/s before that consumer add @@ -726,6 +739,7 @@ class Worker: if not (self._running and self._auto): occ_ewma = util_ewma = tput_ewma = None prev_p, prev_t = self.processed, time.monotonic() + prev_fail = self.transient tick = 0 con_grew = False self._util_smooth = None @@ -764,9 +778,17 @@ class Worker: tput_ewma = inst if tput_ewma is None else ( TPUT_ALPHA * inst + (1 - TPUT_ALPHA) * tput_ewma ) + fail_delta = self.transient - prev_fail + prev_fail = self.transient d0, c0 = self._dl_target, self._consumer_target - if occ_ewma < OCC_LOW: + if fail_delta > 0: + # Downloads are FAILING (curator slow/unreachable), so the empty + # buffer is NOT the GPU starving — growing would stampede a + # struggling curator. Back off toward the floor and let it recover. + self._apply_downloaders(-1) + con_grew = False + elif occ_ewma < OCC_LOW: # Buffer starving → GPU idle waiting on downloads → add a feeder. self._apply_downloaders(+1) con_grew = False