diff --git a/agent/docker-compose.yml b/agent/docker-compose.yml index 79538a4..bad164d 100644 --- a/agent/docker-compose.yml +++ b/agent/docker-compose.yml @@ -38,6 +38,10 @@ services: # 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} + # Aggregate download cap in MB/s (stills + video streams combined), so the + # agent can't saturate the desktop's network and wreck browsing — WiFi + # especially. 0 = unlimited; tunable live in the control UI. + BANDWIDTH_LIMIT_MB_S: ${BANDWIDTH_LIMIT_MB_S:-8} # 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 cc33aca..93795d4 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -21,7 +21,7 @@ log = logging.getLogger("fc_agent.app") # 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-02.3 · poison guard: repeated transient failures fail the job (with reason) instead of releasing it forever" +VERSION = "2026-07-02.4 · bandwidth governor: aggregate download cap (MB/s dial) so the agent can't saturate the desktop's network" logbuf.install() cfg = Config.from_env() @@ -83,6 +83,13 @@ async def auto(request: Request): return JSONResponse(worker.status()) +@app.post("/bandwidth") +async def bandwidth(request: Request): + body = await request.json() + worker.set_bandwidth(float(body.get("value", 0))) + return JSONResponse(worker.status()) + + @app.get("/gpu") def gpu(): # GPU meters poll this on their own fast cadence. It only reads local @@ -161,8 +168,9 @@ _PAGE = """ .step{background:#262c34;color:var(--fg);border:1px solid var(--bd);border-radius:8px; width:30px;height:32px;font:700 16px system-ui;cursor:pointer} .step:hover{border-color:var(--acc)} - #conc{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a; + #conc,#bw{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a; color:var(--fg);border:1px solid var(--bd);border-radius:8px} + .unit{color:var(--mut);font-size:12px;font-weight:600} .hint{color:var(--mut);font-size:12px;margin-top:12px} .tiles{display:grid;grid-template-columns:repeat(6,1fr);gap:8px;margin-bottom:16px} .tile{background:#13171d;border:1px solid var(--bd);border-radius:10px;padding:12px 8px;text-align:center} @@ -216,6 +224,10 @@ _PAGE = """ +
+ + MB/s +
auto-tuning downloaders to keep the GPU fed · max 8
@@ -281,6 +293,11 @@ _PAGE = """ await fetch('/auto',{method:'POST',headers:{'Content-Type':'application/json'}, body:JSON.stringify({value:on})});refresh() } + async function setbw(v){ + v=Math.max(0,parseFloat(v)||0); bw.value=v + await fetch('/bandwidth',{method:'POST',headers:{'Content-Type':'application/json'}, + body:JSON.stringify({value:v})});refresh() + } async function refresh(){ let s; try{ s=await (await fetch('/status')).json() }catch{ return } applyStatus(s) @@ -318,6 +335,9 @@ _PAGE = """ // Instantaneous pool state → demoted to the sub-line, where its jumpiness reads // as live churn rather than a "broken" headline metric. pipe.textContent='downloaders '+(s.downloaders!=null?s.downloaders:'—')+' · consumers '+(s.consumers!=null?s.consumers:'—')+' · on GPU '+(s.active||0) + +' · net '+(s.net_mb_s!=null?s.net_mb_s.toFixed(1):'—')+' MB/s' + +(s.bandwidth_limit_mb_s>0?(' / cap '+s.bandwidth_limit_mb_s):'') + if(document.activeElement!==bw && s.bandwidth_limit_mb_s!=null) bw.value=s.bandwidth_limit_mb_s // Buffer occupancy bar (also driven here so it tracks the /status cadence). if(s.buffer!=null && s.buffer_max){ const p=Math.round(100*s.buffer/s.buffer_max) buflbl.textContent=s.buffer+' / '+s.buffer_max; bufbar.style.width=p+'%' } diff --git a/agent/fc_agent/client.py b/agent/fc_agent/client.py index 1f40150..b2b8cea 100644 --- a/agent/fc_agent/client.py +++ b/agent/fc_agent/client.py @@ -101,15 +101,26 @@ class FcClient: return self._post_quiet("/api/gpu/jobs/release", {"job_ids": job_ids}) - def fetch_image(self, image_url: str) -> bytes: + def fetch_image(self, image_url: str, throttle=None) -> bytes: # image_url is a server-relative path ("/images/..."). # 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 + # With a throttle (the worker's shared TokenBucket), the body is streamed + # in chunks and each chunk is charged to the global bandwidth budget — + # pausing between reads lets TCP flow control pace curator's send side. + with self.s.get( + f"{self.base}{image_url}", timeout=(10, 60), stream=throttle is not None + ) as r: + r.raise_for_status() + if throttle is None: + return r.content + buf = bytearray() + for chunk in r.iter_content(chunk_size=262_144): + throttle.take(len(chunk)) + buf.extend(chunk) + return bytes(buf) def is_reachable(self) -> bool: """Cheap 'is curator responding at all right now?' check. Used to decide, diff --git a/agent/fc_agent/config.py b/agent/fc_agent/config.py index 045114b..098d728 100644 --- a/agent/fc_agent/config.py +++ b/agent/fc_agent/config.py @@ -48,6 +48,9 @@ class Config: # higher keeps more frames, 0 disables ffmpeg_timeout: float # hard ceiling (s) for ffmpeg-from-URL video sampling; # generous so a SLOW media link still completes + bandwidth_limit_mb_s: float # aggregate download cap in MEGABYTES/s across + # all downloaders + video streams (0 = unlimited); + # tunable live from the agent UI @classmethod def from_env(cls) -> Config: @@ -77,4 +80,11 @@ class Config: dedupe_iou=float(os.environ.get("DEDUPE_IOU", "0.85")), frame_dedupe_distance=int(os.environ.get("FRAME_DEDUPE_DISTANCE", "8")), ffmpeg_timeout=float(os.environ.get("FFMPEG_TIMEOUT", "1200")), + # Default 8 MB/s (~64 Mbit/s): ~20% of the measured ~300 Mbit/s home + # WiFi, so browsing stays snappy while the agent works — yet MORE + # sweep throughput than the self-inflicted congestion collapse this + # replaces (2026-07-02: 8 unthrottled downloaders bufferbloated the + # link to ~1-1.5 MB/s per stream, browser included). Raise it (or 0) + # from the agent UI on wired/faster networks. + bandwidth_limit_mb_s=float(os.environ.get("BANDWIDTH_LIMIT_MB_S", "8")), ) diff --git a/agent/fc_agent/media.py b/agent/fc_agent/media.py index 99c49d7..e70b2f8 100644 --- a/agent/fc_agent/media.py +++ b/agent/fc_agent/media.py @@ -4,12 +4,15 @@ instances, each with a timestamp.""" import io import logging import os +import signal import subprocess import tempfile import time from PIL import Image, ImageFile +from .throttle import PidReadMeter + log = logging.getLogger("fc_agent.media") # Load slightly-truncated images (a few missing trailing bytes) instead of @@ -111,6 +114,12 @@ def _collect_frames( def _terminate(proc: subprocess.Popen) -> None: """Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM.""" + try: + # A bandwidth-paused (SIGSTOPped) process can't receive SIGTERM until it + # resumes — always CONT first so termination is prompt, not queued. + proc.send_signal(signal.SIGCONT) + except OSError: + pass proc.terminate() try: proc.wait(timeout=2) @@ -122,9 +131,33 @@ def _terminate(proc: subprocess.Popen) -> None: pass +def _pause(proc: subprocess.Popen, seconds: float, should_stop) -> bool: + """SIGSTOP ffmpeg for ~`seconds` of bandwidth debt, staying responsive to + Stop. While paused, the kernel socket buffer fills and TCP flow control + stalls curator's send side — that's the throttle. SIGCONT is ALWAYS sent + before returning. False = a Stop arrived mid-pause.""" + try: + proc.send_signal(signal.SIGSTOP) + except OSError: + return True # already exited — nothing to pause + try: + end = time.monotonic() + seconds + while (left := end - time.monotonic()) > 0: + if should_stop and should_stop(): + return False + time.sleep(min(0.5, left)) + return True + finally: + try: + proc.send_signal(signal.SIGCONT) + except OSError: + pass + + def sample_frames_from_url( url: str, interval_seconds: float, max_frames: int, *, headers: str = "", timeout: float = 1200.0, should_stop=None, + governor=None, ) -> tuple[list[tuple[float, Image.Image]], str | None]: """Sample frames by pointing ffmpeg STRAIGHT at the media URL — it Range-reads only the video index + up to max_frames worth of content, so the agent never @@ -133,7 +166,10 @@ def sample_frames_from_url( the timeout is the per-video ceiling (a slow/reconnecting stream can otherwise run for minutes). `should_stop` is polled while ffmpeg runs so a Stop KILLS the subprocess at once — otherwise a downloader stuck in a long decode keeps the - agent "working" long after Stop. + agent "working" long after Stop. `governor` (the worker's shared TokenBucket) + meters ffmpeg's network reads from outside via /proc//io and SIGSTOPs + the process into budget, so video streaming honors the same aggregate + bandwidth cap as still downloads. Returns (frames, reason): frames is empty on failure/stop/timeout, and `reason` then carries the SPECIFIC cause (ffmpeg's stderr tail / timeout) so @@ -168,6 +204,7 @@ def sample_frames_from_url( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=errf, ) + meter = PidReadMeter(proc.pid) if governor is not None else None # Poll rather than block, so a Stop (or the per-video timeout) can # kill a slow/wedged ffmpeg promptly instead of waiting it out. start = time.monotonic() @@ -184,6 +221,17 @@ def sample_frames_from_url( log.warning("ffmpeg timed out after %.0fs: %s", timeout, url) return [], f"ffmpeg timed out after {timeout:.0f}s" + if meter is not None: + read = meter.delta() + if read is None: # /proc gone → stop governing + meter = None + elif (debt := governor.charge(read)) > 0: + # Over budget: pause ffmpeg until the bucket + # recovers. Pause time counts toward `timeout` + # (it stays the wedge backstop either way). + if not _pause(proc, debt, should_stop): + _terminate(proc) + return [], "stopped" except (OSError, ValueError) as exc: return [], f"ffmpeg not runnable: {exc}" frames = _collect_frames(tmp, interval, cap) diff --git a/agent/fc_agent/throttle.py b/agent/fc_agent/throttle.py new file mode 100644 index 0000000..6459105 --- /dev/null +++ b/agent/fc_agent/throttle.py @@ -0,0 +1,111 @@ +"""Global download-bandwidth governor (one token bucket for the whole agent). + +The agent lives on someone's desktop and shares that desktop's network — +typically WiFi, where saturating the link doesn't just slow other apps: it +bufferbloats the airtime (RTT 21→45ms) and collapses EVERY connection, +the operator's browser included. Measured 2026-07-02: the idle link moved +~38 MB/s single-stream, but under the 8-downloader sweep every stream on the +machine crawled at ~1-1.5 MB/s. So the cap is on the AGGREGATE, not per +stream: still downloads pump their chunks through take(), and ffmpeg video +streams — whose sockets live in a subprocess we can't wrap — are metered from +outside via /proc//io and paused (SIGSTOP) into budget using charge()'s +debt signal; TCP flow control then stalls the sender while ffmpeg sleeps. + +Accounting is post-paid (charge the bytes first, then wait out any debt): the +bytes have already crossed the network by the time we count them, and it means +a chunk larger than one second of budget can never deadlock the bucket. +Stdlib-only on purpose — unit-tested in CI, where the agent's ML deps +don't exist. +""" +import threading +import time + + +class TokenBucket: + """Thread-safe token bucket in bytes/second. rate 0 = unlimited. + + `consumed` is the monotonic total of bytes charged (throttled or not) — + the worker's rate loop derives the UI's "net MB/s" readout from it. + """ + + def __init__(self, rate_bytes_per_s: float = 0.0): + self._cond = threading.Condition() + self._rate = max(0.0, float(rate_bytes_per_s)) + # Burst = one second of budget: enough that chunked reads stay smooth, + # small enough that a burst can't meaningfully lift the average. + self._level = self._rate + self._stamp = time.monotonic() + self.consumed = 0 + + @property + def rate(self) -> float: + return self._rate + + def set_rate(self, rate_bytes_per_s: float) -> None: + """Retune live (the UI dial). Waiters re-check immediately, so raising + the cap (or lifting it with 0) unblocks a mid-download wait at once.""" + with self._cond: + self._refill_locked() # settle elapsed time at the OLD rate + self._rate = max(0.0, float(rate_bytes_per_s)) + self._level = min(self._level, self._rate) + self._cond.notify_all() + + def _refill_locked(self) -> None: + now = time.monotonic() + self._level = min(self._rate, self._level + (now - self._stamp) * self._rate) + self._stamp = now + + def take(self, n: int) -> None: + """Charge n bytes and block until the budget recovers (stills path).""" + with self._cond: + self.consumed += n + if self._rate <= 0: + return + self._refill_locked() + self._level -= n + while self._level < 0: + # Wake early on set_rate; cap the wait so a big debt is paid in + # re-checked slices rather than one long uninterruptible sleep. + self._cond.wait(min(-self._level / self._rate, 0.5)) + if self._rate <= 0: + return + self._refill_locked() + + def charge(self, n: int) -> float: + """Charge n bytes WITHOUT blocking; return seconds of debt (0 = within + budget). The ffmpeg governor can't block the subprocess's own reads, so + it SIGSTOPs the process for (about) the returned debt instead.""" + with self._cond: + self.consumed += n + if self._rate <= 0: + return 0.0 + self._refill_locked() + self._level -= n + return max(0.0, -self._level / self._rate) + + +class PidReadMeter: + """Cumulative read-bytes meter for a subprocess, via /proc//io. + + `rchar` counts every read() syscall's bytes — for a streaming ffmpeg the + network reads dominate, so the delta is a good-enough aggregate-bandwidth + signal (it's a governor, not a billing meter). Returns None when /proc is + unavailable (process exited, or a non-Linux host): the caller then simply + doesn't govern — degrade to unthrottled rather than break video sampling. + """ + + def __init__(self, pid: int): + self._path = f"/proc/{pid}/io" + self._last = 0 + + def delta(self) -> int | None: + try: + with open(self._path, "rb") as f: + for line in f: + if line.startswith(b"rchar:"): + total = int(line.split()[1]) + d, self._last = total - self._last, total + return max(0, d) + except (OSError, ValueError): + return None + return None diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 3544797..4554703 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -42,6 +42,7 @@ from .client import FcClient from .config import Config from .crops import crop_region from .detectors import dedupe_crops +from .throttle import TokenBucket # Cap on the lease-retry backoff: when curator is unreachable (e.g. you redeploy # it while away), a downloader retries leasing with exponential backoff up to this @@ -181,6 +182,10 @@ class Worker: f"Authorization: Bearer {cfg.token}\r\n" if cfg.token else "" ) self._lock = threading.Lock() + # ONE bandwidth budget for everything the agent pulls (still downloads + # and ffmpeg video streams): the agent shares a desktop's network, so + # the polite bound is on the aggregate — see throttle.py for why. + self.throttle = TokenBucket(cfg.bandwidth_limit_mb_s * 1_048_576) self._running = False # The lifecycle state the UI shows (see the STOPPED/STARTING/... consts). # It stays STOPPING — a truthful "winding down" — from the Stop press until @@ -219,6 +224,7 @@ class Worker: # often the browser polls (see RATE_INTERVAL). Decay to 0 when work stops. self._jpm = 0.0 self._dpm = 0.0 + self._net_mb_s = 0.0 # smoothed aggregate download rate (UI readout) 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 @@ -382,6 +388,7 @@ class Worker: this here (not in the browser) makes the numbers independent of the poll rate, so a throttled/unfocused tab still shows a real rate.""" prev_p, prev_d, prev_t = self.processed, self.downloaded, time.monotonic() + prev_b = self.throttle.consumed while True: time.sleep(RATE_INTERVAL) now = time.monotonic() @@ -389,9 +396,12 @@ class Worker: if dt > 0: jp = max(0.0, 60.0 * (self.processed - prev_p) / dt) dp = max(0.0, 60.0 * (self.downloaded - prev_d) / dt) + nb = max(0.0, (self.throttle.consumed - prev_b) / dt / 1_048_576) self._jpm = RATE_ALPHA * jp + (1 - RATE_ALPHA) * self._jpm self._dpm = RATE_ALPHA * dp + (1 - RATE_ALPHA) * self._dpm + self._net_mb_s = RATE_ALPHA * nb + (1 - RATE_ALPHA) * self._net_mb_s prev_p, prev_d, prev_t = self.processed, self.downloaded, now + prev_b = self.throttle.consumed # --- control ----------------------------------------------------------- def start(self): @@ -494,6 +504,14 @@ class Worker: with self._lock: self._auto = bool(on) + def set_bandwidth(self, mb_s: float): + # Live-retunes the shared bucket; a downloader blocked mid-wait re-checks + # immediately (set_rate notifies), so raising the cap takes effect now. + self.throttle.set_rate(max(0.0, float(mb_s)) * 1_048_576) + log.info("bandwidth cap set to %s", + "unlimited" if self.throttle.rate <= 0 + else f"{self.throttle.rate / 1_048_576:g} MB/s") + def set_concurrency(self, n: int): # The UI dial tunes the DOWNLOADER count. A manual set is an override → # leave Auto so the autoscaler stops fighting the operator. @@ -564,6 +582,8 @@ class Worker: "downloads_per_min": round(self._dpm, 1), "errors": self.errors, "transient": self.transient, + "bandwidth_limit_mb_s": round(self.throttle.rate / 1_048_576, 1), + "net_mb_s": round(self._net_mb_s, 1), # observed aggregate rate } def _bump(self, *, processed=0, downloaded=0, errors=0, active=0, transient=0): @@ -682,6 +702,7 @@ class Worker: job.get("max_frames", 64), headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout, should_stop=lambda: self._stopped(stop_evt), + governor=self.throttle, ) if not frames: # Stop killed ffmpeg → NOT the job's fault; raise transient so the @@ -709,7 +730,7 @@ class Worker: return frames # Stills: download the bytes and decode. with self._timed("download"): - data = self.client.fetch_image(job["image_url"]) + data = self.client.fetch_image(job["image_url"], throttle=self.throttle) with self._timed("decode"): frames = [(None, media.load_image(data))] return frames diff --git a/tests/test_agent_throttle.py b/tests/test_agent_throttle.py new file mode 100644 index 0000000..73ab0e1 --- /dev/null +++ b/tests/test_agent_throttle.py @@ -0,0 +1,71 @@ +"""Agent bandwidth governor (TokenBucket + PidReadMeter) — pure stdlib, so it +runs in the unit lane even though the rest of the agent (torch/ultralytics) +can't be imported in CI. Timing asserts use generous margins: they check the +throttle's ORDER of magnitude, not scheduler precision.""" +import os +import threading +import time + +from agent.fc_agent.throttle import PidReadMeter, TokenBucket + + +def test_unlimited_never_blocks_and_counts(): + b = TokenBucket(0) + t0 = time.monotonic() + for _ in range(100): + b.take(10_000_000) + assert time.monotonic() - t0 < 0.2 + assert b.consumed == 1_000_000_000 + + +def test_take_enforces_average_rate(): + # 400 KB/s with a 1s burst: the first 400KB is free (burst), the next + # 200KB is 0.5s of debt the caller must wait out. + b = TokenBucket(400_000) + b.take(400_000) + t0 = time.monotonic() + b.take(200_000) + elapsed = time.monotonic() - t0 + assert 0.35 <= elapsed < 3.0 + + +def test_set_rate_zero_unblocks_a_waiter(): + b = TokenBucket(1_000) # 1 KB/s → a 1MB take would wait ~17 min + started = threading.Event() + + def blocked_take(): + started.set() + b.take(1_000_000) + + th = threading.Thread(target=blocked_take, daemon=True) + th.start() + started.wait(1.0) + time.sleep(0.1) # let it enter the wait loop + b.set_rate(0) # lift the cap live (the UI dial) + th.join(timeout=2.0) + assert not th.is_alive() + + +def test_charge_reports_debt_without_blocking(): + b = TokenBucket(1_000) + t0 = time.monotonic() + assert b.charge(1_000) == 0.0 # covered by the 1s burst + debt = b.charge(2_000) # 2s over budget → ~2s of debt + assert time.monotonic() - t0 < 0.2 + assert 1.5 <= debt <= 2.1 + assert b.consumed == 3_000 + + +def test_pid_read_meter_tracks_own_reads(): + m = PidReadMeter(os.getpid()) + first = m.delta() # cumulative rchar since process start + assert first is not None and first > 0 + with open("/dev/zero", "rb") as f: + f.read(1_048_576) + grown = m.delta() + assert grown is not None and grown >= 1_048_576 + + +def test_pid_read_meter_missing_pid_degrades_to_none(): + # PID 2^22+ is above the default pid_max — guaranteed absent. + assert PidReadMeter(2**22 + 12345).delta() is None