diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md index 9703dcdf..7ac413c0 100644 --- a/docs/guide/downloading.md +++ b/docs/guide/downloading.md @@ -875,7 +875,7 @@ Three related flags: | --- | --- | | `--workers N` | Threads used per track for segment downloads. Default depends on the downloader. | | `--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`. | +| `--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 a per-job 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/docs/reference/configuration/services.md b/docs/reference/configuration/services.md index 0131eff3..f20447e3 100644 --- a/docs/reference/configuration/services.md +++ b/docs/reference/configuration/services.md @@ -147,7 +147,7 @@ is the [REST API](../../dev/rest-api/index.md) section. These are the config key | `compression_level` | int | `1` | gzip level for responses. | | `services_refresh_interval` | int (s) | `0` | How often the server pulls the git-backed service repositories in `directories.services` and hot-reloads the services that changed. `0` turns it off. A service with a running or queued job, or a live remote session, is staged and swaps to the new code as soon as its last job and session finish. | | `services_staged_max_age` | int (s) | `0` | How long a staged service waits for its jobs and sessions before the server swaps it in anyway. `0` waits forever. Set it on a server whose popular services never have a quiet moment. A session that is live at the swap keeps the old code until it ends; a running job keeps its old code too. | -| `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). One shared budget across all concurrent jobs; the server ignores per-job speed limits while it is set. | +| `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). Each download job gets an equal share: the cap divided by the number of download job slots (2 by default, see [job lifecycle](../../dev/rest-api/job-lifecycle.md)). A job that runs alone still gets only its share. The server ignores per-job speed limits while it is set. | | `cdm_overrides` | list or bool | *(unset)* | Allowed per-request CDM overrides: a list of permitted device names, or `true` for any. Unset rejects every override. | | `allow_job_credentials` | bool | `false` | Permit clients to supply credentials per job. | | `server_accounts` | dict | `{}` | Services whose remote sessions use the server's own `credentials` and cookie files instead of the client's (see below). | @@ -168,7 +168,8 @@ list. continue; an API key you removed gets `401` on its next request. A file that does not parse or validate keeps the running settings and logs the reason. A change to `remote_only`, `services_refresh_interval`, the first PlayReady device, or any command-line flag needs - a restart, and the log says so. Windows has no `SIGHUP`: restart the server there. + a restart, and the log says so. A new `global_speed_limit` applies to jobs that start after the + reload; running jobs keep their share. Windows has no `SIGHUP`: restart the server there. `server_cdm` decides whether the server runs the CDM licensing for that API key. Set it to `true` to enable every service, or to a list of service tags to enable only those. It is `false` diff --git a/tests/remote/unit/test_worker_speed_limit.py b/tests/remote/unit/test_worker_speed_limit.py new file mode 100644 index 00000000..d57efd28 --- /dev/null +++ b/tests/remote/unit/test_worker_speed_limit.py @@ -0,0 +1,82 @@ +import asyncio +import importlib +import json +import tempfile +from types import SimpleNamespace + +import pytest + +from unshackle.core.api import download_worker +from unshackle.core.api.download_manager import DownloadQueueManager +from unshackle.core.api.download_worker import apply_payload_speed_limit + +requests_downloader = importlib.import_module("unshackle.core.downloaders.requests") + +pytestmark = pytest.mark.unit + + +@pytest.fixture(autouse=True) +def clean_limiter(): + """Start and end each test with no limiter, whatever an earlier test leaked.""" + requests_downloader.speed_limit_locked = False + requests_downloader.speed_limiter = None + yield + requests_downloader.speed_limit_locked = False + requests_downloader.speed_limiter = None + + +@pytest.mark.parametrize(("lock", "expected"), [(True, 2_000_000.0), (False, None)]) +def test_worker_share_of_limit(lock, expected): + requests_downloader.set_speed_limit(8_000_000, lock=lock) + assert DownloadQueueManager(max_concurrent_downloads=4).worker_speed_limit() == expected + + +def test_worker_locks_payload_limit_against_job_reset(): + apply_payload_speed_limit({"speed_limit": 1_000_000}) + requests_downloader.set_speed_limit(None) + limiter = requests_downloader.speed_limiter + assert limiter is not None and limiter.rate == 1_000_000 + + +def test_worker_without_payload_limit_stays_unlimited(): + apply_payload_speed_limit({"speed_limit": None}) + assert requests_downloader.speed_limiter is None + + +def test_run_download_async_sends_share_in_payload(monkeypatch, tmp_path): + monkeypatch.setattr(tempfile, "tempdir", str(tmp_path)) + + class Stop(Exception): + pass + + seen = {} + + async def fake_spawn(*args, **kwargs): + with open(args[3], encoding="utf-8") as handle: + seen["payload"] = json.load(handle) + raise Stop + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_spawn) + requests_downloader.set_speed_limit(8_000_000, lock=True) + job = SimpleNamespace(job_id="j1", service="SVC", title_id="t", parameters={}) + with pytest.raises(Stop): + asyncio.run(DownloadQueueManager(max_concurrent_downloads=4).run_download_async(job)) + assert seen["payload"]["speed_limit"] == 2_000_000.0 + + +def test_worker_main_applies_share_before_download(monkeypatch, tmp_path): + seen = {} + + def fake_perform(*args, **kwargs): + requests_downloader.set_speed_limit(None) # what dl.result() does + seen["rate"] = requests_downloader.speed_limiter.rate + return [] + + monkeypatch.setattr(download_worker, "perform_download", fake_perform) + payload = tmp_path / "payload.json" + payload.write_text( + json.dumps({"job_id": "j1", "service": "SVC", "title_id": "t", "parameters": {}, "speed_limit": 500_000}), + encoding="utf-8", + ) + assert download_worker.main(["w", str(payload), str(tmp_path / "result.json")]) == 0 + assert seen["rate"] == 500_000 diff --git a/unshackle/commands/serve.py b/unshackle/commands/serve.py index 33603273..b3bc7c0d 100644 --- a/unshackle/commands/serve.py +++ b/unshackle/commands/serve.py @@ -468,7 +468,7 @@ def serve( global_speed_limit = parse_speed_limit(config.serve.get("global_speed_limit")) if global_speed_limit: set_speed_limit(global_speed_limit, lock=True) - log.info(f"Global speed limit: {format_speed(global_speed_limit)} (shared by all jobs)") + log.info(f"Global speed limit: {format_speed(global_speed_limit)} (split evenly across job slots)") global_services = config.serve.get("services") if global_services: diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index 875d3f36..b3537b83 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -1,5 +1,6 @@ import asyncio import hashlib +import importlib import json import logging import os @@ -1054,6 +1055,15 @@ class DownloadQueueManager: log.error(f"Download failed for job {job.job_id}: {e}") raise + def worker_speed_limit(self) -> Optional[float]: + """Return one job worker's share of the locked global speed limit, or None when unset.""" + requests_downloader = importlib.import_module("unshackle.core.downloaders.requests") + + limiter = requests_downloader.speed_limiter + if not (requests_downloader.speed_limit_locked and limiter): + return None + return limiter.rate / max(1, self.max_concurrent_downloads) + async def run_download_async(self, job: DownloadJob) -> List[str]: """Invoke a job worker subprocess to do the download.""" @@ -1062,6 +1072,7 @@ class DownloadQueueManager: "service": job.service, "title_id": job.title_id, "parameters": job.parameters, + "speed_limit": self.worker_speed_limit(), } payload_fd, payload_path = tempfile.mkstemp(prefix=f"unshackle_job_{job.job_id}_", suffix="_payload.json") diff --git a/unshackle/core/api/download_worker.py b/unshackle/core/api/download_worker.py index 12a790b2..ecfd6288 100644 --- a/unshackle/core/api/download_worker.py +++ b/unshackle/core/api/download_worker.py @@ -17,6 +17,7 @@ from uuid import uuid4 from unshackle.core.api.input_bridge import AUTH_INPUT_TIMEOUT from unshackle.core.console import set_notice_handler, set_prompt_handler +from unshackle.core.downloaders import format_speed, set_speed_limit from .download_manager import perform_download @@ -95,6 +96,15 @@ def relay_prompts(progress_callback: Callable[[Dict[str, Any]], None]) -> None: set_notice_handler(lambda message: progress_callback({"notice": message})) +def apply_payload_speed_limit(payload: Dict[str, Any]) -> None: + """Lock this worker's share of the server speed limit so the job cannot clear it.""" + limit = payload.get("speed_limit") + if not limit: + return + set_speed_limit(limit, lock=True) + log.info(f"Job speed limit: {format_speed(limit)} (share of the global speed limit)") + + def main(argv: list[str]) -> int: if len(argv) not in [3, 4]: print( @@ -116,6 +126,7 @@ def main(argv: list[str]) -> int: service = payload["service"] title_id = payload["title_id"] params = payload.get("parameters", {}) + apply_payload_speed_limit(payload) log.info(f"Worker starting job {job_id} ({service}:{title_id})")