perf: benchmark native and Cast startup, seeks, and sustained playback

2026-10-17 01:44:57 +02:00
parent fcad96de62
commit e1284936f9
6 changed files with 572 additions and 2 deletions
@@ -0,0 +1,151 @@
+#!/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):
+    command = [str(binary.resolve())] + (["--playback-benchmark"] if cast else [])
+    command += [server.base_url, source]
+    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, "<loopback>")
+                raise RuntimeError(f"{binary.name}/{source} exited {result}: {detail}")
+        finally:
+            timer.cancel()
+            if process.poll() is None:
+                process.kill()
+                process.wait()
+    if sorted(row["phase"] for row in rows) != ["cached_seek", "cold_start", "sustained", "uncached_seek"]:
+        raise RuntimeError("benchmark must report all four phases exactly once")
+    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])
+    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")
+    fixtures(args.fixtures)
+    print(json.dumps({"environment": platform.platform(), "machine": platform.machine(),
+                      "runs": args.runs, "request_delay_ms": args.request_delay_ms}), flush=True)
+    results = []
+    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()