Add optimization guide for prebuffer→playlist delay reduction

Document détaillé pour réduire le délai de 19s à 1s en pushant à la
playlist immédiatement après le prebuffer, sans attendre pump_future.

Contient:
- Analyse du problème actuel (tokio::join! bloquant)
- 3 solutions possibles avec avantages/inconvénients
- Plan d'implémentation détaillé avec code complet
- Guide de test et validation
- Debugging tips et tests de régression

Ce document permet de reprendre l'optimisation dans une nouvelle session
avec tout le contexte nécessaire.
This commit is contained in:
Claude
2025-11-07 09:39:53 +00:00
parent ed0bbfbf69
commit a8f9e19f4a

View File

@@ -0,0 +1,485 @@
# Optimisation: Réduction du délai prebuffer → playlist (19s → 1s)
## Contexte
Le système de progressive caching fonctionne correctement, mais il y a un délai non optimal entre le moment où le prebuffer est atteint et le moment où la track est ajoutée à la playlist.
### État actuel (branche `claude/fix-play-and-cache-streaming-011CUsMBxH4fsgoadgkiPdoK`)
**Timing mesuré:**
```
t=0.6s : Prebuffer complete (512KB téléchargés) ✅
t=19.2s : tokio::join!() complete (pump_future finit)
t=19.2s : Track added to playlist
t=19.7s : Playback starts
```
**Délai total: ~19 secondes**
### Code actuel problématique
Location: `pmoaudio-ext/src/sinks/flac_cache_sink.rs:167-178`
```rust
// Exécuter pump et add_from_reader en parallèle
let pump_future = pump_track_segments(
first_segment,
&mut rx, // ← emprunte muablement rx
pcm_tx,
bits_per_sample,
sample_rate,
&stop_token,
);
// Attendre les deux tâches en parallèle
let (cache_result, pump_result) = tokio::join!(cache_future, pump_future);
let pk = cache_result.map_err(|e| {
AudioError::ProcessingError(format!("Failed to add to cache: {}", e))
})?;
let (_chunks, _samples, _duration_sec, stop_reason) = pump_result?;
```
**Le problème:** `tokio::join!()` attend que **LES DEUX** futures se terminent:
- `cache_future` retourne après prebuffer (~0.6s) ✅
- `pump_future` lit **toute** la première track du RadioParadiseStreamSource (~19s) ⏱️
Donc même si le prebuffer est atteint en 0.6s, on attend 19s avant de push à la playlist!
## Objectif
Réduire le délai à **~1 seconde** en pushant à la playlist **immédiatement après le prebuffer**, sans attendre que `pump_future` se termine.
**Timing visé:**
```
t=0.6s : Prebuffer complete ✅
t=0.7s : Track added to playlist ← IMMÉDIAT!
t=1.2s : Playback starts ← ~1 seconde!
t=19.2s : pump_future finit en arrière-plan
```
## Contraintes techniques
### 1. Problème du borrow checker
`pump_future` emprunte muablement `rx`:
```rust
async fn pump_track_segments(
first_segment: Arc<AudioSegment>,
rx: &mut mpsc::Receiver<Arc<AudioSegment>>, // ← &mut borrow
// ...
)
```
On ne peut pas faire:
```rust
tokio::pin!(cache_future);
tokio::pin!(pump_future); // ← pump_future contient un &mut rx
let pk = cache_future.await; // cache_future termine
// ❌ ERREUR: on a toujours un borrow mutable de rx dans pump_future
// On ne peut pas continuer à utiliser rx (ou l'objet qui le contient)
playlist_handle.push(pk.clone()).await;
let result = pump_future.await; // pump_future continue
```
Le borrow checker nous empêche d'attendre `cache_future` seul, puis de faire d'autres opérations, puis d'attendre `pump_future`, car `pump_future` garde un borrow mutable de `rx` pendant toute sa durée de vie.
### 2. Contraintes de l'API
- `pump_track_segments()` doit lire `rx` pour recevoir les segments du RadioParadiseStreamSource
- Le FlacCacheSinkLogic doit garder ownership de `rx` pour traiter les tracks suivantes
- `pump_future` ne peut pas être spawné dans un tokio::spawn car il retourne un `StopReason` nécessaire pour la logique métier
## Solutions possibles
### Solution A: Refactoriser pump_track_segments pour prendre ownership de rx
**Approche:**
1. Créer `pump_track_segments_owned` qui prend ownership de `rx`
2. Cette fonction retourne `(result, rx)` - elle rend ownership de `rx`
3. Spawner cette future dans tokio::spawn
4. Attendre cache_future seul, push immédiatement
5. Attendre la task spawnée plus tard
**Signature:**
```rust
async fn pump_track_segments_owned(
first_segment: Arc<AudioSegment>,
rx: mpsc::Receiver<Arc<AudioSegment>>, // ownership!
pcm_tx: mpsc::Sender<Vec<u8>>,
bits_per_sample: u8,
expected_rate: u32,
stop_token: CancellationToken,
) -> Result<(u64, u64, f64, StopReason, mpsc::Receiver<Arc<AudioSegment>>), AudioError>
// ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ rend rx
```
**Utilisation:**
```rust
let pump_handle = tokio::spawn(pump_track_segments_owned(
first_segment,
rx, // move ownership
pcm_tx,
bits_per_sample,
sample_rate,
stop_token.clone(),
));
// Attendre SEULEMENT le prebuffer
let pk = cache_future.await?;
// Push IMMÉDIATEMENT à la playlist
#[cfg(feature = "playlist")]
if let Some(ref playlist_handle) = self.playlist_handle {
playlist_handle.push(pk.clone()).await?;
}
// MAINTENANT attendre que pump finisse
let (result, rx_returned) = pump_handle.await.unwrap()?;
rx = rx_returned; // récupérer rx pour la prochaine track
```
**Avantages:**
- ✅ Pas de problème de borrow checker
- ✅ Push immédiat après prebuffer
- ✅ Délai réduit à ~1s
**Inconvénients:**
- ⚠️ Nécessite de modifier la signature de `pump_track_segments`
- ⚠️ Plus complexe (ownership passé puis rendu)
### Solution B: Utiliser un channel pour signaler le prebuffer
**Approche:**
1. Créer un oneshot channel `(prebuffer_tx, prebuffer_rx)`
2. `cache_future` envoie le pk via `prebuffer_tx` dès le prebuffer atteint
3. Le code principal attend `prebuffer_rx`, push immédiatement
4. Puis attend `tokio::join!()` normalement
**Code:**
```rust
let (prebuffer_tx, prebuffer_rx) = tokio::sync::oneshot::channel();
let cache_future = async {
let pk = self.cache.add_from_reader(...).await?;
let _ = prebuffer_tx.send(pk.clone()); // Signal prebuffer!
Ok(pk)
};
let pump_future = pump_track_segments(...);
// Spawner les deux en parallèle
let cache_handle = tokio::spawn(cache_future);
let pump_handle = tokio::spawn(pump_future);
// Attendre SEULEMENT le signal de prebuffer
let pk = prebuffer_rx.await.unwrap();
// Push IMMÉDIATEMENT à la playlist
playlist_handle.push(pk.clone()).await?;
// Puis attendre que tout finisse
let (cache_result, pump_result) = tokio::join!(cache_handle, pump_handle);
```
**Avantages:**
- ✅ Pas besoin de changer les signatures
- ✅ Push immédiat après prebuffer
**Inconvénients:**
- ⚠️ Nécessite de wrapper cache_future pour envoyer le signal
- ⚠️ Ajoute un oneshot channel
### Solution C: Modifier l'API du cache pour avoir un callback
**Approche:**
1. Ajouter un paramètre callback à `add_from_reader()`
2. Le cache appelle ce callback dès le prebuffer atteint
3. Le callback push à la playlist
**Signature:**
```rust
pub async fn add_from_reader_with_callback<R, F>(
&self,
source_uri: Option<&str>,
reader: R,
length: Option<u64>,
collection: Option<&str>,
on_prebuffer: F, // ← nouveau callback
) -> Result<String>
where
R: AsyncRead + Send + Unpin + 'static,
F: FnOnce(String) + Send + 'static, // F reçoit le pk
```
**Avantages:**
- ✅ API propre et réutilisable
- ✅ Pas de problème de borrow checker
**Inconvénients:**
- ⚠️ Nécessite de modifier l'API du cache (impact sur autres parties du code)
- ⚠️ Ajoute de la complexité à l'API
## Recommandation
**Je recommande la Solution A** (refactoriser `pump_track_segments_owned`):
- Plus explicite et claire
- Pas d'impact sur l'API du cache
- Ownership bien défini (passage puis retour de rx)
- Testable indépendamment
## Plan d'implémentation
### Étape 1: Créer pump_track_segments_owned
Location: `pmoaudio-ext/src/sinks/flac_cache_sink.rs`
```rust
/// 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<AudioSegment>,
mut rx: mpsc::Receiver<Arc<AudioSegment>>,
pcm_tx: mpsc::Sender<Vec<u8>>,
bits_per_sample: u8,
expected_rate: u32,
stop_token: CancellationToken,
) -> Result<(u64, u64, f64, StopReason, mpsc::Receiver<Arc<AudioSegment>>), 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;
}
}
// Loop pour le reste des segments...
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) => {
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;
}
}
_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));
}
_ => continue,
},
}
}
}
```
### Étape 2: Modifier FlacCacheSinkLogic::process
Location: `pmoaudio-ext/src/sinks/flac_cache_sink.rs:~167`
```rust
let collection_ref = self.collection.as_deref();
let cache_future = self.cache.add_from_reader(
None,
flac_stream,
None,
collection_ref,
);
// Spawner pump_future avec ownership de rx
let pump_handle = tokio::spawn(pump_track_segments_owned(
first_segment,
rx, // move ownership!
pcm_tx,
bits_per_sample,
sample_rate,
stop_token.clone(),
));
// Attendre SEULEMENT le prebuffer (cache retourne après 512KB)
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: Prebuffer complete with pk {}, pushing to playlist NOW", pk);
// Copier les métadonnées AVANT push
if let Some(src_metadata) = track_metadata.clone() {
let dest_metadata = self.cache.track_metadata(&pk);
pmometadata::copy_metadata_into(&src_metadata, &dest_metadata)
.await
.map_err(|e| {
AudioError::ProcessingError(format!("Failed to copy metadata to cache: {}", e))
})?;
}
// Push IMMÉDIATEMENT à la playlist (après prebuffer, avant pump complet!)
#[cfg(feature = "playlist")]
if let Some(ref playlist_handle) = self.playlist_handle {
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::debug!("FlacCacheSink: Successfully pushed to playlist");
}
// 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");
// Continuer avec download des covers en arrière-plan...
```
### Étape 3: Tester
```bash
# Nettoyer et rebuild
rm -rf /tmp/pmomusic_test
source setup-env.sh
cargo build --example play_and_cache --features full
# Tester avec logs de timing
RUST_LOG=debug target/debug/examples/play_and_cache 0 --null-audio 2>&1 | \
grep -E "Prebuffer complete|Pushing pk.*to playlist|popped track" | \
head -20
```
**Résultats attendus:**
```
[TIME_A] FlacCacheSink: Prebuffer complete with pk XXX, pushing to playlist NOW
[TIME_B] FlacCacheSink: Successfully pushed to playlist
[TIME_C] PlaylistSourceLogic: popped track from playlist
Délai (TIME_C - TIME_A) devrait être < 1 seconde!
```
### Étape 4: Valider le comportement
Vérifier que:
1. ✅ Le prebuffer est atteint rapidement (~0.6s)
2. ✅ Le push à la playlist est immédiat (~0.1s après prebuffer)
3. ✅ La lecture démarre rapidement (~1s total)
4. ✅ Toutes les tracks se suivent correctement
5. ✅ Les completion markers sont créés
6. ✅ Les tracks suivantes fonctionnent (rx est bien récupéré)
7. ✅ Pas de panic ou deadlock
## Debugging
### Si le borrow checker proteste
Vérifier que:
- `pump_track_segments_owned` prend bien ownership de `rx` (pas `&mut`)
- `rx` est bien retourné dans le tuple de retour
- `rx = rx_returned;` récupère bien ownership après await
### Si les tracks suivantes ne fonctionnent pas
Vérifier que:
- `rx` est bien réassigné après le pump: `rx = rx_returned;`
- La loop dans `process()` continue correctement avec le nouveau `rx`
### Si le timing n'est pas amélioré
Ajouter des logs avec timestamps:
```rust
let start = std::time::Instant::now();
let pk = cache_future.await?;
tracing::info!("Prebuffer took {:?}", start.elapsed());
let start2 = std::time::Instant::now();
playlist_handle.push(pk.clone()).await?;
tracing::info!("Playlist push took {:?}", start2.elapsed());
```
## Fichiers à modifier
1. **pmoaudio-ext/src/sinks/flac_cache_sink.rs**
- Ajouter `pump_track_segments_owned()` (~ligne 432)
- Modifier `FlacCacheSinkLogic::process()` (~ligne 167)
## Tests de régression
Après l'implémentation, tester:
```bash
# Test 1: Premier download (cache vide)
rm -rf /tmp/pmomusic_test
target/debug/examples/play_and_cache 0 --null-audio
# Test 2: Deuxième download (fichier déjà en cache)
# Ne pas supprimer /tmp/pmomusic_test
target/debug/examples/play_and_cache 0 --null-audio
# Test 3: Download interrompu (Ctrl+C)
target/debug/examples/play_and_cache 0 --null-audio
# Appuyer Ctrl+C après 2 secondes
# Test 4: Plusieurs tracks consécutives
# Laisser tourner 1 minute pour voir plusieurs tracks
timeout 60 target/debug/examples/play_and_cache 0 --null-audio
```
## Métriques de succès
- ✅ Délai prebuffer → playlist: **< 1 seconde** (actuellement ~19s)
- ✅ Délai prebuffer → lecture: **< 2 secondes** (actuellement ~19.5s)
- ✅ Pas de régression fonctionnelle
- ✅ Toutes les tracks se suivent correctement
- ✅ Les completion markers sont créés
## Références
- Branche actuelle: `claude/fix-play-and-cache-streaming-011CUsMBxH4fsgoadgkiPdoK`
- Code de référence: commit `ed0bbfb` (Add FlacCacheSink debug logs - system now works!)
- Issue originale: "play_and_cache n'a pas le comportement souhaité"