diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 2169ae7..ba787e1 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -70,6 +70,11 @@ TPUT_ALPHA = 0.5 # throughput EWMA weight TPUT_MARGIN = 0.08 # a grow must lift smoothed jobs/s by this to "help" REPROBE_TICKS = 8 # decisions to hold after settling before re-probing +# How often to log the per-stage timing breakdown (lease/download/decode/gpu/ +# submit) so the operator can see where a job's wall-clock actually goes — the +# data that decides whether a download/compute split is worth building. +STATS_INTERVAL = 30.0 + log = logging.getLogger("fc_agent.worker") @@ -106,6 +111,12 @@ class Worker: self._queue: dict | None = None self._queue_thread = threading.Thread(target=self._queue_poll_loop, daemon=True) self._queue_thread.start() + # Per-stage timing: stage -> [sum_seconds, count], summarised to the log + # every STATS_INTERVAL so we can see where wall-clock goes per job. + self._timing: dict[str, list[float]] = {} + self._timing_lock = threading.Lock() + self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True) + self._stats_thread.start() # The crop embedder (SigLIP-family) is built lazily on the first job that # needs it, from the model the server announces — one shared instance. self._embedder = None @@ -131,6 +142,41 @@ class Worker: def util_smooth(self) -> float | None: return self._util_smooth + def _record(self, stage: str, seconds: float) -> None: + with self._timing_lock: + s = self._timing.get(stage) + if s is None: + self._timing[stage] = [seconds, 1] + else: + s[0] += seconds + s[1] += 1 + + def _stats_loop(self) -> None: + """Log a per-stage timing breakdown every STATS_INTERVAL (only when there + was work), so the operator can see the download/decode/gpu/submit split.""" + while True: + time.sleep(STATS_INTERVAL) + with self._timing_lock: + snap = {k: (v[0], v[1]) for k, v in self._timing.items() if v[1]} + self._timing = {} + if not snap: + continue + order = ["lease", "download", "decode", "gpu", "submit"] + parts = [ + f"{st} {1000 * snap[st][0] / snap[st][1]:.0f}ms" + for st in order if st in snap + ] + jobs = (snap.get("gpu") or snap.get("download") or (0, 0))[1] + # Per-job wall time across the compute path (lease is per-batch, so + # it's shown separately above, not folded into this figure). + per_job = sum( + snap[st][0] for st in ("download", "decode", "gpu", "submit") + if st in snap + ) + pj_ms = 1000 * per_job / jobs if jobs else 0 + log.info("timing/%ds — %s | wall/job %.0fms (%d jobs)", + int(STATS_INTERVAL), " · ".join(parts), pj_ms, jobs) + # --- control ----------------------------------------------------------- def start(self): with self._lock: @@ -209,7 +255,9 @@ class Worker: backoff = self.cfg.poll_idle_seconds while not slot.stop.is_set() and self._running: try: + _t = time.monotonic() jobs = self.client.lease(self.cfg.batch_size) + self._record("lease", time.monotonic() - _t) backoff = self.cfg.poll_idle_seconds # server answered → reset except Exception: # curator unreachable (redeploy, network drop): wait it out with @@ -367,7 +415,11 @@ class Worker: re-leased rather than fail()ed into its attempt budget.""" self._bump(active=1) try: + _t = time.monotonic() data = self.client.fetch_image(job["image_url"]) + self._record("download", time.monotonic() - _t) + + _t = time.monotonic() if media.is_video(job.get("mime", "")): frames = media.sample_frames( data, job.get("frame_interval_seconds", 4.0), @@ -375,6 +427,7 @@ class Worker: ) or [(None, media.load_image(data))] else: frames = [(None, media.load_image(data))] + self._record("decode", time.monotonic() - _t) task = job.get("task") or "ccip" embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION @@ -388,7 +441,8 @@ class Worker: # new model, #1190) → image_record.siglip_embedding. Mean-pool video # frames, matching the server's tag_and_embed. No regions. if task == "embed": - embedder = self._ensure_embedder(model_name) + embedder = self._ensure_embedder(model_name) # one-time model load + _t = time.monotonic() vecs = [embedder.embed(frame) for _, frame in frames] if len(vecs) > 1: vec = np.mean( @@ -396,7 +450,10 @@ class Worker: ).tolist() else: vec = vecs[0] + self._record("gpu", time.monotonic() - _t) + _t = time.monotonic() self.client.submit_embedding(job["job_id"], vec, embed_version) + self._record("submit", time.monotonic() - _t) self._bump(processed=1) return True @@ -419,6 +476,7 @@ class Worker: ccip_ev = self.cfg.ccip_model or "ccip-default" dv = f"person-{self.cfg.detector_level}" + _t_gpu = time.monotonic() # detect + CCIP + batched embed = "gpu" for t, frame in frames: # FIGURE boxes: imgutils detect_person ∪ general COCO person, # NMS-merged → CCIP identity (+ a concept crop). Covers anime + @@ -473,7 +531,10 @@ class Worker: tmpl["siglip_embedding"] = vec tmpl["embedding_version"] = embed_version regions.append(tmpl) + self._record("gpu", time.monotonic() - _t_gpu) + _t = time.monotonic() self.client.submit(job["job_id"], regions, replace_kinds) + self._record("submit", time.monotonic() - _t) self._bump(processed=1) return True except requests.RequestException as exc: