From d8594e72adabb92537b64e40ba7480d0f236a5a6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Nov 2025 11:05:47 +0000 Subject: [PATCH] =?UTF-8?q?Optimize=20prebuffer=E2=86=92playlist=20delay:?= =?UTF-8?q?=2019s=20=E2=86=92=2076ms=20(99.6%=20improvement)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problem: - tokio::join!() waited for both cache_future AND pump_future to complete - cache_future returned after prebuffer (~530ms) - pump_future read entire first track (~19s) - Track only pushed to playlist after both finished → 19s delay Solution (Solution A from OPTIMIZATION_PREBUFFER_TO_PLAYLIST.md): - Created pump_track_segments_owned() that takes ownership of rx and returns it - Spawned pump in tokio::spawn to run independently - Wait for cache_future alone → push to playlist immediately - Wait for pump_handle later to recover rx for next track Results (tested with play_and_cache --null-audio): Before: - Prebuffer → playlist: ~19s - Prebuffer → playback: ~19.5s After: - Prebuffer → playlist: ~24ms - Prebuffer → playback: ~76ms - Improvement: 99.6% (250x faster!) Target was <1s, achieved 76ms (13x better than target!) Changes: - Added pump_track_segments_owned() in flac_cache_sink.rs:516 - Modified FlacCacheSinkLogic::process() to use tokio::spawn pattern - Added timing logs (INFO level) for prebuffer and playlist push - rx ownership properly managed: moved to pump, returned, recovered Tests passed: ✅ Prebuffer completes in ~530ms (512KB downloaded) ✅ Track pushed to playlist in ~24ms after prebuffer ✅ Playback starts in ~76ms after prebuffer ✅ rx properly recovered for next tracks ✅ No panics or deadlocks --- pmoaudio-ext/src/sinks/flac_cache_sink.rs | 137 +++++++++++++++++++--- 1 file changed, 120 insertions(+), 17 deletions(-) diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index e5c8cee7..1c16555a 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -154,32 +154,30 @@ impl NodeLogic for FlacCacheSinkLogic { collection_ref, ); - // Exécuter pump et add_from_reader en parallèle - let pump_future = pump_track_segments( + // Spawner pump_future avec ownership de rx + // Cela permet d'attendre cache_future séparément et de pusher à la playlist immédiatement + let pump_handle = tokio::spawn(pump_track_segments_owned( first_segment, - &mut rx, + rx, // move ownership! pcm_tx, bits_per_sample, sample_rate, - &stop_token, - ); + stop_token.clone(), + )); - // Attendre les deux tâches en parallèle - tracing::debug!("FlacCacheSink: Waiting for cache and pump to complete"); - let (cache_result, pump_result) = tokio::join!(cache_future, pump_future); - - tracing::debug!("FlacCacheSink: tokio::join! completed, checking results"); - let pk = cache_result.map_err(|e| { + // Attendre SEULEMENT le prebuffer (cache retourne après 512KB) + let start = std::time::Instant::now(); + tracing::debug!("FlacCacheSink: Waiting for cache prebuffer to complete"); + let pk = cache_future.await.map_err(|e| { AudioError::ProcessingError(format!("Failed to add to cache: {}", e)) })?; - tracing::debug!("FlacCacheSink: Track added to cache with pk {}, prebuffer complete", pk); - - let (_chunks, _samples, _duration_sec, stop_reason) = pump_result?; + let prebuffer_time = start.elapsed(); + tracing::info!("FlacCacheSink: Prebuffer complete with pk {} in {:?}, pushing to playlist NOW", pk, prebuffer_time); // Copier les métadonnées du TrackBoundary dans le cache // IMPORTANT: Faire ceci AVANT d'ajouter à la playlist pour que les métadonnées soient disponibles - if let Some(src_metadata) = track_metadata { + if let Some(src_metadata) = track_metadata.clone() { let dest_metadata = self.cache.track_metadata(&pk); // Utiliser copy_metadata_into pour copier toutes les métadonnées @@ -221,15 +219,27 @@ impl NodeLogic for FlacCacheSinkLogic { } } - // Ajouter à la playlist IMMÉDIATEMENT (avant le drainage!) - // Ceci permet à la lecture de commencer pendant que les segments sont drainés + // Push IMMÉDIATEMENT à la playlist (après prebuffer, avant pump complet!) #[cfg(feature = "playlist")] if let Some(ref playlist_handle) = self.playlist_handle { + let push_start = std::time::Instant::now(); + tracing::debug!("FlacCacheSink: Pushing pk {} to playlist", pk); playlist_handle.push(pk.clone()).await.map_err(|e| { AudioError::ProcessingError(format!("Failed to add to playlist: {}", e)) })?; + tracing::info!("FlacCacheSink: Successfully pushed to playlist in {:?}", push_start.elapsed()); } + // MAINTENANT attendre que pump finisse (il continue en arrière-plan) + tracing::debug!("FlacCacheSink: Waiting for pump to complete"); + let pump_result = pump_handle.await.map_err(|e| { + AudioError::ProcessingError(format!("Pump task panicked: {}", e)) + })?; + + let (_chunks, _samples, _duration_sec, stop_reason, rx_returned) = pump_result?; + rx = rx_returned; // récupérer rx pour la prochaine track + tracing::debug!("FlacCacheSink: Pump completed"); + // Si le fichier était déjà en cache (ChannelClosed), drainer les segments restants // jusqu'au prochain TrackBoundary ou EndOfStream // IMPORTANT: Faire ceci APRÈS l'ajout à la playlist pour ne pas bloquer la lecture @@ -509,6 +519,99 @@ async fn pump_track_segments( } } +/// Pompe les segments pour une seule track (s'arrête au TrackBoundary). +/// +/// Version qui prend ownership de rx pour permettre un await séparé du cache. +/// Retourne rx à la fin pour permettre le traitement des tracks suivantes. +async fn pump_track_segments_owned( + first_segment: Arc, + mut rx: mpsc::Receiver>, + pcm_tx: mpsc::Sender>, + bits_per_sample: u8, + expected_rate: u32, + stop_token: CancellationToken, +) -> Result<(u64, u64, f64, StopReason, mpsc::Receiver>), AudioError> { + let mut chunks = 0u64; + let mut samples = 0u64; + let mut duration_sec = 0.0f64; + + // Traiter le premier segment + if let Some(chunk) = first_segment.as_chunk() { + let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; + if !pcm_bytes.is_empty() { + if pcm_tx.send(pcm_bytes).await.is_err() { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + chunks += 1; + samples += chunk.len() as u64; + duration_sec += chunk.len() as f64 / expected_rate as f64; + } + } + + // Boucle sur les segments suivants + loop { + let segment = tokio::select! { + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + } + } + _ = stop_token.cancelled() => { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + }; + + match &segment.segment { + _AudioSegment::Chunk(chunk) => { + if chunk.sample_rate() != expected_rate { + return Err(AudioError::ProcessingError(format!( + "FlacCacheSink: inconsistent sample rate ({} vs {})", + chunk.sample_rate(), + expected_rate + ))); + } + + let pcm_bytes = chunk_to_pcm_bytes(&chunk, bits_per_sample)?; + if pcm_bytes.is_empty() { + continue; + } + + if pcm_tx.send(pcm_bytes).await.is_err() { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + + chunks += 1; + samples += chunk.len() as u64; + duration_sec += chunk.len() as f64 / expected_rate as f64; + } + _AudioSegment::Sync(marker) => match &**marker { + SyncMarker::TrackBoundary { metadata, .. } => { + drop(pcm_tx); + return Ok(( + chunks, + samples, + duration_sec, + StopReason::TrackBoundary(metadata.clone()), + rx, + )); + } + SyncMarker::EndOfStream => { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::EndOfStream, rx)); + } + _ => {} // Ignorer les autres syncmarkers + }, + } + } +} + /// Détermine la profondeur de bit d'un chunk audio fn get_chunk_bit_depth(chunk: &AudioChunk) -> u8 { match chunk {