/
/
/
1"""
2Hermetic Music Assistant server for performance benchmarking.
3
4Boots a real MusicAssistant instance that is fully isolated from the host LAN,
5sized via the fake `test` music provider and `_demo_player_provider` players.
6
7Safety properties (do NOT weaken these - see scripts/perf/README.md):
8
9- webserver and streamserver bind/publish on 127.0.0.1 only
10- zeroconf/SSDP discovery is mocked: nothing is advertised on or discovered from the LAN
11- the sendspin builtin provider is stripped BEFORE builtin load: it runs its own
12 aiosendspin server with its own mDNS advertisement (bypassing the mocked discovery
13 controller), so real devices on the LAN would otherwise connect to the test instance
14- the default device providers are suppressed and only an allowlist of local-only
15 builtin providers is loaded, so no host hardware is bridged into the player registry
16 and no network metadata providers (musicbrainz, fanarttv, ...) are contacted
17
18Exposes a loopback control channel (aiohttp, --control-port) used by run_benchmark.py:
19
20 GET /ping -> liveness/readiness probe
21 GET /sync_active -> whether any provider sync task is pending/running
22 POST /yappi/start -> start CPU profiling (cpu clock)
23 POST /yappi/stop -> stop profiling, return top rows (by tsub), clear stats
24 POST /lag/reset -> reset the event-loop lag window
25 GET /lag -> max event-loop lag (ms) since last reset
26 POST /resolve_stream_url -> resolve the stream URL for a player's current media
27 POST /gc -> run a full garbage collection (for stable RSS checkpoints)
28
29Set PERF_YAPPI_BOOT=1 to start yappi before the music_assistant import, so the first
30/yappi/stop returns import+boot attribution.
31
32An admin user + long-lived token is created on first boot and written to
33<data-dir>/token.txt for the websocket bench client.
34
35Run with the repo venv python:
36 .venv/bin/python scripts/perf/perf_server.py --data-dir /tmp/xxx --port 8098 --control-port 8099
37"""
38
39from __future__ import annotations
40
41import argparse
42import asyncio
43import gc
44import logging
45import os
46import re
47import signal
48import sys
49import time
50from concurrent.futures import ThreadPoolExecutor
51from pathlib import Path
52from typing import TYPE_CHECKING, Any
53from unittest.mock import AsyncMock, MagicMock, NonCallableMagicMock, patch
54
55if TYPE_CHECKING:
56 from music_assistant.mass import MusicAssistant
57
58# ruff: noqa: T201
59
60try:
61 import psutil
62 import yappi
63 from aiohttp import web
64except ImportError as err:
65 print(
66 f"Missing benchmark dependency: {err}.\n"
67 "Install the dev/test extras first: uv pip install -e '.[test]'",
68 file=sys.stderr,
69 )
70 sys.exit(1)
71
72LOGGER = logging.getLogger("perf_server")
73
74REPO_ROOT = Path(__file__).resolve().parents[2]
75
76# Local-only builtin providers that are allowed to load. Everything else with
77# builtin=true in its manifest (sendspin, musicbrainz, fanarttv, ...)
78# is stripped before builtin load: hermetic-by-default, so a future builtin can
79# never silently punch a hole in the isolation of the benchmark instance.
80ALLOWED_BUILTIN_PROVIDERS = {
81 "builtin",
82 "loudness_analysis",
83 "playlist_metadata",
84 "radio_playlist",
85 "sync_group",
86 "universal_player",
87}
88
89YAPPI_TOP_LIMIT = 15
90
91
92class LoopLagMonitor:
93 """Track the maximum event-loop scheduling latency since the last reset."""
94
95 def __init__(self) -> None:
96 """Initialize the monitor."""
97 self.max_lag: float = 0.0
98
99 def reset(self) -> None:
100 """Reset the tracked maximum lag."""
101 self.max_lag = 0.0
102
103 async def run(self) -> None:
104 """Continuously sample the event-loop scheduling latency."""
105 loop = asyncio.get_running_loop()
106 while True:
107 t_start = loop.time()
108 await asyncio.sleep(0.1)
109 lag = loop.time() - t_start - 0.1
110 self.max_lag = max(self.max_lag, lag)
111
112
113class ControlApi:
114 """Loopback HTTP control channel for the benchmark orchestrator."""
115
116 def __init__(self, mass: MusicAssistant, lag_monitor: LoopLagMonitor) -> None:
117 """
118 Initialize the control API.
119
120 :param mass: The running MusicAssistant instance.
121 :param lag_monitor: The event-loop lag monitor to expose.
122 """
123 self.mass = mass
124 self.lag_monitor = lag_monitor
125 self._runner: web.AppRunner | None = None
126
127 async def start(self, port: int) -> None:
128 """Start the control HTTP server on the loopback interface."""
129 app = web.Application()
130 app.router.add_get("/ping", self._handle_ping)
131 app.router.add_get("/sync_active", self._handle_sync_active)
132 app.router.add_post("/yappi/start", self._handle_yappi_start)
133 app.router.add_post("/yappi/stop", self._handle_yappi_stop)
134 app.router.add_post("/lag/reset", self._handle_lag_reset)
135 app.router.add_get("/lag", self._handle_lag)
136 app.router.add_post("/resolve_stream_url", self._handle_resolve_stream_url)
137 app.router.add_post("/gc", self._handle_gc)
138 self._runner = web.AppRunner(app)
139 await self._runner.setup()
140 site = web.TCPSite(self._runner, "127.0.0.1", port)
141 await site.start()
142
143 async def stop(self) -> None:
144 """Stop the control HTTP server."""
145 if self._runner:
146 await self._runner.cleanup()
147
148 async def _handle_ping(self, request: web.Request) -> web.Response:
149 return web.json_response({"status": "ok"})
150
151 async def _handle_sync_active(self, request: web.Request) -> web.Response:
152 return web.json_response({"active": bool(self.mass.music.active_sync_tasks)})
153
154 async def _handle_yappi_start(self, request: web.Request) -> web.Response:
155 yappi.set_clock_type("cpu")
156 yappi.start(builtins=False)
157 return web.json_response({"status": "started"})
158
159 async def _handle_yappi_stop(self, request: web.Request) -> web.Response:
160 yappi.stop()
161 rows = _yappi_top_rows()
162 yappi.clear_stats()
163 return web.json_response({"rows": rows})
164
165 async def _handle_lag_reset(self, request: web.Request) -> web.Response:
166 self.lag_monitor.reset()
167 return web.json_response({"status": "ok"})
168
169 async def _handle_lag(self, request: web.Request) -> web.Response:
170 return web.json_response({"max_lag_ms": round(self.lag_monitor.max_lag * 1000, 2)})
171
172 async def _handle_gc(self, request: web.Request) -> web.Response:
173 collected = gc.collect()
174 return web.json_response({"collected": collected})
175
176 async def _handle_resolve_stream_url(self, request: web.Request) -> web.Response:
177 body = await request.json()
178 player = self.mass.players.get_player(body["player_id"])
179 if player is None or player.current_media is None:
180 return web.json_response({"error": "player not found or no current media"}, status=404)
181 url = await self.mass.streams.resolve_stream_url(player.player_id, player.current_media)
182 return web.json_response({"url": url})
183
184
185def _yappi_top_rows(limit: int = YAPPI_TOP_LIMIT) -> list[dict[str, Any]]:
186 """Return the top yappi function stats (by self-time) as structured records."""
187 stats = yappi.get_func_stats()
188 stats.sort("tsub", "desc")
189 rows: list[dict[str, Any]] = []
190 for stat in stats:
191 # skip the harness' own frames - they are measurement scaffolding
192 if "scripts/perf/" in stat.module.replace(os.sep, "/"):
193 continue
194 rows.append(
195 {
196 "name": f"{_shorten_module(stat.module)}:{stat.lineno} {stat.name}",
197 "ncall": stat.ncall,
198 "tsub_ms": round(stat.tsub * 1000, 1),
199 "ttot_ms": round(stat.ttot * 1000, 1),
200 }
201 )
202 if len(rows) >= limit:
203 break
204 return rows
205
206
207def _shorten_module(module: str) -> str:
208 """Strip machine-specific path prefixes so profile rows compare across machines."""
209 normalized = module.replace(os.sep, "/")
210 if "/site-packages/" in normalized:
211 return normalized.split("/site-packages/")[-1]
212 if match := re.search(r"/lib/python\d+\.\d+/(.*)", normalized):
213 return match.group(1)
214 try:
215 return Path(module).resolve().relative_to(REPO_ROOT).as_posix()
216 except ValueError, OSError:
217 return normalized
218
219
220def _create_mock_zeroconf() -> MagicMock:
221 """Create a mock AsyncZeroconf that prevents real mDNS network I/O."""
222 from zeroconf.asyncio import AsyncZeroconf # noqa: PLC0415
223
224 mock_zc = MagicMock(spec=AsyncZeroconf)
225 mock_inner_zc = NonCallableMagicMock()
226 mock_inner_zc.cache = NonCallableMagicMock()
227 mock_inner_zc.cache.cache = {} # empty cache - no discovered services
228 mock_zc.zeroconf = mock_inner_zc
229 mock_zc.async_register_service = AsyncMock()
230 mock_zc.async_update_service = AsyncMock()
231 mock_zc.async_unregister_service = AsyncMock()
232 mock_zc.async_close = AsyncMock()
233 return mock_zc
234
235
236async def run(args: argparse.Namespace) -> None:
237 """Boot the hermetic server and serve until SIGINT/SIGTERM."""
238 # imported late so PERF_YAPPI_BOOT profiling can start before this import
239 from music_assistant_models.auth import UserRole # noqa: PLC0415
240
241 from music_assistant.controllers.config.controller import ConfigController # noqa: PLC0415
242 from music_assistant.mass import MusicAssistant # noqa: PLC0415
243
244 data_dir = str(Path(args.data_dir).resolve())
245 cache_dir = os.path.join(data_dir, ".cache")
246 Path(data_dir).mkdir(parents=True, exist_ok=True)
247 Path(cache_dir).mkdir(parents=True, exist_ok=True)
248
249 # force local-only webserver/streamserver binding before anything reads the config
250 orig_cfg_setup = ConfigController.setup
251
252 async def patched_cfg_setup(self: ConfigController) -> None:
253 await orig_cfg_setup(self)
254 # dedicated stream port per instance: the default (8097) is shared with any
255 # other MA instance on this machine, so streams could silently end up served
256 # by the wrong process
257 for core_module, key, value in (
258 ("webserver", "bind_ip", "127.0.0.1"),
259 ("webserver", "bind_port", args.port),
260 ("webserver", "base_url", f"http://127.0.0.1:{args.port}"),
261 ("streams", "bind_ip", "127.0.0.1"),
262 ("streams", "publish_ip", "127.0.0.1"),
263 ("streams", "bind_port", args.stream_port),
264 ):
265 self.set_raw_core_config_value(core_module, key, value)
266
267 ConfigController.setup = patched_cfg_setup # type: ignore[method-assign]
268
269 # hermetic: strip every builtin provider manifest not on the allowlist before
270 # builtin load. Most importantly sendspin: it runs its own aiosendspin server with
271 # its own mDNS advertisement (bypassing the mocked discovery controller) and real
272 # devices on the LAN WILL discover and connect to the test instance otherwise.
273 orig_load_builtin = MusicAssistant._load_builtin_providers
274
275 async def patched_load_builtin(self: MusicAssistant) -> None:
276 for domain, manifest in list(self._provider_manifests.items()):
277 if manifest.builtin and domain not in ALLOWED_BUILTIN_PROVIDERS:
278 self._provider_manifests.pop(domain)
279 await orig_load_builtin(self)
280
281 MusicAssistant._load_builtin_providers = patched_load_builtin # type: ignore[method-assign]
282
283 patches = [
284 patch(
285 "music_assistant.controllers.discovery.controller.AsyncZeroconf",
286 return_value=_create_mock_zeroconf(),
287 ),
288 patch(
289 "music_assistant.controllers.discovery.controller.AsyncServiceBrowser",
290 return_value=NonCallableMagicMock(),
291 ),
292 # hermetic: no real SSDP search
293 patch(
294 "music_assistant.controllers.discovery.controller.async_upnp_search",
295 new=AsyncMock(),
296 ),
297 # hermetic: no auto-loaded device providers (dlna/sonos/...)
298 patch("music_assistant.mass.DEFAULT_PROVIDERS", set()),
299 ]
300 for patcher in patches:
301 patcher.start()
302
303 loop = asyncio.get_running_loop()
304 loop.set_default_executor(ThreadPoolExecutor(max_workers=32))
305
306 mass = MusicAssistant(data_dir, cache_dir)
307 # dev mode allows loading the `_`-prefixed demo player provider
308 mass.dev_mode = True
309
310 stop_event = asyncio.Event()
311 for sig in (signal.SIGINT, signal.SIGTERM):
312 loop.add_signal_handler(sig, stop_event.set)
313
314 t_start = time.monotonic()
315 await mass.start()
316 startup_secs = time.monotonic() - t_start
317 # re-apply production-like log level (MA config may have set VERBOSE in dev mode)
318 logging.getLogger("music_assistant").setLevel(getattr(logging, args.log_level.upper()))
319
320 # admin user + long-lived token for the bench client
321 token_file = Path(data_dir) / "token.txt"
322 if not token_file.is_file():
323 user = await mass.webserver.auth.create_user("perfadmin", role=UserRole.ADMIN)
324 token = await mass.webserver.auth.create_token(user, "perfbench", is_long_lived=True)
325 await asyncio.to_thread(token_file.write_text, token)
326
327 lag_monitor = LoopLagMonitor()
328 mass.create_task(lag_monitor.run())
329 control_api = ControlApi(mass, lag_monitor)
330 await control_api.start(args.control_port)
331
332 print(
333 f"PERF_SERVER_READY pid={os.getpid()} port={args.port} "
334 f"control_port={args.control_port} stream_port={args.stream_port} "
335 f"startup_secs={startup_secs:.2f} "
336 f"rss_mb={psutil.Process().memory_info().rss / 1024 / 1024:.1f}",
337 flush=True,
338 )
339
340 await stop_event.wait()
341 LOGGER.info("shutting down")
342 await control_api.stop()
343 await mass.stop()
344 for patcher in patches:
345 patcher.stop()
346
347
348def main() -> None:
349 """Parse arguments and run the hermetic server."""
350 parser = argparse.ArgumentParser(description=__doc__)
351 parser.add_argument("--data-dir", required=True)
352 parser.add_argument("--port", type=int, default=8098)
353 parser.add_argument("--control-port", type=int, default=8099)
354 parser.add_argument("--stream-port", type=int, default=8100)
355 parser.add_argument("--log-level", default="warning")
356 args = parser.parse_args()
357
358 logging.basicConfig(
359 level=getattr(logging, args.log_level.upper()),
360 format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
361 )
362 logging.getLogger("aiosqlite").setLevel(logging.WARNING)
363 logging.getLogger("zeroconf").setLevel(logging.WARNING)
364
365 if os.environ.get("PERF_YAPPI_BOOT") == "1":
366 yappi.set_clock_type("cpu")
367 yappi.start(builtins=False)
368
369 asyncio.run(run(args))
370
371
372if __name__ == "__main__":
373 main()
374