fix(downloader): keep speed-limited batches in a single process

Spawned children re-import the module and start with no TokenBucket, so
--download-processes silently dropped a configured --speed-limit /
global_speed_limit for any 24+ segment batch. Fall back to the in-process
path while a limit is set (same gate as the ranged-parallel paths), log the
fallback, and document the interaction.
This commit is contained in:
imSp4rky
2026-08-07 11:36:17 -06:00
parent cfae8f8882
commit 431aa8fa54
4 changed files with 41 additions and 2 deletions
+1 -1
View File
@@ -193,7 +193,7 @@ order the service used, then renumbers the episodes into the order you asked for
|---|---|---|
| `--workers` | downloader default | Per-track download threads. |
| `--adaptive-workers` | off | Scale per-track segment workers dynamically (up to the `--workers` cap) based on measured CDN throughput and errors. |
| `--download-processes` | `1` | Split large segment batches (24+) across this many download processes to exceed the single-process throughput cap on multi-gigabit connections. |
| `--download-processes` | `1` | Split large segment batches (24+) across this many download processes to exceed the single-process throughput cap on multi-gigabit connections. Ignored while a speed limit is set: the cap is one shared budget, which extra processes cannot share, so the download stays in a single process. |
| `--downloads` | `1` | Number of tracks downloaded concurrently. |
| `--speed-limit` | unlimited | Cap total download speed across all threads and tracks, e.g. `500k`, `5M`, `1.5G` or plain bytes/sec. Values are bytes, not bits (`5M` = 5.0 MB/s). `off` disables a configured limit. |
| `--no-cache` | off | Bypass the title cache. |
+1 -1
View File
@@ -529,7 +529,7 @@ Two related flags:
| --- | --- |
| `--workers N` | Threads used per track for segment downloads. Default depends on the downloader. |
| `--adaptive-workers` | Opt-in: start with a moderate per-track worker count and ramp it up or back off based on measured CDN throughput and errors, capped at `--workers`. Off by default (fixed worker count). |
| `--download-processes N` | Opt-in: split a large segment batch across `N` download processes, each with its own worker pool. A single process tops out around 1.3 Gb/s (Python interpreter limit); `2` reaches 2.5GbE line rate on fast CDNs. Only engages for batches of 24+ segments. Default `1`. |
| `--download-processes N` | Opt-in: split a large segment batch across `N` download processes, each with its own worker pool. A single process tops out around 1.3 Gb/s (Python interpreter limit); `2` reaches 2.5GbE line rate on fast CDNs. Only engages for batches of 24+ segments. Ignored while `--speed-limit` (or serve's `global_speed_limit`) is set — the cap is one shared budget, which extra processes cannot share, so the download stays in a single process. Default `1`. |
| `--downloads N` | Number of tracks downloaded concurrently. Default `1`. |
| `--slow [MIN-MAX]` | Add a delay between titles to look more like a real device. `--slow` alone means 60-120s; `--slow 20-40` sets a custom range. Minimum 20s. |
| `--no-cache` | Bypass the title cache for this download. |
+23
View File
@@ -154,3 +154,26 @@ def test_multiprocess_cancel_terminates_children(server, tmp_path):
while time.time() < deadline and len(multiprocessing.active_children()) > baseline:
time.sleep(0.05)
assert len(multiprocessing.active_children()) <= baseline
def test_speed_limit_forces_single_process(server, tmp_path, monkeypatch):
"""A set speed limit must keep the batch in-process: spawned children re-import the
module and would run with no TokenBucket, silently ignoring the configured cap."""
def _no_mp(**kwargs):
raise AssertionError("multiprocess fan-out engaged despite a speed limit")
monkeypatch.setattr(dl, "_download_multiprocess", _no_mp)
n = dl.MP_MIN_SEGMENTS
urls = [{"url": _url(server, f"/seg/{i}.bin")} for i in range(n)]
dl.set_speed_limit(1_000_000_000) # far above the tiny batch, so the test is not slowed
try:
for _ in dl.requests(urls, output_dir=tmp_path, filename="seg_{i:04}.bin", max_workers=4, processes=2):
pass
finally:
dl.set_speed_limit(None)
files = sorted(tmp_path.glob("seg_*.bin"))
assert len(files) == n
for f in files:
assert f.read_bytes() == _seg_body(int(f.stem.split("_")[1]))
+16
View File
@@ -1176,6 +1176,8 @@ def requests(
beat the single-interpreter throughput cap (GIL in the ssl read path). Engaged
only when > 1 and there are at least MP_MIN_SEGMENTS urls; otherwise the behaviour
is byte-identical to a single process. Each child runs its own worker pool.
Ignored while a download speed limit is set: spawned children cannot share the
process-global rate budget, so the batch stays in a single process.
index_offset: Added to each url's enumerate index (after ``index_stride`` scaling)
when formatting ``filename`` so a child handling part of a larger batch keeps
its urls' original global indices/names. Internal: set by the multiprocess
@@ -1222,6 +1224,20 @@ def requests(
if not max_workers:
max_workers = min(16, (os.cpu_count() or 1) + 4)
# A spawned child re-imports this module and would start with no TokenBucket, so the
# configured cap would silently stop applying; the cap wins over the fan-out, matching
# the _speed_limiter gates on the ranged-parallel paths.
speed_limiter = _speed_limiter
if processes > 1 and len(urls) >= MP_MIN_SEGMENTS and speed_limiter is not None:
processes = 1
if debug_logger:
debug_logger.log(
level="DEBUG",
operation="downloader_mp_fallback",
message="Speed limit is set; spawned children cannot share its budget, using a single process",
context={"url_count": len(urls), "speed_limit": speed_limiter.rate},
)
# Process fan-out: split a large segment batch across spawned children. Single-URL and small
# batches keep the in-process path (a lone file is already parallelized by ranged parts).
if processes > 1 and len(urls) >= MP_MIN_SEGMENTS: