diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index 1f75d071..7ecebc2f 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -157,6 +157,60 @@ _VAAPI_DEVICE = "/dev/dri/renderD128" # full-GPU VAAPI pipeline. A source outside this set (AV1, MPEG-4 ASP/XviD…) falls back to software even # when the render node is present, because the GPU can't decode it to feed its own encoder. _HW_DECODABLE = {"h264", "hevc", "h265", "vp9", "vc1", "mpeg2video"} +# Color-transfer values that mark an HDR source (PQ / HLG). Even a HW-decodable HEVC that is HDR must +# take the software path — see _probe_video_hwsafe. +_HDR_TRANSFERS = {"smpte2084", "arib-std-b67"} + + +def _probe_video_hwsafe(src: Path) -> bool: + """Whether the source is safe for the VAAPI HW transcode pipeline — i.e. 8-bit SDR. A 10-bit (p010) + or HDR source breaks `scale_vaapi=format=nv12` on Gen 9.5 VPP (`VAProfile not supported` → ffmpeg + emits zero segments and the item never plays), so it must take the software libx264 path, which + CPU-decodes and downconverts to 8-bit yuv420p cleanly. Ground truth from ffprobe (bit depth via + pix_fmt / bits_per_raw_sample, HDR via color_transfer); the Plex mirror doesn't store these. + + Returns True (assume HW-safe) on any probe failure: the runtime HW→SW fallback in start_session is + the backstop for a HW session that still produces nothing, so a probe miss only costs the rare + wasted first attempt, never a stranded item.""" + try: + out = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=pix_fmt,bits_per_raw_sample,color_transfer", + "-of", + "default=noprint_wrappers=1", + str(src), + ], + capture_output=True, + text=True, + timeout=10, + ) + except (subprocess.SubprocessError, OSError): + return True + pix_fmt = bits = transfer = "" + for line in out.stdout.splitlines(): + k, _, v = line.strip().partition("=") + v = v.strip().lower() + if k == "pix_fmt": + pix_fmt = v + elif k == "bits_per_raw_sample": + bits = v + elif k == "color_transfer": + transfer = v + if transfer in _HDR_TRANSFERS: + return False # HDR (PQ/HLG) — VPP can't tonemap it to nv12 as configured + if bits.isdigit() and int(bits) > 8: + return False + # pix_fmt bit-depth suffix (yuv420p10le, p010le, yuv444p12le…) signals >8-bit even when + # bits_per_raw_sample is absent (some containers don't set it). + if re.search(r"(?:10|12|16)(?:le|be)$", pix_fmt): + return False + return True def _hwaccel_in_opts(hw: bool) -> list[str]: @@ -220,12 +274,13 @@ def _render_node_present() -> bool: return Path(_VAAPI_DEVICE).exists() -def _want_hw(db: DbSession, item: PlexItem, transcode: bool) -> bool: +def _want_hw(db: DbSession, item: PlexItem, transcode: bool, src: Path) -> bool: """Whether to use the VAAPI hardware encoder for this transcode session, honouring the admin `plex_transcode_hw` policy: "off" → always software; "on" → VAAPI wherever the render node exists - (trusting the admin, skipping the source-decodability gate); "auto" (default) → VAAPI only when the - render node is present AND the iGPU can hardware-decode the source (else the full-GPU pipeline can't - feed itself → software libx264).""" + (trusting the admin, skipping the source gates — the runtime HW→SW fallback still backstops a + session that produces nothing); "auto" (default) → VAAPI only when the render node is present AND + the iGPU can hardware-decode the source AND the source is 8-bit SDR (a 10-bit/HDR source breaks + scale_vaapi=format=nv12 on Gen 9.5 → software libx264, which downconverts cleanly).""" if not transcode: return False mode = (sysconfig.get_str(db, "plex_transcode_hw") or "auto").lower() @@ -233,7 +288,9 @@ def _want_hw(db: DbSession, item: PlexItem, transcode: bool) -> bool: return False if mode == "on": return True - return (item.codec_video or "").lower() in _HW_DECODABLE + if (item.codec_video or "").lower() not in _HW_DECODABLE: + return False + return _probe_video_hwsafe(src) # CPU-nice increment for the SOFTWARE transcode child. The Plex ffmpeg is a Popen in the long-lived API @@ -425,67 +482,82 @@ def start_session( return None key = item.rating_key start_s = max(0.0, float(start_s)) - hw = _want_hw(db, item, transcode) - # Software-fallback encode tuning — read only on the SW transcode path (the VAAPI path and remux - # ignore preset/crf/max_height, and remux must not touch the DB for them). - preset, crf, max_height = "veryfast", 20, 0 - if transcode and not hw: - preset = sysconfig.get_str(db, "plex_transcode_preset") or "veryfast" - crf = sysconfig.get_int(db, "plex_transcode_crf") - max_height = sysconfig.get_int(db, "plex_transcode_max_height") audio_langs = _probe_audio_langs(src) if multi else [] is_multi = len(audio_langs) > 1 # only worth a master playlist when there really are ≥2 tracks - # Nice the SOFTWARE transcode child (not the HW path, and not remux) so a CPU-bound re-encode can't - # starve the co-located API/playback. preexec_fn runs in the forked child (POSIX); on a platform - # without fork (the API always runs in a Linux container in prod) it's simply omitted. - preexec = _sw_child_preexec if (transcode and not hw and os.name == "posix") else None - with _lock: - old = _sessions.pop(key, None) - if old is not None: - _kill(old) - tag = "multi" if is_multi else str(audio_ord) - directory = _HLS_ROOT / f"{key}_{int(start_s)}_{tag}_{aoff:+.2f}" - shutil.rmtree(directory, ignore_errors=True) - directory.mkdir(parents=True, exist_ok=True) - proc = subprocess.Popen( - _ffmpeg_cmd( - src, - start_s, - directory, - audio_ord, - aoff, - audio_langs if is_multi else None, - transcode=transcode, - hw=hw, - preset=preset, - crf=crf, - max_height=max_height, - ), - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - preexec_fn=preexec, - ) - s = HlsSession( - key, - directory, - proc, - start_s, - "master.m3u8" if is_multi else "index.m3u8", - transcode=transcode, - hw=hw, - ) - _sessions[key] = s - # Admin-configurable concurrency cap (Configuration → Plex → Max concurrent transcodes); - # at least 1 so playback can't be capped to zero. - _enforce_cap(max(1, sysconfig.get_int(db, "plex_max_transcodes"))) + tag = "multi" if is_multi else str(audio_ord) + directory = _HLS_ROOT / f"{key}_{int(start_s)}_{tag}_{aoff:+.2f}" - # Wait for the first video segment, then warm up a few more — all outside the lock (this blocks on - # ffmpeg producing them, and we must not hold up other sessions). def _seg(i: int) -> Path: # multi-audio: the video rendition is variant v0 → seg_{i}_0.ts; single-audio → seg_{i}.ts return directory / (f"seg_{i}_0.ts" if is_multi else f"seg_{i}.ts") - if wait_for(_seg(0), timeout=25.0): + def _spawn(hw: bool) -> HlsSession: + # Software-fallback encode tuning — read only on the SW transcode path (the VAAPI path and + # remux ignore preset/crf/max_height, and remux must not touch the DB for them). + preset, crf, max_height = "veryfast", 20, 0 + if transcode and not hw: + preset = sysconfig.get_str(db, "plex_transcode_preset") or "veryfast" + crf = sysconfig.get_int(db, "plex_transcode_crf") + max_height = sysconfig.get_int(db, "plex_transcode_max_height") + # Nice the SOFTWARE transcode child (not the HW path, and not remux) so a CPU-bound re-encode + # can't starve the co-located API/playback. preexec_fn runs in the forked child (POSIX); on a + # platform without fork (the API always runs in a Linux container in prod) it's simply omitted. + preexec = _sw_child_preexec if (transcode and not hw and os.name == "posix") else None + with _lock: + old = _sessions.pop(key, None) + if old is not None: + _kill(old) + shutil.rmtree(directory, ignore_errors=True) + directory.mkdir(parents=True, exist_ok=True) + proc = subprocess.Popen( + _ffmpeg_cmd( + src, + start_s, + directory, + audio_ord, + aoff, + audio_langs if is_multi else None, + transcode=transcode, + hw=hw, + preset=preset, + crf=crf, + max_height=max_height, + ), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + preexec_fn=preexec, + ) + s = HlsSession( + key, + directory, + proc, + start_s, + "master.m3u8" if is_multi else "index.m3u8", + transcode=transcode, + hw=hw, + ) + _sessions[key] = s + # Admin-configurable concurrency cap (Configuration → Plex → Max concurrent transcodes); + # at least 1 so playback can't be capped to zero. + _enforce_cap(max(1, sysconfig.get_int(db, "plex_max_transcodes"))) + return s + + s = _spawn(_want_hw(db, item, transcode, src)) + # Wait for the first video segment outside the lock (this blocks on ffmpeg, and we must not hold up + # other sessions). Pass the proc so a HW session that dies at VAAPI init fails fast (not after 25s). + ready = wait_for(_seg(0), timeout=25.0, proc=s.proc) + if not ready and s.hw: + # Runtime HW→SW fallback: the VAAPI session produced no segments (a VAProfile/init failure the + # _want_hw gates didn't catch — e.g. an admin-forced "on" on a 10-bit source, or an unforeseen + # driver refusal). Respawn on libx264 so a failed GPU session never strands playback; the + # software path CPU-decodes and downconverts every source. s.hw is now False, so the route + # honestly reports "CPU" and the player's Preparing-stream state stays correct. + log.warning( + "plex transcode: HW session produced no segments — falling back to software key=%s", key + ) + s = _spawn(False) + ready = wait_for(_seg(0), timeout=25.0, proc=s.proc) + if ready: # For a seek (start_s>0) learn the REAL start offset K from the first segment's absolute PTS, # so the client's absolute clock + subtitle shift key off the true keyframe, not the requested # offset. On probe failure we keep the provisional requested offset (old behaviour, small drift). @@ -497,7 +569,7 @@ def start_session( # at the first segment that isn't ready in time — ffmpeg produces them faster than realtime once # warm, so a miss means something slow and we proceed rather than block the request further. for i in range(1, _WARMUP_SEGMENTS): - if not wait_for(_seg(i), timeout=_WARMUP_SEG_TIMEOUT_S): + if not wait_for(_seg(i), timeout=_WARMUP_SEG_TIMEOUT_S, proc=s.proc): break log.info( "plex hls session start key=%s req=%.1f real=%.3f multi=%s audio=%s transcode=%s hw=%s", @@ -507,7 +579,7 @@ def start_session( is_multi, audio_ord, transcode, - hw, + s.hw, ) return s @@ -520,10 +592,14 @@ def current_session(key: str) -> HlsSession | None: return s -def wait_for(path: Path, timeout: float = 20.0) -> bool: +def wait_for(path: Path, timeout: float = 20.0, proc: subprocess.Popen | None = None) -> bool: """Wait until a session file (playlist / segment) exists and is non-empty. Segments are produced ~faster than realtime, so a segment just ahead of playback appears quickly; a segment - far beyond the generated region won't (the frontend restarts the session at a seek instead).""" + far beyond the generated region won't (the frontend restarts the session at a seek instead). + + With `proc` given, also stop as soon as that ffmpeg exits: a VAAPI session that fails to init + (10-bit/HDR VAProfile refusal) dies within ~1s producing nothing, and without this we'd block the + whole `timeout` before the HW→SW fallback could kick in — turning the fix into a stall.""" end = time.time() + timeout while time.time() < end: try: @@ -531,6 +607,12 @@ def wait_for(path: Path, timeout: float = 20.0) -> bool: return True except OSError: pass + if proc is not None and proc.poll() is not None: + # the producer exited — no more segments are coming; one last check then give up. + try: + return path.exists() and path.stat().st_size > 0 + except OSError: + return False time.sleep(0.15) return path.exists() diff --git a/backend/tests/test_plex_stream.py b/backend/tests/test_plex_stream.py index 00c55ece..7db5acb3 100644 --- a/backend/tests/test_plex_stream.py +++ b/backend/tests/test_plex_stream.py @@ -269,48 +269,92 @@ class TestTranscodeCommand: class TestWantHw: - """VAAPI is used only where the render node was passed in AND (in auto) the source is HW-decodable, - with an admin `plex_transcode_hw` policy (auto/on/off) on top.""" + """VAAPI is used only where the render node was passed in AND (in auto) the source is HW-decodable + AND 8-bit SDR, with an admin `plex_transcode_hw` policy (auto/on/off) on top.""" + + SRC = Path("/media/movie.mkv") @staticmethod def _policy(monkeypatch, mode): monkeypatch.setattr(stream.sysconfig, "get_str", lambda db, k: mode) + @staticmethod + def _hwsafe(monkeypatch, safe): + # Default the bit-depth probe so the auto-decodable tests don't shell out to ffprobe. + monkeypatch.setattr(stream, "_probe_video_hwsafe", lambda src: safe) + def test_no_hw_when_not_transcoding(self, monkeypatch): self._policy(monkeypatch, "auto") monkeypatch.setattr(stream, "_render_node_present", lambda: True) - assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=False) is False + assert ( + stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=False, src=self.SRC) + is False + ) def test_no_hw_without_a_render_node(self, monkeypatch): self._policy(monkeypatch, "auto") monkeypatch.setattr(stream, "_render_node_present", lambda: False) - assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False + assert ( + stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC) + is False + ) - def test_auto_uses_hw_for_a_decodable_source_with_the_render_node(self, monkeypatch): + def test_auto_uses_hw_for_a_decodable_8bit_source_with_the_render_node(self, monkeypatch): self._policy(monkeypatch, "auto") + self._hwsafe(monkeypatch, True) monkeypatch.setattr(stream, "_render_node_present", lambda: True) - assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is True + assert ( + stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC) + is True + ) def test_auto_falls_back_to_software_for_a_non_decodable_source(self, monkeypatch): # AV1 / MPEG-4 ASP (XviD) aren't decodable on Gen 9.5 → the GPU can't feed its own encoder. self._policy(monkeypatch, "auto") + self._hwsafe(monkeypatch, True) # decodability gate must fire BEFORE the probe monkeypatch.setattr(stream, "_render_node_present", lambda: True) - assert stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True) is False - assert stream._want_hw(None, _FakeItem(codec_video="mpeg4"), transcode=True) is False + assert ( + stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True, src=self.SRC) + is False + ) + assert ( + stream._want_hw(None, _FakeItem(codec_video="mpeg4"), transcode=True, src=self.SRC) + is False + ) + + def test_auto_falls_back_to_software_for_a_10bit_or_hdr_source(self, monkeypatch): + # BUG 1: a decodable HEVC that is 10-bit/HDR still breaks scale_vaapi=format=nv12 → software. + self._policy(monkeypatch, "auto") + self._hwsafe(monkeypatch, False) + monkeypatch.setattr(stream, "_render_node_present", lambda: True) + assert ( + stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC) + is False + ) def test_off_forces_software_even_for_a_decodable_source(self, monkeypatch): self._policy(monkeypatch, "off") monkeypatch.setattr(stream, "_render_node_present", lambda: True) - assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False + assert ( + stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC) + is False + ) - def test_on_forces_hw_even_for_a_non_decodable_source(self, monkeypatch): - # The admin override trusts the GPU to decode it; skips the decodability gate (render node still - # required — HW is impossible without it). + def test_on_forces_hw_and_skips_both_source_gates(self, monkeypatch): + # The admin override trusts the GPU; skips the decodability AND the bit-depth gate (the runtime + # HW→SW fallback still backstops it). Render node still required — HW is impossible without it. self._policy(monkeypatch, "on") + self._hwsafe(monkeypatch, False) # would veto in auto; "on" ignores it monkeypatch.setattr(stream, "_render_node_present", lambda: True) - assert stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True) is True + assert ( + stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True, src=self.SRC) + is True + ) monkeypatch.setattr(stream, "_render_node_present", lambda: False) - assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False + assert ( + stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC) + is False + ) class TestColdStartWarmup: @@ -333,7 +377,7 @@ class TestColdStartWarmup: def test_warmup_waits_for_the_first_few_segments(self, fake_env, monkeypatch): waited = [] - def spy(path, timeout=20.0): + def spy(path, timeout=20.0, proc=None): waited.append(path.name) return True @@ -346,7 +390,7 @@ class TestColdStartWarmup: def test_warmup_stops_at_the_first_segment_that_isnt_ready(self, fake_env, monkeypatch): waited = [] - def spy(path, timeout=20.0): + def spy(path, timeout=20.0, proc=None): waited.append(path.name) return path.name != "seg_1.ts" # seg_1 never appears → warm-up gives up there @@ -360,7 +404,7 @@ class TestColdStartWarmup: # still returned so the client's own segment requests can drive it, as before. waited = [] - def spy(path, timeout=20.0): + def spy(path, timeout=20.0, proc=None): waited.append(path.name) return False @@ -398,7 +442,7 @@ class TestTranscodeStartSession: "get_str", lambda db, k: {"plex_transcode_preset": "slow"}.get(k, "auto"), ) - monkeypatch.setattr(stream, "wait_for", lambda path, timeout=20.0: True) + monkeypatch.setattr(stream, "wait_for", lambda path, timeout=20.0, proc=None: True) _CapturingProc.calls = [] monkeypatch.setattr(stream.subprocess, "Popen", _CapturingProc) return tmp_path @@ -406,7 +450,7 @@ class TestTranscodeStartSession: def test_software_transcode_threads_the_knobs_into_the_command( self, capturing_env, monkeypatch ): - monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: False) + monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode, src: False) s = stream.start_session( db=None, item=_FakeItem(codec_video="mpeg4"), start_s=0.0, transcode=True ) @@ -421,11 +465,12 @@ class TestTranscodeStartSession: assert kwargs.get("preexec_fn") is stream._sw_child_preexec def test_hardware_transcode_uses_the_gpu_and_never_nices(self, capturing_env, monkeypatch): - monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: True) + monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode, src: True) s = stream.start_session( db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True ) assert s is not None and s.transcode is True and s.hw is True + assert len(_CapturingProc.calls) == 1 # first segment arrived — no HW→SW fallback respawn (cmd, *_), kwargs = _CapturingProc.calls[0] assert _venc(cmd) == "h264_vaapi" assert "libx264" not in cmd @@ -434,12 +479,82 @@ class TestTranscodeStartSession: ) # no SW cap assert kwargs.get("preexec_fn") is None # GPU encode is cheap — never niced + def test_a_hw_session_that_produces_nothing_falls_back_to_software( + self, capturing_env, monkeypatch + ): + # BUG 1 backstop: _want_hw picked HW but the VAAPI session emits no first segment (a VAProfile + # refusal the gates didn't catch). start_session must respawn on libx264 and report hw=False so + # the item still plays and the route/player honestly show CPU. + monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode, src: True) + seen = {"n": 0} + + def flaky_wait(path, timeout=20.0, proc=None): + seen["n"] += 1 + return seen["n"] > 1 # the first seg-0 wait (HW) fails; every wait after succeeds (SW) + + monkeypatch.setattr(stream, "wait_for", flaky_wait) + s = stream.start_session( + db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True + ) + assert ( + s is not None and s.transcode is True and s.hw is False + ) # honest CPU after the fallback + assert len(_CapturingProc.calls) == 2 # the HW attempt, then the SW respawn + (hw_cmd, *_), _ = _CapturingProc.calls[0] + (sw_cmd, *_), sw_kwargs = _CapturingProc.calls[1] + assert _venc(hw_cmd) == "h264_vaapi" + assert _venc(sw_cmd) == "libx264" + if os.name == "posix": + assert sw_kwargs.get("preexec_fn") is stream._sw_child_preexec # the SW child is niced + def test_remux_reads_no_transcode_knobs_and_stream_copies(self, capturing_env, monkeypatch): # A remux start must stream-copy the video and spawn no niced child (the transcode knobs and # the preexec nice are gated behind `transcode=True`). - monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: False) + monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode, src: False) s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0) # transcode defaults False assert s is not None and s.transcode is False (cmd, *_), kwargs = _CapturingProc.calls[0] assert _venc(cmd) == "copy" assert kwargs.get("preexec_fn") is None + + +class TestProbeVideoHwsafe: + """The ffprobe-based 8-bit-SDR gate that keeps 10-bit/HDR sources off the VAAPI pipeline (BUG 1): + scale_vaapi=format=nv12 can't convert them on Gen 9.5, so they must take software libx264.""" + + @staticmethod + def _probe(monkeypatch, stdout): + class _R: + def __init__(self): + self.stdout = stdout + + monkeypatch.setattr(stream.subprocess, "run", lambda *a, **k: _R()) + + def test_8bit_sdr_is_hw_safe(self, monkeypatch): + self._probe(monkeypatch, "pix_fmt=yuv420p\nbits_per_raw_sample=8\ncolor_transfer=bt709\n") + assert stream._probe_video_hwsafe(Path("/x")) is True + + def test_10bit_pix_fmt_is_not_hw_safe(self, monkeypatch): + # p010/yuv420p10le breaks the VPP format pass even when bits_per_raw_sample is absent. + self._probe(monkeypatch, "pix_fmt=yuv420p10le\ncolor_transfer=bt2020-10\n") + assert stream._probe_video_hwsafe(Path("/x")) is False + + def test_bits_per_raw_sample_over_8_is_not_hw_safe(self, monkeypatch): + self._probe(monkeypatch, "pix_fmt=yuv420p\nbits_per_raw_sample=10\n") + assert stream._probe_video_hwsafe(Path("/x")) is False + + def test_hdr_transfer_is_not_hw_safe(self, monkeypatch): + # Even an 8-bit-tagged PQ/HLG source can't be tonemapped to nv12 as configured. + self._probe( + monkeypatch, "pix_fmt=yuv420p\nbits_per_raw_sample=8\ncolor_transfer=smpte2084\n" + ) + assert stream._probe_video_hwsafe(Path("/x")) is False + + def test_probe_failure_assumes_hw_safe(self, monkeypatch): + # ffprobe missing/erroring → assume safe; the runtime HW→SW fallback is the real backstop, so a + # probe miss costs at most one wasted HW attempt, never a stranded item. + def boom(*a, **k): + raise OSError("no ffprobe") + + monkeypatch.setattr(stream.subprocess, "run", boom) + assert stream._probe_video_hwsafe(Path("/x")) is True