diff --git a/backend/app/config.py b/backend/app/config.py index edb511eb..d48d6de9 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -186,6 +186,21 @@ class Settings(BaseSettings): download_supervisor_poll_seconds: int = 3 # Grace between SIGTERM and SIGKILL when stopping a child (matches run_ffmpeg's escalation). download_kill_grace_seconds: int = 5 + # --- worker-child resource de-prioritization --- + # The download/edit child (yt-dlp + ffmpeg) shares ONE LAN box with the API, Postgres, and Plex + # playback. Without a priority handicap a CPU re-encode pegs every core for ~30s and starves web + # requests + playback until it finishes. The child lowers its OWN priority at startup — inherited + # by every ffmpeg/yt-dlp descendant — so any foreground work always wins. Each knob is best-effort + # and independently disable-able with 0. + download_child_nice: int = 10 # CPU nice increment applied to the child; 0 = don't renice + download_child_ionice_class: int = 3 # best-effort IO class (3 = idle); 0 = leave IO untouched + # Cap the browser-safe re-encode's encoder threads so a transcode can't claim every core. 0 = + # reserve one core for foreground work (max(1, cpu_count - 1)); >0 pins exactly that many. + download_ffmpeg_threads: int = 0 + # Throttle the retention GC's whole-tree orphan file sweep so its rglob can't monopolize disk IO + # in the API process: sleep `pause_ms` after every `batch` files touched. pause_ms=0 disables. + download_gc_sweep_pause_ms: int = 20 + download_gc_sweep_batch: int = 400 # Total cache size cap across ALL users (bytes). When >0 and exceeded, the GC evicts the # least-recently-accessed ready assets until under it. 0 = unlimited (disk is the limit). download_total_max_bytes: int = 0 diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py index 7f4d6a7a..a4a37b8d 100644 --- a/backend/app/downloads/edit.py +++ b/backend/app/downloads/edit.py @@ -25,6 +25,7 @@ import hashlib import json import logging import math +import os import queue import shutil import subprocess @@ -32,9 +33,21 @@ import threading import time from pathlib import Path +from app.config import settings + log = logging.getLogger("siftlode.edit") +def encode_thread_cap() -> int: + """How many encoder threads a browser-safe re-encode may use. `download_ffmpeg_threads` when the + admin pinned one (>0), else reserve a core for foreground work (max(1, cpu_count - 1)). Shared by + the ffmpeg `-threads` flag and the child's `OMP_NUM_THREADS` so both honour the same budget.""" + n = settings.download_ffmpeg_threads + if n > 0: + return n + return max(1, (os.cpu_count() or 2) - 1) + + class EditAborted(Exception): """Raised by run_ffmpeg when the job was paused/canceled mid-encode.""" @@ -510,11 +523,12 @@ def browser_transcode_flags( def build_browser_safe_cmd( - src: Path, dest: Path, reencode_video: bool, reencode_audio: bool + src: Path, dest: Path, reencode_video: bool, reencode_audio: bool, threads: int | None = None ) -> list[str]: """ffmpeg command that lands `src` as a browser-safe mp4 (faststart): re-encode only the stream(s) that need it (libx264 / aac), stream-copy the rest, always into mp4. `-progress - pipe:1` streams machine-readable progress for run_ffmpeg.""" + pipe:1` streams machine-readable progress for run_ffmpeg. `threads` caps the encoder so a + re-encode can't claim every core on the shared box (see the worker-child de-prioritization).""" cmd = [ "ffmpeg", "-y", @@ -534,6 +548,8 @@ def build_browser_safe_cmd( else ["-c:v", "copy"] ) cmd += ["-c:a", "aac", "-b:a", "192k"] if reencode_audio else ["-c:a", "copy"] + if threads and threads > 0: + cmd += ["-threads", str(threads)] cmd += ["-movflags", "+faststart", str(dest)] return cmd diff --git a/backend/app/downloads/gc.py b/backend/app/downloads/gc.py index 9ce499af..a640c22e 100644 --- a/backend/app/downloads/gc.py +++ b/backend/app/downloads/gc.py @@ -86,11 +86,25 @@ def _sweep_orphan_files(db: Session, download_root: str) -> int: cutoff = time.time() - _ORPHAN_MIN_AGE_S removed = 0 + # Throttle the whole-tree walk: this runs in the API process, and an unpaced rglob+stat over a + # large download root monopolizes disk IO long enough to make web requests + playback stutter. + # Yield the IO channel after every `batch` files touched so foreground work interleaves. + pause_s = max(0, settings.download_gc_sweep_pause_ms) / 1000.0 + batch = max(1, settings.download_gc_sweep_batch) + touched = 0 + + def _throttle() -> None: + nonlocal touched + touched += 1 + if pause_s and touched % batch == 0: + time.sleep(pause_s) + for entry in root.iterdir(): if entry.name.startswith("."): # .edits / .plex-* / .staging / .storyboards + any dotfile continue files = [entry] if entry.is_file() else [p for p in entry.rglob("*") if p.is_file()] for f in files: + _throttle() try: if f.resolve() in keep or f.stat().st_mtime > cutoff: continue diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py index e0ed42a1..6b8088c6 100644 --- a/backend/app/downloads/pipeline.py +++ b/backend/app/downloads/pipeline.py @@ -381,7 +381,9 @@ def _ensure_browser_playable(job_id: int, produced: Path, spec: dict, info: dict # re-encode still shows real progress instead of a frozen 0%. total_s = int(info["duration"]) if info.get("duration") else editmod.probe_duration(produced) editmod.run_ffmpeg( - editmod.build_browser_safe_cmd(produced, dest, reenc_v, reenc_a), + editmod.build_browser_safe_cmd( + produced, dest, reenc_v, reenc_a, threads=editmod.encode_thread_cap() + ), total_s=total_s, on_progress=_ffmpeg_progress_writer(job_id, "processing"), should_cancel=_cancel_poll(job_id), diff --git a/backend/app/downloads/runner.py b/backend/app/downloads/runner.py index c049208b..f3ff581c 100644 --- a/backend/app/downloads/runner.py +++ b/backend/app/downloads/runner.py @@ -12,16 +12,48 @@ progress hook), then exits 0. An unexpected crash exits non-zero and the parent """ import logging +import os +import shutil +import subprocess import sys +from app.config import settings from app.db import SessionLocal +from app.downloads import edit as editmod from app.models import DownloadJob log = logging.getLogger("siftlode.worker.runner") +def _deprioritize() -> None: + """Lower THIS child's CPU + IO priority so its yt-dlp/ffmpeg work — every descendant inherits the + handicap — can never starve the co-located API, Postgres, or Plex playback. Run once at startup, + before any heavy work. Every step is best-effort: a failed renice must never fail a download.""" + nice_inc = settings.download_child_nice + if nice_inc > 0: + try: + os.nice(nice_inc) # relative increment; the child starts at the worker's nice (0) + except OSError as exc: + log.info("could not renice download child: %s", str(exc)[:120]) + cls = settings.download_child_ionice_class + if cls > 0 and shutil.which("ionice"): # util-linux; absent on some minimal images + try: + subprocess.run( + ["ionice", "-c", str(cls), "-p", str(os.getpid())], + capture_output=True, + timeout=5, + check=False, + ) + except (OSError, subprocess.SubprocessError) as exc: + log.info("could not ionice download child: %s", str(exc)[:120]) + # Cap OpenMP-parallel ffmpeg filters to the same budget as the encoder threads, so a filter + # graph can't fan out across every core behind ffmpeg's own `-threads` cap. + os.environ.setdefault("OMP_NUM_THREADS", str(editmod.encode_thread_cap())) + + def main() -> None: logging.basicConfig(level=logging.INFO) + _deprioritize() if len(sys.argv) != 2: print("usage: python -m app.downloads.runner ", file=sys.stderr) raise SystemExit(2) diff --git a/backend/tests/test_download_deprioritize.py b/backend/tests/test_download_deprioritize.py new file mode 100644 index 00000000..cfd769f1 --- /dev/null +++ b/backend/tests/test_download_deprioritize.py @@ -0,0 +1,103 @@ +"""Worker-child resource de-prioritization (S4): the encoder thread cap, the `-threads` flag on the +browser-safe re-encode, and the child's startup renice/ionice/OMP handicap. These keep a CPU-heavy +transcode from pegging every core on the box it shares with the API, Postgres, and Plex playback.""" + +import subprocess + +import pytest + +from app.config import settings +from app.downloads import edit as editmod +from app.downloads import runner + + +class TestEncodeThreadCap: + def test_pinned_setting_wins(self, monkeypatch): + monkeypatch.setattr(settings, "download_ffmpeg_threads", 3) + assert editmod.encode_thread_cap() == 3 + + def test_zero_reserves_one_core(self, monkeypatch): + monkeypatch.setattr(settings, "download_ffmpeg_threads", 0) + monkeypatch.setattr(editmod.os, "cpu_count", lambda: 8) + assert editmod.encode_thread_cap() == 7 + + def test_single_core_box_never_caps_below_one(self, monkeypatch): + monkeypatch.setattr(settings, "download_ffmpeg_threads", 0) + monkeypatch.setattr(editmod.os, "cpu_count", lambda: 1) + assert editmod.encode_thread_cap() == 1 + + def test_unknown_cpu_count_falls_back(self, monkeypatch): + monkeypatch.setattr(settings, "download_ffmpeg_threads", 0) + monkeypatch.setattr(editmod.os, "cpu_count", lambda: None) + assert editmod.encode_thread_cap() == 1 # max(1, 2 - 1) + + +class TestBrowserSafeThreads: + def test_threads_flag_added_when_capped(self): + cmd = editmod.build_browser_safe_cmd( + "/in.mkv", "/out.mp4", reencode_video=True, reencode_audio=False, threads=2 + ) + assert "-threads" in cmd + assert cmd[cmd.index("-threads") + 1] == "2" + # The cap precedes the output file, so it binds the encoder (not input decoding). + assert cmd.index("-threads") < cmd.index("/out.mp4") + + @pytest.mark.parametrize("threads", [0, None]) + def test_no_threads_flag_when_uncapped(self, threads): + cmd = editmod.build_browser_safe_cmd( + "/in.mkv", "/out.mp4", reencode_video=True, reencode_audio=False, threads=threads + ) + assert "-threads" not in cmd + + +class TestDeprioritize: + def test_renices_by_configured_increment(self, monkeypatch): + monkeypatch.setattr(settings, "download_child_nice", 12) + monkeypatch.setattr(settings, "download_child_ionice_class", 0) # skip ionice branch + calls = [] + monkeypatch.setattr(runner.os, "nice", lambda n: calls.append(n)) + monkeypatch.setattr(runner.os, "environ", {}) + runner._deprioritize() + assert calls == [12] + + def test_nice_zero_does_not_renice(self, monkeypatch): + monkeypatch.setattr(settings, "download_child_nice", 0) + monkeypatch.setattr(settings, "download_child_ionice_class", 0) + calls = [] + monkeypatch.setattr(runner.os, "nice", lambda n: calls.append(n)) + monkeypatch.setattr(runner.os, "environ", {}) + runner._deprioritize() + assert calls == [] + + def test_sets_omp_budget_to_thread_cap(self, monkeypatch): + monkeypatch.setattr(settings, "download_child_nice", 0) + monkeypatch.setattr(settings, "download_child_ionice_class", 0) + monkeypatch.setattr(settings, "download_ffmpeg_threads", 4) + env: dict = {} + monkeypatch.setattr(runner.os, "environ", env) + runner._deprioritize() + assert env["OMP_NUM_THREADS"] == "4" + + def test_ionice_invoked_when_binary_present(self, monkeypatch): + monkeypatch.setattr(settings, "download_child_nice", 0) + monkeypatch.setattr(settings, "download_child_ionice_class", 3) + monkeypatch.setattr(runner.os, "environ", {}) + monkeypatch.setattr(runner.shutil, "which", lambda name: "/usr/bin/ionice") + ran = [] + monkeypatch.setattr( + runner.subprocess, + "run", + lambda cmd, **kw: ran.append(cmd) or subprocess.CompletedProcess(cmd, 0), + ) + runner._deprioritize() + assert ran and ran[0][:3] == ["ionice", "-c", "3"] + + def test_ionice_skipped_when_binary_absent(self, monkeypatch): + monkeypatch.setattr(settings, "download_child_nice", 0) + monkeypatch.setattr(settings, "download_child_ionice_class", 3) + monkeypatch.setattr(runner.os, "environ", {}) + monkeypatch.setattr(runner.shutil, "which", lambda name: None) + ran = [] + monkeypatch.setattr(runner.subprocess, "run", lambda cmd, **kw: ran.append(cmd)) + runner._deprioritize() + assert ran == [] diff --git a/backend/tests/test_gc.py b/backend/tests/test_gc.py index 30da6bf2..3b2ac10e 100644 --- a/backend/tests/test_gc.py +++ b/backend/tests/test_gc.py @@ -5,6 +5,8 @@ import os import time from pathlib import Path +from app.config import settings +from app.downloads import gc as gcmod from app.downloads.gc import _sweep_orphan_files @@ -69,3 +71,38 @@ def test_sweep_removes_orphans_but_keeps_owned_system_and_fresh(tmp_path: Path): assert not (root / "OldChan").exists() assert sysfile.exists() # system tree untouched assert fresh.exists() # too new to sweep + + +def test_sweep_throttles_after_each_batch(tmp_path: Path, monkeypatch): + """The whole-tree walk yields the IO channel after every `batch` files so it can't monopolize + disk IO in the API process. batch=2 over 5 touched files → 2 pauses (at file 2 and 4).""" + root = tmp_path + d = root / "Chan" + d.mkdir() + for i in range(5): + (d / f"orphan_{i}.dat").write_bytes(b"x") + for f in d.iterdir(): + _aged(f, 5) # old enough that every file is "touched" by the walk + + monkeypatch.setattr(settings, "download_gc_sweep_pause_ms", 10) + monkeypatch.setattr(settings, "download_gc_sweep_batch", 2) + pauses = [] + monkeypatch.setattr(gcmod.time, "sleep", lambda s: pauses.append(s)) + + _sweep_orphan_files(_StubDB([]), str(root)) + + assert pauses == [0.01, 0.01] # 5 files, batch 2 → yield at #2 and #4 + + +def test_sweep_no_throttle_when_pause_disabled(tmp_path: Path, monkeypatch): + (tmp_path / "Chan").mkdir() + f = tmp_path / "Chan" / "orphan.dat" + f.write_bytes(b"x") + _aged(f, 5) + monkeypatch.setattr(settings, "download_gc_sweep_pause_ms", 0) + pauses = [] + monkeypatch.setattr(gcmod.time, "sleep", lambda s: pauses.append(s)) + + _sweep_orphan_files(_StubDB([]), str(tmp_path)) + + assert pauses == [] diff --git a/frontend/src/components/DownloadCenter.tsx b/frontend/src/components/DownloadCenter.tsx index b17c21b7..0fb569ac 100644 --- a/frontend/src/components/DownloadCenter.tsx +++ b/frontend/src/components/DownloadCenter.tsx @@ -51,7 +51,13 @@ export default function DownloadCenter() { const [editJob, setEditJob] = useState(null); const [errorJob, setErrorJob] = useState(null); - const jobsQ = useLiveQuery(qk.downloads(), api.downloads, { intervalMs: 2000 }); + // Poll fast (2s) while work is actually in flight; back off to 15s once the queue is settled + // (only done/error/paused rows left, which don't advance on their own) so an idle Download Center + // isn't hitting the API — and adding to the shared box's load — every 2s for nothing. + const jobsQ = useLiveQuery(qk.downloads(), api.downloads, { + intervalMs: (data) => + data?.some((j) => j.status === "queued" || j.status === "running") ? 2000 : 15000, + }); const sharedQ = useQuery({ queryKey: qk.downloadsShared(), queryFn: api.downloadsShared, diff --git a/frontend/src/components/downloadRow.tsx b/frontend/src/components/downloadRow.tsx index c9227d3e..53f00711 100644 --- a/frontend/src/components/downloadRow.tsx +++ b/frontend/src/components/downloadRow.tsx @@ -278,8 +278,12 @@ export function UsageBar() { const { t } = useTranslation(); // Poll live (like the library list) so the footprint updates the moment the worker finishes an // edit/download — enqueue only creates a queued job (0 bytes), the size lands asynchronously, so - // a one-shot fetch at enqueue time would show a stale 0 B until a manual reload. - const usage = useLiveQuery(qk.downloadUsage(), api.downloadUsage, { intervalMs: 2000 }); + // a one-shot fetch at enqueue time would show a stale 0 B until a manual reload. The footprint + // only moves while a job is in flight, so poll fast (2s) then and idle-back-off to 15s otherwise + // (running_jobs counts queued+running — it flips back to 2s on the first poll after a new enqueue). + const usage = useLiveQuery(qk.downloadUsage(), api.downloadUsage, { + intervalMs: (u) => (u?.running_jobs ? 2000 : 15000), + }); const u = usage.data; if (!u) return null; const pct =