diff --git a/hyperwall/__init__.py b/hyperwall/__init__.py index 57ba886..c2756c4 100644 --- a/hyperwall/__init__.py +++ b/hyperwall/__init__.py @@ -7,7 +7,7 @@ from __future__ import annotations -__version__ = "10.13.2" +__version__ = "10.14.0" # Short "major.minor" form, derived — used for User-Agent / Emby auth Version / # window titles so a version bump touches exactly ONE line (this file). VERSION_SHORT = ".".join(__version__.split(".")[:2]) diff --git a/hyperwall/cell.py b/hyperwall/cell.py index ade3ccb..543f395 100644 --- a/hyperwall/cell.py +++ b/hyperwall/cell.py @@ -234,6 +234,11 @@ def __init__(self, controller: Any): self._buffering_card = False self._retry_count = 0 self._force_transcode = False + # True while this cell's CURRENT stream is a server transcode (HLS). + # Read by the controller's transcode-concurrency gate. Kept drift-free + # by re-deriving it from the URL at every stream commit (play + + # advance_to_prefetched); a transcode URL carries an .m3u8 playlist. + self._is_transcoding = False self._played_anything = False self._paused = False # main-thread cache; safe to read cross-thread self._last_next_request_ts = 0.0 @@ -662,6 +667,7 @@ def play(self, item: dict[str, Any], url: str) -> None: # loadfile (replace) clears the mpv playlist tail, so any queued # prefetch entry is gone with it (probed live 2026-07-13). self._prefetched = None + self._is_transcoding = ".m3u8" in url # transcode = HLS playlist URL self._begin_track(item) # Determine if we need to recreate mpv @@ -766,6 +772,7 @@ def advance_to_prefetched(self) -> bool: return False item, _url, sid = self._prefetched self._prefetched = None + self._is_transcoding = ".m3u8" in _url # transcode = HLS playlist URL if STATS_ENABLED: self._flush_stats() if self.muted and self._audio_started: diff --git a/hyperwall/constants.py b/hyperwall/constants.py index b79193e..8c1e384 100644 --- a/hyperwall/constants.py +++ b/hyperwall/constants.py @@ -111,18 +111,24 @@ def _int_env(name: str, default: int, lo: int, hi: int) -> int: # changes — e.g. 2.5 GbE ≈ 2000, 10 GbE ≈ 8000. LINK_MBPS = _int_env("HYPERWALL_LINK_MBPS", 800, 50, 100_000) +# Max cells that may transcode simultaneously. greg's Arc A310 media engine +# handles a few concurrent 1080p transcodes fine, but 6-8 cold starts at once +# (8-cell startup) stampede it → HTTP 500s + partial/pixelated segments +# (2026-07-15). Over this ceiling, heavy clips direct-play instead. 0 = no gate. +MAX_CONCURRENT_TRANSCODES = _int_env("HYPERWALL_MAX_CONCURRENT_TRANSCODES", 4, 0, 64) + def effective_bitrate_budget_mbps(n_cells: int) -> int: """Cell-count-aware direct-play bitrate cap. An explicit HYPERWALL_MAX_DIRECT_BITRATE_MBPS wins verbatim. Otherwise - the default 60 is scaled down as cells go up (8 cells → ~33 Mbps): the + the default 60 is scaled down as cells go up (8 cells → ~50 Mbps): the graduated middle ground between transcode-everything and direct-everything. The cap divides LINK_MBPS across cells with burst - headroom, so the aggregate steady-state stays well inside the link (8 - cells × 33 Mbps ≈ 264 Mbps of an 800 Mbps budget). High-bitrate outliers - transcode server-side, where Emby throttles delivery to ~realtime — - smooth by construction — while the bulk of the library stays direct. + headroom, so the aggregate steady-state stays inside the link (8 + cells × 50 Mbps ≈ 400 Mbps of an 800 Mbps budget). Heavy outliers + transcode server-side (capped in count by MAX_CONCURRENT_TRANSCODES); + the bulk of the library stays direct — the reliable path on this setup. ≤4 cells resolve to the base (measured clean at 4 cells). """ if os.environ.get("HYPERWALL_MAX_DIRECT_BITRATE_MBPS"): diff --git a/hyperwall/reliability.py b/hyperwall/reliability.py index 184cbdf..74b25b0 100644 --- a/hyperwall/reliability.py +++ b/hyperwall/reliability.py @@ -98,18 +98,42 @@ def scale_bitrate_budget_mbps( """Cell-count-aware direct-play bitrate cap (the graduated middle ground between transcode-everything and direct-everything). - Items above the cap transcode server-side, where Emby throttles - delivery to ~realtime — smooth by construction. The cap divides the - usable link (`link_mbps`) across cells with 3x burst headroom; at - <=4 cells it resolves to the configured base (measured clean), at - 8 cells ~33 Mbps so the rare 40-60+ Mbps outliers transcode while the - bulk of the library stays direct. - - `link_mbps` defaults to 800 (greg→skyhawk 1 GbE usable goodput); + Items above the cap transcode server-side. The cap divides the usable + link (`link_mbps`) across cells with 2x burst headroom; at <=4 cells it + resolves to the configured base (measured clean), at 8 cells ~50 Mbps so + only genuinely heavy 50-60+ Mbps outliers transcode while the bulk of the + library stays direct. + + The headroom was relaxed 3x->2x (33->50 Mbps at 8 cells, 2026-07-15): + the original 33 sent ~6 of 8 startup cells to transcode at once and + stampeded greg's media engine (HTTP 500s + partial, pixelated segments). + Fewer files now transcode, and a concurrency gate (see gate_auto_transcode) + caps how many run at once — direct-play is the reliable path on this setup. + + `link_mbps` defaults to 800 (greg->skyhawk 1 GbE usable goodput); callers pass constants.LINK_MBPS so it retunes with the real link. """ n = max(1, int(n_cells)) - return min(base_mbps, max(8, link_mbps // (3 * n))) + return min(base_mbps, max(8, link_mbps // (2 * n))) + + +def gate_auto_transcode( + want_auto: bool, active_transcodes: int, max_concurrent: int, +) -> bool: + """Whether an AUTO transcode should proceed given how many cells are + already transcoding. + + Forced retries (a file that failed direct play and MUST transcode) bypass + this in the caller. Protects greg's Arc A310 / QuickSync media engine from + a cold-start stampede — 8 cells escalating at once produced HTTP 500s and + partial, pixelated HLS segments (2026-07-15). Over the cap, the heavy clip + direct-plays instead (reliable). max_concurrent <= 0 disables the gate. + """ + if not want_auto: + return False + if max_concurrent <= 0: + return True + return active_transcodes < max_concurrent def apply_jitter(delay_s: float, rand: float) -> float: diff --git a/hyperwall/wall.py b/hyperwall/wall.py index 28a601d..4afc452 100644 --- a/hyperwall/wall.py +++ b/hyperwall/wall.py @@ -38,6 +38,7 @@ from .perftrace import traced from .constants import ( MAX_DIRECT_FPS, + MAX_CONCURRENT_TRANSCODES, effective_bitrate_budget_mbps, OUTAGE_MIN_CELLS, OUTAGE_WINDOW_S, @@ -52,7 +53,7 @@ ) from .emby import EmbyClient, ContentLoader from .urls import needs_transcode as _needs_transcode_pure -from .reliability import is_systemic_outage +from .reliability import is_systemic_outage, gate_auto_transcode from .urls import build_stream_url, tag_names from .playlist import PlaylistManager, DEFAULT_GROUP @@ -286,7 +287,7 @@ def _on_items_loaded(self, items: list[dict[str, Any]]) -> None: def _build_url( self, item: dict[str, Any], force_transcode: bool = False, - prefetch: bool = False, + prefetch: bool = False, cell: "VideoCell | None" = None, ) -> tuple[str, str]: iid = item["Id"] key = self.client.access_token @@ -300,7 +301,21 @@ def _build_url( max_fps=MAX_DIRECT_FPS, max_bitrate_mbps=self._bitrate_budget_mbps, ) - transcode = bool(force_transcode or auto_transcode) + # Concurrency gate: a forced retry (failed direct) must transcode, but + # an AUTO escalation defers to direct-play when the transcode engine is + # already busy — never stampede greg's media engine (2026-07-15). + gated = False + if force_transcode: + transcode = True + else: + active = sum( + 1 for c in self.cells + if c is not cell and getattr(c, "_is_transcoding", False) + ) + transcode = gate_auto_transcode( + auto_transcode, active, MAX_CONCURRENT_TRANSCODES, + ) + gated = auto_transcode and not transcode url = build_stream_url( base=base, item_id=iid, api_key=key, session_id=sid, transcode=transcode, @@ -309,7 +324,7 @@ def _build_url( if transcode: tag = "TRANSCODE/retry" if force_transcode else "TRANSCODE/auto" else: - tag = "DIRECT" + tag = "DIRECT/gated" if gated else "DIRECT" if prefetch: tag += "/prefetch" logger.info("[%s] %s", tag, item.get("Name")) @@ -369,7 +384,7 @@ def _hand_off( # with the new stream creation, and Emby can kill both when it sees # a session-stop from the same device. Sessions are cleaned up on # wall shutdown via _cleanup(). - url, sid = self._build_url(item, force_transcode) + url, sid = self._build_url(item, force_transcode, cell=cell) cell._emby_session_id = sid cell._emby_item_id = item["Id"] cell.play(item, url) @@ -384,7 +399,7 @@ def _arm_prefetch(self, cell: VideoCell) -> None: item = self.playlists.next(self._cell_group(cell)) if item is None: return - url, sid = self._build_url(item, prefetch=True) + url, sid = self._build_url(item, prefetch=True, cell=cell) if not cell.prefetch(item, url, sid): logger.debug("Prefetch declined for %s.", item.get("Name", "?")) diff --git a/tests/run_repo_guards.py b/tests/run_repo_guards.py index 0a127cc..44898b4 100644 --- a/tests/run_repo_guards.py +++ b/tests/run_repo_guards.py @@ -46,10 +46,10 @@ def test_01_entry_point_imports(): def test_02_package_identity(): """Package has version and banner.""" from hyperwall import __version__, runtime_banner - assert __version__ == "10.13.2" + assert __version__ == "10.14.0" banner = runtime_banner() assert "Hyperwall" in banner - assert "10.13.2" in banner + assert "10.14.0" in banner def test_03_config_loads(): diff --git a/tests/test_reliability.py b/tests/test_reliability.py index 2e8f70a..17dc013 100644 --- a/tests/test_reliability.py +++ b/tests/test_reliability.py @@ -312,30 +312,30 @@ def test_scale_readahead_shrinks_beyond_four_cells(): def test_scale_bitrate_budget_graduated(): from hyperwall.reliability import scale_bitrate_budget_mbps assert scale_bitrate_budget_mbps(4, 60) == 60 # <=4 cells: base wins - assert scale_bitrate_budget_mbps(8, 60) == 33 # 800/(3*8) - assert scale_bitrate_budget_mbps(16, 60) == 16 + assert scale_bitrate_budget_mbps(8, 60) == 50 # 800/(2*8) — relaxed 3x→2x + assert scale_bitrate_budget_mbps(16, 60) == 25 assert scale_bitrate_budget_mbps(100, 60) == 8 # floor def test_scale_bitrate_budget_tracks_link_speed(): - # The per-file cap divides the usable link across cells with 3x burst + # The per-file cap divides the usable link across cells with 2x burst # headroom — so a faster link raises the cap (up to the base) and a # slower one lowers it. Guards the greg→skyhawk 1 GbE default (800). from hyperwall.reliability import scale_bitrate_budget_mbps - assert scale_bitrate_budget_mbps(8, 60, link_mbps=800) == 33 # 1 GbE + assert scale_bitrate_budget_mbps(8, 60, link_mbps=800) == 50 # 1 GbE assert scale_bitrate_budget_mbps(8, 60, link_mbps=2000) == 60 # 2.5 GbE → base - assert scale_bitrate_budget_mbps(8, 60, link_mbps=400) == 16 # half-link + assert scale_bitrate_budget_mbps(8, 60, link_mbps=400) == 25 # half-link def test_link_mbps_constant_wired_into_effective_budget(): # LINK_MBPS is the single knob; effective_bitrate_budget_mbps must feed it - # through (default 800 → 33 Mbps at 8 cells). + # through (default 800 → 50 Mbps at 8 cells). import os from hyperwall import constants as c assert c.LINK_MBPS == 800 old = os.environ.pop("HYPERWALL_MAX_DIRECT_BITRATE_MBPS", None) try: - assert c.effective_bitrate_budget_mbps(8) == 33 + assert c.effective_bitrate_budget_mbps(8) == 50 finally: if old is not None: os.environ["HYPERWALL_MAX_DIRECT_BITRATE_MBPS"] = old @@ -346,7 +346,7 @@ def test_effective_budget_env_override_wins(monkey=None): from hyperwall import constants as c old = os.environ.pop("HYPERWALL_MAX_DIRECT_BITRATE_MBPS", None) try: - assert c.effective_bitrate_budget_mbps(8) == 33 + assert c.effective_bitrate_budget_mbps(8) == 50 os.environ["HYPERWALL_MAX_DIRECT_BITRATE_MBPS"] = "60" # explicit env wins verbatim even at 8 cells (module constant was # parsed at import; presence of the var is the override signal) @@ -357,6 +357,28 @@ def test_effective_budget_env_override_wins(monkey=None): os.environ["HYPERWALL_MAX_DIRECT_BITRATE_MBPS"] = old +def test_gate_auto_transcode_caps_concurrency(): + from hyperwall.reliability import gate_auto_transcode + # Under the ceiling → allowed; at/over → deferred to direct-play. + assert gate_auto_transcode(True, active_transcodes=0, max_concurrent=4) + assert gate_auto_transcode(True, active_transcodes=3, max_concurrent=4) + assert not gate_auto_transcode(True, active_transcodes=4, max_concurrent=4) + assert not gate_auto_transcode(True, active_transcodes=7, max_concurrent=4) + + +def test_gate_auto_transcode_passthrough_and_disable(): + from hyperwall.reliability import gate_auto_transcode + # A clip that doesn't want transcode never transcodes, gate or not. + assert not gate_auto_transcode(False, active_transcodes=0, max_concurrent=4) + # max_concurrent <= 0 disables the gate entirely. + assert gate_auto_transcode(True, active_transcodes=99, max_concurrent=0) + + +def test_max_concurrent_transcodes_constant(): + from hyperwall import constants as c + assert c.MAX_CONCURRENT_TRANSCODES == 4 + + def test_apply_cache_budget_shape(): from hyperwall.constants import apply_cache_budget out = apply_cache_budget({"demuxer_max_bytes": "512MiB", "vo": "gpu-next"}, 36)