Prevent players buffering ahead too much

music_assistant/server/controllers/player_queues.py

@@ -1129,6 +1129,8 @@ def track_loaded_in_buffer(self, queue_id: str, item_id: str) -> None:
  if not queue:
  msg = f"PlayerQueue {queue_id} is not available"
  raise PlayerUnavailableError(msg)
+ # store the index of the item that is currently (being) loaded in the buffer
+ # which helps us a bit to determine how far the player has buffered ahead
  queue.index_in_buffer = self.index_by_id(queue_id, item_id)
  if queue.flow_mode:
  return # nothing to do when flow mode is active

music_assistant/server/controllers/streams.py

@@ -383,6 +383,8 @@ async def serve_queue_item_stream(self, request: web.Request) -> web.Response:
  input_format=pcm_format,
  output_format=output_format,
  filter_params=get_player_filter_params(self.mass, queue_player.player_id),
+ # we don't allow the player to buffer too much ahead so we use readrate limiting
+ extra_input_args=["-readrate", "1.1", "-readrate_initial_burst", "10"],
  ):
  try:
  await resp.write(chunk)
@@ -472,6 +474,8 @@ async def serve_queue_flow_stream(self, request: web.Request) -> web.Response:
  output_format=output_format,
  filter_params=get_player_filter_params(self.mass, queue_player.player_id),
  chunk_size=icy_meta_interval if enable_icy else None,
+ # we don't allow the player to buffer too much ahead so we use readrate limiting
+ extra_input_args=["-readrate", "1.1", "-readrate_initial_burst", "10"],
  ):
  try:
  await resp.write(chunk)

music_assistant/server/helpers/ffmpeg.py

@@ -17,6 +17,7 @@
 from .util import TimedAsyncGenerator, close_async_generator
 
 LOGGER = logging.getLogger("ffmpeg")
+MINIMAL_FFMPEG_VERSION = 6
 
 
 class FFMpeg(AsyncProcess):
@@ -175,7 +176,7 @@ async def get_ffmpeg_stream(
  yield chunk
 
 
-def get_ffmpeg_args(
+def get_ffmpeg_args( # noqa: PLR0915
  input_format: AudioFormat,
  output_format: AudioFormat,
  filter_params: list[str],
@@ -199,6 +200,12 @@ def get_ffmpeg_args(
  )
 
  major_version = int("".join(char for char in version.split(".")[0] if not char.isalpha()))
+ if major_version < MINIMAL_FFMPEG_VERSION:
+ msg = (
+ f"FFmpeg version {version} is not supported. "
+ f"Minimal version required is {MINIMAL_FFMPEG_VERSION}."
+ )
+ raise AudioError(msg)
 
  # generic args
  generic_args = [
@@ -227,18 +234,14 @@ def get_ffmpeg_args(
  # If set then even streamed/non seekable streams will be reconnected on errors.
  "-reconnect_streamed",
  "1",
+ # Reconnect automatically in case of TCP/TLS errors during connect.
+ "-reconnect_on_network_error",
+ "1",
+ # A comma separated list of HTTP status codes to reconnect on.
+ # The list can include specific status codes (e.g. 503) or the strings 4xx / 5xx.
+ "-reconnect_on_http_error",
+ "5xx,4xx",
  ]
- if major_version > 4:
- # these options are only supported in ffmpeg > 5
- input_args += [
- # Reconnect automatically in case of TCP/TLS errors during connect.
- "-reconnect_on_network_error",
- "1",
- # A comma separated list of HTTP status codes to reconnect on.
- # The list can include specific status codes (e.g. 503) or the strings 4xx / 5xx.
- "-reconnect_on_http_error",
- "5xx,4xx",
- ]
  if input_format.content_type.is_pcm():
  input_args += [
  "-ac",

music_assistant/server/providers/player_group/ugp_stream.py

@@ -83,9 +83,8 @@ async def _runner(self) -> None:
  audio_input=self.audio_source,
  input_format=self.input_format,
  output_format=self.output_format,
- # enable realtime to prevent too much buffering ahead
- # TODO: enable initial burst once we have a newer ffmpeg version
- extra_input_args=["-re"],
+ # we don't allow the player to buffer too much ahead so we use readrate limiting
+ extra_input_args=["-readrate", "1.1", "-readrate_initial_burst", "10"],
  ):
  await asyncio.gather(
  *[sub(chunk) for sub in self.subscribers],