#!/usr/bin/env python3 """Measure real native/Cast source paths; no resolver, WAN, TLS or physical receiver.""" from __future__ import annotations import argparse import json import platform import statistics import subprocess import threading from pathlib import Path from streaming import StreamingServer def fixtures(directory: Path): directory.mkdir(parents=True, exist_ok=True) if (directory / "bench.m3u8").is_file(): return ffmpeg = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y"] combined = directory / "bench.mp4" subprocess.run(ffmpeg + [ "-f", "lavfi", "-i", "testsrc2=size=1280x720:rate=30:duration=120", "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=120", "-c:v", "libx264", "-preset", "ultrafast", "-threads", "2", "-bf", "2", "-g", "60", "-keyint_min", "60", "-sc_threshold", "0", "-pix_fmt", "yuv420p", "-c:a", "aac", "-movflags", "+faststart", str(combined), ], check=True, timeout=180) for track, name in (("0:v:0", "bench-video.mp4"), ("0:a:0", "bench-audio.m4a")): subprocess.run(ffmpeg + ["-i", str(combined), "-map", track, "-c", "copy", "-movflags", "+faststart", str(directory / name)], check=True, timeout=60) subprocess.run(ffmpeg + ["-i", str(combined), "-map", "0", "-c", "copy", "-hls_time", "6", "-hls_playlist_type", "vod", "-hls_segment_filename", str(directory / "bench-%03d.ts"), str(directory / "bench.m3u8")], check=True, timeout=60) class BenchmarkServer(StreamingServer): def __init__(self, directory: Path, delay_ms: float): super().__init__(directory) self.headers = {path.name: "combined" for path in directory.iterdir() if path.is_file()} self.headers.update({"bench-video.mp4": "video", "bench-audio.m4a": "audio"}) self.delay_seconds = delay_ms / 1000 self.connections: set[tuple] = set() def _get(self, request): if request.path.startswith("/media/"): with self.lock: self.connections.add(request.client_address) # Controlled per-request response latency, not a simulated bandwidth or RTT. if self.delay_seconds: self.stopping.wait(self.delay_seconds) super()._get(request) def snapshot(self): with self.lock: return {**{key: sum(row[key] for row in self.stats.values()) for key in ("requests", "ranges", "bytes", "denied")}, "connections": len(self.connections)} def sample(binary: Path, cast: bool, source: str, server: BenchmarkServer, *, seek_work: bool = False): command = [str(binary.resolve())] + (["--playback-benchmark"] if cast else []) command += [server.base_url, source] if seek_work: command.append("--seek-work") rows = [] # A process deadline is infrastructure protection, never a measured playback clock. with subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process: timer = threading.Timer(120, process.kill) timer.start() errors = [] reader = threading.Thread(target=lambda: errors.append(process.stderr.read()), daemon=True) reader.start() before = None phase = None try: for line in process.stdout: row = json.loads(line) after = server.snapshot() if row.get("event") == "begin": if before is not None: raise RuntimeError("nested benchmark phase") before, phase = after, row["phase"] else: if before is None or phase != row["phase"]: raise RuntimeError("missing benchmark phase boundary") row.update({"upstream_" + key: after[key] - before[key] for key in after}) if row["upstream_denied"]: raise RuntimeError("benchmark source headers rejected") rows.append(row) before = None process.stdin.write("\n") process.stdin.flush() result = process.wait() reader.join() if result: # Diagnostics can contain fixture addresses; do not publish addresses/headers. detail = "".join(errors).replace(server.base_url, "") raise RuntimeError(f"{binary.name}/{source} exited {result}: {detail}") finally: timer.cancel() if process.poll() is None: process.kill() process.wait() expected_phases = (["cold_start", "uncached_seek"] if seek_work else ["cached_seek", "cold_start", "sustained", "uncached_seek"]) if sorted(row["phase"] for row in rows) != expected_phases: raise RuntimeError(f"benchmark must report each expected phase exactly once: {expected_phases}") return rows def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("native", type=Path) parser.add_argument("cast", type=Path) parser.add_argument("fixtures", type=Path) parser.add_argument("--runs", type=int, default=3) parser.add_argument("--request-delay-ms", type=float, nargs="+", default=[0, 25]) parser.add_argument("--compare-native", type=Path, metavar="BASELINE_BINARY") parser.add_argument("--seek-work", action="store_true") args = parser.parse_args() if args.runs < 1 or any(value < 0 for value in args.request_delay_ms): parser.error("runs must be positive and delays nonnegative") if args.seek_work and not args.compare_native: parser.error("--seek-work requires --compare-native") fixtures(args.fixtures) print(json.dumps({"environment": platform.platform(), "machine": platform.machine(), "runs": args.runs, "request_delay_ms": args.request_delay_ms, **({"baseline": str(args.compare_native), "candidate": str(args.native), "seek_work": args.seek_work} if args.compare_native else {})}), flush=True) results = [] if args.compare_native: for delay in args.request_delay_ms: for run in range(1, args.runs + 1): variants = [("baseline", args.compare_native), ("candidate", args.native)] if run % 2 == 0: variants.reverse() for source in ("split", "hls"): for variant, binary in variants: with BenchmarkServer(args.fixtures, delay) as server: for row in sample(binary, False, source, server, seek_work=args.seek_work): row.update(engine="native", variant=variant, source=source, request_delay_ms=delay, run=run) results.append(row) print(json.dumps(row, sort_keys=True), flush=True) print("\nMedians [min, max]; CPU is process CPU; peak RSS is process-lifetime high water; " "upstream counts include concurrent read-ahead.") print("| Delay ms | Variant | Source | Phase | Wall ms | CPU ms | Peak MiB | Requests | Upstream bytes |") print("| --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: |") for delay in args.request_delay_ms: for variant in ("baseline", "candidate"): for source in ("split", "hls"): phases = dict.fromkeys(row["phase"] for row in results) for phase in phases: group = [row for row in results if (row["request_delay_ms"], row["variant"], row["source"], row["phase"]) == (delay, variant, source, phase)] def summary(key): values = [row[key] for row in group] return f"{statistics.median(values):.2f} [{min(values):.2f}, {max(values):.2f}]" print(f"| {delay:g} | {variant} | {source} | {phase} | " f"{summary('wall_ms')} | {summary('cpu_ms')} | {summary('peak_rss_mib')} | " f"{summary('upstream_requests')} | {summary('upstream_bytes')} |") return for delay in args.request_delay_ms: for run in range(1, args.runs + 1): for source in ("split", "hls"): for engine, binary in (("native", args.native), ("cast", args.cast)): with BenchmarkServer(args.fixtures, delay) as server: for row in sample(binary, engine == "cast", source, server): row.update(engine=engine, source=source, request_delay_ms=delay, run=run) results.append(row) print(json.dumps(row, sort_keys=True), flush=True) print("\nMedians; upstream counts include concurrent read-ahead; peak RSS is process-lifetime high water.") print("| Delay ms | Engine | Source | Phase | Wall ms | CPU ms | Peak MiB | Requests | Connections | Upstream MiB | Media s |") print("| --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |") for delay in args.request_delay_ms: for engine in ("native", "cast"): for source in ("split", "hls"): for phase in ("cold_start", "cached_seek", "uncached_seek", "sustained"): group = [row for row in results if (row["request_delay_ms"], row["engine"], row["source"], row["phase"]) == (delay, engine, source, phase)] def median(key): return statistics.median(row[key] for row in group) print(f"| {delay:g} | {engine} | {source} | {phase} | {median('wall_ms'):.2f} | " f"{median('cpu_ms'):.2f} | {median('peak_rss_mib'):.1f} | " f"{median('upstream_requests'):g} | {median('upstream_connections'):g} | " f"{median('upstream_bytes') / 1048576:.2f} | {median('media_seconds'):.2f} |") if __name__ == "__main__": main()