/
/
/
1"""
2Repeatable performance benchmark suite for the Music Assistant server.
3
4Boots a hermetic server (see perf_server.py for the isolation guarantees), runs a
5fixed scenario suite against it and emits one self-describing JSON document with
6all metrics (units embedded in the key names), designed to be diffed by a tool or
7pasted into an LLM.
8
9Methodology: two passes. Pass 1 runs every scenario unprofiled for accurate
10wall/CPU/RSS metrics; pass 2 boots a fresh server with yappi enabled and repeats
11the CPU-relevant scenarios to capture per-scenario hotspot attribution
12(yappi_top), so profiler overhead never pollutes the timing metrics.
13
14Usage (from the repo root, with the repo venv):
15 .venv/bin/python scripts/perf/run_benchmark.py [--quick] [--out report.json]
16 [--save-baseline] [--compare [baseline.json]] [--markdown] [--keep-data]
17
18Baselines are machine-specific, so they are stored per machine (and per mode) in
19~/.musicassistant-perf/ rather than in the repo: run once with --save-baseline,
20then use a bare --compare to check for regressions against it.
21
22Exit code is non-zero when --compare detects a regression beyond the thresholds
23(see report.py) or when the suite itself fails.
24"""
25
26from __future__ import annotations
27
28import argparse
29import asyncio
30import contextlib
31import importlib.util
32import json
33import os
34import platform
35import re
36import select
37import shutil
38import socket
39import subprocess
40import sys
41import tempfile
42import threading
43import time
44from dataclasses import dataclass
45from datetime import UTC, datetime
46from pathlib import Path
47from typing import IO, Any
48
49# ruff: noqa: T201
50
51try:
52 import aiohttp
53 import psutil
54except ImportError as err:
55 sys.exit(
56 f"Missing benchmark dependency: {err}.\n"
57 "Install the dev/test extras first: uv pip install -e '.[test]'"
58 )
59
60try:
61 from scripts.perf.report import compare_reports, render_markdown
62 from scripts.perf.ws_client import PerfWsClient, summarize_latencies
63except ImportError: # direct file execution: python scripts/perf/run_benchmark.py
64 from report import compare_reports, render_markdown # type: ignore[no-redef]
65 from ws_client import PerfWsClient, summarize_latencies # type: ignore[no-redef]
66
67SCRIPT_DIR = Path(__file__).resolve().parent
68REPO_ROOT = SCRIPT_DIR.parents[1]
69PERF_SERVER = SCRIPT_DIR / "perf_server.py"
70BASELINE_DIR = Path.home() / ".musicassistant-perf"
71
72SCHEMA_VERSION = 1
73SERVER_READY_TIMEOUT = 120
74SYNC_TIMEOUT = 1800
75PLAYBACK_READY_TIMEOUT = 30
76SYNC_POLL_INTERVAL = 0.25
77# control-channel handlers are near-instant; a bounded per-request timeout keeps a
78# stuck server from hanging the suite past its scenario deadlines
79CONTROL_REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=60)
80# 44.1 kHz / 16 bit / stereo WAV is 176400 bytes/s; limiting curl to ~1x realtime
81# makes the stream consumption deterministic and realistic for the whole window
82STREAM_RATE_BYTES_PER_SEC = 176 * 1024
83STREAM_RATE_LIMIT = f"{STREAM_RATE_BYTES_PER_SEC // 1024}k"
84STREAMING_PLAYERS = 2
85MEDIA_ITEM_EVENTS_SETTLE_SECONDS = 2.0
86
87
88@dataclass(frozen=True)
89class SuiteParams:
90 """Sizing parameters for one benchmark mode."""
91
92 quick: bool
93 artists: int
94 albums: int
95 tracks: int
96 podcasts: int
97 audiobooks: int
98 players: int
99 api_iterations: int
100 streaming_seconds: int
101 memory_requests_per_round: int
102 profiled_api_iterations: int = 2
103 profiled_streaming_seconds: int = 10
104
105 @property
106 def total_tracks(self) -> int:
107 """Total number of generated tracks in the test library."""
108 return self.artists * self.albums * self.tracks
109
110
111FULL_PARAMS = SuiteParams(
112 quick=False,
113 artists=50,
114 albums=10,
115 tracks=20,
116 podcasts=10,
117 audiobooks=10,
118 players=4,
119 api_iterations=5,
120 streaming_seconds=30,
121 memory_requests_per_round=60,
122)
123QUICK_PARAMS = SuiteParams(
124 quick=True,
125 artists=25,
126 albums=8,
127 tracks=5,
128 podcasts=2,
129 audiobooks=2,
130 players=4,
131 api_iterations=3,
132 streaming_seconds=12,
133 memory_requests_per_round=20,
134)
135
136
137class BenchmarkError(RuntimeError):
138 """Raised when the benchmark cannot produce valid results."""
139
140
141class ServerHandle:
142 """A running hermetic perf server subprocess."""
143
144 def __init__(self, proc: subprocess.Popen[str], data_dir: Path, log_file: IO[str]) -> None:
145 """Wrap a spawned perf_server process; call wait_ready() before use."""
146 self.proc = proc
147 self.data_dir = data_dir
148 self.ps = psutil.Process(proc.pid)
149 self.port = 0
150 self.control_port = 0
151 self.stream_port = 0
152 self.boot_wall_seconds = 0.0
153 self.boot_cpu_seconds = 0.0
154 self.boot_rss_mb = 0.0
155 self._log_file = log_file
156 self._spawned_at = time.perf_counter()
157
158 def wait_ready(self) -> None:
159 """Block until the server prints its READY line (or fail with the log tail)."""
160 assert self.proc.stdout is not None
161 deadline = time.monotonic() + SERVER_READY_TIMEOUT
162 while time.monotonic() < deadline:
163 if self.proc.poll() is not None:
164 raise BenchmarkError(
165 f"perf server exited during boot (rc={self.proc.returncode}), "
166 f"see {self.data_dir / 'server.log'}"
167 )
168 readable, _, _ = select.select([self.proc.stdout], [], [], 1.0)
169 if not readable:
170 continue
171 line = self.proc.stdout.readline()
172 if not line.startswith("PERF_SERVER_READY"):
173 continue
174 self.boot_wall_seconds = round(time.perf_counter() - self._spawned_at, 2)
175 fields = dict(part.split("=") for part in line.split()[1:])
176 self.port = int(fields["port"])
177 self.control_port = int(fields["control_port"])
178 self.stream_port = int(fields["stream_port"])
179 self.boot_rss_mb = float(fields["rss_mb"])
180 self.boot_cpu_seconds = round(cpu_seconds(self.ps), 2)
181 # keep draining stdout so the pipe can never fill up and block the server
182 threading.Thread(target=self._drain_stdout, daemon=True).start()
183 return
184 raise BenchmarkError(f"perf server not ready within {SERVER_READY_TIMEOUT}s")
185
186 def stop(self) -> None:
187 """Terminate the server subprocess (SIGTERM, then SIGKILL after a grace period)."""
188 if self.proc.poll() is None:
189 self.proc.terminate()
190 try:
191 self.proc.wait(timeout=30)
192 except subprocess.TimeoutExpired:
193 self.proc.kill()
194 self.proc.wait(timeout=10)
195 self._log_file.close()
196
197 def token(self) -> str:
198 """Return the admin API token created by the server."""
199 return (self.data_dir / "token.txt").read_text().strip()
200
201 def library_db_mb(self) -> float:
202 """Return the on-disk size of library.db (incl. WAL) in MB."""
203 total = 0
204 for suffix in ("", "-wal", "-shm"):
205 path = self.data_dir / f"library.db{suffix}"
206 if path.is_file():
207 total += path.stat().st_size
208 return round(total / 1024 / 1024, 1)
209
210 def _drain_stdout(self) -> None:
211 """Forward any further stdout lines to the server log file."""
212 assert self.proc.stdout is not None
213 for line in self.proc.stdout:
214 with contextlib.suppress(ValueError):
215 self._log_file.write(line)
216
217
218class ControlClient:
219 """Client for the perf server's loopback control channel."""
220
221 def __init__(self, session: aiohttp.ClientSession, control_port: int) -> None:
222 """Initialize with a shared aiohttp session and the server's control port."""
223 self._session = session
224 self._base_url = f"http://127.0.0.1:{control_port}"
225
226 async def get(self, path: str) -> dict[str, Any]:
227 """Perform a GET request against the control channel."""
228 async with self._session.get(
229 f"{self._base_url}{path}", timeout=CONTROL_REQUEST_TIMEOUT
230 ) as resp:
231 resp.raise_for_status()
232 return await resp.json()
233
234 async def post(self, path: str, body: dict[str, Any] | None = None) -> dict[str, Any]:
235 """Perform a POST request against the control channel."""
236 async with self._session.post(
237 f"{self._base_url}{path}", json=body, timeout=CONTROL_REQUEST_TIMEOUT
238 ) as resp:
239 resp.raise_for_status()
240 return await resp.json()
241
242 async def yappi_start(self) -> None:
243 """Start CPU profiling in the server process."""
244 await self.post("/yappi/start")
245
246 async def yappi_stop(self) -> list[dict[str, Any]]:
247 """Stop CPU profiling and return the top hotspot rows."""
248 return (await self.post("/yappi/stop"))["rows"]
249
250 async def wait_sync_active(self, timeout: float = 60.0) -> None:
251 """Wait until a provider sync task is pending or running."""
252 deadline = time.monotonic() + timeout
253 while time.monotonic() < deadline:
254 if (await self.get("/sync_active"))["active"]:
255 return
256 await asyncio.sleep(SYNC_POLL_INTERVAL)
257 raise BenchmarkError("sync did not start within timeout")
258
259 async def wait_sync_idle(self, timeout: float = SYNC_TIMEOUT) -> float:
260 """
261 Wait until no provider sync task is pending or running.
262
263 :return: perf_counter timestamp of the first idle observation.
264 """
265 deadline = time.monotonic() + timeout
266 first_idle: float | None = None
267 idle_streak = 0
268 while time.monotonic() < deadline:
269 if (await self.get("/sync_active"))["active"]:
270 first_idle = None
271 idle_streak = 0
272 else:
273 first_idle = first_idle or time.perf_counter()
274 idle_streak += 1
275 # require a few consecutive idle samples so a gap between two queued
276 # sync tasks is not mistaken for completion
277 if idle_streak >= 3:
278 return first_idle
279 await asyncio.sleep(SYNC_POLL_INTERVAL)
280 raise BenchmarkError(f"sync did not complete within {timeout}s")
281
282
283class PeakRssSampler:
284 """Track the peak RSS of a process while active."""
285
286 def __init__(self, ps: psutil.Process) -> None:
287 """Initialize the sampler for the given process."""
288 self._ps = ps
289 self.peak_mb = 0.0
290 self._task: asyncio.Task[None] | None = None
291
292 def start(self) -> None:
293 """Start sampling."""
294 self.peak_mb = self.current_mb()
295 self._task = asyncio.create_task(self._run())
296
297 def stop(self) -> float:
298 """Stop sampling and return the observed peak RSS in MB."""
299 if self._task:
300 self._task.cancel()
301 return round(max(self.peak_mb, self.current_mb()), 1)
302
303 def current_mb(self) -> float:
304 """Return the current RSS in MB."""
305 return self._ps.memory_info().rss / 1024 / 1024
306
307 async def _run(self) -> None:
308 while True:
309 self.peak_mb = max(self.peak_mb, self.current_mb())
310 await asyncio.sleep(0.2)
311
312
313class FfmpegCpuSampler:
314 """Accumulate CPU seconds of all ffmpeg child processes of the server."""
315
316 def __init__(self, ps: psutil.Process) -> None:
317 """Initialize the sampler for the given (server) process."""
318 self._ps = ps
319 self._seen: dict[int, float] = {}
320 self._task: asyncio.Task[None] | None = None
321
322 def start(self) -> None:
323 """Start sampling."""
324 self._task = asyncio.create_task(self._run())
325
326 def stop(self) -> float:
327 """Stop sampling and return the summed ffmpeg CPU seconds."""
328 if self._task:
329 self._task.cancel()
330 self._sample()
331 return round(sum(self._seen.values()), 2)
332
333 def _sample(self) -> None:
334 with contextlib.suppress(psutil.Error):
335 for child in self._ps.children(recursive=True):
336 with contextlib.suppress(psutil.Error):
337 if "ffmpeg" not in child.name():
338 continue
339 times = child.cpu_times()
340 self._seen[child.pid] = max(
341 self._seen.get(child.pid, 0.0), times.user + times.system
342 )
343
344 async def _run(self) -> None:
345 while True:
346 self._sample()
347 await asyncio.sleep(0.5)
348
349
350def cpu_seconds(ps: psutil.Process) -> float:
351 """Return the total (user+system) CPU seconds consumed by the process."""
352 times = ps.cpu_times()
353 return times.user + times.system
354
355
356async def rss_checkpoint_mb(server: ServerHandle, control: ControlClient) -> float:
357 """Return the server RSS in MB after a full garbage collection."""
358 await control.post("/gc")
359 return round(server.ps.memory_info().rss / 1024 / 1024, 1)
360
361
362def start_server(data_dir: Path, *, yappi_boot: bool = False) -> ServerHandle:
363 """
364 Spawn the hermetic perf server and wait until it is ready.
365
366 :param data_dir: The (isolated) data directory for this server instance.
367 :param yappi_boot: Start yappi before the music_assistant import for boot profiling.
368 """
369 port, control_port, stream_port = _free_ports()
370 env = os.environ.copy()
371 if yappi_boot:
372 env["PERF_YAPPI_BOOT"] = "1"
373 log_file = (data_dir / "server.log").open("a")
374 proc = subprocess.Popen( # noqa: S603
375 [
376 sys.executable,
377 str(PERF_SERVER),
378 "--data-dir",
379 str(data_dir),
380 "--port",
381 str(port),
382 "--control-port",
383 str(control_port),
384 "--stream-port",
385 str(stream_port),
386 ],
387 stdout=subprocess.PIPE,
388 stderr=log_file,
389 text=True,
390 cwd=REPO_ROOT,
391 env=env,
392 )
393 handle = ServerHandle(proc, data_dir, log_file)
394 try:
395 handle.wait_ready()
396 except BaseException:
397 handle.stop()
398 raise
399 return handle
400
401
402# --------------------------------------------------------------------------------------
403# scenarios
404# --------------------------------------------------------------------------------------
405
406
407def scenario_import_time() -> dict[str, Any]:
408 """Measure the import cost of music_assistant.mass in a fresh interpreter."""
409 print("scenario: startup / import time ...", file=sys.stderr)
410 code = (
411 "import resource, time\n"
412 "t0 = time.perf_counter(); c0 = time.process_time()\n"
413 "import music_assistant.mass\n"
414 "wall = time.perf_counter() - t0; cpu = time.process_time() - c0\n"
415 "rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss\n"
416 "print(f'IMPORT_RESULT wall={wall:.3f} cpu={cpu:.3f} maxrss={rss}')\n"
417 )
418 result = subprocess.run( # noqa: S603
419 [sys.executable, "-X", "importtime", "-c", code],
420 capture_output=True,
421 text=True,
422 cwd=REPO_ROOT,
423 check=True,
424 timeout=300,
425 )
426 match = re.search(r"IMPORT_RESULT wall=([\d.]+) cpu=([\d.]+) maxrss=(\d+)", result.stdout)
427 if not match:
428 raise BenchmarkError(f"importtime probe produced no result: {result.stdout[-500:]}")
429 # ru_maxrss is bytes on macOS, kilobytes on Linux
430 maxrss = int(match.group(3))
431 maxrss_mb = maxrss / 1024 / 1024 if sys.platform == "darwin" else maxrss / 1024
432 rows: list[tuple[int, int, str]] = []
433 for line in result.stderr.splitlines():
434 if parsed := re.match(r"import time:\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(.+)$", line):
435 rows.append((int(parsed.group(1)), int(parsed.group(2)), parsed.group(3).strip()))
436 rows.sort(key=lambda row: -row[0])
437 return {
438 "import_wall_seconds": float(match.group(1)),
439 "import_cpu_seconds": float(match.group(2)),
440 "import_peak_rss_mb": round(maxrss_mb, 1),
441 "importtime_top": [
442 {
443 "module": module,
444 "self_ms": round(self_us / 1000, 1),
445 "cumulative_ms": round(cum_us / 1000, 1),
446 }
447 for self_us, cum_us, module in rows[:10]
448 ],
449 }
450
451
452async def configure_demo_players(ws: PerfWsClient, params: SuiteParams) -> list[str]:
453 """Configure the demo player provider and wait for its players to register."""
454 await ws.command(
455 "config/providers/save",
456 {
457 "provider_domain": "_demo_player_provider",
458 "values": {"number_of_players": params.players},
459 },
460 )
461 deadline = time.monotonic() + 60
462 while time.monotonic() < deadline:
463 players = (await ws.command("players/all")).result or []
464 demo_ids = sorted(p["player_id"] for p in players if p["player_id"].startswith("demo_"))
465 if len(demo_ids) >= params.players:
466 return demo_ids
467 await asyncio.sleep(0.5)
468 raise BenchmarkError("demo players did not register within timeout")
469
470
471async def scenario_initial_sync(
472 server: ServerHandle, ws: PerfWsClient, control: ControlClient, params: SuiteParams
473) -> dict[str, Any]:
474 """Configure the test music provider and measure its first full library sync."""
475 print(f"scenario: initial_sync ({params.total_tracks} tracks) ...", file=sys.stderr)
476 rss_sampler = PeakRssSampler(server.ps)
477 rss_sampler.start()
478 cpu_start = cpu_seconds(server.ps)
479 wall_start = time.perf_counter()
480 await ws.command(
481 "config/providers/save",
482 {
483 "provider_domain": "test",
484 "values": {
485 "num_artists": params.artists,
486 "num_albums": params.albums,
487 "num_tracks": params.tracks,
488 "num_podcasts": params.podcasts,
489 "num_audiobooks": params.audiobooks,
490 },
491 },
492 )
493 await control.wait_sync_active()
494 wall_end = await control.wait_sync_idle()
495 cpu_end = cpu_seconds(server.ps)
496 peak_rss_mb = rss_sampler.stop()
497 return {
498 "wall_seconds": round(wall_end - wall_start, 2),
499 "cpu_seconds": round(cpu_end - cpu_start, 2),
500 "peak_rss_mb": peak_rss_mb,
501 "library_db_mb": server.library_db_mb(),
502 "library_tracks": params.total_tracks,
503 }
504
505
506async def scenario_noop_resync(
507 server: ServerHandle, ws: PerfWsClient, control: ControlClient
508) -> dict[str, Any]:
509 """Measure a full re-sync of the already-synced library (the no-change case)."""
510 print("scenario: noop_resync ...", file=sys.stderr)
511 cpu_start = cpu_seconds(server.ps)
512 wall_start = time.perf_counter()
513 await ws.command("music/sync")
514 wall_end = await control.wait_sync_idle()
515 cpu_end = cpu_seconds(server.ps)
516 return {
517 "wall_seconds": round(wall_end - wall_start, 2),
518 "cpu_seconds": round(cpu_end - cpu_start, 2),
519 }
520
521
522def _api_suite(params: SuiteParams) -> list[tuple[str, str, dict[str, Any] | None]]:
523 """Return the fixed API command suite as (metric_key, command, args) tuples."""
524 deep_offset = max(params.total_tracks - 500, 0)
525 return [
526 ("artists_page_500", "music/artists/library_items", {"limit": 500}),
527 ("albums_page_500", "music/albums/library_items", {"limit": 500}),
528 ("tracks_page_500", "music/tracks/library_items", {"limit": 500}),
529 (
530 "tracks_page_deep_offset",
531 "music/tracks/library_items",
532 {"limit": 500, "offset": deep_offset},
533 ),
534 ("tracks_search", "music/tracks/library_items", {"search": "Test Track 1", "limit": 100}),
535 (
536 "tracks_ordered_by_name",
537 "music/tracks/library_items",
538 {"order_by": "name", "limit": 200},
539 ),
540 ("players_all", "players/all", None),
541 ("player_queues_all", "player_queues/all", None),
542 ("providers_list", "providers", None),
543 ("browse_root", "music/browse", None),
544 ]
545
546
547async def scenario_api_bench(
548 ws: PerfWsClient, params: SuiteParams, iterations: int
549) -> dict[str, Any]:
550 """Measure latency and payload size for the fixed API command suite."""
551 print(f"scenario: api_bench ({iterations} iterations) ...", file=sys.stderr)
552 results: dict[str, Any] = {}
553 for key, command, args in _api_suite(params):
554 durations_ms: list[float] = []
555 payload_bytes = 0
556 items = 0
557 for _ in range(iterations):
558 result = await ws.command(command, args)
559 durations_ms.append(result.duration_seconds * 1000)
560 payload_bytes = result.payload_bytes
561 items = result.items
562 if "library_items" in command and items == 0:
563 raise BenchmarkError(f"api_bench sanity check failed: {key} returned 0 items")
564 median_ms, p95_ms = summarize_latencies(durations_ms)
565 results[key] = {
566 "median_ms": median_ms,
567 "p95_ms": p95_ms,
568 "payload_kb": round(payload_bytes / 1024, 1),
569 "items": items,
570 }
571 return results
572
573
574async def start_flow_playback(
575 ws: PerfWsClient, control: ControlClient, player_ids: list[str], server: ServerHandle
576) -> list[str]:
577 """
578 Start flow-mode playback on the given players and return their stream URLs.
579
580 Players are switched to flow mode with WAV output so consumption at the WAV
581 byterate equals realtime playback.
582 """
583 for player_id in player_ids:
584 await ws.command(
585 "config/players/save",
586 {"player_id": player_id, "values": {"flow_mode": True, "output_codec": "wav"}},
587 )
588 # allow the config change to be applied to the registered players
589 await asyncio.sleep(1)
590 # a few tracks so the flow stream comfortably outlasts the capture window
591 tracks = (await ws.command("music/tracks/library_items", {"limit": 3})).result
592 track_uris = [track["uri"] for track in tracks]
593 for player_id in player_ids:
594 await ws.command(
595 "player_queues/play_media",
596 {"queue_id": player_id, "media": track_uris},
597 )
598 deadline = time.monotonic() + PLAYBACK_READY_TIMEOUT
599 while time.monotonic() < deadline:
600 players = {p["player_id"]: p for p in (await ws.command("players/all")).result or []}
601 if all(
602 players.get(pid, {}).get("playback_state") == "playing"
603 and players.get(pid, {}).get("current_media")
604 for pid in player_ids
605 ):
606 break
607 await asyncio.sleep(0.5)
608 else:
609 raise BenchmarkError("players did not reach playing state within timeout")
610 urls: list[str] = []
611 for player_id in player_ids:
612 url = (await control.post("/resolve_stream_url", {"player_id": player_id}))["url"]
613 if "/flow/" not in url:
614 raise BenchmarkError(f"expected a flow stream URL for {player_id}, got {url}")
615 if f":{server.stream_port}/" not in url:
616 raise BenchmarkError(
617 f"stream URL {url} does not use this instance's stream port "
618 f"{server.stream_port}; refusing to fetch from a foreign server"
619 )
620 urls.append(url)
621 return urls
622
623
624async def stop_playback(ws: PerfWsClient, player_ids: list[str]) -> None:
625 """Stop playback on the given players."""
626 for player_id in player_ids:
627 await ws.command("players/cmd/stop", {"player_id": player_id})
628
629
630async def _consume_stream(url: str, duration: int) -> None:
631 """Consume a stream URL rate-limited to ~realtime for the given duration."""
632 started = time.perf_counter()
633 proc = await asyncio.create_subprocess_exec(
634 "curl",
635 "-sS",
636 "--fail",
637 "-N",
638 "--limit-rate",
639 STREAM_RATE_LIMIT,
640 "--max-time",
641 str(duration),
642 "-o",
643 "/dev/null",
644 "-w",
645 "%{http_code} %{size_download}",
646 url,
647 stdout=asyncio.subprocess.PIPE,
648 stderr=asyncio.subprocess.PIPE,
649 )
650 try:
651 stdout, stderr = await proc.communicate()
652 finally:
653 # on cancellation (ctrl-c or a sibling stream failing) curl is not
654 # killed automatically and would keep downloading until --max-time
655 if proc.returncode is None:
656 with contextlib.suppress(OSError):
657 proc.kill()
658 with contextlib.suppress(asyncio.CancelledError):
659 await proc.wait()
660 elapsed = time.perf_counter() - started
661 # 28 = --max-time reached (expected), 18 = partial transfer on close
662 if proc.returncode not in (0, 18, 28):
663 raise BenchmarkError(f"curl failed (rc={proc.returncode}): {stderr.decode()[:300]}")
664 if elapsed < duration * 0.8:
665 raise BenchmarkError(
666 f"stream ended prematurely after {elapsed:.1f}s "
667 f"(rc={proc.returncode}, http/bytes={stdout.decode()}): {stderr.decode()[:300]}"
668 )
669 # a connection that stays open but stops sending data would otherwise pass
670 http_code, _, downloaded = stdout.decode().strip().partition(" ")
671 min_bytes = int(STREAM_RATE_BYTES_PER_SEC * duration * 0.5)
672 if http_code != "200" or int(downloaded or 0) < min_bytes:
673 raise BenchmarkError(
674 f"stream stalled (http={http_code}, downloaded {downloaded} bytes, "
675 f"expected >={min_bytes}): {stderr.decode()[:300]}"
676 )
677
678
679async def scenario_streaming(
680 server: ServerHandle,
681 ws: PerfWsClient,
682 control: ControlClient,
683 player_ids: list[str],
684 duration: int,
685) -> dict[str, Any]:
686 """Measure CPU cost and loop lag while serving concurrent flow streams."""
687 print(f"scenario: streaming ({len(player_ids)} flow streams, {duration}s) ...", file=sys.stderr)
688 urls = await start_flow_playback(ws, control, player_ids, server)
689 ffmpeg_sampler = FfmpegCpuSampler(server.ps)
690 await control.post("/lag/reset")
691 cpu_start = cpu_seconds(server.ps)
692 ffmpeg_sampler.start()
693 wall_start = time.perf_counter()
694 # TaskGroup cancels the sibling streams as soon as one fails
695 async with asyncio.TaskGroup() as tg:
696 for url in urls:
697 tg.create_task(_consume_stream(url, duration))
698 wall_seconds = time.perf_counter() - wall_start
699 cpu_end = cpu_seconds(server.ps)
700 ffmpeg_cpu = ffmpeg_sampler.stop()
701 max_lag_ms = (await control.get("/lag"))["max_lag_ms"]
702 await stop_playback(ws, player_ids)
703 return {
704 "streams": len(player_ids),
705 "capture_seconds": round(wall_seconds, 1),
706 "python_cpu_seconds": round(cpu_end - cpu_start, 2),
707 "ffmpeg_cpu_seconds": ffmpeg_cpu,
708 "max_loop_lag_ms": max_lag_ms,
709 }
710
711
712async def scenario_memory(
713 server: ServerHandle,
714 ws: PerfWsClient,
715 control: ControlClient,
716 params: SuiteParams,
717 rss_post_boot_mb: float,
718 rss_post_sync_mb: float,
719) -> dict[str, Any]:
720 """
721 Measure RSS growth under sustained heavy listing load.
722
723 Two identical rounds of heavy listing requests are performed; the second round
724 hitting an already-ratcheted allocator should leave RSS approximately flat -
725 growth in round two indicates a leak.
726 """
727 print(
728 f"scenario: memory (2 rounds x {params.memory_requests_per_round} listings) ...",
729 file=sys.stderr,
730 )
731 listing_commands = [
732 ("music/tracks/library_items", {"limit": 500}),
733 ("music/albums/library_items", {"limit": 500}),
734 ("music/artists/library_items", {"limit": 500}),
735 ]
736
737 async def _round() -> float:
738 for i in range(params.memory_requests_per_round):
739 command, args = listing_commands[i % len(listing_commands)]
740 await ws.command(command, args)
741 await asyncio.sleep(MEDIA_ITEM_EVENTS_SETTLE_SECONDS)
742 return await rss_checkpoint_mb(server, control)
743
744 rss_round1 = await _round()
745 rss_round2 = await _round()
746 return {
747 "rss_post_boot_mb": rss_post_boot_mb,
748 "rss_post_sync_mb": rss_post_sync_mb,
749 "rss_after_round1_mb": rss_round1,
750 "rss_after_round2_mb": rss_round2,
751 "round2_growth_mb": round(rss_round2 - rss_round1, 1),
752 "requests_per_round": params.memory_requests_per_round,
753 }
754
755
756# --------------------------------------------------------------------------------------
757# passes
758# --------------------------------------------------------------------------------------
759
760
761async def run_metrics_pass(data_dir: Path, params: SuiteParams) -> dict[str, Any]:
762 """Run pass 1: all scenarios unprofiled, on a fresh server, for accurate metrics."""
763 scenarios: dict[str, Any] = {}
764 print("pass 1/2: metrics (unprofiled) - booting hermetic server ...", file=sys.stderr)
765 server = start_server(data_dir)
766 try:
767 async with aiohttp.ClientSession() as session:
768 control = ControlClient(session, server.control_port)
769 ws = PerfWsClient(server.port, server.token())
770 await ws.connect()
771 try:
772 startup = {
773 "cold_boot_wall_seconds": server.boot_wall_seconds,
774 "cold_boot_cpu_seconds": server.boot_cpu_seconds,
775 "cold_boot_rss_mb": server.boot_rss_mb,
776 }
777 demo_players = await configure_demo_players(ws, params)
778 scenarios["initial_sync"] = await scenario_initial_sync(server, ws, control, params)
779 rss_post_sync_mb = await rss_checkpoint_mb(server, control)
780 scenarios["noop_resync"] = await scenario_noop_resync(server, ws, control)
781 scenarios["api_bench"] = await scenario_api_bench(ws, params, params.api_iterations)
782 scenarios["streaming"] = await scenario_streaming(
783 server,
784 ws,
785 control,
786 demo_players[:STREAMING_PLAYERS],
787 params.streaming_seconds,
788 )
789 scenarios["memory"] = await scenario_memory(
790 server, ws, control, params, server.boot_rss_mb, rss_post_sync_mb
791 )
792 finally:
793 await ws.close()
794 finally:
795 server.stop()
796
797 # warm boot: restart on the now-populated data dir (the realistic restart case)
798 print("pass 1/2: warm boot on populated library ...", file=sys.stderr)
799 server = start_server(data_dir)
800 try:
801 startup["warm_boot_wall_seconds"] = server.boot_wall_seconds
802 startup["warm_boot_cpu_seconds"] = server.boot_cpu_seconds
803 startup["warm_boot_rss_mb"] = server.boot_rss_mb
804 finally:
805 server.stop()
806
807 scenarios["startup"] = {**scenario_import_time(), **startup}
808 # fixed scenario order in the report
809 order = ["startup", "initial_sync", "noop_resync", "api_bench", "streaming", "memory"]
810 return {name: scenarios[name] for name in order}
811
812
813async def run_profiled_pass(data_dir: Path, params: SuiteParams) -> dict[str, Any]:
814 """Run pass 2: fresh server with yappi, capturing per-scenario hotspot attribution."""
815 yappi_top: dict[str, Any] = {}
816 print("pass 2/2: yappi attribution - booting fresh profiled server ...", file=sys.stderr)
817 server = start_server(data_dir, yappi_boot=True)
818 try:
819 async with aiohttp.ClientSession() as session:
820 control = ControlClient(session, server.control_port)
821 yappi_top["startup_boot"] = await control.yappi_stop()
822 ws = PerfWsClient(server.port, server.token())
823 await ws.connect()
824 try:
825 demo_players = await configure_demo_players(ws, params)
826
827 await control.yappi_start()
828 await scenario_initial_sync(server, ws, control, params)
829 yappi_top["initial_sync"] = await control.yappi_stop()
830
831 await control.yappi_start()
832 await scenario_noop_resync(server, ws, control)
833 yappi_top["noop_resync"] = await control.yappi_stop()
834
835 await control.yappi_start()
836 await scenario_api_bench(ws, params, params.profiled_api_iterations)
837 yappi_top["api_bench"] = await control.yappi_stop()
838
839 player_ids = demo_players[:STREAMING_PLAYERS]
840 urls = await start_flow_playback(ws, control, player_ids, server)
841 await control.yappi_start()
842 async with asyncio.TaskGroup() as tg:
843 for url in urls:
844 tg.create_task(_consume_stream(url, params.profiled_streaming_seconds))
845 yappi_top["streaming"] = await control.yappi_stop()
846 await stop_playback(ws, player_ids)
847 finally:
848 await ws.close()
849 finally:
850 server.stop()
851 return yappi_top
852
853
854# --------------------------------------------------------------------------------------
855# env & main
856# --------------------------------------------------------------------------------------
857
858
859def collect_env() -> dict[str, Any]:
860 """Collect environment/context info for the report header."""
861
862 def _git(*args: str) -> str:
863 return subprocess.run( # noqa: S603
864 ["git", *args], # noqa: S607
865 capture_output=True,
866 text=True,
867 cwd=REPO_ROOT,
868 check=True,
869 ).stdout.strip()
870
871 if sys.platform == "darwin":
872 cpu = subprocess.run(
873 ["sysctl", "-n", "machdep.cpu.brand_string"], # noqa: S607
874 capture_output=True,
875 text=True,
876 check=False,
877 ).stdout.strip()
878 else:
879 cpu = platform.processor() or platform.machine()
880 with contextlib.suppress(OSError):
881 for line in Path("/proc/cpuinfo").read_text().splitlines():
882 if line.lower().startswith("model name"):
883 cpu = line.split(":", 1)[1].strip()
884 break
885 ffmpeg = subprocess.run(
886 ["ffmpeg", "-version"], # noqa: S607
887 capture_output=True,
888 text=True,
889 check=False,
890 ).stdout.splitlines()
891 return {
892 "git_sha": _git("rev-parse", "HEAD"),
893 "dirty": bool(_git("status", "--porcelain", "--untracked-files=no")),
894 "python": platform.python_version(),
895 "platform": platform.platform(),
896 "cpu": cpu,
897 "cpu_count": os.cpu_count(),
898 "ffmpeg": ffmpeg[0].split(" version ")[1].split(" ")[0] if ffmpeg else "unknown",
899 }
900
901
902def _free_ports() -> tuple[int, int, int]:
903 """
904 Pick three distinct currently-free loopback TCP ports.
905
906 The ports are not reserved after this returns; the tiny window until the server
907 binds them is an accepted race â a collision makes the server exit and the boot
908 fail loudly.
909 """
910 with socket.socket() as sock1, socket.socket() as sock2, socket.socket() as sock3:
911 sock1.bind(("127.0.0.1", 0))
912 sock2.bind(("127.0.0.1", 0))
913 sock3.bind(("127.0.0.1", 0))
914 return sock1.getsockname()[1], sock2.getsockname()[1], sock3.getsockname()[1]
915
916
917def _check_prerequisites() -> None:
918 """Fail fast with a clear message when a required tool or package is missing."""
919 if importlib.util.find_spec("yappi") is None:
920 sys.exit(
921 "The benchmark requires yappi (dev/test extra). "
922 "Install with: uv pip install -e '.[test]'"
923 )
924 for tool in ("curl", "ffmpeg", "git"):
925 if shutil.which(tool) is None:
926 sys.exit(f"The benchmark requires `{tool}` on PATH.")
927 if importlib.util.find_spec("music_assistant") is None:
928 sys.exit("music_assistant is not importable; run from the repo venv.")
929
930
931async def run_suite(params: SuiteParams, keep_data: bool) -> dict[str, Any]:
932 """Run the complete two-pass benchmark suite and return the report document."""
933 suite_start = time.perf_counter()
934 tmp_root = Path(tempfile.mkdtemp(prefix="ma-perf-"))
935 try:
936 metrics_dir = tmp_root / "metrics"
937 profiled_dir = tmp_root / "profiled"
938 metrics_dir.mkdir()
939 profiled_dir.mkdir()
940 scenarios = await run_metrics_pass(metrics_dir, params)
941 yappi_top = await run_profiled_pass(profiled_dir, params)
942 finally:
943 if keep_data:
944 print(f"benchmark data kept at {tmp_root}", file=sys.stderr)
945 else:
946 shutil.rmtree(tmp_root, ignore_errors=True)
947 return {
948 "meta": {
949 "schema_version": SCHEMA_VERSION,
950 "quick": params.quick,
951 "generated_at": datetime.now(UTC).isoformat(timespec="seconds"),
952 "suite_runtime_seconds": round(time.perf_counter() - suite_start, 1),
953 },
954 "env": collect_env(),
955 "scenarios": scenarios,
956 "yappi_top": yappi_top,
957 }
958
959
960def main() -> None:
961 """Run the benchmark suite CLI."""
962 parser = argparse.ArgumentParser(description="Music Assistant performance benchmark suite")
963 parser.add_argument("--quick", action="store_true", help="small library, short scenarios")
964 parser.add_argument("--out", type=Path, help="write the JSON report to this file")
965 parser.add_argument(
966 "--save-baseline",
967 action="store_true",
968 help="save this report as the machine-local baseline for future --compare runs",
969 )
970 parser.add_argument(
971 "--compare",
972 nargs="?",
973 const="",
974 metavar="BASELINE",
975 help="compare against a baseline report (default: the machine-local baseline)",
976 )
977 parser.add_argument(
978 "--markdown", action="store_true", help="print a human-readable markdown report"
979 )
980 parser.add_argument(
981 "--keep-data", action="store_true", help="keep the temporary server data dirs"
982 )
983 args = parser.parse_args()
984 _check_prerequisites()
985
986 # resolve and read the baseline upfront: fail before the suite runs, and
987 # read the old baseline before --save-baseline may overwrite it
988 baseline: dict[str, Any] | None = None
989 if args.compare is not None:
990 baseline_path = Path(args.compare) if args.compare else _baseline_path(args.quick)
991 if not baseline_path.is_file():
992 sys.exit(
993 f"no baseline found at {baseline_path} - generate one first with --save-baseline"
994 )
995 baseline = json.loads(baseline_path.read_text())
996 if baseline.get("meta", {}).get("quick") != args.quick:
997 print(
998 "warning: comparing a quick report against a full baseline (or vice versa)",
999 file=sys.stderr,
1000 )
1001
1002 params = QUICK_PARAMS if args.quick else FULL_PARAMS
1003 report = asyncio.run(run_suite(params, args.keep_data))
1004
1005 if args.out:
1006 args.out.write_text(json.dumps(report, indent=2) + "\n")
1007 print(f"report written to {args.out}", file=sys.stderr)
1008 if args.markdown:
1009 print(render_markdown(report))
1010 elif not args.out:
1011 print(json.dumps(report, indent=2))
1012
1013 if args.save_baseline:
1014 path = _baseline_path(args.quick)
1015 path.parent.mkdir(parents=True, exist_ok=True)
1016 path.write_text(json.dumps(report, indent=2) + "\n")
1017 print(f"baseline saved to {path}", file=sys.stderr)
1018
1019 if baseline is not None:
1020 lines, regressions = compare_reports(baseline, report)
1021 print("\n".join(lines))
1022 if regressions:
1023 sys.exit(1)
1024
1025
1026def _baseline_path(quick: bool) -> Path:
1027 """Return the machine-local baseline path for the given suite mode."""
1028 return BASELINE_DIR / ("baseline-quick.json" if quick else "baseline-full.json")
1029
1030
1031if __name__ == "__main__":
1032 main()
1033