From 62f3f22002b256c085de089a5e4df82584a410d2 Mon Sep 17 00:00:00 2001 From: Marcel van der Veldt Date: Sat, 20 Jul 2024 19:02:09 +0200 Subject: [PATCH] Fix AirPlay streams get mixed up when started in quick succession (#1508) --- music_assistant/common/models/player_queue.py | 2 - .../server/controllers/player_queues.py | 7 ++- music_assistant/server/controllers/players.py | 29 ++++++--- music_assistant/server/controllers/streams.py | 6 ++ music_assistant/server/helpers/util.py | 4 +- .../server/providers/airplay/__init__.py | 63 +++++-------------- .../server/providers/builtin/__init__.py | 5 +- 7 files changed, 52 insertions(+), 64 deletions(-) diff --git a/music_assistant/common/models/player_queue.py b/music_assistant/common/models/player_queue.py index 87e37e8e..8f16281a 100644 --- a/music_assistant/common/models/player_queue.py +++ b/music_assistant/common/models/player_queue.py @@ -53,7 +53,6 @@ class PlayerQueue(DataClassDictMixin): d.pop("current_item", None) d.pop("next_item", None) d.pop("index_in_buffer", None) - d.pop("announcement_in_progress", None) d.pop("flow_mode", None) d.pop("flow_mode_start_index", None) return d @@ -64,7 +63,6 @@ class PlayerQueue(DataClassDictMixin): d.pop("current_item", None) d.pop("next_item", None) d.pop("index_in_buffer", None) - d.pop("announcement_in_progress", None) d.pop("flow_mode", None) d.pop("flow_mode_start_index", None) return cls.from_dict(d) diff --git a/music_assistant/server/controllers/player_queues.py b/music_assistant/server/controllers/player_queues.py index e4643910..d3a6e6bd 100644 --- a/music_assistant/server/controllers/player_queues.py +++ b/music_assistant/server/controllers/player_queues.py @@ -743,10 +743,15 @@ class PlayerQueuesController(CoreController): self.mass, queue_item, seek_position=seek_position, fade_in=fade_in ) # send play_media request to player - await self.mass.players.play_media( + # NOTE that we debounce this a bit to account for someone hitting the next button + # like a madman. This will prevent the player from being overloaded with requests. + self.mass.call_later( + 0.25, + self.mass.players.play_media, player_id=queue_id, # transform into PlayerMedia to send to the actual player implementation media=self.player_media_from_queue_item(queue_item, queue.flow_mode), + task_id=f"play_media_{queue_id}", ) # Interaction with player diff --git a/music_assistant/server/controllers/players.py b/music_assistant/server/controllers/players.py index 3d46f171..ccfcf74f 100644 --- a/music_assistant/server/controllers/players.py +++ b/music_assistant/server/controllers/players.py @@ -106,6 +106,7 @@ class PlayerController(CoreController): ) self.manifest.icon = "speaker-multiple" self._poll_task: asyncio.Task | None = None + self._player_locks: dict[str, asyncio.Lock] = {} async def setup(self, config: CoreConfig) -> None: """Async initialize of module.""" @@ -199,7 +200,8 @@ class PlayerController(CoreController): await self.mass.player_queues.play(player_id) return player_provider = self.get_player_provider(player_id) - await player_provider.cmd_play(player_id) + async with self._player_locks[player_id]: + await player_provider.cmd_play(player_id) @api_command("players/cmd/pause") @handle_player_command @@ -242,7 +244,6 @@ class PlayerController(CoreController): self.mass.create_task(_watch_pause(player_id)) @api_command("players/cmd/play_pause") - @handle_player_command async def cmd_play_pause(self, player_id: str) -> None: """Toggle play/pause on given player. @@ -292,7 +293,8 @@ class PlayerController(CoreController): if PlayerFeature.POWER in player.supported_features: # forward to player provider player_provider = self.get_player_provider(player_id) - await player_provider.cmd_power(player_id, powered) + async with self._player_locks[player_id]: + await player_provider.cmd_power(player_id, powered) else: # allow the stop command to process and prevent race conditions await asyncio.sleep(0.2) @@ -332,7 +334,8 @@ class PlayerController(CoreController): msg = f"Player {player.display_name} does not support volume_set" raise UnsupportedFeaturedException(msg) player_provider = self.get_player_provider(player_id) - await player_provider.cmd_volume_set(player_id, volume_level) + async with self._player_locks[player_id]: + await player_provider.cmd_volume_set(player_id, volume_level) @api_command("players/cmd/volume_up") @handle_player_command @@ -438,7 +441,8 @@ class PlayerController(CoreController): msg = f"Player {player.display_name} does not support muting" raise UnsupportedFeaturedException(msg) player_provider = self.get_player_provider(player_id) - await player_provider.cmd_volume_mute(player_id, muted) + async with self._player_locks[player_id]: + await player_provider.cmd_volume_mute(player_id, muted) @api_command("players/cmd/seek") async def cmd_seek(self, player_id: str, position: int) -> None: @@ -466,8 +470,8 @@ class PlayerController(CoreController): ) -> None: """Handle playback of an announcement (url) on given player.""" player = self.get(player_id, True) - if player.announcement_in_progress: - return + while player.announcement_in_progress: + await asyncio.sleep(0.5) if not url.startswith("http"): raise PlayerCommandFailed("Only URLs are supported for announcements") try: @@ -576,7 +580,8 @@ class PlayerController(CoreController): ) return player_prov = self.mass.players.get_player_provider(player_id) - await player_prov.enqueue_next_media(player_id=player_id, media=media) + async with self._player_locks[player_id]: + await player_prov.enqueue_next_media(player_id=player_id, media=media) @api_command("players/cmd/sync") @handle_player_command @@ -628,7 +633,7 @@ class PlayerController(CoreController): continue if child_player.synced_to and child_player.synced_to == target_player: continue # already synced to this target - if child_player.synced_to and child_player.synced_to != target_player: + elif child_player.synced_to: # player already synced to another player, unsync first self.logger.warning( "Player %s is already synced, unsyncing first", child_player.name @@ -649,7 +654,8 @@ class PlayerController(CoreController): # forward command to the player provider after all (base) sanity checks player_provider = self.get_player_provider(target_player) - await player_provider.cmd_sync_many(target_player, child_player_ids) + async with self._player_locks[target_player]: + await player_provider.cmd_sync_many(target_player, child_player_ids) @api_command("players/cmd/unsync_many") async def cmd_unsync_many(self, player_ids: list[str]) -> None: @@ -735,6 +741,9 @@ class PlayerController(CoreController): # register playerqueue for this player self.mass.create_task(self.mass.player_queues.on_player_register(player)) + # register lock for this player + self._player_locks[player_id] = asyncio.Lock() + self._players[player_id] = player # ignore disabled players diff --git a/music_assistant/server/controllers/streams.py b/music_assistant/server/controllers/streams.py index 33451e26..b480fca4 100644 --- a/music_assistant/server/controllers/streams.py +++ b/music_assistant/server/controllers/streams.py @@ -498,6 +498,9 @@ class StreamsController(CoreController): use_crossfade = await self.mass.config.get_player_config_value( queue.queue_id, CONF_CROSSFADE ) + if not start_queue_item: + # this can happen in some (edge case) race conditions + return if start_queue_item.media_type != MediaType.TRACK: use_crossfade = False pcm_sample_size = int( @@ -775,6 +778,9 @@ class StreamsController(CoreController): # del chunk finished = True finally: + if "ffmpeg_proc" not in locals(): + # edge case: ffmpeg process was not yet started + return # noqa: B012 if finished and not ffmpeg_proc.closed: await asyncio.wait_for(ffmpeg_proc.wait(), 60) elif not ffmpeg_proc.closed: diff --git a/music_assistant/server/helpers/util.py b/music_assistant/server/helpers/util.py index b5afcc26..b1be4202 100644 --- a/music_assistant/server/helpers/util.py +++ b/music_assistant/server/helpers/util.py @@ -149,9 +149,9 @@ class TaskManager: self.mass = mass self._tasks: list[asyncio.Task] = [] - def create_task(self, coro: Coroutine) -> None: + def create_task(self, coro: Coroutine, eager_start: bool = False) -> None: """Create a new task and add it to the manager.""" - task = self.mass.create_task(coro) + task = self.mass.create_task(coro, eager_start=eager_start) self._tasks.append(task) async def __aenter__(self) -> Self: diff --git a/music_assistant/server/providers/airplay/__init__.py b/music_assistant/server/providers/airplay/__init__.py index db046689..52aee918 100644 --- a/music_assistant/server/providers/airplay/__init__.py +++ b/music_assistant/server/providers/airplay/__init__.py @@ -17,7 +17,7 @@ from zeroconf import IPVersion, ServiceStateChange from zeroconf.asyncio import AsyncServiceInfo from music_assistant.common.helpers.datetime import utc -from music_assistant.common.helpers.util import empty_queue, get_ip_pton, select_free_port +from music_assistant.common.helpers.util import get_ip_pton, select_free_port from music_assistant.common.models.config_entries import ( CONF_ENTRY_CROSSFADE, CONF_ENTRY_CROSSFADE_DURATION, @@ -299,7 +299,7 @@ class AirplayStream: await self._cliraop_proc.start() await asyncio.to_thread(os.close, read) self._started.set() - self._log_reader_task = asyncio.create_task(self._log_watcher()) + self._log_reader_task = self.mass.create_task(self._log_watcher()) async def stop(self): """Stop playback and cleanup.""" @@ -307,19 +307,9 @@ class AirplayStream: return if self.audio_source_task and not self.audio_source_task.done(): self.audio_source_task.cancel() - if self._cliraop_proc.proc and not self._cliraop_proc.closed: - await self.send_cli_command("ACTION=STOP") - self._stopped = True # set after send_cli command! if self._cliraop_proc.proc: - try: - await asyncio.wait_for(self._cliraop_proc.wait(), 5) - except TimeoutError: - self.prov.logger.warning( - "Raop process for %s did not stop in time, is the player offline?", - self.airplay_player.player_id, - ) - await self._cliraop_proc.close(True) - + await self._cliraop_proc.close(True) + self._stopped = True # set after close command! # ffmpeg can sometimes hang due to the connected pipes # we handle closing it but it can be a bit slow so do that in the background if not self._ffmpeg_proc.closed: @@ -609,7 +599,9 @@ class AirplayProvider(PlayerProvider): for airplay_player in self._get_sync_clients(player_id): if airplay_player.active_stream and airplay_player.active_stream.running: # prefer interactive command to our streamer - tg.create_task(airplay_player.active_stream.send_cli_command("ACTION=PLAY")) + tg.create_task( + airplay_player.active_stream.send_cli_command("ACTION=PLAY"), + ) async def cmd_pause(self, player_id: str) -> None: """Send PAUSE command to given player. @@ -699,38 +691,18 @@ class AirplayProvider(PlayerProvider): self, airplay_player, input_format=input_format ) - # use a buffer here to consume small hiccups as the - # raop streaming is pretty much realtime and without a buffer to stdin - buffer: asyncio.Queue[bytes] = asyncio.Queue(10) - - async def fill_buffer() -> None: - async for chunk in audio_source: - await buffer.put(chunk) - await buffer.put(b"EOF") - - fill_buffer_task = asyncio.create_task(fill_buffer()) - async def audio_streamer() -> None: - try: - while True: - chunk = await buffer.get() - if chunk == b"EOF": - break - await asyncio.gather( - *[x.active_stream.write_chunk(chunk) for x in sync_clients], - return_exceptions=True, - ) - - # entire stream consumed: send EOF + async for chunk in audio_source: await asyncio.gather( - *[x.active_stream.write_eof() for x in sync_clients], + *[x.active_stream.write_chunk(chunk) for x in sync_clients], return_exceptions=True, ) - finally: - if not fill_buffer_task.done(): - fill_buffer_task.cancel() - empty_queue(buffer) + # entire stream consumed: send EOF + await asyncio.gather( + *[x.active_stream.write_eof() for x in sync_clients], + return_exceptions=True, + ) # get current ntp and start cliraop _, stdout = await check_output(self.cliraop_bin, "-ntp") @@ -740,7 +712,7 @@ class AirplayProvider(PlayerProvider): *[x.active_stream.start(start_ntp, wait_start) for x in sync_clients], return_exceptions=True, ) - self._players[player_id].active_stream.audio_source_task = asyncio.create_task( + self._players[player_id].active_stream.audio_source_task = self.mass.create_task( audio_streamer() ) @@ -767,6 +739,8 @@ class AirplayProvider(PlayerProvider): - player_id: player_id of the player to handle the command. - target_player: player_id of the syncgroup master or group player. """ + if player_id == target_player: + return child_player = self.mass.players.get(player_id) assert child_player # guard parent_player = self.mass.players.get(target_player) @@ -813,9 +787,6 @@ class AirplayProvider(PlayerProvider): group_leader = self.mass.players.get(player.synced_to, raise_unavailable=True) group_leader.group_childs.remove(player_id) player.synced_to = None - # guard if this was the last sync child of the group player - if group_leader.group_childs == {group_leader.player_id}: - group_leader.group_childs.remove(group_leader.player_id) await self.cmd_stop(player_id) self.mass.players.update(player_id) diff --git a/music_assistant/server/providers/builtin/__init__.py b/music_assistant/server/providers/builtin/__init__.py index 02517fbd..8ab9368c 100644 --- a/music_assistant/server/providers/builtin/__init__.py +++ b/music_assistant/server/providers/builtin/__init__.py @@ -6,7 +6,7 @@ import asyncio import os import time from collections.abc import AsyncGenerator -from typing import TYPE_CHECKING, NotRequired, TypedDict +from typing import TYPE_CHECKING, NotRequired, TypedDict, cast import aiofiles import shortuuid @@ -162,8 +162,7 @@ class BuiltinProvider(MusicProvider): async def get_track(self, prov_track_id: str) -> Track: """Get full track details by id.""" - parsed_item = await self.parse_item(prov_track_id) - assert isinstance(parsed_item, Track) + parsed_item = cast(Track, await self.parse_item(prov_track_id)) stored_items: list[StoredItem] = self.mass.config.get(CONF_KEY_TRACKS, []) if stored_item := next((x for x in stored_items if x["item_id"] == prov_track_id), None): # always prefer the stored info, such as the name -- 2.34.1