|
|
|
@@ -61,10 +61,14 @@ def _is_transient(exc: requests.RequestException) -> bool:
|
|
|
|
|
return resp.status_code >= 500 or resp.status_code in (401, 403, 408, 409, 429)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Pipeline sizing. Downloaders are I/O-bound so the ceiling is generous; consumers
|
|
|
|
|
# are GPU-bound so a couple saturate the card. The buffer is small on purpose —
|
|
|
|
|
# each slot can hold many decoded video frames, so it bounds RAM, not just depth.
|
|
|
|
|
DL_MAX = 24 # max downloader threads
|
|
|
|
|
# Pipeline sizing. Downloaders are I/O-bound, but every download streams a full
|
|
|
|
|
# original (large videos included) THROUGH curator's single Python file-serving
|
|
|
|
|
# path — so the ceiling is deliberately modest: too many concurrent large-file
|
|
|
|
|
# GETs saturate curator's web workers + NFS and slow everything (including the
|
|
|
|
|
# browser). 8 keeps a fast GPU fed without stampeding curator. Consumers are
|
|
|
|
|
# GPU-bound so a couple saturate the card. The buffer is small on purpose — each
|
|
|
|
|
# slot can hold many decoded video frames, so it bounds RAM, not just depth.
|
|
|
|
|
DL_MAX = 8 # max downloader threads
|
|
|
|
|
CONSUMER_MAX = 2 # max GPU consumer threads
|
|
|
|
|
BUFFER_MAX = 12 # bounded decoded-frame buffer (backpressure + RAM cap)
|
|
|
|
|
|
|
|
|
@@ -712,13 +716,22 @@ class Worker:
|
|
|
|
|
(the GPU is the bottleneck → add a 2nd consumer if it has headroom and the
|
|
|
|
|
add lifts throughput, else trim a downloader). Occupancy, util and
|
|
|
|
|
throughput are EWMA-smoothed and decisions spaced so moves ride averaged
|
|
|
|
|
signals, not tick-to-tick noise. VRAM pressure sheds a consumer at once."""
|
|
|
|
|
signals, not tick-to-tick noise. VRAM pressure sheds a consumer at once.
|
|
|
|
|
|
|
|
|
|
Failure guard (critical): an empty buffer can mean the GPU is starving OR
|
|
|
|
|
that downloads are FAILING (curator slow/unreachable). In the latter case
|
|
|
|
|
adding downloaders piles more concurrent large-file requests onto a
|
|
|
|
|
struggling curator — a congestion collapse that slows curator (and its
|
|
|
|
|
browser) further and never recovers. So if transient download failures
|
|
|
|
|
rose since the last decision, SHRINK toward the floor instead of growing,
|
|
|
|
|
and let the pool ramp back up only once downloads succeed again."""
|
|
|
|
|
from . import gpu as gpumod
|
|
|
|
|
|
|
|
|
|
occ_ewma: float | None = None
|
|
|
|
|
util_ewma: float | None = None
|
|
|
|
|
tput_ewma: float | None = None
|
|
|
|
|
prev_p, prev_t = self.processed, time.monotonic()
|
|
|
|
|
prev_fail = self.transient
|
|
|
|
|
tick = 0
|
|
|
|
|
con_grew = False # did the previous decision add a consumer?
|
|
|
|
|
tput_before = 0.0 # smoothed jobs/s before that consumer add
|
|
|
|
@@ -726,6 +739,7 @@ class Worker:
|
|
|
|
|
if not (self._running and self._auto):
|
|
|
|
|
occ_ewma = util_ewma = tput_ewma = None
|
|
|
|
|
prev_p, prev_t = self.processed, time.monotonic()
|
|
|
|
|
prev_fail = self.transient
|
|
|
|
|
tick = 0
|
|
|
|
|
con_grew = False
|
|
|
|
|
self._util_smooth = None
|
|
|
|
@@ -764,9 +778,17 @@ class Worker:
|
|
|
|
|
tput_ewma = inst if tput_ewma is None else (
|
|
|
|
|
TPUT_ALPHA * inst + (1 - TPUT_ALPHA) * tput_ewma
|
|
|
|
|
)
|
|
|
|
|
fail_delta = self.transient - prev_fail
|
|
|
|
|
prev_fail = self.transient
|
|
|
|
|
|
|
|
|
|
d0, c0 = self._dl_target, self._consumer_target
|
|
|
|
|
if occ_ewma < OCC_LOW:
|
|
|
|
|
if fail_delta > 0:
|
|
|
|
|
# Downloads are FAILING (curator slow/unreachable), so the empty
|
|
|
|
|
# buffer is NOT the GPU starving — growing would stampede a
|
|
|
|
|
# struggling curator. Back off toward the floor and let it recover.
|
|
|
|
|
self._apply_downloaders(-1)
|
|
|
|
|
con_grew = False
|
|
|
|
|
elif occ_ewma < OCC_LOW:
|
|
|
|
|
# Buffer starving → GPU idle waiting on downloads → add a feeder.
|
|
|
|
|
self._apply_downloaders(+1)
|
|
|
|
|
con_grew = False
|
|
|
|
|