diff --git a/docs/REPO_OVERVIEW.md b/docs/REPO_OVERVIEW.md index 2984260..c4c0591 100644 --- a/docs/REPO_OVERVIEW.md +++ b/docs/REPO_OVERVIEW.md @@ -123,3 +123,7 @@ reported at once. and size, so **a change to how the scanner derives a title, artist or series is invisible until the cache is invalidated** — bump `_INDEX_VERSION` in `musicmouse/library/cache.py` when you touch that logic. +- Track analysis (librosa) runs in a pool of worker *processes* - see + `musicmouse/library/workers.py`. Anything an `Analyzer` returns therefore has to be + picklable, and an analyzer that records state in its own instance (a test double + counting calls) only behaves as written with `analysis_workers=1`. diff --git a/python-backend/config.yml.example b/python-backend/config.yml.example index 486a32b..695014e 100644 --- a/python-backend/config.yml.example +++ b/python-backend/config.yml.example @@ -23,6 +23,12 @@ general: # is rebuilt on the next start. Deleting it does throw away track analysis, which # is expensive to recompute. cache: .musicmouse-cache + # How many tracks the background analyzer may work on at once, each in its own + # worker process. Omitted means one per core bar one (capped at 8), which is what + # turns a first-time pass over a whole library from an overnight job into a coffee + # break on a desktop. Set it to 1 on a machine that has better things to do, or to + # a specific number to cap how much of it analysis may take. + # analysis_workers: 4 # Serial port the ESP32 firmware is on. A dropped link is retried, not fatal. # Required - use "simulate" to run without the mouse attached, which is a complete diff --git a/python-backend/musicmouse/__main__.py b/python-backend/musicmouse/__main__.py index e227cfa..91a0965 100644 --- a/python-backend/musicmouse/__main__.py +++ b/python-backend/musicmouse/__main__.py @@ -32,7 +32,7 @@ from musicmouse.devices.mouse import MusicMouseDevice from musicmouse.devices.null_transport import NullTransport from musicmouse.devices.player import Player, VlcPlayer from musicmouse.devices.serial_link import SerialLink -from musicmouse.library import MusicLibrary +from musicmouse.library import MusicLibrary, default_worker_count from musicmouse.library.analysis import build_analyzer from musicmouse.reactions import register_all from musicmouse.services.base import Service @@ -283,11 +283,16 @@ def _build_player(bus: EventBus, general: GeneralConfig, *, clock: RealClock) -> async def build_library(config: Config) -> MusicLibrary: library_config = config.general.library + workers = library_config.analysis_workers return await MusicLibrary.build( library_config.root, library_config.cache, frozenset(config.general.audio_extensions), analyzer=build_analyzer(), + # Unset in the config means "use the machine": a first pass over an unanalyzed + # library is hours of DSP, and there is no reason for a desktop to do it one + # core at a time. `MusicLibrary` itself defaults to 1 - see its docstring. + analysis_workers=default_worker_count() if workers is None else workers, figure_kinds=config.figure_kinds, ) diff --git a/python-backend/musicmouse/config.py b/python-backend/musicmouse/config.py index e79fd53..c9315f2 100644 --- a/python-backend/musicmouse/config.py +++ b/python-backend/musicmouse/config.py @@ -172,6 +172,11 @@ class LibraryConfig(_Strict): root: Path #: Scan results, extracted cover art and track analysis. Relative to this file. cache: Path = Path(".musicmouse-cache") + #: How many tracks background analysis may work on at once, each in its own worker + #: process. Omit for one per core bar one (see + #: :func:`musicmouse.library.workers.default_worker_count`); set it to 1 to keep + #: analysis to a single process on a machine that has other work to do. + analysis_workers: int | None = Field(default=None, ge=1) @field_validator("root") @classmethod diff --git a/python-backend/musicmouse/library/__init__.py b/python-backend/musicmouse/library/__init__.py index 606da34..80b38cd 100644 --- a/python-backend/musicmouse/library/__init__.py +++ b/python-backend/musicmouse/library/__init__.py @@ -8,14 +8,14 @@ plain immutable data. from __future__ import annotations import asyncio -import contextlib import logging -import os import time +from collections import deque from collections.abc import Awaitable, Callable, Collection, Mapping +from concurrent.futures import BrokenExecutor, Executor from dataclasses import replace from pathlib import Path -from typing import Final +from typing import Any, Final from musicmouse.library.analysis import ( ANALYZER_VERSION, @@ -29,6 +29,7 @@ from musicmouse.library.cache import Fingerprint, LibraryCache from musicmouse.library.models import Album, LibraryTrack, album_id, track_key from musicmouse.library.scanner import scan_library from musicmouse.library.sections import SECTIONS, AlbumKind +from musicmouse.library.workers import analysis_pool, analyze_one, default_worker_count from musicmouse.media import Playlist _log = logging.getLogger(__name__) @@ -45,6 +46,7 @@ __all__ = [ "NullAnalyzer", "TrackCurves", "album_id", + "default_worker_count", "track_key", ] @@ -69,19 +71,12 @@ _PUBLISH_BATCH_SIZE: Final = 25 _PROGRESS_INTERVAL_SECONDS: Final = 5.0 -def _analyze_one( - analyzer: Analyzer, path: Path -) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]: - """Runs in a worker thread. Lowers this thread's own scheduling priority first. - - On Linux, ``os.nice`` affects only the calling thread, not the whole process - so - this makes idle-time analysis yield CPU to anything else without touching threads - used for other work. Niceness only ever increases and clamps at the OS maximum - (19), so calling this repeatedly on a reused pool thread is harmless. - """ - with contextlib.suppress(OSError): - os.nice(1) - return analyzer.analyze(path) +def _discard(future: asyncio.Future[Any]) -> None: + """Drop a result nobody is going to read, without leaving a warning behind.""" + if not future.done(): + future.cancel() + elif not future.cancelled(): + future.exception() class MusicLibrary: @@ -94,12 +89,19 @@ class MusicLibrary: extensions: frozenset[str], *, analyzer: Analyzer | None = None, + analysis_workers: int = 1, figure_kinds: Mapping[str, AlbumKind] | None = None, ) -> None: self.root = root self.cache = cache self.extensions = extensions self.analyzer: Analyzer = analyzer or NullAnalyzer() + #: How many tracks background analysis may work on at once. The default of 1 + #: keeps the analyzer in this process, where an analyzer that holds state + #: still behaves as written; the app passes `default_worker_count()`, which is + #: what makes a first-time pass finish in hours rather than days on a desktop. + #: See `musicmouse.library.workers`. + self.analysis_workers = analysis_workers #: What each figure holds. The only thing a folder name cannot say. self.figure_kinds: Mapping[str, AlbumKind] = figure_kinds or {} self._entries: dict[str, tuple[Album, Fingerprint]] = {} @@ -146,9 +148,7 @@ class MusicLibrary: ] if not candidates: return None - return max( - candidates, key=lambda album: album.tracks[0].path.name if album.tracks else "" - ) + return max(candidates, key=lambda album: album.tracks[0].path.name if album.tracks else "") def beats(self, identifier: str, index: int) -> BeatGrid | None: album = self.get(identifier) @@ -219,10 +219,18 @@ class MusicLibrary: extensions: frozenset[str], *, analyzer: Analyzer | None = None, + analysis_workers: int = 1, figure_kinds: Mapping[str, AlbumKind] | None = None, ) -> MusicLibrary: cache = LibraryCache(cache_dir) - library = cls(root, cache, extensions, analyzer=analyzer, figure_kinds=figure_kinds) + library = cls( + root, + cache, + extensions, + analyzer=analyzer, + analysis_workers=analysis_workers, + figure_kinds=figure_kinds, + ) library._entries = await asyncio.to_thread(cache.load_index) await library.refresh() return library @@ -261,64 +269,145 @@ class MusicLibrary: is_busy: Callable[[], bool] = lambda: False, on_batch: Callable[[], Awaitable[None]] | None = None, batch_size: int = _PUBLISH_BATCH_SIZE, + workers: int | None = None, ) -> int: """Run the analyzer over tracks of `kinds` that have no current result. - Restricted to music by default - see `_ANALYZED_KINDS`. Checked before every - track, `is_busy()` pauses the whole pass rather than one file: analysis must - never compete with audio decoding for CPU, and a children's player is idle most - of the day, so the pass simply resumes next time it is. A track the analyzer - fails on (corrupt file, DRM, zero length) is still recorded as attempted - with - every scalar left `None` - so it is never retried forever and the frontend falls - back to the un-analyzed baseline for it. Results are folded into the live index - and persisted every `batch_size` tracks, so a long first run is visible in open - browser tabs as it goes rather than only once it finishes. + Restricted to music by default - see `_ANALYZED_KINDS`. Up to `workers` tracks + are analyzed at once (`self.analysis_workers` when not given), each in its own + worker process - see `musicmouse.library.workers` for why processes and how the + machine is kept usable while they run. Checked before every track is handed out, + `is_busy()` pauses the whole pass rather than one file: analysis must never + compete with audio decoding for CPU, and a children's player is idle most of the + day, so the pass simply resumes next time it is. Tracks already in flight when + it goes busy are allowed to finish - their results are already paid for - and + the workers are then shut down for the duration rather than sitting idle with + a librosa apiece resident on a machine that is now playing music. + + A track the analyzer fails on (corrupt file, DRM, zero length) is still recorded + as attempted - with every scalar left `None` - so it is never retried forever and + the frontend falls back to the un-analyzed baseline for it. A worker *process* + dying, though, says nothing about the track it was on, so that ends the pass + without recording anything: the next one picks the same tracks up again. + + Results are folded into the live index and persisted every `batch_size` tracks, + so a long first run is visible in open browser tabs as it goes rather than only + once it finishes. """ analyzer = self.analyzer if analyzer.version < ANALYZER_VERSION: return 0 + pending = deque(self._pending_tracks(kinds, analyzer.version)) + if not pending: + return 0 + workers = self.analysis_workers if workers is None else max(1, workers) + total = len(pending) + _log.info("Analyzing %d tracks with %d worker(s)", total, workers) + + loop = asyncio.get_running_loop() + in_flight: dict[asyncio.Future[Any], tuple[str, Path]] = {} done = 0 last_report = time.monotonic() - for album in self.albums: - if album.kind not in kinds: - continue - for track in album.tracks: + try: + while pending: while is_busy(): + # Waited out *between* pools, so a half-hour album is not played + # with a houseful of idle worker processes holding onto librosa. await asyncio.sleep(_BUSY_POLL_SECONDS) - key = track_key(track.path) - cached = self.cache.load_analysis(key) - if cached is not None and cached.version >= analyzer.version: - continue - now = time.monotonic() - if now - last_report >= _PROGRESS_INTERVAL_SECONDS: - _log.info("Analyzing library: %d tracks done so far, now on %s", done, track.path) - last_report = now - try: - analysis, grid, curve = await asyncio.to_thread( - _analyze_one, analyzer, track.path - ) - except Exception: - _log.warning( - "Analyzer raised on %s; marking it attempted so it is not retried forever", - track.path, - exc_info=True, - ) - analysis, grid, curve = TrackAnalysis(version=analyzer.version), None, None - if grid is not None: - self.cache.store_beats(key, grid) - if curve is not None: - self.cache.store_curve(key, curve) - self.cache.store_analysis(key, analysis) - done += 1 - if done % batch_size == 0: - await self._publish_analysis(kinds, on_batch) + with analysis_pool(workers) as pool: + try: + while in_flight or (pending and not is_busy()): + while pending and len(in_flight) < workers and not is_busy(): + key, path = pending.popleft() + in_flight[self._submit(pool, loop, path)] = (key, path) + if not in_flight: + break # gone busy: drop the pool and wait above + finished, _ = await asyncio.wait( + in_flight, return_when=asyncio.FIRST_COMPLETED + ) + for future in finished: + key, path = in_flight.pop(future) + self._store_result(key, path, future) + done += 1 + if done % batch_size == 0: + await self._publish_analysis(kinds, on_batch) + now = time.monotonic() + if now - last_report >= _PROGRESS_INTERVAL_SECONDS: + _log.info("Analyzing library: %d/%d tracks done", done, total) + last_report = now + finally: + # Nothing is left to read these - without this, a pass that ends + # early (cancelled at shutdown, or a dead pool) leaves "exception + # was never retrieved" behind for every track still in flight. + for future in in_flight: + _discard(future) + in_flight.clear() + except BrokenExecutor: + _log.error( + "An analysis worker process died (out of memory?) after %d of %d tracks; " + "stopping this pass. The rest are retried on the next one.", + done, + total, + ) if done % batch_size: await self._publish_analysis(kinds, on_batch) if done: _log.info("Analyzed %d tracks", done) return done + def _pending_tracks(self, kinds: Collection[AlbumKind], version: int) -> list[tuple[str, Path]]: + """The whole pass's worklist, as (cache key, path), worked out up front. + + Up front rather than per track, because a pool has to have the next file ready + the moment a worker frees up, and because it is what lets the progress line say + "37/412". Deduplicated by cache key: the same file can sit in two albums, and + two workers analyzing it at once would be pure waste. + """ + worklist: list[tuple[str, Path]] = [] + seen: set[str] = set() + for album in self.albums: + if album.kind not in kinds: + continue + for track in album.tracks: + key = track_key(track.path) + if key in seen: + continue + seen.add(key) + cached = self.cache.load_analysis(key) + if cached is not None and cached.version >= version: + continue + worklist.append((key, track.path)) + return worklist + + def _submit( + self, pool: Executor | None, loop: asyncio.AbstractEventLoop, path: Path + ) -> asyncio.Future[Any]: + """Start one track, in a worker process or - with no pool - on a thread here.""" + if pool is None: + return asyncio.ensure_future(asyncio.to_thread(analyze_one, self.analyzer, path)) + return loop.run_in_executor(pool, analyze_one, self.analyzer, path) + + def _store_result(self, key: str, path: Path, future: asyncio.Future[Any]) -> None: + """Persist one finished track. Re-raises only what ends the whole pass.""" + error = future.exception() + if isinstance(error, BrokenExecutor): + raise error + if error is not None: + _log.warning( + "Analyzer raised on %s; marking it attempted so it is not retried forever", + path, + exc_info=error, + ) + analysis, grid, curve = TrackAnalysis(version=self.analyzer.version), None, None + else: + analysis, grid, curve = future.result() + if grid is not None: + self.cache.store_beats(key, grid) + if curve is not None: + self.cache.store_curve(key, curve) + self.cache.store_analysis(key, analysis) + async def _publish_analysis( self, kinds: Collection[AlbumKind], on_batch: Callable[[], Awaitable[None]] | None ) -> None: diff --git a/python-backend/musicmouse/library/workers.py b/python-backend/musicmouse/library/workers.py new file mode 100644 index 0000000..7992cdc --- /dev/null +++ b/python-backend/musicmouse/library/workers.py @@ -0,0 +1,143 @@ +"""Where analysis actually burns CPU, and how it is kept from taking the machine over. + +Analyzing one track is a few seconds of single-threaded DSP that the GIL will not let +another Python thread overlap with - librosa's work is numba-jitted and numpy glue, not +long C calls that release it. So a library-sized pass parallelizes across *processes*: +:func:`analysis_pool` hands :meth:`~musicmouse.library.MusicLibrary.analyze_pending` an +executor whose workers are separate interpreters, and a 12-core desktop chews through a +first-time scan roughly an order of magnitude faster than the Raspberry Pi this also +has to stay polite on. + +Polite means three things, all of them set here rather than at the call site: + +* **One core stays free** (:func:`default_worker_count`), so the audio thread and the + web server never have to fight a full house of analyzers for a timeslice. +* **Workers run niced**, so even the cores they do own yield to playback instantly. +* **Each worker stays single-threaded** - numpy's BLAS and numba would each happily + start one thread per core *inside* every worker, and N x N threads on a 4-core Pi is + slower than N, not faster. + +Whatever an analyzer returns has to survive the trip back from a worker process, which +is what keeps :class:`~musicmouse.library.analysis.TrackAnalysis` and friends plain +frozen dataclasses of floats. An analyzer that records state in its own instance - +a test double counting calls, say - only sees that state in the worker, so such an +analyzer must be run with ``workers=1``, where everything stays in this process on a +thread. +""" + +from __future__ import annotations + +import contextlib +import logging +import multiprocessing +import os +from collections.abc import Iterator +from concurrent.futures import Executor, ProcessPoolExecutor +from pathlib import Path +from typing import Final + +from musicmouse.library.analysis import Analyzer, BeatGrid, TrackAnalysis, TrackCurves + +_log = logging.getLogger(__name__) + +__all__ = ["analysis_pool", "analyze_one", "default_worker_count"] + +#: Ceiling on the automatic worker count. Every worker is a fresh interpreter with its +#: own librosa, numpy and a decoded track in memory - a few hundred MB each - so on a +#: big machine the limit that bites first is RAM, not cores. An explicit +#: ``analysis_workers`` in the config overrides this; the default stays conservative. +_MAX_AUTO_WORKERS: Final = 8 + +#: How much worse than everything else analysis schedules. Niceness clamps at the OS +#: maximum (19) and only ever increases, so re-applying it to a reused worker is +#: harmless. +_NICENESS: Final = 5 + +#: Forced into every worker *before* it imports numpy, which reads these once at import +#: time. Without them each worker opens its own BLAS thread pool sized for the whole +#: machine and the pool oversubscribes every core several times over. +_SINGLE_THREADED: Final = { + "OMP_NUM_THREADS": "1", + "OPENBLAS_NUM_THREADS": "1", + "MKL_NUM_THREADS": "1", + "NUMEXPR_NUM_THREADS": "1", + "NUMBA_NUM_THREADS": "1", +} + + +def default_worker_count() -> int: + """One worker per core bar one, capped at :data:`_MAX_AUTO_WORKERS`. + + The core left over is for the rest of the app: audio decoding, the web server and + the serial link all have to stay responsive while a first-time pass runs for hours. + Single-core machines get 1, which :func:`analysis_pool` turns into the in-process + path rather than a pool of one. + """ + return max(1, min(_MAX_AUTO_WORKERS, (os.cpu_count() or 1) - 1)) + + +def analyze_one( + analyzer: Analyzer, path: Path +) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]: + """Analyze one file off the event loop - in a worker process, or on a thread here. + + Lowers the caller's own scheduling priority first. On Linux ``os.nice`` affects only + the calling *thread*, so on the in-process path this makes idle-time analysis yield + CPU without touching threads doing other work; in a worker process there is nothing + else in the process to slow down anyway. + """ + with contextlib.suppress(OSError): + os.nice(_NICENESS) + return analyzer.analyze(path) + + +def _init_worker() -> None: + """Runs once per worker process, before it imports librosa or numpy. + + A spawned worker starts from a bare interpreter and pulls the analyzer in when it + unpickles its first task, so setting the thread-count variables here still lands + ahead of numpy reading them. + """ + os.environ.update(_SINGLE_THREADED) + with contextlib.suppress(OSError): + os.nice(_NICENESS) + + +@contextlib.contextmanager +def analysis_pool(workers: int) -> Iterator[Executor | None]: + """A pool of `workers` analyzer processes, or ``None`` for "stay in this process". + + ``None`` - for ``workers <= 1``, and as the fallback when a pool cannot be started + at all - means the caller should run each track on a thread instead. That path is + what the tests and single-core devices use, and it is the only one where an + analyzer holding state in its own instance behaves as written. + + Workers are *spawned*, never forked: this process has an asyncio loop, a serial + reader and libVLC's own threads running, and forking that is a well-known way to + inherit a held lock and deadlock in a child. The price is one librosa import per + worker, a few seconds paid once per pass - nothing next to the hours of DSP a + first-time pass over a real library costs. + """ + if workers <= 1: + yield None + return + try: + executor = ProcessPoolExecutor( + max_workers=workers, + mp_context=multiprocessing.get_context("spawn"), + initializer=_init_worker, + ) + except (OSError, ValueError): + _log.warning( + "Could not start %d analysis workers; analyzing in this process instead", + workers, + exc_info=True, + ) + yield None + return + try: + yield executor + finally: + # `wait=False`: shutdown happens on cancellation too (the app is stopping), and + # waiting there would hold it up for however long the tracks in flight take. + executor.shutdown(wait=False, cancel_futures=True) diff --git a/python-backend/tests/test_library.py b/python-backend/tests/test_library.py index 5c30b2b..7334a11 100644 --- a/python-backend/tests/test_library.py +++ b/python-backend/tests/test_library.py @@ -5,14 +5,16 @@ from __future__ import annotations import asyncio import contextlib import json +import os import threading from collections.abc import Callable +from dataclasses import replace from pathlib import Path import pytest from musicmouse.config import DEFAULT_AUDIO_EXTENSIONS, load_config -from musicmouse.library import Album, Analyzer, MusicLibrary +from musicmouse.library import Album, Analyzer, MusicLibrary, default_worker_count from musicmouse.library.analysis import ANALYZER_VERSION, BeatGrid, TrackAnalysis, TrackCurves from musicmouse.library.cache import LibraryCache from musicmouse.library.colors import colors_from_id @@ -52,6 +54,19 @@ class FakeAnalyzer: return analysis, BeatGrid((0.1,), (1.0,)), curves +class PidAnalyzer(FakeAnalyzer): + """Reports *which process* analyzed each track, as `tempo`. + + The point of a pool is that this is never the process running the test - and a + class at module scope is also the only kind of analyzer a worker can be handed, + since it has to survive being pickled over to one. + """ + + def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]: + analysis, grid, curves = super().analyze(path) + return replace(analysis, tempo=float(os.getpid())), grid, curves + + def album_named(library: MusicLibrary, title: str) -> Album: return next(album for album in library.albums if album.title == title) @@ -641,3 +656,73 @@ async def test_requests_raised_during_a_pass_coalesce_into_one_more_pass( assert pass_count == 2 assert len(analyzer.calls) == 5 + + +# ------------------------------------------------------------------ worker processes + + +async def test_a_parallel_pass_analyzes_every_track_in_worker_processes( + config_dir: Path, +) -> None: + """The whole point of `analysis_workers`: a machine with cores gets to use them. + + Two workers rather than the real default, because what is being checked is that + the work left this process at all - not how fast a five-track fixture goes. + """ + library = await build(config_dir, analyzer=PidAnalyzer(fails={"00 - lied.mp3"})) + + done = await library.analyze_pending(workers=2) + + assert done == 5 + analyzed = [ + track.analysis + for album in library.albums + if album.kind == "music" + for track in album.tracks + ] + assert all(a is not None and a.version == ANALYZER_VERSION for a in analyzed) + # The one file the analyzer chokes on is recorded as attempted, exactly as on the + # in-process path - a raise in a worker must not abort the other four. + failed = [a for a in analyzed if a is not None and a.tempo is None] + assert len(failed) == 1 + workers = {a.tempo for a in analyzed if a is not None and a.tempo is not None} + assert workers + assert os.getpid() not in workers + + +async def test_a_parallel_pass_starts_no_workers_while_playback_is_busy( + config_dir: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Being busy holds the pass *before* the pool exists, so a device playing an album + is not also hosting a set of idle analyzer processes.""" + import musicmouse.library as library_module + + monkeypatch.setattr(library_module, "_BUSY_POLL_SECONDS", 0.01) + library = await build(config_dir, analyzer=PidAnalyzer()) + + busy = True + task = asyncio.create_task(library.analyze_pending(workers=2, is_busy=lambda: busy)) + try: + await asyncio.sleep(0.2) + assert list(library.cache.analysis.glob("*.json")) == [] + + busy = False + assert await asyncio.wait_for(task, timeout=30) == 5 + finally: + if not task.done(): + await _cancel(task) + + +async def test_a_parallel_pass_is_cancellable_mid_flight(config_dir: Path) -> None: + """Shutdown cancels this task like any other, and it must not hang waiting for + worker processes to finish the tracks they are on.""" + library = await build(config_dir, analyzer=PidAnalyzer()) + + task = asyncio.create_task(library.analyze_pending(workers=2)) + await asyncio.sleep(0.1) # long enough to have workers starting up + await asyncio.wait_for(_cancel(task), timeout=10) + + +def test_the_default_worker_count_leaves_a_core_for_everything_else() -> None: + count = default_worker_count() + assert 1 <= count <= max(1, (os.cpu_count() or 1) - 1)