yum/3ner

A toon shader for Unity's BIRP.

git clone https://git.yummers.dev/yum/3ner

yumImpostors: begin optimization work50e4514

master
branchesmasterall refs →
11.6 KiB403 linesraw
1#!/usr/bin/env python3
2"""Spawn multiple ffmpeg readers against a single HLS playlist.
3
4Usage:
5    python load_test.py https://example/hls/abc/stream.m3u8 5
6
7The positional arguments control the target playlist URL and how many
8concurrent ffmpeg processes are started. Each client retries on failure with a
9short backoff so it behaves like a persistent video player. A health summary is
10printed periodically showing how many workers are streaming, buffering, or
11backing off. When the script exits (normally or via Ctrl+C) all spawned ffmpeg
12children are terminated.
13"""
14
15from __future__ import annotations
16
17import argparse
18import os
19import random
20import signal
21import subprocess
22import sys
23import tempfile
24import time
25from collections import deque
26from dataclasses import dataclass
27
28
29STOP_REQUESTED = False
30BUFFER_KEYWORDS = (
31    "buffer",
32    "stall",
33    "timeout",
34    "timed out",
35    "drop",
36    "late",
37    "overrun",
38    "input/output error",
39    "connection refused",
40)
41SUMMARY_INTERVAL = float(os.environ.get("LOAD_TEST_SUMMARY_INTERVAL", "5"))
42STABLE_AFTER = float(os.environ.get("LOAD_TEST_STABLE_AFTER", "5"))
43LOG_CHECK_INTERVAL = float(os.environ.get("LOAD_TEST_LOG_CHECK_INTERVAL", "2"))
44MIN_BACKOFF = float(os.environ.get("LOAD_TEST_MIN_BACKOFF", "0.5"))
45MAX_BACKOFF = float(os.environ.get("LOAD_TEST_MAX_BACKOFF", "30"))
46
47
48def parse_args() -> argparse.Namespace:
49    parser = argparse.ArgumentParser(
50        description="Spawn N concurrent ffmpeg readers against an HLS playlist."
51    )
52    parser.add_argument(
53        "playlist_url",
54        help="HLS playlist URL to probe."
55    )
56    parser.add_argument(
57        "count",
58        type=int,
59        help="Number of concurrent ffmpeg processes to launch."
60    )
61    return parser.parse_args()
62
63
64@dataclass
65class Client:
66    index: int
67    command: list[str]
68    attempts: int = 0
69    process: subprocess.Popen[str] | None = None
70    log_path: str | None = None
71    resume_at: float = 0.0
72    last_pid: int | None = None
73    start_time: float = 0.0
74    is_buffering: bool = False
75    last_warning: str | None = None
76    last_error: str | None = None
77    state: str = "idle"
78    last_log_check: float = 0.0
79    consecutive_failures: int = 0
80
81    def spawn(self) -> None:
82        log_handle = tempfile.NamedTemporaryFile(
83            prefix=f"hls_client_{self.index}_",
84            suffix=".log",
85            delete=False,
86            mode="w",
87            encoding="utf-8",
88        )
89
90        try:
91            proc = subprocess.Popen(
92                self.command,
93                stdout=subprocess.DEVNULL,
94                stderr=log_handle,
95                text=False,
96            )
97        except FileNotFoundError:  # pragma: no cover - requires missing ffmpeg
98            sys.exit("ffmpeg binary not found; install ffmpeg before running this script.")
99        finally:
100            log_handle.close()
101
102        self.attempts += 1
103        self.process = proc
104        self.log_path = log_handle.name
105        self.resume_at = 0.0
106        self.last_pid = proc.pid
107        self.start_time = time.time()
108        self.is_buffering = False
109        self.last_warning = None
110        self.state = "starting"
111        self.last_log_check = 0.0
112        print(
113            f"Spawned client {self.index} attempt {self.attempts} (pid={proc.pid})"
114        )
115
116    def poll(self) -> int | None:
117        if self.process is None:
118            return None
119        return self.process.poll()
120
121    def terminate(self) -> None:
122        if self.process and self.process.poll() is None:
123            self.process.terminate()
124
125
126def spawn_clients(target_url: str, count: int) -> list[Client]:
127    command = [
128        "ffmpeg",
129        "-nostdin",
130        "-loglevel",
131        os.environ.get("FFMPEG_LOGLEVEL", "warning"),
132        "-i",
133        target_url,
134        "-f",
135        "null",
136        "-",
137    ]
138
139    clients: list[Client] = []
140    for index in range(1, count + 1):
141        client = Client(index=index, command=command.copy())
142        client.spawn()
143        clients.append(client)
144
145    return clients
146
147
148def terminate_processes(clients: list[Client]) -> None:
149    for client in clients:
150        client.terminate()
151
152    deadline = time.time() + 5
153    for client in clients:
154        proc = client.process
155        if proc and proc.poll() is None:
156            remaining = max(0, deadline - time.time())
157            try:
158                proc.wait(timeout=remaining)
159            except subprocess.TimeoutExpired:
160                proc.kill()
161
162
163def install_signal_handlers(clients: list[Client]) -> None:
164    def handler(signum: int, _frame) -> None:
165        global STOP_REQUESTED
166        if STOP_REQUESTED:
167            return
168        STOP_REQUESTED = True
169        print(f"Received signal {signum}, shutting down clients…")
170        terminate_processes(clients)
171        cleanup_logs(clients)
172        sys.exit(0)
173
174    signal.signal(signal.SIGINT, handler)
175    signal.signal(signal.SIGTERM, handler)
176
177
178def explain_exit_code(return_code: int | None) -> str:
179    if return_code is None:
180        return "process still running"
181    if return_code == 0:
182        return "exited cleanly"
183
184    signal_num: int | None = None
185    if return_code < 0:
186        signal_num = -return_code
187    elif return_code >= 128:
188        signal_num = return_code - 128
189
190    if signal_num:
191        try:
192            sig_name = signal.Signals(signal_num).name
193        except ValueError:
194            sig_name = f"signal {signal_num}"
195        return f"terminated by {sig_name} (exit code {return_code})"
196
197    return f"exited with status {return_code}"
198
199
200def read_log_tail(path: str, max_lines: int = 5) -> str:
201    try:
202        with open(path, "r", encoding="utf-8", errors="replace") as handle:
203            tail_lines = ''.join(deque(handle, maxlen=max_lines)).strip()
204            return tail_lines or "<no log output>"
205    except OSError as exc:
206        return f"<unable to read log: {exc}>"
207
208
209def report_exit(client: Client, return_code: int | None) -> None:
210    log_path = client.log_path
211    description = explain_exit_code(return_code)
212    pid_info = f"pid {client.last_pid}" if client.last_pid is not None else "pid unknown"
213    print(f"Client {client.index} ({pid_info}) {description}.")
214    if log_path:
215        log_tail = read_log_tail(log_path)
216        if log_tail:
217            print("Recent log:")
218            print(log_tail)
219        client.last_error = log_tail if (return_code and log_tail) else None
220        try:
221            os.remove(log_path)
222        except OSError:
223            pass
224    client.log_path = None
225    client.is_buffering = False
226    client.last_warning = None
227
228
229def cleanup_logs(clients: list[Client]) -> None:
230    for client in clients:
231        if client.log_path:
232            report_exit(client, client.process.returncode if client.process else None)
233
234
235def compute_backoff(failures: int) -> float:
236    if failures <= 0:
237        return MIN_BACKOFF
238
239    min_backoff = max(0.0, MIN_BACKOFF)
240    max_backoff = max(min_backoff, MAX_BACKOFF)
241
242    lower = min(max_backoff, min_backoff * (2 ** (failures - 1)))
243    upper = min(max_backoff, min_backoff * (2 ** failures))
244
245    if upper < lower:
246        lower, upper = upper, lower
247
248    if lower == upper:
249        return lower
250
251    return random.uniform(lower, upper)
252
253
254def truncate(text: str, limit: int = 120) -> str:
255    return text if len(text) <= limit else text[: limit - 3] + "..."
256
257
258def detect_buffering(client: Client, now: float) -> None:
259    if not client.log_path:
260        client.is_buffering = False
261        return
262
263    if now - client.last_log_check < LOG_CHECK_INTERVAL:
264        return
265
266    client.last_log_check = now
267    tail = read_log_tail(client.log_path)
268    if tail and tail != "<no log output>":
269        lower_tail = tail.lower()
270        if any(keyword in lower_tail for keyword in BUFFER_KEYWORDS):
271            client.is_buffering = True
272            client.last_warning = tail
273            return
274
275    client.is_buffering = False
276
277
278def print_summary(clients: list[Client], now: float) -> None:
279    counts = {
280        "streaming": 0,
281        "buffering": 0,
282        "starting": 0,
283        "backoff": 0,
284        "idle": 0,
285    }
286
287    buffering_notes = []
288    recent_errors = []
289
290    for client in clients:
291        proc = client.process
292        if proc and proc.poll() is None:
293            uptime = now - client.start_time
294            detect_buffering(client, now)
295            if client.is_buffering:
296                client.state = "buffering"
297            elif uptime >= STABLE_AFTER:
298                client.state = "streaming"
299                if client.consecutive_failures:
300                    client.consecutive_failures = 0
301                    client.last_error = None
302            else:
303                client.state = "starting"
304        else:
305            if STOP_REQUESTED:
306                client.state = "idle"
307            elif client.resume_at > now:
308                client.state = "backoff"
309            else:
310                client.state = "idle"
311
312        counts.setdefault(client.state, 0)
313        counts[client.state] += 1
314
315        if client.is_buffering and client.last_warning:
316            buffering_notes.append(
317                f"#{client.index} {truncate(client.last_warning)}"
318            )
319        if client.last_error and client.state != "streaming":
320            recent_errors.append(
321                f"#{client.index} {truncate(client.last_error)}"
322            )
323
324    restarts = sum(max(0, client.attempts - 1) for client in clients)
325    summary_parts = [
326        f"Streaming:{counts['streaming']}",
327        f"Buffering:{counts['buffering']}",
328        f"Starting:{counts['starting']}",
329        f"Backoff:{counts['backoff']}",
330        f"Idle:{counts['idle']}",
331        f"Restarts:{restarts}",
332    ]
333
334    print("[Summary] " + " | ".join(summary_parts))
335
336    if buffering_notes:
337        print("  Buffering clients:")
338        for note in buffering_notes:
339            print(f"    {note}")
340
341    if recent_errors:
342        print("  Recent errors:")
343        for error in recent_errors[:5]:
344            print(f"    {error}")
345
346
347def monitor(clients: list[Client]) -> None:
348    global STOP_REQUESTED
349    try:
350        next_summary = time.time() + SUMMARY_INTERVAL
351        while True:
352            now = time.time()
353            for client in clients:
354                proc = client.process
355                if proc and proc.poll() is not None:
356                    return_code = proc.returncode
357                    report_exit(client, return_code)
358                    client.process = None
359                    if STOP_REQUESTED:
360                        continue
361
362                    if return_code and return_code != 0:
363                        client.consecutive_failures += 1
364                        backoff = compute_backoff(client.consecutive_failures)
365                    else:
366                        client.consecutive_failures = 0
367                        backoff = MIN_BACKOFF
368                    client.resume_at = now + backoff
369                    print(
370                        f"Scheduling restart for client {client.index} in {backoff:.1f}s…"
371                    )
372                    client.state = "backoff"
373
374                if client.process is None and not STOP_REQUESTED and now >= client.resume_at:
375                    client.spawn()
376
377            if now >= next_summary:
378                print_summary(clients, now)
379                next_summary = now + SUMMARY_INTERVAL
380
381            if STOP_REQUESTED:
382                break
383
384            time.sleep(1)
385    finally:
386        STOP_REQUESTED = True
387        terminate_processes(clients)
388        cleanup_logs(clients)
389
390
391def main() -> None:
392    args = parse_args()
393    if args.count < 1:
394        sys.exit("Count must be at least 1.")
395
396    clients = spawn_clients(args.playlist_url, args.count)
397    install_signal_handlers(clients)
398    print("Press Ctrl+C to stop all clients.")
399    monitor(clients)
400
401
402if __name__ == "__main__":
403    main()