eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Add TimerNode for rate limiting and improve progressive cache handling

Browse Source 
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.
This commit is contained in:
Claude
2025-11-07 13:44:25 +00:00
parent 58e6753a81
commit 7e81a8e777
5 changed files with 312 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
pmoaudio-ext/src/sources/playlist_source.rs
View File

@@ -279,6 +279,28 @@ async fn decode_and_emit_track(
cache: &Arc<AudioCache>,
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;
}

1
pmoaudio/src/lib.rs
View File

@@ -124,6 +124,7 @@ pub use nodes::{
flac_file_sink::{FlacFileSink, FlacFileSinkStats},
http_source::HttpSource,
resampling_node::ResamplingNode,
timer_node::TimerNode,
AudioError, AudioNode, TypedAudioNode,
};

2
pmoaudio/src/nodes/mod.rs
View File

@@ -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;
*/

263
pmoaudio/src/nodes/timer_node.rs Normal file
View File

@@ -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<Instant>,
}
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<mpsc::Receiver<Arc<AudioSegment>>>,
output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
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<TimerNodeLogic>
// ═══════════════════════════════════════════════════════════════════════════
pub struct TimerNode {
inner: Node<TimerNodeLogic>,
}
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<mpsc::Sender<Arc<AudioSegment>>> {
self.inner.get_tx()
}
fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
self.inner.register(child);
}
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.inner).run(stop_token).await
}
}
impl TypedAudioNode for TimerNode {
fn input_type(&self) -> Option<TypeRequirement> {
// Accepte n'importe quel type
Some(TypeRequirement::any())
}
fn output_type(&self) -> Option<TypeRequirement> {
// Passthrough: produit le même type qu'il consomme
Some(TypeRequirement::any())
}
}

24
pmoparadise/examples/play_and_cache.rs
View File

@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
};
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