Analyse tracks in a pool of worker processes
A first pass over an unanalysed library is hours of librosa, and there was no reason for a desktop to spend them one core at a time. Analysis now runs in a process pool sized by `general.library.analysis_workers`, defaulting to one per core bar one (capped at 8) when the key is absent. Two consequences worth knowing before touching this: whatever an `Analyzer` returns has to be picklable, and an analyzer that keeps state on its own instance - a test double counting calls - only behaves as written with `analysis_workers=1`. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -123,3 +123,7 @@ reported at once.
|
|||||||
and size, so **a change to how the scanner derives a title, artist or series is
|
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
|
invisible until the cache is invalidated** — bump `_INDEX_VERSION` in
|
||||||
`musicmouse/library/cache.py` when you touch that logic.
|
`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`.
|
||||||
|
|||||||
@@ -23,6 +23,12 @@ general:
|
|||||||
# is rebuilt on the next start. Deleting it does throw away track analysis, which
|
# is rebuilt on the next start. Deleting it does throw away track analysis, which
|
||||||
# is expensive to recompute.
|
# is expensive to recompute.
|
||||||
cache: .musicmouse-cache
|
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.
|
# 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
|
# Required - use "simulate" to run without the mouse attached, which is a complete
|
||||||
|
|||||||
@@ -32,7 +32,7 @@ from musicmouse.devices.mouse import MusicMouseDevice
|
|||||||
from musicmouse.devices.null_transport import NullTransport
|
from musicmouse.devices.null_transport import NullTransport
|
||||||
from musicmouse.devices.player import Player, VlcPlayer
|
from musicmouse.devices.player import Player, VlcPlayer
|
||||||
from musicmouse.devices.serial_link import SerialLink
|
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.library.analysis import build_analyzer
|
||||||
from musicmouse.reactions import register_all
|
from musicmouse.reactions import register_all
|
||||||
from musicmouse.services.base import Service
|
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:
|
async def build_library(config: Config) -> MusicLibrary:
|
||||||
library_config = config.general.library
|
library_config = config.general.library
|
||||||
|
workers = library_config.analysis_workers
|
||||||
return await MusicLibrary.build(
|
return await MusicLibrary.build(
|
||||||
library_config.root,
|
library_config.root,
|
||||||
library_config.cache,
|
library_config.cache,
|
||||||
frozenset(config.general.audio_extensions),
|
frozenset(config.general.audio_extensions),
|
||||||
analyzer=build_analyzer(),
|
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,
|
figure_kinds=config.figure_kinds,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -172,6 +172,11 @@ class LibraryConfig(_Strict):
|
|||||||
root: Path
|
root: Path
|
||||||
#: Scan results, extracted cover art and track analysis. Relative to this file.
|
#: Scan results, extracted cover art and track analysis. Relative to this file.
|
||||||
cache: Path = Path(".musicmouse-cache")
|
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")
|
@field_validator("root")
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|||||||
@@ -8,14 +8,14 @@ plain immutable data.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import contextlib
|
|
||||||
import logging
|
import logging
|
||||||
import os
|
|
||||||
import time
|
import time
|
||||||
|
from collections import deque
|
||||||
from collections.abc import Awaitable, Callable, Collection, Mapping
|
from collections.abc import Awaitable, Callable, Collection, Mapping
|
||||||
|
from concurrent.futures import BrokenExecutor, Executor
|
||||||
from dataclasses import replace
|
from dataclasses import replace
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Final
|
from typing import Any, Final
|
||||||
|
|
||||||
from musicmouse.library.analysis import (
|
from musicmouse.library.analysis import (
|
||||||
ANALYZER_VERSION,
|
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.models import Album, LibraryTrack, album_id, track_key
|
||||||
from musicmouse.library.scanner import scan_library
|
from musicmouse.library.scanner import scan_library
|
||||||
from musicmouse.library.sections import SECTIONS, AlbumKind
|
from musicmouse.library.sections import SECTIONS, AlbumKind
|
||||||
|
from musicmouse.library.workers import analysis_pool, analyze_one, default_worker_count
|
||||||
from musicmouse.media import Playlist
|
from musicmouse.media import Playlist
|
||||||
|
|
||||||
_log = logging.getLogger(__name__)
|
_log = logging.getLogger(__name__)
|
||||||
@@ -45,6 +46,7 @@ __all__ = [
|
|||||||
"NullAnalyzer",
|
"NullAnalyzer",
|
||||||
"TrackCurves",
|
"TrackCurves",
|
||||||
"album_id",
|
"album_id",
|
||||||
|
"default_worker_count",
|
||||||
"track_key",
|
"track_key",
|
||||||
]
|
]
|
||||||
|
|
||||||
@@ -69,19 +71,12 @@ _PUBLISH_BATCH_SIZE: Final = 25
|
|||||||
_PROGRESS_INTERVAL_SECONDS: Final = 5.0
|
_PROGRESS_INTERVAL_SECONDS: Final = 5.0
|
||||||
|
|
||||||
|
|
||||||
def _analyze_one(
|
def _discard(future: asyncio.Future[Any]) -> None:
|
||||||
analyzer: Analyzer, path: Path
|
"""Drop a result nobody is going to read, without leaving a warning behind."""
|
||||||
) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]:
|
if not future.done():
|
||||||
"""Runs in a worker thread. Lowers this thread's own scheduling priority first.
|
future.cancel()
|
||||||
|
elif not future.cancelled():
|
||||||
On Linux, ``os.nice`` affects only the calling thread, not the whole process - so
|
future.exception()
|
||||||
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)
|
|
||||||
|
|
||||||
|
|
||||||
class MusicLibrary:
|
class MusicLibrary:
|
||||||
@@ -94,12 +89,19 @@ class MusicLibrary:
|
|||||||
extensions: frozenset[str],
|
extensions: frozenset[str],
|
||||||
*,
|
*,
|
||||||
analyzer: Analyzer | None = None,
|
analyzer: Analyzer | None = None,
|
||||||
|
analysis_workers: int = 1,
|
||||||
figure_kinds: Mapping[str, AlbumKind] | None = None,
|
figure_kinds: Mapping[str, AlbumKind] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
self.root = root
|
self.root = root
|
||||||
self.cache = cache
|
self.cache = cache
|
||||||
self.extensions = extensions
|
self.extensions = extensions
|
||||||
self.analyzer: Analyzer = analyzer or NullAnalyzer()
|
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.
|
#: What each figure holds. The only thing a folder name cannot say.
|
||||||
self.figure_kinds: Mapping[str, AlbumKind] = figure_kinds or {}
|
self.figure_kinds: Mapping[str, AlbumKind] = figure_kinds or {}
|
||||||
self._entries: dict[str, tuple[Album, Fingerprint]] = {}
|
self._entries: dict[str, tuple[Album, Fingerprint]] = {}
|
||||||
@@ -146,9 +148,7 @@ class MusicLibrary:
|
|||||||
]
|
]
|
||||||
if not candidates:
|
if not candidates:
|
||||||
return None
|
return None
|
||||||
return max(
|
return max(candidates, key=lambda album: album.tracks[0].path.name if album.tracks else "")
|
||||||
candidates, key=lambda album: album.tracks[0].path.name if album.tracks else ""
|
|
||||||
)
|
|
||||||
|
|
||||||
def beats(self, identifier: str, index: int) -> BeatGrid | None:
|
def beats(self, identifier: str, index: int) -> BeatGrid | None:
|
||||||
album = self.get(identifier)
|
album = self.get(identifier)
|
||||||
@@ -219,10 +219,18 @@ class MusicLibrary:
|
|||||||
extensions: frozenset[str],
|
extensions: frozenset[str],
|
||||||
*,
|
*,
|
||||||
analyzer: Analyzer | None = None,
|
analyzer: Analyzer | None = None,
|
||||||
|
analysis_workers: int = 1,
|
||||||
figure_kinds: Mapping[str, AlbumKind] | None = None,
|
figure_kinds: Mapping[str, AlbumKind] | None = None,
|
||||||
) -> MusicLibrary:
|
) -> MusicLibrary:
|
||||||
cache = LibraryCache(cache_dir)
|
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)
|
library._entries = await asyncio.to_thread(cache.load_index)
|
||||||
await library.refresh()
|
await library.refresh()
|
||||||
return library
|
return library
|
||||||
@@ -261,64 +269,145 @@ class MusicLibrary:
|
|||||||
is_busy: Callable[[], bool] = lambda: False,
|
is_busy: Callable[[], bool] = lambda: False,
|
||||||
on_batch: Callable[[], Awaitable[None]] | None = None,
|
on_batch: Callable[[], Awaitable[None]] | None = None,
|
||||||
batch_size: int = _PUBLISH_BATCH_SIZE,
|
batch_size: int = _PUBLISH_BATCH_SIZE,
|
||||||
|
workers: int | None = None,
|
||||||
) -> int:
|
) -> int:
|
||||||
"""Run the analyzer over tracks of `kinds` that have no current result.
|
"""Run the analyzer over tracks of `kinds` that have no current result.
|
||||||
|
|
||||||
Restricted to music by default - see `_ANALYZED_KINDS`. Checked before every
|
Restricted to music by default - see `_ANALYZED_KINDS`. Up to `workers` tracks
|
||||||
track, `is_busy()` pauses the whole pass rather than one file: analysis must
|
are analyzed at once (`self.analysis_workers` when not given), each in its own
|
||||||
never compete with audio decoding for CPU, and a children's player is idle most
|
worker process - see `musicmouse.library.workers` for why processes and how the
|
||||||
of the day, so the pass simply resumes next time it is. A track the analyzer
|
machine is kept usable while they run. Checked before every track is handed out,
|
||||||
fails on (corrupt file, DRM, zero length) is still recorded as attempted - with
|
`is_busy()` pauses the whole pass rather than one file: analysis must never
|
||||||
every scalar left `None` - so it is never retried forever and the frontend falls
|
compete with audio decoding for CPU, and a children's player is idle most of the
|
||||||
back to the un-analyzed baseline for it. Results are folded into the live index
|
day, so the pass simply resumes next time it is. Tracks already in flight when
|
||||||
and persisted every `batch_size` tracks, so a long first run is visible in open
|
it goes busy are allowed to finish - their results are already paid for - and
|
||||||
browser tabs as it goes rather than only once it finishes.
|
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
|
analyzer = self.analyzer
|
||||||
if analyzer.version < ANALYZER_VERSION:
|
if analyzer.version < ANALYZER_VERSION:
|
||||||
return 0
|
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
|
done = 0
|
||||||
last_report = time.monotonic()
|
last_report = time.monotonic()
|
||||||
for album in self.albums:
|
try:
|
||||||
if album.kind not in kinds:
|
while pending:
|
||||||
continue
|
|
||||||
for track in album.tracks:
|
|
||||||
while is_busy():
|
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)
|
await asyncio.sleep(_BUSY_POLL_SECONDS)
|
||||||
key = track_key(track.path)
|
with analysis_pool(workers) as pool:
|
||||||
cached = self.cache.load_analysis(key)
|
try:
|
||||||
if cached is not None and cached.version >= analyzer.version:
|
while in_flight or (pending and not is_busy()):
|
||||||
continue
|
while pending and len(in_flight) < workers and not is_busy():
|
||||||
now = time.monotonic()
|
key, path = pending.popleft()
|
||||||
if now - last_report >= _PROGRESS_INTERVAL_SECONDS:
|
in_flight[self._submit(pool, loop, path)] = (key, path)
|
||||||
_log.info("Analyzing library: %d tracks done so far, now on %s", done, track.path)
|
if not in_flight:
|
||||||
last_report = now
|
break # gone busy: drop the pool and wait above
|
||||||
try:
|
finished, _ = await asyncio.wait(
|
||||||
analysis, grid, curve = await asyncio.to_thread(
|
in_flight, return_when=asyncio.FIRST_COMPLETED
|
||||||
_analyze_one, analyzer, track.path
|
)
|
||||||
)
|
for future in finished:
|
||||||
except Exception:
|
key, path = in_flight.pop(future)
|
||||||
_log.warning(
|
self._store_result(key, path, future)
|
||||||
"Analyzer raised on %s; marking it attempted so it is not retried forever",
|
done += 1
|
||||||
track.path,
|
if done % batch_size == 0:
|
||||||
exc_info=True,
|
await self._publish_analysis(kinds, on_batch)
|
||||||
)
|
now = time.monotonic()
|
||||||
analysis, grid, curve = TrackAnalysis(version=analyzer.version), None, None
|
if now - last_report >= _PROGRESS_INTERVAL_SECONDS:
|
||||||
if grid is not None:
|
_log.info("Analyzing library: %d/%d tracks done", done, total)
|
||||||
self.cache.store_beats(key, grid)
|
last_report = now
|
||||||
if curve is not None:
|
finally:
|
||||||
self.cache.store_curve(key, curve)
|
# Nothing is left to read these - without this, a pass that ends
|
||||||
self.cache.store_analysis(key, analysis)
|
# early (cancelled at shutdown, or a dead pool) leaves "exception
|
||||||
done += 1
|
# was never retrieved" behind for every track still in flight.
|
||||||
if done % batch_size == 0:
|
for future in in_flight:
|
||||||
await self._publish_analysis(kinds, on_batch)
|
_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:
|
if done % batch_size:
|
||||||
await self._publish_analysis(kinds, on_batch)
|
await self._publish_analysis(kinds, on_batch)
|
||||||
if done:
|
if done:
|
||||||
_log.info("Analyzed %d tracks", done)
|
_log.info("Analyzed %d tracks", done)
|
||||||
return 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(
|
async def _publish_analysis(
|
||||||
self, kinds: Collection[AlbumKind], on_batch: Callable[[], Awaitable[None]] | None
|
self, kinds: Collection[AlbumKind], on_batch: Callable[[], Awaitable[None]] | None
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|||||||
143
python-backend/musicmouse/library/workers.py
Normal file
143
python-backend/musicmouse/library/workers.py
Normal file
@@ -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)
|
||||||
@@ -5,14 +5,16 @@ from __future__ import annotations
|
|||||||
import asyncio
|
import asyncio
|
||||||
import contextlib
|
import contextlib
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
import threading
|
import threading
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
|
from dataclasses import replace
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from musicmouse.config import DEFAULT_AUDIO_EXTENSIONS, load_config
|
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.analysis import ANALYZER_VERSION, BeatGrid, TrackAnalysis, TrackCurves
|
||||||
from musicmouse.library.cache import LibraryCache
|
from musicmouse.library.cache import LibraryCache
|
||||||
from musicmouse.library.colors import colors_from_id
|
from musicmouse.library.colors import colors_from_id
|
||||||
@@ -52,6 +54,19 @@ class FakeAnalyzer:
|
|||||||
return analysis, BeatGrid((0.1,), (1.0,)), curves
|
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:
|
def album_named(library: MusicLibrary, title: str) -> Album:
|
||||||
return next(album for album in library.albums if album.title == title)
|
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 pass_count == 2
|
||||||
assert len(analyzer.calls) == 5
|
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)
|
||||||
|
|||||||
Reference in New Issue
Block a user