bvandeusen
a4df279343
The GPU lease now carries the crop-proposer config from MLSettings in a per-job 'detectors' block (same pattern as embed_model_name). The agent's worker builds its Proposers from the announced config via _effective_cfg (lease block overlaid on env) + _proposers_for (rebuilds only when a config signature changes) — so an operator's UI edit takes effect on the next lease with NO restart, and env is now just the bootstrap fallback until the server announces. enabled-off maps to empty weights (proposer skipped); dedupe_iou + max_regions also come from the effective cfg. Test: lease announces the detectors block with the seeded default weights. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
1154 lines
59 KiB
Python
1154 lines
59 KiB
Python
"""The lease → download → detect+embed → submit pipeline.
|
||
|
||
The workload is DOWNLOAD-BOUND (operator timing 2026-07-01: download 400–5462ms,
|
||
GPU ~300–600ms), so a design where each worker runs the whole serial chain leaves
|
||
the fast GPU idle during every download. This splits the chain into a producer/
|
||
consumer pipeline instead:
|
||
|
||
downloader pool (N threads) ── lease→download→decode ──▶ [bounded buffer]
|
||
│
|
||
┌────────────────────┘
|
||
▼
|
||
GPU consumer(s) (1–2) ── detect+embed(batched)→submit
|
||
|
||
* DOWNLOADERS are I/O-bound so many overlap well; the autoscaler tunes their
|
||
count by BUFFER OCCUPANCY — a near-empty buffer means the GPU is starving
|
||
(add downloaders); a near-full buffer means they outpace the GPU (hold/trim,
|
||
or add a 2nd consumer if the GPU has headroom).
|
||
* The BOUNDED BUFFER is backpressure: decoded frames are big, so a full buffer
|
||
blocks downloaders — capping RAM and how far leases run ahead of the GPU.
|
||
* CONSUMERS are GPU-bound and fast, so one usually keeps up; a 2nd is added
|
||
only when the buffer stays full and the GPU has spare util/VRAM.
|
||
* A HEARTBEAT thread keeps every still-held lease alive (buffered jobs wait for
|
||
the GPU and would otherwise hit curator's 180s lease TTL and be reclaimed).
|
||
|
||
Resilience carried over from the slot model: lease exponential backoff (ride out
|
||
a curator redeploy), submit-path retry (client.py — never discard finished GPU
|
||
work on a blip), release-on-stop (hand leases back at once), region caps + video
|
||
early-exit (bound pathological jobs). Stop drains BOTH pools and releases every
|
||
held lease immediately so orphaned work is re-picked without waiting out the TTL.
|
||
"""
|
||
import contextlib
|
||
import logging
|
||
import queue
|
||
import threading
|
||
import time
|
||
|
||
import numpy as np
|
||
import requests
|
||
|
||
from . import media, models
|
||
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
|
||
# many seconds, then resumes within this window once the server is back — no
|
||
# restart needed.
|
||
MAX_BACKOFF_SECONDS = 60.0
|
||
|
||
# Sleep mode (operator 2026-07-02): an EMPTY queue must cost almost nothing —
|
||
# the autoscaler sheds to one downloader and that one backs its lease poll off
|
||
# exponentially to this ceiling. New work is therefore noticed within at most
|
||
# ~15 minutes (accepted trade-off), after which polling and the pool ramp back
|
||
# up on their own.
|
||
IDLE_POLL_MAX_SECONDS = 900.0
|
||
|
||
# A job whose fetch dies transiently this many times IN ONE SESSION stops being
|
||
# handed back and is failed instead. Transient handbacks (release) burn no
|
||
# attempts on the server, so a poisoned transfer — an original that stalls the
|
||
# download every single time — would otherwise release→re-lease forever, churning
|
||
# bandwidth without ever landing in the error queue (operator-observed jobs
|
||
# 99044/125288/131594/143131, 2026-07-01). Failing it lets curator's attempt cap
|
||
# tombstone it WITH the real reason. A genuine curator outage is unaffected:
|
||
# every job takes at most one strike per outage, and strikes clear on success.
|
||
TRANSIENT_JOB_CAP = 3
|
||
|
||
|
||
def _is_transient(exc: requests.RequestException) -> bool:
|
||
"""A server/transport problem (wait it out) vs a job-specific fault (fail it).
|
||
No response → connection refused/timeout → curator is down → transient. With
|
||
a response: 5xx, auth (401/403, e.g. a token blip on redeploy), 408/409/429
|
||
(timeout / our lease reclaimed / rate-limited) are all 'not this job's fault'.
|
||
A specific 4xx like 404 (image gone) / 400 IS the job's fault → fail it."""
|
||
resp = getattr(exc, "response", None)
|
||
if resp is None:
|
||
return True
|
||
return resp.status_code >= 500 or resp.status_code in (401, 403, 408, 409, 429)
|
||
|
||
|
||
def _transient_reason(exc: requests.RequestException) -> str:
|
||
"""A SPECIFIC label for a transient failure — so the log distinguishes a
|
||
stalled/slow transfer from an actually-unreachable curator, which need
|
||
different fixes. `HTTP <code>` for a 5xx/auth/conflict, else the exception
|
||
class: ReadTimeout (transfer stalled >60s between bytes — curator up, this
|
||
file/stream is slow), ConnectTimeout (curator didn't accept in 10s → web
|
||
workers/pool exhausted or down), ConnectionError (reset mid-transfer)."""
|
||
resp = getattr(exc, "response", None)
|
||
if resp is not None:
|
||
return f"HTTP {resp.status_code}"
|
||
return type(exc).__name__
|
||
|
||
|
||
def _ewma(prev: float | None, x: float, alpha: float) -> float:
|
||
"""Exponentially-weighted moving average: seed with the first sample, then
|
||
fold each new sample in at weight `alpha`. Smooths the noisy control signals
|
||
(buffer occupancy, GPU util, throughput) the autoscaler decides on."""
|
||
return x if prev is None else alpha * x + (1 - alpha) * prev
|
||
|
||
|
||
# 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)
|
||
|
||
# Fallbacks only — the server ANNOUNCES the embedding model (name + version) in
|
||
# the lease so the agent stays model-agnostic and in lock-step with the space
|
||
# the heads were trained in. These cover an older server that doesn't send them.
|
||
DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384"
|
||
DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384"
|
||
|
||
# Autoscaler (Auto mode): scale DOWNLOADERS by buffer occupancy — the elegant
|
||
# control signal, since the buffer sits exactly between the two stages. Buffer
|
||
# mostly EMPTY → GPU starving → add downloaders. Buffer mostly FULL → downloaders
|
||
# outpace the GPU → the GPU is the bottleneck: add a 2nd consumer if it has
|
||
# util/VRAM headroom and doing so lifts throughput, else trim a downloader (it's
|
||
# only adding lease pressure). Occupancy + util are EWMA-smoothed (both are noisy
|
||
# tick-to-tick), and decisions are spaced so a move is judged on averaged signals.
|
||
CONTROL_INTERVAL = 2.0 # sampling cadence (seconds)
|
||
SAMPLES_PER_DECISION = 6 # decide ~every 12s on averaged signals
|
||
OCC_ALPHA = 0.3 # buffer-occupancy EWMA weight on the newest sample
|
||
OCC_LOW = 0.25 # below this = buffer starving → add a downloader
|
||
OCC_HIGH = 0.80 # above this = downloaders outpace the GPU
|
||
UTIL_ALPHA = 0.25 # GPU-util EWMA weight
|
||
UTIL_START = 85 # GPU has headroom below this (gate a 2nd consumer)
|
||
# Bandwidth-cap awareness (operator 2026-07-02): with the aggregate governor in
|
||
# place, the occupancy signal alone would peg downloaders at DL_MAX while the
|
||
# CAP — not concurrency — is the real constraint: 8 streams sharing 8 MB/s move
|
||
# no more data than 4, they just hold more leases + RAM and stretch every job's
|
||
# latency. So growth is gated on the pipe having headroom, and a pipe pinned at
|
||
# the cap sheds streams down to BW_MIN_DL (enough overlap to keep the cap
|
||
# filled through TTFB + decode gaps). The dead band between the two thresholds
|
||
# prevents add/trim flapping.
|
||
BW_ADD_HEADROOM = 0.85 # add a downloader only while net < 85% of the cap
|
||
BW_TRIM_AT = 0.95 # net ≥ 95% of the cap → shed toward BW_MIN_DL
|
||
BW_MIN_DL = 3
|
||
VRAM_HI = 0.90 # memory pressure → shed a consumer
|
||
VRAM_GROW_MAX = 0.82 # don't add a consumer past this VRAM
|
||
TPUT_ALPHA = 0.5 # throughput EWMA weight
|
||
TPUT_MARGIN = 0.08 # a consumer add must lift smoothed j/s by this to keep
|
||
|
||
# Keep buffered-but-unprocessed leases alive: they hold curator leases while they
|
||
# wait for the GPU, so heartbeat them well inside curator's 180s lease TTL.
|
||
HEARTBEAT_INTERVAL = 45.0
|
||
|
||
# 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.
|
||
STATS_INTERVAL = 30.0
|
||
|
||
# The queue snapshot exists only to populate the UI's counts, so it's polled
|
||
# lazily — only while a browser is actually watching (a /status hit in the last
|
||
# UI_IDLE_GRACE seconds), and not on a tight loop. The pipeline's own lease/
|
||
# submit calls are the real "is curator up?" signal; nothing polls just to poll.
|
||
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.
|
||
# If a thread wedges (a stuck submit/release to an overloaded curator), NEVER
|
||
# hang in "stopping": declare "stopped" after this many seconds and let the lone
|
||
# straggler finish detached. This is the guarantee the old active>0 UI hack lacked.
|
||
STOPPING_TIMEOUT = 20.0
|
||
|
||
# The lifecycle the operator sees. `_state` is the source of truth the UI reads;
|
||
# `_running` is the low-level flag the worker THREADS gate on (True while
|
||
# starting/running; False from the Stop press onward, so they wind down at once).
|
||
# stopped → starting (spun up, not yet reached curator)
|
||
# → running (leased at least once — curator is answering)
|
||
# → stopping (Stop pressed; threads winding down + handing work back)
|
||
# → stopped (threads exited, or the timeout backstop fired)
|
||
STOPPED, STARTING, RUNNING, STOPPING = "stopped", "starting", "running", "stopping"
|
||
|
||
log = logging.getLogger("fc_agent.worker")
|
||
|
||
|
||
class Worker:
|
||
def __init__(self, cfg: Config):
|
||
self.cfg = cfg
|
||
self.client = FcClient(cfg.fc_url, cfg.token, cfg.agent_id)
|
||
# Sent to ffmpeg-from-URL so video sampling works whether or not a
|
||
# deployment gates /images behind the bearer token (FC's is public).
|
||
self._auth_header = (
|
||
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
|
||
# the worker threads have actually exited (or the timeout backstop fires),
|
||
# then STOPPED. `_running` flips False the instant Stop is pressed.
|
||
self._state = STOPPED
|
||
self._shutdown_thread: threading.Thread | None = None
|
||
self._auto = bool(cfg.auto_scale) # autoscale the downloader count
|
||
self._dl_target = max(1, min(DL_MAX, cfg.concurrency))
|
||
self._consumer_target = 1 # GPU is fast — start with one
|
||
self._dls: list[tuple[threading.Thread, threading.Event]] = []
|
||
self._consumers: list[tuple[threading.Thread, threading.Event]] = []
|
||
self._ctrl_stop = threading.Event()
|
||
self._ctrl_thread: threading.Thread | None = None
|
||
# Decoded jobs waiting for the GPU: (job, frames). Bounded = backpressure.
|
||
self._buffer: queue.Queue = queue.Queue(maxsize=BUFFER_MAX)
|
||
# Every job leased and not yet terminal (submitted / failed / released) is
|
||
# "held" — the heartbeat thread keeps these alive, and stop() releases them
|
||
# all at once. Add on lease, discard on every terminal client call.
|
||
self._held: set[int] = set()
|
||
self._held_lock = threading.Lock()
|
||
# job_id → in-session transient-handback count (guarded by _held_lock).
|
||
# Cleared on success or terminal fail; KEPT across releases — persisting
|
||
# through the release→re-lease cycle is what lets TRANSIENT_JOB_CAP
|
||
# catch a poisoned transfer.
|
||
self._transient_seen: dict[int, int] = {}
|
||
self.processed = 0
|
||
self.downloaded = 0 # jobs fetched+decoded into the buffer (monotonic);
|
||
# feeds the server-side downloads/min rate below.
|
||
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._net_mb_s = 0.0 # smoothed aggregate download rate (UI readout)
|
||
self._bw_capped = False # autoscaler is holding/shedding at the cap (UI)
|
||
self._idle = False # last lease came back empty → sleep mode (UI)
|
||
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
|
||
# stalls the whole status view when curator is busy).
|
||
self._queue: dict | None = None
|
||
self._ui_seen = 0.0 # monotonic time of the last UI /status hit
|
||
threading.Thread(target=self._queue_poll_loop, daemon=True).start()
|
||
threading.Thread(target=self._heartbeat_loop, daemon=True).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()
|
||
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.
|
||
self._embedder = None
|
||
self._embedder_lock = threading.Lock()
|
||
self._proposers = None
|
||
self._proposers_sig = None # detector-config signature the current
|
||
# proposers were built for (#134)
|
||
self._proposers_lock = threading.Lock()
|
||
|
||
# --- held-lease bookkeeping --------------------------------------------
|
||
def _hold(self, job_ids) -> None:
|
||
with self._held_lock:
|
||
self._held.update(job_ids)
|
||
|
||
def _unhold(self, job_id: int) -> None:
|
||
with self._held_lock:
|
||
self._held.discard(job_id)
|
||
|
||
def _strike(self, jid: int) -> int:
|
||
"""Record a transient bounce for this job; returns the in-session total
|
||
(compared against TRANSIENT_JOB_CAP by both the fetch and submit paths)."""
|
||
with self._held_lock:
|
||
n = self._transient_seen.get(jid, 0) + 1
|
||
self._transient_seen[jid] = n
|
||
return n
|
||
|
||
def _forget_strikes(self, jid: int) -> None:
|
||
"""Terminal outcome (submitted or failed) → this job's strike count no
|
||
longer matters. Deliberately NOT called on release: strikes surviving
|
||
the release→re-lease cycle is what makes TRANSIENT_JOB_CAP work."""
|
||
with self._held_lock:
|
||
self._transient_seen.pop(jid, None)
|
||
|
||
def _release(self, job_ids: list[int]) -> None:
|
||
"""Hand still-held leases back to curator and drop them from the held set —
|
||
the single hand-back path for both a downloader exiting (stop/shrink) with
|
||
unbuffered leases and a consumer releasing one job it can't finish. Empty
|
||
list is a no-op; a duplicate release is a harmless server no-op."""
|
||
if not job_ids:
|
||
return
|
||
self.client.release(job_ids)
|
||
for jid in job_ids:
|
||
self._unhold(jid)
|
||
|
||
def _fail(self, jid: int, image_id, exc: Exception, verb: str = "failed") -> None:
|
||
"""Terminal job-fault path: count the error, log it, tell curator the job
|
||
failed, and drop the lease. `verb` lets a caller say 'failed to decode'."""
|
||
self._bump(errors=1)
|
||
log.warning("job %s (image %s) %s: %s", jid, image_id, verb, str(exc)[:200])
|
||
self.client.fail(jid, str(exc)[:500])
|
||
self._unhold(jid)
|
||
self._forget_strikes(jid)
|
||
|
||
def _stopped(self, stop_evt: threading.Event) -> bool:
|
||
"""The shared 'should I bail now?' check — the worker is stopping (global
|
||
flag) or this thread's own stop event is set. Lock-free atomic reads."""
|
||
return not self._running or stop_evt.is_set()
|
||
|
||
def _abort_if_stopped(self, jid: int, stop_evt: threading.Event) -> bool:
|
||
"""If we're stopping, hand this job's lease back (so a Stop drains promptly
|
||
instead of finishing heavy work) and return True so the caller bails."""
|
||
if self._stopped(stop_evt):
|
||
self._release([jid])
|
||
return True
|
||
return False
|
||
|
||
# --- background loops ---------------------------------------------------
|
||
def _heartbeat_loop(self) -> None:
|
||
"""Keep every held lease alive so buffered jobs waiting on the GPU aren't
|
||
reclaimed by curator's 180s TTL. Errors are swallowed by client.heartbeat;
|
||
a reclaimed lease just re-leases elsewhere — never fatal."""
|
||
while True:
|
||
if self._running:
|
||
with self._held_lock:
|
||
ids = list(self._held)
|
||
if ids:
|
||
self.client.heartbeat(ids)
|
||
time.sleep(HEARTBEAT_INTERVAL)
|
||
|
||
def _queue_poll_loop(self):
|
||
"""Refresh the curator queue snapshot so /status is a pure in-memory read
|
||
— but ONLY while the UI is being watched (a recent /status hit). No
|
||
browser open → no polling; the pipeline is curator's only visitor.
|
||
Errors just leave the last snapshot (or None) — never blocks the UI."""
|
||
while True:
|
||
if time.monotonic() - self._ui_seen <= UI_IDLE_GRACE:
|
||
try:
|
||
self._queue = self.client.queue_status()
|
||
except Exception:
|
||
self._queue = None
|
||
time.sleep(QUEUE_POLL_INTERVAL)
|
||
|
||
def note_ui(self) -> None:
|
||
"""The UI polled /status — keep the queue snapshot warm for a while."""
|
||
self._ui_seen = time.monotonic()
|
||
|
||
def latest_queue(self) -> dict | None:
|
||
return self._queue
|
||
|
||
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
|
||
|
||
@contextlib.contextmanager
|
||
def _timed(self, stage: str):
|
||
"""Time a pipeline stage and fold it into the per-stage breakdown — but
|
||
ONLY on a clean exit: an exception (a failed download, a stopped GPU pass)
|
||
propagates through the `yield` and skips the record, so it never skews the
|
||
stage average with work that didn't complete. Matches the old inline
|
||
monotonic()/_record() pairs, where _record ran only if no exception fired."""
|
||
t = time.monotonic()
|
||
yield
|
||
self._record(stage, time.monotonic() - t)
|
||
|
||
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.
|
||
In the pipeline these stages run on DIFFERENT threads concurrently, so the
|
||
figures are per-stage averages, not a single job's serial wall-clock."""
|
||
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]
|
||
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()
|
||
prev_b = self.throttle.consumed
|
||
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)
|
||
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):
|
||
# Only a fully-stopped worker can start — ignore a click while already
|
||
# up or still winding down (the UI also disables the button then). This
|
||
# keeps a Start during "stopping" from racing the shutdown monitor's
|
||
# lease-release against a fresh pool's new leases.
|
||
with self._lock:
|
||
if self._state != STOPPED:
|
||
return
|
||
self._running = True
|
||
self._state = STARTING # → RUNNING on the first successful lease
|
||
self._dl_target = max(1, self._dl_target)
|
||
self._consumer_target = max(1, self._consumer_target)
|
||
self._reconcile_locked()
|
||
log.info("▶ start → starting (spinning up, awaiting curator)")
|
||
# (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 _mark_running(self) -> None:
|
||
"""STARTING → RUNNING once a downloader has actually leased from curator —
|
||
so 'running' means 'curator is answering', not just 'Start was clicked'.
|
||
First caller wins; the rest are a cheap no-op."""
|
||
if self._state != STARTING:
|
||
return
|
||
with self._lock:
|
||
if self._state != STARTING:
|
||
return
|
||
self._state = RUNNING
|
||
log.info("state → running (curator reachable — leasing)")
|
||
|
||
def stop(self):
|
||
# Enter STOPPING and signal every worker thread to wind down. The handler
|
||
# returns at once — the actual wait-for-exit + lease-release runs in a
|
||
# background monitor, so a slow curator can NEVER block the Stop button or
|
||
# wedge the state (the bug: the old UI faked "stopping" from active>0 and
|
||
# sat there forever when a consumer hung mid-release).
|
||
with self._lock:
|
||
if self._state not in (STARTING, RUNNING):
|
||
return # already stopping/stopped
|
||
self._state = STOPPING
|
||
self._running = False # every thread's _stopped() now trips → wind down
|
||
dls, self._dls = self._dls, []
|
||
cons, self._consumers = self._consumers, []
|
||
self._active = 0
|
||
self._ctrl_stop.set()
|
||
for _, ev in dls:
|
||
ev.set()
|
||
for _, ev in cons:
|
||
ev.set()
|
||
# Wake any consumer blocked on an empty buffer so it sees the stop at once.
|
||
for _ in range(CONSUMER_MAX):
|
||
try:
|
||
self._buffer.put_nowait(None)
|
||
except queue.Full:
|
||
break
|
||
log.info("■ stop → stopping (winding down, handing work back)")
|
||
threads = [t for t, _ in dls] + [t for t, _ in cons]
|
||
self._shutdown_thread = threading.Thread(
|
||
target=self._finish_stop, args=(threads,), daemon=True)
|
||
self._shutdown_thread.start()
|
||
|
||
def _finish_stop(self, threads: list[threading.Thread]) -> None:
|
||
"""Wait for the signalled worker threads to actually exit, then land on
|
||
STOPPED — so 'stopped' is truthful. Bounded by STOPPING_TIMEOUT so a
|
||
wedged submit/release can never hold the UI in 'stopping' forever."""
|
||
deadline = time.monotonic() + STOPPING_TIMEOUT
|
||
for t in threads:
|
||
t.join(timeout=max(0.0, deadline - time.monotonic()))
|
||
# Hand back every lease still held in one shot (background — no handler
|
||
# block); a straggler releasing its own job later is a harmless dup.
|
||
self._drain_and_release()
|
||
with self._lock:
|
||
if self._state == STOPPING: # a Start may have re-entered; if so, leave it
|
||
self._state = STOPPED
|
||
self._active = 0 # any straggler's -1 is clamped (see _bump)
|
||
stragglers = sum(1 for t in threads if t.is_alive())
|
||
if stragglers:
|
||
log.info("state → stopped (%d thread(s) still winding down past %ds — detached)",
|
||
stragglers, int(STOPPING_TIMEOUT))
|
||
else:
|
||
log.info("state → stopped (all work handed back)")
|
||
|
||
def _drain_and_release(self) -> None:
|
||
while True:
|
||
try:
|
||
self._buffer.get_nowait()
|
||
except queue.Empty:
|
||
break
|
||
with self._held_lock:
|
||
ids = list(self._held)
|
||
self._held.clear()
|
||
if ids:
|
||
self.client.release(ids)
|
||
|
||
def set_auto(self, on: bool):
|
||
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.
|
||
with self._lock:
|
||
self._auto = False
|
||
self._dl_target = max(1, min(DL_MAX, int(n)))
|
||
if self._running:
|
||
self._reconcile_locked()
|
||
|
||
def _apply_downloaders(self, delta: int) -> bool:
|
||
with self._lock:
|
||
new = max(1, min(DL_MAX, self._dl_target + delta))
|
||
if new == self._dl_target:
|
||
return False
|
||
self._dl_target = new
|
||
if self._running:
|
||
self._reconcile_locked()
|
||
return True
|
||
|
||
def _apply_consumers(self, delta: int) -> bool:
|
||
with self._lock:
|
||
new = max(1, min(CONSUMER_MAX, self._consumer_target + delta))
|
||
if new == self._consumer_target:
|
||
return False
|
||
self._consumer_target = new
|
||
if self._running:
|
||
self._reconcile_locked()
|
||
return True
|
||
|
||
def _reconcile_locked(self):
|
||
"""Bring both thread pools to their target counts. New threads start; a
|
||
shrink sets a thread's stop event (it exits after its current iteration,
|
||
releasing any lease it still owns)."""
|
||
while len(self._dls) < self._dl_target:
|
||
ev = threading.Event()
|
||
th = threading.Thread(target=self._downloader, args=(ev,), daemon=True)
|
||
self._dls.append((th, ev))
|
||
th.start()
|
||
while len(self._dls) > self._dl_target:
|
||
_, ev = self._dls.pop()
|
||
ev.set()
|
||
while len(self._consumers) < self._consumer_target:
|
||
ev = threading.Event()
|
||
th = threading.Thread(target=self._consumer, args=(ev,), daemon=True)
|
||
self._consumers.append((th, ev))
|
||
th.start()
|
||
while len(self._consumers) > self._consumer_target:
|
||
_, ev = self._consumers.pop()
|
||
ev.set()
|
||
|
||
def status(self) -> dict:
|
||
# Lock-free on purpose: these are plain int / bool / len reads (atomic
|
||
# under the GIL) and this backs the UI poll — it must NEVER be able to
|
||
# block behind a thread holding _lock, or the whole status view freezes.
|
||
return {
|
||
"state": self._state, # stopped|starting|running|stopping (truthful)
|
||
"concurrency": self._dl_target, # the UI dial = downloader count
|
||
"max_concurrency": DL_MAX,
|
||
"auto": self._auto,
|
||
"downloaders": len(self._dls),
|
||
"consumers": len(self._consumers),
|
||
"buffer": self._buffer.qsize(),
|
||
"buffer_max": BUFFER_MAX,
|
||
"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,
|
||
"bandwidth_limit_mb_s": round(self.throttle.rate / 1_048_576, 1),
|
||
"net_mb_s": round(self._net_mb_s, 1), # observed aggregate rate
|
||
"bw_capped": self._bw_capped, # autoscaler holding at the cap (UI hint)
|
||
"idle": self._idle, # queue empty → poll backed off (UI hint)
|
||
}
|
||
|
||
def _bump(self, *, processed=0, downloaded=0, errors=0, active=0, transient=0):
|
||
with self._lock:
|
||
self.processed += processed
|
||
self.downloaded += downloaded
|
||
self.errors += errors
|
||
self.transient += transient
|
||
# Clamp at 0: a Stop resets _active to 0, so a consumer that was
|
||
# mid-image decrements afterwards — that must not go negative.
|
||
self._active = max(0, self._active + active)
|
||
|
||
# --- downloader pool ---------------------------------------------------
|
||
def _downloader(self, stop_evt: threading.Event):
|
||
"""Lease a batch, download + decode each job, and hand it to the GPU
|
||
consumers via the bounded buffer. Owns its leases until they're buffered;
|
||
on any exit path it releases whatever it still owns so nothing is stranded
|
||
holding a lease."""
|
||
backoff = self.cfg.poll_idle_seconds
|
||
idle_wait = self.cfg.poll_idle_seconds
|
||
while not self._stopped(stop_evt):
|
||
try:
|
||
with self._timed("lease"):
|
||
jobs = self.client.lease(self.cfg.batch_size)
|
||
backoff = self.cfg.poll_idle_seconds # server answered → reset
|
||
self._mark_running() # curator answered → leave "starting"
|
||
except Exception:
|
||
# curator unreachable (redeploy, network drop): wait it out with
|
||
# exponential backoff, capped — resume on our own when it returns.
|
||
if stop_evt.wait(backoff):
|
||
break
|
||
backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
|
||
continue
|
||
if not jobs:
|
||
# Empty queue → sleep mode (operator 2026-07-02): back the lease
|
||
# poll off exponentially to IDLE_POLL_MAX_SECONDS instead of
|
||
# chattering every 10s all night. The autoscaler reads _idle and
|
||
# sheds to ONE polling downloader; work appearing resets both
|
||
# within one wake-up (≤15 min latency, operator-accepted).
|
||
self._idle = True
|
||
if stop_evt.wait(idle_wait):
|
||
break
|
||
idle_wait = min(idle_wait * 2, IDLE_POLL_MAX_SECONDS)
|
||
continue
|
||
self._idle = False
|
||
idle_wait = self.cfg.poll_idle_seconds
|
||
self._hold(j["job_id"] for j in jobs)
|
||
owned = [j["job_id"] for j in jobs] # released on any early exit
|
||
for job in jobs:
|
||
jid = job["job_id"]
|
||
if self._stopped(stop_evt):
|
||
break
|
||
try:
|
||
frames = self._download_decode(job, stop_evt)
|
||
except requests.RequestException as exc:
|
||
owned.remove(jid)
|
||
if _is_transient(exc):
|
||
# A Stop mid-fetch lands here too (a killed ffmpeg is
|
||
# not the job's fault) — no strike for that; only count
|
||
# bounces taken while genuinely running.
|
||
if (not self._stopped(stop_evt)
|
||
and self._strike(jid) >= TRANSIENT_JOB_CAP):
|
||
# THIS job keeps dying while others move: a poisoned
|
||
# transfer, not a curator outage. Fail it so the
|
||
# server's attempt cap tombstones it with the real
|
||
# reason instead of cycling it forever.
|
||
self._fail(
|
||
jid, job.get("image_id"), exc,
|
||
verb="gave up after repeated transient failures",
|
||
)
|
||
continue
|
||
# curator down/redeploying or our lease was reclaimed —
|
||
# NOT the job's fault. Hand back this job + the rest of the
|
||
# batch and back the whole loop off.
|
||
self._bump(transient=1)
|
||
self._release([jid])
|
||
# Name the ACTUAL failure — "curator unreachable" was
|
||
# printed for every transient, hiding whether a single
|
||
# file's transfer stalled (ReadTimeout, curator fine) or
|
||
# curator itself is down (ConnectTimeout/ConnectionError).
|
||
log.info("fetch failed job %s (image %s, %s) — released, backing off",
|
||
jid, job.get("image_id"), _transient_reason(exc))
|
||
self._release(owned)
|
||
owned = []
|
||
if not stop_evt.wait(backoff):
|
||
backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
|
||
break
|
||
# a job-specific HTTP fault (404 image gone, 400) → fail it
|
||
self._fail(jid, job.get("image_id"), exc)
|
||
continue
|
||
except Exception as exc: # noqa: BLE001 — bad media → the job's fault
|
||
owned.remove(jid)
|
||
self._fail(jid, job.get("image_id"), exc, verb="failed to decode")
|
||
continue
|
||
# Blocks on a full buffer (backpressure) but wakes promptly on stop.
|
||
if self._put((job, frames), stop_evt):
|
||
self._bump(downloaded=1) # fetched+decoded into the buffer
|
||
owned.remove(jid) # ownership handed to the buffer/consumer
|
||
else:
|
||
break # stopped while waiting for buffer space
|
||
self._release(owned)
|
||
|
||
def _put(self, item, stop_evt: threading.Event) -> bool:
|
||
"""Push onto the bounded buffer, blocking while it's full but rechecking
|
||
stop so a shrink/Stop can't hang a full-buffer window. False = stopped."""
|
||
while not self._stopped(stop_evt):
|
||
try:
|
||
self._buffer.put(item, timeout=0.5)
|
||
return True
|
||
except queue.Full:
|
||
continue
|
||
return False
|
||
|
||
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", "")):
|
||
# Stream the video: ffmpeg reads the media URL directly and Range-reads
|
||
# only the frames it needs, so we NEVER pull the whole file (VR/4K
|
||
# originals are 800MB+ — buffering that in RAM and getting cut off
|
||
# mid-download was the failure loop). Environment-agnostic + resilient.
|
||
url = f"{self.cfg.fc_url}{job['image_url']}"
|
||
with self._timed("decode"):
|
||
frames, ffmpeg_err = media.sample_frames_from_url(
|
||
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),
|
||
governor=self.throttle,
|
||
)
|
||
if not frames:
|
||
# Stop killed ffmpeg → NOT the job's fault; raise transient so the
|
||
# 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 WITH ffmpeg's reason, so the job's stored
|
||
# error says e.g. "moov atom not found" instead of a bare
|
||
# "unprocessable". If curator is unreachable, it's transient → let
|
||
# the loop back off + retry (ConnectionError is caught as such).
|
||
if self.client.is_reachable():
|
||
raise RuntimeError(
|
||
f"no frames sampled from video — {ffmpeg_err or 'unknown reason'}"
|
||
)
|
||
raise requests.ConnectionError("curator unreachable during video sampling")
|
||
# Temporal dedup: a near-static video re-runs the whole detect+embed
|
||
# chain on ~identical frames — drop near-dups HERE (CPU) pre-GPU.
|
||
if len(frames) > 1 and self.cfg.frame_dedupe_distance > 0:
|
||
kept = media.dedupe_frames(frames, self.cfg.frame_dedupe_distance)
|
||
if len(kept) < len(frames):
|
||
log.info("job %s: video frames %d→%d (near-dup dedup)",
|
||
job.get("job_id"), len(frames), len(kept))
|
||
frames = kept
|
||
return frames
|
||
# Stills: download the bytes and decode.
|
||
with self._timed("download"):
|
||
data = self.client.fetch_image(job["image_url"], throttle=self.throttle)
|
||
with self._timed("decode"):
|
||
frames = [(None, media.load_image(data))]
|
||
return frames
|
||
|
||
# --- GPU consumer pool -------------------------------------------------
|
||
def _consumer(self, stop_evt: threading.Event):
|
||
"""Pull decoded jobs off the buffer and run detect + embed + submit."""
|
||
while not self._stopped(stop_evt):
|
||
try:
|
||
item = self._buffer.get(timeout=1.0)
|
||
except queue.Empty:
|
||
continue
|
||
if item is None: # stop sentinel
|
||
continue
|
||
job, frames = item
|
||
if self._abort_if_stopped(job["job_id"], stop_evt):
|
||
continue
|
||
self._bump(active=1)
|
||
try:
|
||
if self._consume(job, frames, stop_evt):
|
||
self._bump(processed=1)
|
||
finally:
|
||
self._bump(active=-1)
|
||
|
||
def _ensure_embedder(self, model_name: str):
|
||
if self._embedder is not None:
|
||
return self._embedder
|
||
with self._embedder_lock:
|
||
if self._embedder is None:
|
||
from .embedder import CropEmbedder
|
||
self._embedder = CropEmbedder(model_name, self.cfg.embed_dtype)
|
||
return self._embedder
|
||
|
||
def _effective_cfg(self, det: dict | None):
|
||
"""The detector config for a job: the LEASE-ANNOUNCED block (the DB/UI
|
||
source of truth, #134) overlaid on the env cfg, or the env cfg unchanged
|
||
when the server announced nothing (bootstrap / older backend). An enabled
|
||
toggle off maps to empty weights — the 'weights == "" → proposer skipped'
|
||
contract in detectors.py. detector_level + ccip_model stay env (separate
|
||
settings, not part of the crop-proposer config)."""
|
||
if not det:
|
||
return self.cfg
|
||
|
||
def _w(p: str) -> str:
|
||
slot = det.get(p) or {}
|
||
return (slot.get("weights") or "") if slot.get("enabled", True) else ""
|
||
|
||
from dataclasses import replace
|
||
return replace(
|
||
self.cfg,
|
||
person_weights=_w("person"),
|
||
person_conf=float(det["person"]["conf"]),
|
||
anatomy_weights=_w("anatomy"),
|
||
anatomy_conf=float(det["anatomy"]["conf"]),
|
||
panel_weights=_w("panel"),
|
||
panel_conf=float(det["panel"]["conf"]),
|
||
max_figures=int(det["max_figures"]),
|
||
max_components=int(det["max_components"]),
|
||
max_panels=int(det["max_panels"]),
|
||
max_regions=int(det["max_regions"]),
|
||
dedupe_iou=float(det["dedupe_iou"]),
|
||
)
|
||
|
||
def _proposers_for(self, eff):
|
||
"""Proposers built from the effective detector config, rebuilt ONLY when
|
||
that config changes — so an operator's UI edit takes effect on the next
|
||
lease with no restart, and steady-state reuses the loaded YOLO models."""
|
||
sig = (
|
||
eff.person_weights, eff.person_conf,
|
||
eff.anatomy_weights, eff.anatomy_conf,
|
||
eff.panel_weights, eff.panel_conf,
|
||
eff.max_figures, eff.max_components, eff.max_panels,
|
||
)
|
||
with self._proposers_lock:
|
||
if self._proposers is None or self._proposers_sig != sig:
|
||
from .detectors import Proposers
|
||
self._proposers = Proposers(eff)
|
||
self._proposers_sig = sig
|
||
return self._proposers
|
||
|
||
def _consume(self, job: dict, frames: list, stop_evt: threading.Event) -> bool:
|
||
"""Detect + embed the decoded frames and submit the result. Returns True
|
||
when the job was completed (→ count it processed), False otherwise: a
|
||
transient transport fault releases the job (counted 'waited out'); a
|
||
job-specific fault fails it (counted an error); a stop mid-flight releases
|
||
it so a Stop drains promptly instead of finishing heavy GPU work."""
|
||
jid = job["job_id"]
|
||
try:
|
||
if self._abort_if_stopped(jid, stop_evt):
|
||
return False
|
||
|
||
task = job.get("task") or "ccip"
|
||
embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION
|
||
model_name = (
|
||
self.cfg.embed_model_override
|
||
or job.get("embed_model_name")
|
||
or DEFAULT_EMBED_MODEL
|
||
)
|
||
|
||
# 'embed' = WHOLE-IMAGE SigLIP embedding (re-embed the library under a
|
||
# 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) # one-time model load
|
||
with self._timed("gpu"):
|
||
vecs = [embedder.embed(frame) for _, frame in frames]
|
||
vec = (
|
||
np.mean(np.asarray(vecs, dtype=np.float32), axis=0).tolist()
|
||
if len(vecs) > 1 else vecs[0]
|
||
)
|
||
if self._abort_if_stopped(jid, stop_evt):
|
||
return False
|
||
with self._timed("submit"):
|
||
self.client.submit_embedding(jid, vec, embed_version)
|
||
self._unhold(jid)
|
||
self._forget_strikes(jid)
|
||
return True
|
||
|
||
# task picks what to produce per crop:
|
||
# 'siglip' (backfill existing images) → concept (SigLIP) regions
|
||
# ONLY, so it never churns their figure/CCIP regions or the
|
||
# character-reference cache.
|
||
# 'ccip' / 'both' (a new image's first pass) → figure (CCIP) AND
|
||
# concept (SigLIP) in one go, off the same crop.
|
||
want_ccip = task in ("ccip", "both")
|
||
want_siglip = task in ("ccip", "siglip", "both")
|
||
replace_kinds = (
|
||
["concept", "panel"] if task == "siglip"
|
||
else ["figure", "face", "concept", "panel"]
|
||
)
|
||
embedder = self._ensure_embedder(model_name) if want_siglip else None
|
||
eff = self._effective_cfg(job.get("detectors"))
|
||
proposers = self._proposers_for(eff)
|
||
|
||
regions = []
|
||
ccip_ev = self.cfg.ccip_model or "ccip-default"
|
||
dv = f"person-{self.cfg.detector_level}"
|
||
|
||
# detect + CCIP + batched embed across every (deduped) frame = "gpu".
|
||
with self._timed("gpu"):
|
||
for t, frame in frames:
|
||
# Bail promptly on Stop instead of grinding through every frame
|
||
# of a long video before the caller can hand the job back.
|
||
if self._stopped(stop_evt):
|
||
break
|
||
# FIGURE boxes: imgutils detect_person ∪ general COCO person,
|
||
# NMS-merged → CCIP identity (+ a concept crop). Covers anime +
|
||
# Western/realistic figures.
|
||
base = models.detect_figures(frame, self.cfg.detector_level)
|
||
figs = proposers.figures(frame, base)
|
||
if not figs:
|
||
figs = [((0.0, 0.0, 1.0, 1.0), 1.0, "whole")] # whole-frame fallback
|
||
|
||
# Collect every crop that needs a SigLIP embedding, then embed
|
||
# them in ONE batched forward pass (huge GPU-util + throughput
|
||
# win vs one forward per crop). CCIP runs per figure inline.
|
||
pending = [] # (crop, region-template-without-embedding)
|
||
for bbox, score, _label in figs:
|
||
crop = crop_region(frame, bbox)
|
||
if crop is None:
|
||
continue
|
||
if want_ccip:
|
||
regions.append({
|
||
"kind": "figure", "bbox": list(bbox), "frame_time": t,
|
||
"score": score,
|
||
"ccip_embedding": models.ccip_vector(
|
||
crop, self.cfg.ccip_model or None
|
||
),
|
||
"embedding_version": ccip_ev, "detector_version": dv,
|
||
})
|
||
if want_siglip:
|
||
pending.append((crop, {
|
||
"kind": "concept", "bbox": list(bbox), "frame_time": t,
|
||
"score": score, "detector_version": dv,
|
||
}))
|
||
if not want_siglip:
|
||
continue
|
||
# ANATOMY components (booru_yolo) + PANELS → concept/panel crops.
|
||
for bbox, score, label in proposers.components(frame):
|
||
crop = crop_region(frame, bbox)
|
||
if crop is not None:
|
||
pending.append((crop, {
|
||
"kind": "concept", "bbox": list(bbox), "frame_time": t,
|
||
"score": score, "detector_version": f"booru:{label}",
|
||
}))
|
||
for bbox, score, _label in proposers.panels(frame):
|
||
crop = crop_region(frame, bbox)
|
||
if crop is not None:
|
||
pending.append((crop, {
|
||
"kind": "panel", "bbox": list(bbox), "frame_time": t,
|
||
"score": score, "detector_version": "panel",
|
||
}))
|
||
if pending:
|
||
# Drop near-duplicate crops (a figure box ≈ an anatomy
|
||
# component, overlapping booru classes) before the embed so
|
||
# we never SigLIP the same region twice — saves GPU and a
|
||
# slot against max_regions. High-IoU + kind-aware, so
|
||
# intentional nested crops (figure ⊃ head) survive.
|
||
pending = dedupe_crops(pending, eff.dedupe_iou)
|
||
vecs = embedder.embed_batch([c for c, _ in pending])
|
||
for (_c, tmpl), vec in zip(pending, vecs, strict=True):
|
||
tmpl["siglip_embedding"] = vec
|
||
tmpl["embedding_version"] = embed_version
|
||
regions.append(tmpl)
|
||
# Stop once we have enough: a long video (image 81602 = a 156 MB
|
||
# mp4, 64 sampled frames × ~32 regions) would otherwise burn
|
||
# ~38s of GPU across every frame before the submit is even
|
||
# truncated. Bounds the WORK, not just the POST body.
|
||
if len(regions) >= eff.max_regions:
|
||
break
|
||
|
||
# A Stop mid-frame-loop leaves partial regions — don't submit those;
|
||
# hand the whole job back so another agent redoes it cleanly.
|
||
if self._abort_if_stopped(jid, stop_evt):
|
||
return False
|
||
|
||
# Backstop: never submit an unbounded pile of regions (a pathological
|
||
# image / long video). Keep the highest-scoring max_regions so the
|
||
# POST body stays sane — curator rejects an oversized one with 413
|
||
# (operator-flagged: image 81602 looped on 413).
|
||
if len(regions) > self.cfg.max_regions:
|
||
regions.sort(key=lambda r: r.get("score", 0.0) or 0.0, reverse=True)
|
||
dropped = len(regions) - self.cfg.max_regions
|
||
regions = regions[: self.cfg.max_regions]
|
||
log.info("job %s: capped regions %d→%d (dropped %d)",
|
||
jid, len(regions) + dropped, len(regions), dropped)
|
||
with self._timed("submit"):
|
||
self.client.submit(jid, regions, replace_kinds)
|
||
self._unhold(jid)
|
||
self._forget_strikes(jid)
|
||
return True
|
||
except requests.RequestException as exc:
|
||
if _is_transient(exc):
|
||
if (not self._stopped(stop_evt)
|
||
and self._strike(jid) >= TRANSIENT_JOB_CAP):
|
||
# Same poison rationale as the fetch path: a job whose
|
||
# submit keeps dying transiently would otherwise re-lease →
|
||
# re-download → re-GPU forever.
|
||
self._fail(jid, job.get("image_id"), exc,
|
||
verb="gave up after repeated transient failures")
|
||
return False
|
||
# curator down/redeploying, a 5xx, or our lease was reclaimed
|
||
# while we worked. NOT the job's fault — hand it back (best
|
||
# effort; then the server's orphan-recovery reclaims it if down).
|
||
self._bump(transient=1)
|
||
log.info("submit failed job %s (%s) — released, re-lease later",
|
||
jid, _transient_reason(exc))
|
||
self._release([jid])
|
||
return False
|
||
self._fail(jid, job.get("image_id"), exc)
|
||
return False
|
||
except Exception as exc: # noqa: BLE001 — a genuine job fault: report it
|
||
self._fail(jid, job.get("image_id"), exc)
|
||
return False
|
||
|
||
# --- autoscaler --------------------------------------------------------
|
||
def _control_loop(self):
|
||
"""Scale DOWNLOADERS by buffer occupancy (Auto mode). The buffer sits
|
||
between the two stages, so its fill level is the direct signal: empty =
|
||
the GPU is starving (add downloaders); full = downloaders outpace the GPU
|
||
(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.
|
||
|
||
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
|
||
while not self._ctrl_stop.wait(CONTROL_INTERVAL):
|
||
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
|
||
self._bw_capped = False
|
||
self._idle = False
|
||
continue
|
||
|
||
occ = self._buffer.qsize() / BUFFER_MAX
|
||
occ_ewma = _ewma(occ_ewma, occ, OCC_ALPHA)
|
||
# Bandwidth-cap position: compare the observed aggregate (the same
|
||
# EWMA the UI shows) against the governor's cap. `soft` gates
|
||
# growth; `hard` sheds streams (see BW_* rationale above).
|
||
bw_rate = self.throttle.rate
|
||
net_bytes = self._net_mb_s * 1_048_576
|
||
bw_soft = bw_rate > 0 and net_bytes >= BW_ADD_HEADROOM * bw_rate
|
||
bw_hard = bw_rate > 0 and net_bytes >= BW_TRIM_AT * bw_rate
|
||
self._bw_capped = bw_soft
|
||
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
|
||
util_ewma = _ewma(util_ewma, util, UTIL_ALPHA)
|
||
self._util_smooth = util_ewma
|
||
|
||
# Memory pressure overrides the cadence — react immediately.
|
||
if vram >= VRAM_HI and self._consumer_target > 1:
|
||
if self._apply_consumers(-1):
|
||
log.info("autoscale: consumers→%d (vram %d%% — memory pressure)",
|
||
self._consumer_target, round(vram * 100))
|
||
tick = 0
|
||
con_grew = False
|
||
continue
|
||
|
||
tick += 1
|
||
if tick < SAMPLES_PER_DECISION:
|
||
continue
|
||
tick = 0
|
||
|
||
now = time.monotonic()
|
||
inst = (self.processed - prev_p) / max(1e-3, now - prev_t)
|
||
prev_p, prev_t = self.processed, now
|
||
tput_ewma = _ewma(tput_ewma, inst, TPUT_ALPHA)
|
||
fail_delta = self.transient - prev_fail
|
||
prev_fail = self.transient
|
||
|
||
d0, c0 = self._dl_target, self._consumer_target
|
||
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 → WHY decides the move (operator-flagged
|
||
# 2026-07-02: the pool scaled to 8 against an EMPTY queue):
|
||
# - queue empty (_idle): nothing to download — shed to ONE
|
||
# polling downloader; growth would just add idle threads.
|
||
# - pinned at the bandwidth cap: more streams split the same
|
||
# budget and stretch per-job latency — shed toward BW_MIN_DL.
|
||
# - cap headroom + work flowing: concurrency is genuinely
|
||
# short — add a feeder.
|
||
if self._idle:
|
||
if self._dl_target > 1:
|
||
self._apply_downloaders(-1)
|
||
elif bw_hard and self._dl_target > BW_MIN_DL:
|
||
self._apply_downloaders(-1)
|
||
elif not bw_soft:
|
||
self._apply_downloaders(+1)
|
||
con_grew = False
|
||
elif occ_ewma > OCC_HIGH:
|
||
# Downloaders outpace the GPU. Prefer helping the GPU (add a 2nd
|
||
# consumer) when it has util + VRAM headroom and the last add
|
||
# actually paid off; otherwise trim a downloader.
|
||
if con_grew:
|
||
if tput_ewma > tput_before * (1 + TPUT_MARGIN):
|
||
con_grew = False # it helped → keep it, stop probing
|
||
else:
|
||
self._apply_consumers(-1) # no gain → revert
|
||
con_grew = False
|
||
elif (self._consumer_target < CONSUMER_MAX
|
||
and util_ewma < UTIL_START and vram < VRAM_GROW_MAX):
|
||
tput_before = tput_ewma
|
||
con_grew = self._apply_consumers(+1)
|
||
if not con_grew: # already maxed → trim a feeder
|
||
self._apply_downloaders(-1)
|
||
else:
|
||
self._apply_downloaders(-1)
|
||
else:
|
||
con_grew = False # balanced → settle
|
||
|
||
if self._dl_target != d0 or self._consumer_target != c0:
|
||
log.info(
|
||
"autoscale: dl %d→%d · consumers %d→%d "
|
||
"(buf %d%% · util~%d%% · %.2f j/s · vram %d%% · "
|
||
"net %.1f MB/s%s)",
|
||
d0, self._dl_target, c0, self._consumer_target,
|
||
round(occ_ewma * 100), round(util_ewma), tput_ewma,
|
||
round(vram * 100), self._net_mb_s,
|
||
" — at cap" if bw_soft else "")
|