From 85503fd87cbe25363a0ecfc0b4cfe1c8bf8a01dd Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 6 Sep 2026 20:03:46 -0600 Subject: [PATCH] test(downloader): cover probe judging and the submission window Assert that a probe is judged on the trailing tick, that a backoff is temporary and climbs back to the cap, and that a worker always finds its next segment staged. Documents adaptive-workers behaviour in the downloading guide. --- docs/guide/downloading.md | 4 +- tests/core/test_adaptive_workers.py | 70 +++++++++++++ tests/downloaders/test_submission_window.py | 105 ++++++++++++++++++++ 3 files changed, 177 insertions(+), 2 deletions(-) create mode 100644 tests/downloaders/test_submission_window.py diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md index c218c4e6..678a0ce5 100644 --- a/docs/guide/downloading.md +++ b/docs/guide/downloading.md @@ -734,8 +734,8 @@ Two related flags: | Flag | Purpose | | --- | --- | | `--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. 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`. | +| `--adaptive-workers` | Opt-in: start the per-track worker count at `--workers`, reduce it when the CDN returns errors, then raise it again while the extra workers add measured throughput. The count never goes above `--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. 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. Also ignored when the service HTTP session carries state a child process cannot rebuild, such as a custom TLS adapter. The download stays in a single process instead of running with the wrong settings. Default `1`. | | `--continue-downloads` | Opt-in: keep completed segment files when a download fails so the next run resumes instead of restarting from zero. One-off enable of the [`continue_downloads`](../reference/configuration/download.md#continue_downloads) config option, which documents what can and cannot resume. | | `--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. | diff --git a/tests/core/test_adaptive_workers.py b/tests/core/test_adaptive_workers.py index 2d9ecc81..d39fd47b 100644 --- a/tests/core/test_adaptive_workers.py +++ b/tests/core/test_adaptive_workers.py @@ -18,6 +18,7 @@ from unshackle.core.downloaders.requests import ( plan_tail_parts, requests, split_ranges, + submission_window, tail_boost_engages, ) from unshackle.core.tracks.track import DownloadContext @@ -176,6 +177,75 @@ def test_tail_guard_still_halves_on_error_burst() -> None: assert c.update(now, 2) == 3 # guard active (2 < 6) but 3 errors -> 6 // 2 +PER_WORKER = 1_000_000.0 # bytes/sec each worker adds, so throughput scales with the target + + +def run_scaling(c: AdaptiveWorkerController, now: float, seconds: float, work: int = 10_000) -> float: + """Advance ``seconds`` of a download whose throughput scales with the live target. + + Records samples on the 0.5s grid at ``target * PER_WORKER`` bytes/sec and calls ``update`` + on every step, so the speed window mixes old and new rates the way a real run does. ``work`` + stands for the segments in flight or not yet started, kept high to hold the tail guard off. + """ + end = now + seconds + while now < end - 1e-9: + now += DT + c.record_bytes(int(c.target * PER_WORKER * DT), now) + c.update(now, work) + return now + + +def test_probe_kept_when_the_true_gain_clears_the_bar() -> None: + c = AdaptiveWorkerController(cap=16, start=10, tick=TICK, window=WINDOW) + now = 0.0 + c.update(now) + rate = 1_000_000.0 + now = fill_window(c, rate, now) + assert c.update(now) == 12 # first evaluation probes 10 -> 12 + target, now = run_tick(c, rate * 1.2, now) + assert target == 14 # probe held, and the controller climbs again + + +def test_recovers_to_the_cap_after_an_error_burst() -> None: + c = AdaptiveWorkerController(cap=16, start=16, tick=TICK, window=WINDOW) + now = 0.0 + c.update(now) + now = run_scaling(c, now, WINDOW) + for _ in range(3): + c.record_error(now) + now = run_scaling(c, now, TICK) + assert c.target == 8, "error burst must halve the target 16 -> 8" + + now = run_scaling(c, now, 60.0) + assert c.target == 16, "controller never returned to its cap after backing off" + now = run_scaling(c, now, 40.0) + assert c.target == 16, "controller left the cap on a link where every worker still pays" + + +def test_reverts_when_the_probe_stops_paying_on_a_saturated_link() -> None: + c = AdaptiveWorkerController(cap=16, start=8, tick=TICK, window=WINDOW) + now = 0.0 + c.update(now) + rate = 8 * PER_WORKER + now = fill_window(c, rate, now) + seen = [c.update(now, 10_000)] + for _ in range(8): + rate = min(c.target, 10) * PER_WORKER + end = now + TICK + while now < end - 1e-9: + now += DT + c.record_bytes(int(rate * DT), now) + seen.append(c.update(now, 10_000)) + assert max(seen) <= 12 # one probe past the knee is the most it may hold + assert seen[-1] <= 12 + + +def test_submission_window_stages_a_spare_segment_for_every_worker() -> None: + assert submission_window(16) == 32 + for target in (2, 4, 6, 16): + assert submission_window(target) == 2 * target + + def test_plan_tail_parts_full_coverage_no_gaps() -> None: size = 20 * 1024 * 1024 parts = plan_tail_parts(size, spare=8) diff --git a/tests/downloaders/test_submission_window.py b/tests/downloaders/test_submission_window.py new file mode 100644 index 00000000..0723e7d8 --- /dev/null +++ b/tests/downloaders/test_submission_window.py @@ -0,0 +1,105 @@ +"""Submission-window invariants for the segmented drain loop. + +Segments are not all submitted up front: that convoys on the pool's global lock and starves +the event drain. The loop meters submission through a window it tops up. The window must hold +two segments for every worker, so a worker that finishes one finds the next already staged +instead of idling until the next drain cycle. Fixed and adaptive mode must agree on that +ratio; adaptive sizes it from its live target instead of from ``max_workers``. + +The test stalls every response, so no future can complete and the queue settles at exactly +the window. Counting ``pool.submit`` calls makes the queue depth observable: a blocked +server only ever sees ``max_workers`` requests, whatever is staged behind them. +""" + +import importlib +import threading +import time +from concurrent.futures.thread import ThreadPoolExecutor +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +dl = importlib.import_module("unshackle.core.downloaders.requests") + +BODY = b"segment-body" * 64 +MAX_WORKERS = 4 +SEG_COUNT = 20 # more than the window, so queue depth is the only bound + +pytestmark = pytest.mark.unit + + +class _Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.server.release.wait(timeout=10) + self.send_response(200) + self.send_header("Content-Length", str(len(BODY))) + self.end_headers() + self.wfile.write(BODY) + + def log_message(self, *args): + pass + + +@pytest.fixture +def server(): + srv = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + srv.release = threading.Event() + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + yield srv + srv.release.set() + srv.shutdown() + thread.join(timeout=5) + + +@pytest.fixture(autouse=True) +def clean_cancel(): + dl.DOWNLOAD_CANCELLED.clear() + yield + dl.DOWNLOAD_CANCELLED.clear() + + +@pytest.fixture +def submissions(monkeypatch): + counter: list[int] = [] + + class _CountingPool(ThreadPoolExecutor): + def submit(self, fn, /, *args, **kwargs): + counter.append(1) # list.append is atomic, so no lock is needed here + return super().submit(fn, *args, **kwargs) + + monkeypatch.setattr(dl, "ThreadPoolExecutor", _CountingPool) + return counter + + +@pytest.mark.parametrize("adaptive", [False, True]) +def test_queue_holds_two_segments_per_worker(server, submissions, tmp_path, adaptive): + urls = [{"url": f"http://127.0.0.1:{server.server_address[1]}/seg{i}.bin"} for i in range(SEG_COUNT)] + events: list[dict] = [] + + def run(): + events.extend( + dl.requests( + urls, + output_dir=tmp_path, + filename="seg_{i:04}.bin", + max_workers=MAX_WORKERS, + adaptive=adaptive, + ) + ) + + worker = threading.Thread(target=run, daemon=True) + worker.start() + try: + deadline = time.time() + 3.0 + while time.time() < deadline and len(submissions) < 2 * MAX_WORKERS: + time.sleep(0.02) + time.sleep(0.2) # settle, so an over-filled window is caught as well as an under-filled one + assert len(submissions) == 2 * MAX_WORKERS + finally: + server.release.set() + worker.join(timeout=30) + + assert not worker.is_alive() + assert sum(event.get("advance", 0) for event in events) == SEG_COUNT + assert len(sorted(tmp_path.glob("seg_*.bin"))) == SEG_COUNT