Merge pull request 'Agent: server-side throughput rates + killable-on-stop ffmpeg' (#183) from dev into main

2026-07-01 20:01:00 -04:00
parent ed20df905b 713a11e394
commit 489e6aaaee
3 changed files with 91 additions and 40 deletions
@@ -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-01.8 · real start/stop state machine (no more stuck 'stopping')"
+VERSION = "2026-07-01.9 · server-computed jobs/min + downloads/min (poll-rate independent)"
 logbuf.install()
 cfg = Config.from_env()
@@ -252,11 +252,6 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 <script>
 const PAGE_BUILD="__BUILD__"
 let CAP=8
 // Smoothed derived rates (jobs/min, downloads/min) + the deltas they're built
 // from. The raw buffer/on-GPU/downloader counts change many times a second, so
 // a poll only ever samples noise; these EWMA the two monotonic counters into a
 // stable, readable throughput instead. Reset to null on load → re-converge.
 let jpmv=null, dpmv=null, lastP=null, lastD=null, lastRateT=null
 // Optimistic transitional state on click, then apply the POST's own status
 // response (it returns worker.status()) for instant feedback — don't wait on the
 // separate /status poll, which can lag behind the curator queue call.
@@ -308,23 +303,12 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  // backend truthfully lands on "stopped".
  startbtn.disabled=(st!=='stopped')
  stopbtn.disabled=!(running||st==='starting')
-  // Derived rates over the poll deltas — j/min ≈ GPU throughput, dl/min ≈ fetch
+  // Throughput rates arrive READY from the backend (jobs/min ≈ GPU throughput,
-  // throughput. EWMA-smoothed so a lull between jobs doesn't blank them; clamped
+  // dl/min ≈ fetch throughput), computed there on a fixed cadence — so they show
-  // at 0 so an agent restart (counters reset to 0 → negative delta) can't spike
+  // a real number no matter how often this tab polls (a backgrounded tab throttles
-  // them; the dt<30s guard drops a backgrounded-tab gap that would read as huge.
+  // its timers, which used to leave a client-side delta-rate blank forever).
-  const now=performance.now()
+  jpm.textContent=(s.jobs_per_min!=null)?Math.round(s.jobs_per_min):'—'
-  if(lastRateT!=null && s.processed!=null){
+  dpm.textContent=(s.downloads_per_min!=null)?Math.round(s.downloads_per_min):'—'
   const dt=(now-lastRateT)/1000
   if(dt>0.5 && dt<30){
    const jp=Math.max(0,60*(s.processed-lastP)/dt)
    const dp=Math.max(0,60*((s.downloaded||0)-lastD)/dt)
    jpmv=(jpmv==null)?jp:0.4*jp+0.6*jpmv
    dpmv=(dpmv==null)?dp:0.4*dp+0.6*dpmv
   }
  }
  lastP=s.processed; lastD=(s.downloaded||0); lastRateT=now
  jpm.textContent=(jpmv==null)?'—':Math.round(jpmv)
  dpm.textContent=(dpmv==null)?'—':Math.round(dpmv)
  done.textContent=s.processed
  err.textContent=s.errors; err.className='n'+(s.errors>0?' warn':'')
  waited.textContent=s.transient||0
@@ -5,6 +5,7 @@ import io
 import os
 import subprocess
 import tempfile
 import time
 from PIL import Image, ImageFile
@@ -105,28 +106,57 @@ def _collect_frames(
    return out
 def _terminate(proc: subprocess.Popen) -> None:
    """Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM."""
    proc.terminate()
    try:
        proc.wait(timeout=2)
    except subprocess.TimeoutExpired:
        proc.kill()
        try:
            proc.wait(timeout=2)
        except subprocess.TimeoutExpired:
            pass
 def sample_frames_from_url(
    url: str, interval_seconds: float, max_frames: int,
-    *, headers: str = "", timeout: float = 1200.0,
+    *, headers: str = "", timeout: float = 1200.0, should_stop=None,
 ) -> list[tuple[float, Image.Image]]:
    """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
    downloads the whole file (VR/4K originals run 800MB+ and would buffer ~1GB in
    RAM and get cut off mid-download). Reconnect flags resume a dropped transfer;
-    the generous timeout lets it finish however slow the link is — completion over
+    the timeout is the per-video ceiling (a slow/reconnecting stream can otherwise
-    speed, and environment-agnostic (any HTTP+Range source). Empty on failure."""
+    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. Empty on failure / stop / timeout."""
    interval = max(0.5, float(interval_seconds or 4.0))
    cap = max(1, int(max_frames or 64))
    hdr = ["-headers", headers] if headers else []
    with tempfile.TemporaryDirectory() as tmp:
        pattern = os.path.join(tmp, "f_%05d.jpg")
        cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
               "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
               "-q:v", "3", pattern]
        try:
-            subprocess.run(
+            proc = subprocess.Popen(
-                ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
+                cmd, stdin=subprocess.DEVNULL,
-                 "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                 "-q:v", "3", pattern],
                check=True, timeout=timeout,
            )
-        except (subprocess.SubprocessError, FileNotFoundError):
+        except (OSError, ValueError):
            return []
        # 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()
        while True:
            try:
                proc.wait(timeout=0.5)
                break
            except subprocess.TimeoutExpired:
                if (should_stop and should_stop()) or (time.monotonic() - start > timeout):
                    _terminate(proc)
                    return []
        if proc.returncode != 0:
            return []
        return _collect_frames(tmp, interval, cap)
@@ -133,6 +133,14 @@ STATS_INTERVAL = 30.0
 QUEUE_POLL_INTERVAL = 5.0
 UI_IDLE_GRACE = 20.0
 # Throughput rates (jobs/min, downloads/min) are computed HERE, on a fixed
 # wall-clock cadence, and reported ready-to-show — NOT derived in the browser
 # from poll deltas. A background/unfocused tab throttles its timers to ~1/min,
 # which made a client-side delta-rate reject every sample and read blank forever;
 # a server-side EWMA is immune to how often (or seldom) the UI actually polls.
 RATE_INTERVAL = 3.0
 RATE_ALPHA = 0.3
 # Stop backstop: pressing Stop signals every worker thread to wind down and hand
 # its lease back; the UI shows "stopping" until those threads have actually
 # exited, then "stopped" — so the state the operator sees is always truthful.
@@ -185,14 +193,17 @@ class Worker:
        self._held: set[int] = set()
        self._held_lock = threading.Lock()
        self.processed = 0
-        self.downloaded = 0       # jobs fetched+decoded into the buffer (monotonic)
+        self.downloaded = 0       # jobs fetched+decoded into the buffer (monotonic);
-        #                           — the UI derives a smoothed downloads/min from it,
+        #                           feeds the server-side downloads/min rate below.
        #                           since the instantaneous buffer/active gauges move
        #                           faster than any sane poll can show.
        self.errors = 0
        self.transient = 0        # jobs handed back due to a server outage (NOT
        #                           failed) — the "waiting out curator" counter
        self._active = 0          # jobs currently mid-GPU (consumers busy)
        # Smoothed throughput rates the UI just displays (jobs/min, downloads/min),
        # computed on a fixed cadence by _rate_loop so they're independent of how
        # often the browser polls (see RATE_INTERVAL). Decay to 0 when work stops.
        self._jpm = 0.0
        self._dpm = 0.0
        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
@@ -206,6 +217,7 @@ class Worker:
        self._timing: dict[str, list[float]] = {}
        self._timing_lock = threading.Lock()
        threading.Thread(target=self._stats_loop, daemon=True).start()
        threading.Thread(target=self._rate_loop, daemon=True).start()
        # The crop embedder (SigLIP-family) and region proposers are built lazily
        # on the first job that needs them and SHARED across all consumers — one
        # instance, so a 2nd consumer adds concurrent inference, not N× VRAM.
@@ -332,6 +344,24 @@ class Worker:
            log.info("timing/%ds — %s (%d jobs)",
                     int(STATS_INTERVAL), " · ".join(parts), jobs)
    def _rate_loop(self) -> None:
        """Compute the jobs/min + downloads/min the UI shows, on a fixed cadence
        from the monotonic counters — EWMA-smoothed, clamped so a counter reset
        (agent restart) can't spike them, decaying to 0 when work stops. Doing
        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()
        while True:
            time.sleep(RATE_INTERVAL)
            now = time.monotonic()
            dt = now - prev_t
            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)
                self._jpm = RATE_ALPHA * jp + (1 - RATE_ALPHA) * self._jpm
                self._dpm = RATE_ALPHA * dp + (1 - RATE_ALPHA) * self._dpm
            prev_p, prev_d, prev_t = self.processed, self.downloaded, now
    # --- control -----------------------------------------------------------
    def start(self):
        # Only a fully-stopped worker can start — ignore a click while already
@@ -499,6 +529,8 @@ class Worker:
            "active": self._active,
            "processed": self.processed,
            "downloaded": self.downloaded,
            "jobs_per_min": round(self._jpm, 1),        # ready-to-show throughput
            "downloads_per_min": round(self._dpm, 1),
            "errors": self.errors,
            "transient": self.transient,
        }
@@ -544,7 +576,7 @@ class Worker:
                if self._stopped(stop_evt):
                    break
                try:
-                    frames = self._download_decode(job)
+                    frames = self._download_decode(job, stop_evt)
                except requests.RequestException as exc:
                    owned.remove(jid)
                    if _is_transient(exc):
@@ -590,7 +622,7 @@ class Worker:
                continue
        return False
-    def _download_decode(self, job: dict):
+    def _download_decode(self, job: dict, stop_evt: threading.Event):
        """Fetch the image bytes and decode → [(frame_time, PIL.Image)]. Videos
        are sampled into frames (ffmpeg). Records the download + decode timings."""
        if media.is_video(job.get("mime", "")):
@@ -604,10 +636,15 @@ class Worker:
                    url, job.get("frame_interval_seconds", 4.0),
                    job.get("max_frames", 64),
                    headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
                    should_stop=lambda: self._stopped(stop_evt),
                )
            if not frames:
-                # Couldn't sample. If curator is up, the file is unprocessable →
+                # Stop killed ffmpeg → NOT the job's fault; raise transient so the
-                # a job fault (fail it, don't re-lease forever). If curator is
+                # downloader releases (not fails) it as it winds down.
                if self._stopped(stop_evt):
                    raise requests.ConnectionError("stopped during video sampling")
                # Else couldn't sample. If curator is up, the file is unprocessable
                # → a job fault (fail it, don't re-lease forever). If curator is
                # unreachable, it's transient → let the loop back off + retry
                # (survives a redeploy). ConnectionError is caught as transient.
                if self.client.is_reachable():