From 7e81a8e7777b702095a9aacff7f265a3eab16644 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Nov 2025 13:44:25 +0000 Subject: [PATCH] Add TimerNode for rate limiting and improve progressive cache handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Changes: - Add TimerNode (pmoaudio/src/nodes/timer_node.rs): Rate-limits audio chunk flow based on timestamps with configurable max_lead_time - Integrate TimerNode into play_and_cache.rs pipeline: PlaylistSource → TimerNode (3s pacing) → AudioSink - Improve EOF retry in playlist_source.rs: Wait for prebuffer (512KB) before decoding, retry on temporary EOF with 200ms delay - Export TimerNode in pmoaudio lib.rs and nodes/mod.rs Known issue: Cache files may still be truncated when TrackBoundary arrives before pump completes flushing. This requires allowing parallel write tasks as suggested. --- pmoaudio-ext/src/sources/playlist_source.rs | 36 ++- pmoaudio/src/lib.rs | 1 + pmoaudio/src/nodes/mod.rs | 2 +- pmoaudio/src/nodes/timer_node.rs | 263 ++++++++++++++++++++ pmoparadise/examples/play_and_cache.rs | 24 +- 5 files changed, 312 insertions(+), 14 deletions(-) create mode 100644 pmoaudio/src/nodes/timer_node.rs diff --git a/pmoaudio-ext/src/sources/playlist_source.rs b/pmoaudio-ext/src/sources/playlist_source.rs index f2896e11..5bca30a9 100644 --- a/pmoaudio-ext/src/sources/playlist_source.rs +++ b/pmoaudio-ext/src/sources/playlist_source.rs @@ -279,6 +279,28 @@ async fn decode_and_emit_track( cache: &Arc, cache_pk: &str, ) -> Result<(), AudioError> { + // Attendre que le fichier soit suffisamment gros pour le sniffing + // Le cache progressif permet de commencer la lecture après le prebuffer (512 KB) + loop { + let metadata = tokio::fs::metadata(path) + .await + .map_err(|e| AudioError::IoError(format!("Failed to stat {:?}: {}", path, e)))?; + + let file_size = metadata.len(); + const MIN_FILE_SIZE: u64 = 512 * 1024; // 512 KB (prebuffer size) + + if file_size >= MIN_FILE_SIZE || cache.is_download_complete(cache_pk) { + tracing::trace!("decode_and_emit_track: file ready ({} bytes), starting decode", file_size); + break; + } + + tracing::trace!( + "decode_and_emit_track: file too small ({} bytes), waiting 50ms...", + file_size + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + // Ouvrir et décoder let file = File::open(path) .await @@ -333,16 +355,16 @@ async fn decode_and_emit_track( // Si EOF atteint (read == 0) if read == 0 { // Vérifier si le fichier est complètement écrit (completion marker existe) - // Si pas de marker, le fichier est encore en cours d'écriture (cache progressif) if !cache.is_download_complete(cache_pk) { - // Fichier encore en cours d'écriture - attendre un peu et réessayer - tracing::trace!("decode_and_emit_track: EOF reached but file not complete (no marker), waiting 50ms..."); - tokio::time::sleep(Duration::from_millis(50)).await; - continue; // Retry la lecture + // Fichier encore en cours d'écriture - attendre et réessayer + // Retry plus longtemps pour le cache progressif + tracing::trace!("decode_and_emit_track: EOF but file incomplete, waiting 200ms..."); + tokio::time::sleep(Duration::from_millis(200)).await; + continue; // Retry } - // Completion marker existe - c'est vraiment la fin du fichier - tracing::trace!("decode_and_emit_track: EOF reached and file is complete (marker exists)"); + // Completion marker existe - vraie fin du fichier + tracing::trace!("decode_and_emit_track: EOF and file complete"); if pending.is_empty() { break; } diff --git a/pmoaudio/src/lib.rs b/pmoaudio/src/lib.rs index 6c0e3edd..dd3c9a6c 100755 --- a/pmoaudio/src/lib.rs +++ b/pmoaudio/src/lib.rs @@ -124,6 +124,7 @@ pub use nodes::{ flac_file_sink::{FlacFileSink, FlacFileSinkStats}, http_source::HttpSource, resampling_node::ResamplingNode, + timer_node::TimerNode, AudioError, AudioNode, TypedAudioNode, }; diff --git a/pmoaudio/src/nodes/mod.rs b/pmoaudio/src/nodes/mod.rs index 9f51198a..877073bf 100755 --- a/pmoaudio/src/nodes/mod.rs +++ b/pmoaudio/src/nodes/mod.rs @@ -25,6 +25,7 @@ pub mod file_source; pub mod flac_file_sink; pub mod http_source; pub mod resampling_node; +pub mod timer_node; // Modules temporairement désactivés /* @@ -36,7 +37,6 @@ pub mod dsp_node; pub mod mpd_sink; pub mod sink_node; pub mod source_node; -pub mod timer_node; pub mod volume_node; */ diff --git a/pmoaudio/src/nodes/timer_node.rs b/pmoaudio/src/nodes/timer_node.rs new file mode 100644 index 00000000..794e3b59 --- /dev/null +++ b/pmoaudio/src/nodes/timer_node.rs @@ -0,0 +1,263 @@ +//! TimerNode - Régule le débit des chunks audio en fonction de leurs timestamps +//! +//! Ce node implémente un pacing temporel pour éviter que les sources rapides +//! saturent les sinks lents. Il tolère une avance configurable (buffer) et +//! attend activement pour maintenir la synchronisation temps réel. +//! +//! # Use Cases +//! +//! - **Progressive caching**: Empêche PlaylistSource de lire plus vite que FlacCacheSink n'écrit +//! - **Rate limiting**: Contrôle le débit de n'importe quel pipeline audio +//! - **Streaming**: Synchronise la production avec la consommation temps réel +//! +//! # Exemple +//! +//! ```no_run +//! use pmoaudio::{PlaylistSource, TimerNode, FlacCacheSink}; +//! +//! let mut source = PlaylistSource::new(reader, cache); +//! let mut timer = TimerNode::new(3.0); // 3s d'avance max +//! let mut sink = FlacCacheSink::new(cache, covers); +//! +//! source.register(Box::new(timer)); +//! timer.register(Box::new(sink)); +//! ``` +//! +//! # Architecture +//! +//! ```text +//! PlaylistSource → TimerNode → FlacCacheSink +//! ↓ ↓ ↓ +//! Lit à fond Régule en Écrit au +//! temps réel bon rythme +//! ``` +//! +//! Le TimerNode: +//! 1. Reçoit des chunks avec timestamps +//! 2. Compare `chunk.timestamp_sec` avec le temps écoulé depuis `TopZeroSync` +//! 3. Si l'avance > `max_lead_time_sec`, attend: `sleep(avance - max_lead_time)` +//! 4. Transmet le chunk aux enfants +//! +//! # Markers Supportés +//! +//! - **TopZeroSync**: Reset le timer de référence (instant zero) +//! - **TrackBoundary**: Passthrough transparent +//! - **Heartbeat**: Passthrough transparent +//! - **EndOfStream**: Passthrough transparent +//! +//! # Performance +//! +//! - **CPU**: Quasi-nul (tokio::time::sleep efficace) +//! - **Latency**: Ajoute `max_lead_time_sec` de buffering +//! - **Memory**: Minimal (pas de buffer de chunks) + +use crate::{ + nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE}, + pipeline::{AudioPipelineNode, Node, NodeLogic}, + type_constraints::TypeRequirement, + AudioSegment, SyncMarker, _AudioSegment, +}; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio::time::{Duration, Instant}; +use tokio_util::sync::CancellationToken; + +// ═══════════════════════════════════════════════════════════════════════════ +// TimerNodeLogic - Logique pure de pacing temporel +// ═══════════════════════════════════════════════════════════════════════════ + +/// Logique pure de régulation temporelle +/// +/// Contrôle le débit des chunks audio pour éviter qu'une source rapide +/// sature un sink lent (ex: progressive caching). +pub struct TimerNodeLogic { + /// Avance maximale tolérée en secondes (buffer) + max_lead_time_sec: f64, + /// Instant de référence (reset au TopZeroSync) + start_time: Option, +} + +impl TimerNodeLogic { + pub fn new(max_lead_time_sec: f64) -> Self { + Self { + max_lead_time_sec: max_lead_time_sec.max(0.0), + start_time: None, + } + } +} + +#[async_trait::async_trait] +impl NodeLogic for TimerNodeLogic { + async fn process( + &mut self, + input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut rx = input.expect("TimerNode must have input"); + tracing::debug!( + "TimerNodeLogic::process started (max_lead_time={:.1}s), {} children", + self.max_lead_time_sec, + output.len() + ); + + // Macro helper pour envoyer à tous les enfants + macro_rules! send_to_children { + ($segment:expr) => { + for tx in &output { + tx.send($segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + }; + } + + loop { + let segment = tokio::select! { + _ = stop_token.cancelled() => { + tracing::debug!("TimerNodeLogic cancelled"); + break; + } + + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + tracing::debug!("TimerNodeLogic received EOF"); + break; + } + } + } + }; + + // Traitement selon le type de segment + match &segment.segment { + _AudioSegment::Sync(marker) => { + match &**marker { + SyncMarker::TopZeroSync => { + // Reset le timer de référence + self.start_time = Some(Instant::now()); + tracing::debug!("TimerNodeLogic: TopZeroSync received, timer reset"); + } + _ => { + // Autres markers: passthrough transparent + } + } + send_to_children!(segment); + } + + _AudioSegment::Chunk(_) => { + // Vérifier le pacing seulement si on a un timer de référence + if let Some(start) = self.start_time { + let chunk_timestamp = segment.timestamp_sec; + let elapsed = start.elapsed().as_secs_f64(); + let lead_time = chunk_timestamp - elapsed; + + if lead_time > self.max_lead_time_sec { + // On est trop en avance, attendre + let sleep_duration = lead_time - self.max_lead_time_sec; + tracing::trace!( + "TimerNodeLogic: lead_time={:.3}s > max={:.1}s, sleeping {:.3}s", + lead_time, + self.max_lead_time_sec, + sleep_duration + ); + + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs_f64(sleep_duration)) => {} + _ = stop_token.cancelled() => { + tracing::debug!("TimerNodeLogic cancelled during sleep"); + break; + } + } + } else if lead_time < -0.5 { + // On est en retard de plus de 500ms, log warning + tracing::warn!( + "TimerNodeLogic: lagging behind by {:.3}s (chunk ts={:.3}s, elapsed={:.3}s)", + -lead_time, + chunk_timestamp, + elapsed + ); + } + } else { + // Pas encore de TopZeroSync reçu, passthrough sans pacing + tracing::trace!("TimerNodeLogic: no timer set yet, passthrough"); + } + + send_to_children!(segment); + } + } + } + + tracing::debug!("TimerNodeLogic::process finished"); + Ok(()) + } +} + +// ═══════════════════════════════════════════════════════════════════════════ +// TimerNode - Wrapper utilisant Node +// ═══════════════════════════════════════════════════════════════════════════ + +pub struct TimerNode { + inner: Node, +} + +impl TimerNode { + /// Crée un TimerNode avec une avance maximale tolérée + /// + /// # Arguments + /// + /// * `max_lead_time_sec` - Avance maximale en secondes (ex: 3.0 pour 3s de buffer) + /// + /// # Exemples + /// + /// ```no_run + /// use pmoaudio::TimerNode; + /// + /// // Tolérer 3 secondes d'avance + /// let timer = TimerNode::new(3.0); + /// ``` + pub fn new(max_lead_time_sec: f64) -> Self { + Self::with_channel_size(max_lead_time_sec, DEFAULT_CHANNEL_SIZE) + } + + /// Crée un TimerNode avec une taille de buffer MPSC personnalisée + /// + /// # Arguments + /// + /// * `max_lead_time_sec` - Avance maximale en secondes + /// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente) + pub fn with_channel_size(max_lead_time_sec: f64, channel_size: usize) -> Self { + let logic = TimerNodeLogic::new(max_lead_time_sec); + Self { + inner: Node::new_with_input(logic, channel_size), + } + } +} + +#[async_trait::async_trait] +impl AudioPipelineNode for TimerNode { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, child: Box) { + self.inner.register(child); + } + + async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { + Box::new(self.inner).run(stop_token).await + } +} + +impl TypedAudioNode for TimerNode { + fn input_type(&self) -> Option { + // Accepte n'importe quel type + Some(TypeRequirement::any()) + } + + fn output_type(&self) -> Option { + // Passthrough: produit le même type qu'il consomme + Some(TypeRequirement::any()) + } +} diff --git a/pmoparadise/examples/play_and_cache.rs b/pmoparadise/examples/play_and_cache.rs index ae1fd10f..41a72173 100644 --- a/pmoparadise/examples/play_and_cache.rs +++ b/pmoparadise/examples/play_and_cache.rs @@ -4,7 +4,8 @@ //! 1. RadioParadiseStreamSource - Télécharge et décode un bloc FLAC //! 2. FlacCacheSink - Cache chaque piste en FLAC et alimente une playlist //! 3. PlaylistSource - Lit la playlist pendant le téléchargement -//! 4. AudioSink - Joue l'audio sur la sortie standard +//! 4. TimerNode - Régule le débit pour éviter EOF prématurés (progressive cache) +//! 5. AudioSink - Joue l'audio sur la sortie standard //! //! Architecture : //! ```text @@ -12,7 +13,10 @@ //! RadioParadiseStreamSource → FlacCacheSink (avec playlist abonnée) //! //! Pipeline 2 (Playback): -//! PlaylistSource (lit la playlist) → AudioSink (joue l'audio) +//! PlaylistSource → TimerNode (rate limiting) → AudioSink +//! ↓ +//! Prévention EOF +//! (3s max lead) //! ``` //! //! Usage: @@ -22,7 +26,7 @@ //! cargo run --example play_and_cache --features full -- 0 # Main Mix //! cargo run --example play_and_cache --features full -- 2 # Rock Mix -use pmoaudio::{AudioPipelineNode, AudioSink}; +use pmoaudio::{AudioPipelineNode, AudioSink, TimerNode}; use pmoaudio_ext::{FlacCacheSink, PlaylistSource}; use pmoaudiocache::Cache as AudioCache; use pmocovers::Cache as CoverCache; @@ -195,6 +199,11 @@ async fn main() -> Result<(), Box> { let mut playlist_source = PlaylistSource::new(reader, audio_cache.clone()); tracing::debug!("PlaylistSource created"); + // Créer le timer node pour réguler le débit (empêche EOF prématurés) + // Tolère 3 secondes d'avance max pour permettre le buffering + let mut timer = TimerNode::new(3.0); + tracing::debug!("TimerNode created (max_lead_time=3.0s)"); + // Créer le sink audio let audio_sink = if use_null_audio { AudioSink::with_null_output() @@ -203,9 +212,12 @@ async fn main() -> Result<(), Box> { }; tracing::debug!("AudioSink created"); - // Connecter playlist → audio - playlist_source.register(Box::new(audio_sink)); - tracing::info!("Playback pipeline connected: PlaylistSource → AudioSink"); + // Connecter timer → audio (AVANT de mettre timer dans une Box) + timer.register(Box::new(audio_sink)); + + // Connecter playlist → timer + playlist_source.register(Box::new(timer)); + tracing::info!("Playback pipeline connected: PlaylistSource → TimerNode → AudioSink"); // ═══════════════════════════════════════════════════════════════════════════ // Lancer les deux pipelines en parallèle