perf(dash): collapse single-URL byte-range segments into one direct download

Also throttle hedge-median recompute to 0.5s, cache replace_fields format
patterns and short-circuit when no $ token, and stream HLS merge with
copyfileobj. Adds test_dash_collapse and live A/B tooling.
This commit is contained in:
imSp4rky
2026-07-16 12:42:16 -06:00
parent 78d80eb68a
commit 7eee34014e
6 changed files with 428 additions and 97 deletions
+53 -11
View File
@@ -7,6 +7,9 @@ The controller logs its worker-target changes to a JSON-lines debug log; this sc
reads them back and prints the scaling sequence per stream, so you can see whether the
downloader ramps workers up against each CDN or finishes before the controller acts.
STREAMS carries a public 4K/HDR/Dolby-Vision tier (labels uhd-hevc, dv-atmos, hevc-hls,
h265-cmaf) for exercising large high-bitrate tracks; filter it with --only uhd / dv / hevc / h265.
Throwaway-ish operational tool: it makes real network requests to public test assets.
Run: uv run python scripts/live_manifest_test.py [--workers N] [--track smallest|largest]
"""
@@ -17,7 +20,6 @@ import json
import shutil
import sys
import time
from functools import partial
from pathlib import Path
from tempfile import mkdtemp
from typing import Any, Optional
@@ -33,16 +35,51 @@ from unshackle.core.utilities import close_debug_logger, init_debug_logger
# all three formats on Unified Streaming, so the formats are comparable on one CDN.
STREAMS: list[tuple[str, Any, str, str]] = [
("HLS", HLS, "mux", "https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8"),
("HLS", HLS, "unified", "https://demo.unified-streaming.com/k8s/features/stable/video/tears-of-steel/tears-of-steel.ism/.m3u8"),
(
"HLS",
HLS,
"unified",
"https://demo.unified-streaming.com/k8s/features/stable/video/tears-of-steel/tears-of-steel.ism/.m3u8",
),
("DASH", DASH, "akamai", "https://dash.akamaized.net/akamai/bbb_30fps/bbb_30fps.mpd"),
("DASH", DASH, "unified", "https://demo.unified-streaming.com/k8s/features/stable/video/tears-of-steel/tears-of-steel.ism/.mpd"),
(
"DASH",
DASH,
"unified",
"https://demo.unified-streaming.com/k8s/features/stable/video/tears-of-steel/tears-of-steel.ism/.mpd",
),
("DASH", DASH, "dashif", "https://livesim2.dashif.org/vod/testpic_2s/Manifest.mpd"),
("ISM", ISM, "unified", "https://demo.unified-streaming.com/k8s/features/stable/video/tears-of-steel/tears-of-steel.ism/Manifest"),
("HLS", HLS, "apple", "https://devstreaming-cdn.apple.com/videos/streaming/examples/img_bipbop_adv_example_fmp4/master.m3u8"),
(
"ISM",
ISM,
"unified",
"https://demo.unified-streaming.com/k8s/features/stable/video/tears-of-steel/tears-of-steel.ism/Manifest",
),
(
"HLS",
HLS,
"apple",
"https://devstreaming-cdn.apple.com/videos/streaming/examples/img_bipbop_adv_example_fmp4/master.m3u8",
),
("HLS", HLS, "gcs", "https://storage.googleapis.com/shaka-demo-assets/angel-one-hls/hls.m3u8"),
("DASH", DASH, "gcs", "https://storage.googleapis.com/shaka-demo-assets/angel-one/dash.mpd"),
("DASH", DASH, "axinom", "https://media.axprod.net/TestVectors/v7-Clear/Manifest_1080p.mpd"),
("ISM", ISM, "dtaps", "https://playready.directtaps.net/smoothstreaming/SSWSS720H264/SuperSpeedway_720.ism/Manifest"),
(
"ISM",
ISM,
"dtaps",
"https://playready.directtaps.net/smoothstreaming/SSWSS720H264/SuperSpeedway_720.ism/Manifest",
),
# public 4K / HDR / Dolby-Vision tier: large high-bitrate tracks, no auth
("DASH", DASH, "uhd-hevc", "https://dash.akamaized.net/dash264/TestCasesUHD/2b/11/MultiRate.mpd"),
("HLS", HLS, "dv-atmos", "https://devstreaming-cdn.apple.com/videos/streaming/examples/adv_dv_atmos/main.m3u8"),
(
"HLS",
HLS,
"hevc-hls",
"https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_adv_example_hevc/master.m3u8",
),
("DASH", DASH, "h265-cmaf", "https://media.axprod.net/TestVectors/H265/clear_cmaf_1080p_h265/manifest.mpd"),
]
@@ -97,18 +134,20 @@ def run_one(label: str, parser: Any, cdn: str, url: str, workers: int, which: st
save_path = work / "out.mp4"
counter = Counter()
# adaptive_workers only exists on the downloader-speed branch; skip on older checkouts
extra = {"adaptive_workers": True} if "adaptive_workers" in getattr(DownloadContext, "__dataclass_fields__", {}) else {}
extra = (
{"adaptive_workers": True} if "adaptive_workers" in getattr(DownloadContext, "__dataclass_fields__", {}) else {}
)
ctx = DownloadContext(
save_path=save_path,
save_dir=work / "segs",
progress=partial(counter.__call__),
progress=counter,
session=session,
proxy=None,
max_workers=workers,
**extra,
)
ctx.progress = counter # type: ignore[assignment]
track.drm = None # download-speed only: time raw segment download, no license/decrypt
t0 = time.time()
err = ""
try:
@@ -123,14 +162,17 @@ def run_one(label: str, parser: Any, cdn: str, url: str, workers: int, which: st
moves, _ = read_new_transitions(log_path, line0)
scaling = " ".join(moves) if moves else "none (finished under controller tick)"
res = f"{getattr(track, 'width', '?')}x{getattr(track, 'height', '?')}"
return f"{label:5} {cdn:8} {dt:6.1f}s {size/1e6:7.1f}MB {mbps:6.1f}MB/s [{res}] workers: {scaling}{err}"
return f"{label:5} {cdn:8} {dt:6.1f}s {size / 1e6:7.1f}MB {mbps:6.1f}MB/s [{res}] workers: {scaling}{err}"
@click.command()
@click.option("--workers", default=16, help="max workers (the adaptive ceiling)")
@click.option("--track", "which", type=click.Choice(["smallest", "largest"]), default="largest")
@click.option("--only", default="", help="substring filter on CDN/format labels (e.g. 'gcs' or 'ISM')")
@click.option("--only", default="", help="substring filter on CDN/format labels (e.g. 'apple' or 'ISM')")
def main(workers: int, which: str, only: str) -> None:
# download-speed only: skip HLS key handling so segments just download, no license server
HLS.get_drm = staticmethod(lambda *a, **k: None) # type: ignore[assignment]
log_path = Path(mkdtemp(prefix="live_dbg_")) / "debug.jsonl"
init_debug_logger(log_path=log_path, enabled=True)
print(f"debug log: {log_path}\ntrack: {which}, max_workers: {workers}\n")
+130
View File
@@ -0,0 +1,130 @@
"""Live A/B test for SegmentBase (byte-range) DASH downloads.
Drives the real DASH pipeline against public SegmentBase test manifests and reports
wall time, size, throughput and output validity (ffprobe). On branches that have the
single-URL collapse (DASH._collapsible_single_url) it also reports whether the
collapse fired; on checkouts without it (e.g. dev) the same script runs the classic
per-segment + merge path, so running it from both checkouts gives an A/B comparison:
uv run python scripts/live_segmentbase_ab.py --track largest
/path/to/dev/.venv/bin/python scripts/live_segmentbase_ab.py --track largest
Throwaway-ish operational tool: it makes real network requests to public test assets.
"""
from __future__ import annotations
import shutil
import subprocess
import time
from pathlib import Path
from tempfile import mkdtemp
from typing import Any, Optional
import click
from requests import Session
from unshackle.core.manifests import DASH
from unshackle.core.tracks.track import DownloadContext
# (label, SegmentBase MPD). netflix-1a is collapse-eligible (Initialization range from
# byte 0); sony-1a has no Initialization range so every checkout takes the classic path.
STREAMS: list[tuple[str, str]] = [
("netflix-1a", "https://dash.akamaized.net/dash264/TestCases/1a/netflix/exMPD_BIP_TC1.mpd"),
("sony-1a", "https://dash.akamaized.net/dash264/TestCases/1a/sony/SNE_DASH_SD_CASE1A_REVISED.mpd"),
]
guard_results: list[bool] = []
def install_guard_spy() -> bool:
"""Record collapse decisions when this checkout has the guard; no-op otherwise."""
orig = getattr(DASH, "_collapsible_single_url", None)
if orig is None:
return False
def spy(is_subtitle: bool, segments: Any, init_len: Optional[int]) -> bool:
result = orig(is_subtitle, segments, init_len)
guard_results.append(result)
return result
DASH._collapsible_single_url = staticmethod(spy) # type: ignore[method-assign]
return True
def ffprobe(path: Path) -> str:
if not path.exists() or not shutil.which("ffprobe"):
return "n/a"
r = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(path)],
capture_output=True,
text=True,
)
return f"{float(r.stdout.strip()):.0f}s" if r.stdout.strip() else "INVALID"
def run_one(label: str, url: str, which: str, workers: int, has_guard: bool) -> str:
guard_results.clear()
session = Session()
try:
tracks = DASH.from_url(url=url, session=session).to_tracks(language="en")
except Exception as e: # noqa: BLE001
return f"{label:12} PARSE ERROR {type(e).__name__}: {e}"
vids = list(tracks.videos)
if not vids:
return f"{label:12} NO VIDEO TRACKS"
key = lambda t: t.bitrate or 0 # noqa: E731
track = min(vids, key=key) if which == "smallest" else max(vids, key=key)
work = Path(mkdtemp(prefix="segbase_ab_"))
save_path = work / "out.mp4"
# adaptive_workers only exists on the downloader-speed branch; skip elsewhere
extra = {"adaptive_workers": True} if "adaptive_workers" in DownloadContext.__dataclass_fields__ else {}
ctx = DownloadContext(
save_path=save_path,
save_dir=work / "segs",
progress=lambda **ev: None,
session=session,
proxy=None,
max_workers=workers,
**extra,
)
t0 = time.time()
err = ""
try:
DASH.download_track(track=track, ctx=ctx)
except Exception as e: # noqa: BLE001
err = f" ERROR {type(e).__name__}: {e}"
dt = time.time() - t0
size = save_path.stat().st_size if save_path.exists() else 0
mbps = (size / 1e6 / dt) if dt > 0 and size else 0.0
probe = ffprobe(save_path)
shutil.rmtree(work, ignore_errors=True)
if has_guard:
collapse = "yes" if guard_results and guard_results[0] else "no"
else:
collapse = "unsupported"
res = f"{getattr(track, 'width', '?')}x{getattr(track, 'height', '?')}"
return (
f"{label:12} {dt:6.1f}s {size / 1e6:8.1f}MB {mbps:6.1f}MB/s [{res}] "
f"collapse: {collapse:11} duration: {probe}{err}"
)
@click.command()
@click.option("--workers", default=16, show_default=True, help="max workers (adaptive cap where supported)")
@click.option("--track", "which", type=click.Choice(["smallest", "largest"]), default="smallest", show_default=True)
@click.option("--only", default="", help="substring filter on stream labels")
def main(workers: int, which: str, only: str) -> None:
has_guard = install_guard_spy()
print(f"track: {which}, max_workers: {workers}, collapse guard present: {has_guard}\n")
for label, url in STREAMS:
if only and only.lower() not in label.lower():
continue
print(run_one(label, url, which, workers, has_guard))
if __name__ == "__main__":
main()
+68
View File
@@ -0,0 +1,68 @@
"""Offline unit tests for DASH._collapsible_single_url, the guard that decides when a
byte-range track (SegmentBase / ranged SegmentList) can be downloaded as one whole
resource instead of per-segment files + merge."""
from __future__ import annotations
from typing import Optional
from unshackle.core.manifests import DASH
URL = "https://cdn/media.mp4"
def collapsible(segments, init_len: Optional[int], is_subtitle: bool = False) -> bool:
return DASH._collapsible_single_url(is_subtitle, segments, init_len)
def test_segmentbase_single_range_eligible():
# init occupies [0, 863); one media segment covers the rest of the same URL
assert collapsible([(URL, "863-999999")], init_len=863) is True
def test_ranged_segmentlist_contiguous_eligible():
segments = [(URL, "863-1999"), (URL, "2000-3999"), (URL, "4000-5999")]
assert collapsible(segments, init_len=863) is True
def test_subtitle_never_eligible():
assert collapsible([(URL, "863-999999")], init_len=863, is_subtitle=True) is False
def test_missing_init_len_ineligible():
assert collapsible([(URL, "863-999999")], init_len=None) is False
def test_missing_range_ineligible():
assert collapsible([(URL, None)], init_len=863) is False
assert collapsible([(URL, "863-1999"), (URL, None)], init_len=863) is False
def test_mixed_urls_ineligible():
segments = [(URL, "863-1999"), ("https://cdn/other.mp4", "2000-3999")]
assert collapsible(segments, init_len=863) is False
def test_first_segment_gap_after_init_ineligible():
# first media byte does not immediately follow the init prefix
assert collapsible([(URL, "900-1999")], init_len=863) is False
def test_overlapping_ranges_ineligible():
segments = [(URL, "863-2500"), (URL, "2000-3999")]
assert collapsible(segments, init_len=863) is False
def test_non_monotonic_order_ineligible():
segments = [(URL, "4000-5999"), (URL, "863-1999")]
assert collapsible(segments, init_len=863) is False
def test_malformed_range_ineligible():
assert collapsible([(URL, "863")], init_len=863) is False
assert collapsible([(URL, "abc-def")], init_len=863) is False
assert collapsible([(URL, "1999-863")], init_len=863) is False
def test_empty_segments_ineligible():
assert collapsible([], init_len=863) is False
+15 -2
View File
@@ -1084,6 +1084,9 @@ def requests(
total_bytes = 0
start_time = time.time()
last_speed_report = start_time
last_hedge_check = 0.0
hedge_median = 0.0
hedge_median_len = 0
pool = ThreadPoolExecutor(max_workers=max_workers)
event_queue: Queue[dict[str, Any]] = Queue()
@@ -1362,8 +1365,18 @@ def requests(
last_speed_report = now
# hedge stuck segments once spare workers exist within the active target
if len(pending) < target and seg_durations and not DOWNLOAD_CANCELLED.is_set():
threshold = max(HEDGE_FACTOR * statistics.median(seg_durations), HEDGE_MIN_WAIT)
if (
len(pending) < target
and seg_durations
and now - last_hedge_check > 0.5
and not DOWNLOAD_CANCELLED.is_set()
):
last_hedge_check = now
cur_len = len(seg_durations) # len() is safe here even though workers append under seg_lock
if cur_len != hedge_median_len:
hedge_median = statistics.median(seg_durations)
hedge_median_len = cur_len
threshold = max(HEDGE_FACTOR * hedge_median, HEDGE_MIN_WAIT)
with seg_lock:
stuck = [
i
+161 -83
View File
@@ -8,7 +8,7 @@ import re
import shutil
import sys
from copy import deepcopy
from functools import partial
from functools import lru_cache, partial
from typing import Any, Callable, Optional, Union
from urllib.parse import urljoin, urlparse
from uuid import UUID
@@ -448,22 +448,46 @@ class DASH:
downloader = track.downloader
downloader_args = dict(
urls=[
{"url": url, "headers": {"Range": f"bytes={bytes_range}"} if bytes_range else {}}
for url, bytes_range in segments
],
output_dir=save_dir,
filename="{i:0%d}.mp4" % (len(str(len(segments)))),
headers=session.headers,
cookies=session.cookies,
proxy=proxy,
max_workers=max_workers,
session=session,
adaptive=adaptive,
processes=processes,
# When every segment is a byte range of one parent resource and init is its [0, len)
# prefix, the whole resource is itself a valid MP4, so one direct download to save_path
# replaces the per-segment files, init write and merge pass.
collapse_single_url = DASH._collapsible_single_url(
isinstance(track, Subtitle),
segments,
len(init_data) if init_data is not None else None,
)
if collapse_single_url:
# must carry no Range header, or the downloader would skip its ranged-parallel path
downloader_args = dict(
urls=[{"url": segments[0][0]}],
output_dir=save_path.parent,
filename=save_path.name,
headers=session.headers,
cookies=session.cookies,
proxy=proxy,
max_workers=max_workers,
session=session,
adaptive=adaptive,
processes=processes,
)
else:
downloader_args = dict(
urls=[
{"url": url, "headers": {"Range": f"bytes={bytes_range}"} if bytes_range else {}}
for url, bytes_range in segments
],
output_dir=save_dir,
filename="{i:0%d}.mp4" % (len(str(len(segments)))),
headers=session.headers,
cookies=session.cookies,
proxy=proxy,
max_workers=max_workers,
session=session,
adaptive=adaptive,
processes=processes,
)
log_event(
"manifest_dash_download_start",
level="DEBUG",
@@ -489,84 +513,87 @@ class DASH:
status_update["downloaded"] = f"DASH {downloaded}"
progress(**status_update)
# Verify output directory exists and contains files
if not save_dir.exists():
error_msg = f"Output directory does not exist: {save_dir}"
if not collapse_single_url:
# Verify output directory exists and contains files
if not save_dir.exists():
error_msg = f"Output directory does not exist: {save_dir}"
log_event(
"manifest_dash_download_output_missing",
level="ERROR",
message=error_msg,
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"save_dir": str(save_dir),
"save_path": str(save_path),
"downloader": "requests",
},
)
raise FileNotFoundError(error_msg)
for control_file in save_dir.glob("*.!dev"):
control_file.unlink(missing_ok=True)
segments_to_merge = [x for x in sorted(save_dir.iterdir()) if x.is_file()]
log_event(
"manifest_dash_download_output_missing",
level="ERROR",
message=error_msg,
"manifest_dash_download_complete",
level="DEBUG",
message="DASH download complete, preparing to merge",
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"save_dir": str(save_dir),
"save_path": str(save_path),
"save_dir_exists": save_dir.exists(),
"segments_found": len(segments_to_merge),
"segment_files": [f.name for f in segments_to_merge[:10]], # Limit to first 10
"downloader": "requests",
},
)
raise FileNotFoundError(error_msg)
for control_file in save_dir.glob("*.!dev"):
control_file.unlink(missing_ok=True)
if not segments_to_merge:
error_msg = f"No segment files found in output directory: {save_dir}"
# List all contents of the directory for debugging
all_contents = list(save_dir.iterdir()) if save_dir.exists() else []
log_event(
"manifest_dash_download_no_segments",
level="ERROR",
message=error_msg,
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"save_dir": str(save_dir),
"directory_contents": [str(p) for p in all_contents],
"downloader": "requests",
},
)
raise FileNotFoundError(error_msg)
segments_to_merge = [x for x in sorted(save_dir.iterdir()) if x.is_file()]
log_event(
"manifest_dash_download_complete",
level="DEBUG",
message="DASH download complete, preparing to merge",
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"save_dir": str(save_dir),
"save_dir_exists": save_dir.exists(),
"segments_found": len(segments_to_merge),
"segment_files": [f.name for f in segments_to_merge[:10]], # Limit to first 10
"downloader": "requests",
},
)
if not segments_to_merge:
error_msg = f"No segment files found in output directory: {save_dir}"
# List all contents of the directory for debugging
all_contents = list(save_dir.iterdir()) if save_dir.exists() else []
log_event(
"manifest_dash_download_no_segments",
level="ERROR",
message=error_msg,
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"save_dir": str(save_dir),
"directory_contents": [str(p) for p in all_contents],
"downloader": "requests",
},
is_text_subtitle = (
not drm
and isinstance(track, Subtitle)
and track.codec not in (Subtitle.Codec.fVTT, Subtitle.Codec.fTTML)
)
raise FileNotFoundError(error_msg)
is_text_subtitle = (
not drm and isinstance(track, Subtitle) and track.codec not in (Subtitle.Codec.fVTT, Subtitle.Codec.fTTML)
)
with open(save_path, "wb") as f:
if init_data:
f.write(init_data)
if len(segments_to_merge) > 1:
progress(downloaded="Merging", completed=0, total=len(segments_to_merge))
for segment_file in segments_to_merge:
if is_text_subtitle:
segment_data = try_ensure_utf8(segment_file.read_bytes())
segment_data = (
segment_data.decode("utf8")
.replace("&lrm;", html.unescape("&lrm;"))
.replace("&rlm;", html.unescape("&rlm;"))
.encode("utf8")
)
f.write(segment_data)
else:
with open(segment_file, "rb") as src:
shutil.copyfileobj(src, f, 1024 * 1024)
segment_file.unlink()
progress(advance=1)
with open(save_path, "wb") as f:
if init_data:
f.write(init_data)
if len(segments_to_merge) > 1:
progress(downloaded="Merging", completed=0, total=len(segments_to_merge))
for segment_file in segments_to_merge:
if is_text_subtitle:
segment_data = try_ensure_utf8(segment_file.read_bytes())
segment_data = (
segment_data.decode("utf8")
.replace("&lrm;", html.unescape("&lrm;"))
.replace("&rlm;", html.unescape("&rlm;"))
.encode("utf8")
)
f.write(segment_data)
else:
with open(segment_file, "rb") as src:
shutil.copyfileobj(src, f, 1024 * 1024)
segment_file.unlink()
progress(advance=1)
track.path = save_path
events.emit(events.Types.TRACK_DOWNLOADED, track=track)
@@ -588,6 +615,49 @@ class DASH:
progress(downloaded="Downloaded")
@staticmethod
def _collapsible_single_url(
is_subtitle: bool,
segments: list[tuple[str, Optional[str]]],
init_len: Optional[int],
) -> bool:
"""
Whether every segment is a byte range of one parent resource whose [0, init_len)
prefix is the init segment, so the whole resource can be downloaded in one pass.
Returns False for subtitle tracks, an unknown init length, mixed URLs, malformed
ranges, a gap between the init prefix and the first media byte, and ranges that
overlap or regress in document order.
"""
if is_subtitle or init_len is None or not segments:
return False
first_url = segments[0][0]
ranges: list[tuple[int, int]] = []
for url, bytes_range in segments:
if url != first_url or not bytes_range:
return False
parts = bytes_range.split("-")
if len(parts) != 2 or not parts[0] or not parts[1]:
return False
try:
start, end = int(parts[0]), int(parts[1])
except ValueError:
return False
if end < start:
return False
ranges.append((start, end))
if min(start for start, _ in ranges) != init_len:
return False
prev_end = init_len - 1
for start, end in ranges:
if start <= prev_end:
return False
prev_end = end
return True
@staticmethod
def _is_content_period(period: Element, filtered_period_ids: list[str]) -> bool:
"""Check if a period is a valid content period (not an ad, not filtered, not trick mode)."""
@@ -1107,11 +1177,19 @@ class DASH:
m = re.findall(r"([\d.]+.)", d)
return sum(float(x[0:-1]) * {"H": 60 * 60, "M": 60, "S": 1}[x[-1].upper()] for x in m)
@staticmethod
@lru_cache(maxsize=None)
def _field_format_pattern(field: str) -> re.Pattern:
# matches printf-style `$Field%fmt$` format tokens
return re.compile(rf"\${re.escape(field)}%([a-z0-9]+)\$", flags=re.I)
@staticmethod
def replace_fields(url: str, **kwargs: Any) -> str:
if "$" not in url:
return url
for field, value in kwargs.items():
url = url.replace(f"${field}$", str(value))
m = re.search(rf"\${re.escape(field)}%([a-z0-9]+)\$", url, flags=re.I)
m = DASH._field_format_pattern(field).search(url)
if m:
url = url.replace(m.group(), f"{value:{m.group(1)}}")
return url
+1 -1
View File
@@ -1110,7 +1110,7 @@ class HLS:
with open(save_path, "wb") as output_file:
for segment in segments:
with open(segment, "rb") as segment_file:
output_file.write(segment_file.read())
shutil.copyfileobj(segment_file, output_file, 1024 * 1024)
cleanup_segments_and_dirs()
return save_path.stat().st_size