perf(downloads): de-prioritize worker transcodes + idle-back-off polls to stop LAN slowdown
The download/edit child (yt-dlp + ffmpeg) shares one LAN box with the API, Postgres, and Plex playback. A CPU re-encode pegged every core for ~30s and starved web requests + playback until it finished. Full-scope S4 mitigations: - Child self-de-prioritizes at startup (inherited by every ffmpeg/yt-dlp descendant): CPU nice increment + best-effort idle ionice class + an OMP_NUM_THREADS budget, all config-gated and independently disable-able. - Cap the browser-safe re-encode's encoder threads (`-threads`) so a transcode reserves a core for foreground work instead of claiming all of them. - Throttle the retention GC's whole-tree orphan sweep (yield IO after each batch of files) so its rglob can't monopolize disk IO in the API process. - Idle-back-off the Download Center (2s→15s when nothing queued/running) and UsageBar (2s→15s when running_jobs is 0) polls; both snap back to 2s on the next fetch after an enqueue (invalidateDownloads refetches immediately). Tests: encode_thread_cap (pin/reserve-core/edge), -threads flag presence, _deprioritize (nice/ionice/OMP branches), GC sweep throttle batching.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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 <job_id>", file=sys.stderr)
|
||||
raise SystemExit(2)
|
||||
|
||||
@@ -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 == []
|
||||
@@ -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 == []
|
||||
|
||||
@@ -51,7 +51,13 @@ export default function DownloadCenter() {
|
||||
const [editJob, setEditJob] = useState<DownloadJob | null>(null);
|
||||
const [errorJob, setErrorJob] = useState<DownloadJob | null>(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,
|
||||
|
||||
@@ -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 =
|
||||
|
||||
Reference in New Issue
Block a user