From 8eafff0f0ca2274d1d4c2921f01cd21e0233b5a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Nov 2025 11:56:58 +0000 Subject: [PATCH] Add node statistics tracking + reduce MPSC buffer to 8 chunks --- pmoparadise/examples/stream_block.rs | 10 +- pmoparadise/src/lib.rs | 3 + pmoparadise/src/node_stats.rs | 137 ++++++++++++++++++ .../src/radio_paradise_stream_source.rs | 20 +++ 4 files changed, 165 insertions(+), 5 deletions(-) create mode 100644 pmoparadise/src/node_stats.rs diff --git a/pmoparadise/examples/stream_block.rs b/pmoparadise/examples/stream_block.rs index 458614fb..5447ff80 100644 --- a/pmoparadise/examples/stream_block.rs +++ b/pmoparadise/examples/stream_block.rs @@ -212,12 +212,12 @@ async fn main() -> Result<(), Box> { source_flac.push_block_id(END_OF_BLOCKS_SIGNAL); // Signal: no more blocks after this one tracing::debug!("RadioParadiseStreamSource (FLAC) created with block {} + END signal", block.event); - // Calculate channel size to match max_lead_time - // With 50ms chunks and 3.0s lead time: 3.0 / 0.05 = 60 chunks + // Use SMALL channel size to make backpressure more reactive + // Instead of trying to buffer 3s of audio (60 chunks), use a much smaller buffer + // This forces tighter backpressure control let max_lead_time = 3.0; - let chunk_duration_sec = 0.05; - let channel_size = ((max_lead_time / chunk_duration_sec) as usize).max(16); - tracing::debug!("Calculated channel size: {} chunks ({:.1}s buffer)", channel_size, channel_size as f64 * chunk_duration_sec); + let channel_size = 8; // Small buffer for reactive backpressure + tracing::debug!("Using channel size: {} chunks ({:.1}s buffer at 50ms/chunk)", channel_size, channel_size as f64 * 0.05); let mut timer_flac = TimerNode::with_channel_size(max_lead_time, channel_size); tracing::debug!("TimerNode (FLAC) created with {:.1}s max lead time, {} chunk buffer", max_lead_time, channel_size); diff --git a/pmoparadise/src/lib.rs b/pmoparadise/src/lib.rs index 8a901131..376d97d9 100644 --- a/pmoparadise/src/lib.rs +++ b/pmoparadise/src/lib.rs @@ -216,6 +216,9 @@ pub mod error; pub mod models; pub mod source; +#[cfg(feature = "pmoaudio")] +pub mod node_stats; + #[cfg(feature = "pmoserver")] pub mod pmoserver_ext; diff --git a/pmoparadise/src/node_stats.rs b/pmoparadise/src/node_stats.rs new file mode 100644 index 00000000..2bc0b27b --- /dev/null +++ b/pmoparadise/src/node_stats.rs @@ -0,0 +1,137 @@ +//! Node statistics tracking +//! +//! Provides detailed statistics for pipeline nodes to understand +//! data flow, backpressure behavior, and timing. + +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Instant; + +/// Statistics pour un node audio +#[derive(Debug)] +pub struct NodeStats { + /// Nom du node pour identification + pub name: String, + + /// Instant de démarrage du node + pub start_time: Instant, + + /// Nombre total de segments reçus + pub segments_received: AtomicUsize, + + /// Nombre total de segments envoyés + pub segments_sent: AtomicUsize, + + /// Nombre total de bytes traités + pub bytes_processed: AtomicU64, + + /// Nombre de fois où l'envoi a été bloqué (backpressure) + pub backpressure_blocks: AtomicUsize, + + /// Temps total passé bloqué en millisecondes + pub backpressure_time_ms: AtomicU64, + + /// Timestamp du premier segment (secondes) + pub first_segment_timestamp: AtomicU64, // Stocké comme u64 * 1000 pour précision + + /// Timestamp du dernier segment (secondes) + pub last_segment_timestamp: AtomicU64, // Stocké comme u64 * 1000 pour précision +} + +impl NodeStats { + pub fn new(name: impl Into) -> Arc { + Arc::new(Self { + name: name.into(), + start_time: Instant::now(), + segments_received: AtomicUsize::new(0), + segments_sent: AtomicUsize::new(0), + bytes_processed: AtomicU64::new(0), + backpressure_blocks: AtomicUsize::new(0), + backpressure_time_ms: AtomicU64::new(0), + first_segment_timestamp: AtomicU64::new(u64::MAX), + last_segment_timestamp: AtomicU64::new(0), + }) + } + + /// Enregistre la réception d'un segment + pub fn record_segment_received(&self, timestamp_sec: f64) { + self.segments_received.fetch_add(1, Ordering::Relaxed); + + let ts_millis = (timestamp_sec * 1000.0) as u64; + + // Update first timestamp (atomic min) + let mut current = self.first_segment_timestamp.load(Ordering::Relaxed); + while current > ts_millis { + match self.first_segment_timestamp.compare_exchange_weak( + current, + ts_millis, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(x) => current = x, + } + } + + // Update last timestamp (atomic max) + let mut current = self.last_segment_timestamp.load(Ordering::Relaxed); + while current < ts_millis { + match self.last_segment_timestamp.compare_exchange_weak( + current, + ts_millis, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(x) => current = x, + } + } + } + + /// Enregistre l'envoi d'un segment + pub fn record_segment_sent(&self, bytes: usize) { + self.segments_sent.fetch_add(1, Ordering::Relaxed); + self.bytes_processed.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Enregistre un événement de backpressure + pub fn record_backpressure(&self, duration_ms: u64) { + self.backpressure_blocks.fetch_add(1, Ordering::Relaxed); + self.backpressure_time_ms.fetch_add(duration_ms, Ordering::Relaxed); + } + + /// Retourne un rapport formaté des statistiques + pub fn report(&self) -> String { + let elapsed = self.start_time.elapsed().as_secs_f64(); + let received = self.segments_received.load(Ordering::Relaxed); + let sent = self.segments_sent.load(Ordering::Relaxed); + let bytes = self.bytes_processed.load(Ordering::Relaxed); + let bp_blocks = self.backpressure_blocks.load(Ordering::Relaxed); + let bp_time_ms = self.backpressure_time_ms.load(Ordering::Relaxed); + + let first_ts = self.first_segment_timestamp.load(Ordering::Relaxed); + let last_ts = self.last_segment_timestamp.load(Ordering::Relaxed); + + let first_ts_sec = if first_ts == u64::MAX { 0.0 } else { first_ts as f64 / 1000.0 }; + let last_ts_sec = last_ts as f64 / 1000.0; + let audio_duration = last_ts_sec - first_ts_sec; + + let mb = bytes as f64 / 1_048_576.0; + let throughput_mbps = if elapsed > 0.0 { mb / elapsed } else { 0.0 }; + + format!( + "[{}]\n\ + Elapsed: {:.1}s | Received: {} | Sent: {} | Lost: {}\n\ + Data: {:.1} MB | Throughput: {:.2} MB/s\n\ + Audio: {:.1}s (first: {:.1}s, last: {:.1}s) | Real-time ratio: {:.1}%\n\ + Backpressure: {} blocks, {:.2}s total ({:.1}% of time)", + self.name, + elapsed, received, sent, received.saturating_sub(sent), + mb, throughput_mbps, + audio_duration, first_ts_sec, last_ts_sec, + if audio_duration > 0.0 { (elapsed / audio_duration) * 100.0 } else { 0.0 }, + bp_blocks, bp_time_ms as f64 / 1000.0, + if elapsed > 0.0 { (bp_time_ms as f64 / 1000.0 / elapsed) * 100.0 } else { 0.0 } + ) + } +} diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index 17bb3795..0f72bd7f 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -6,6 +6,7 @@ use crate::{ client::RadioParadiseClient, models::{Block, EventId, Song}, + node_stats::NodeStats, }; use futures_util::StreamExt; use pmoaudio::{ @@ -43,6 +44,7 @@ pub struct RadioParadiseStreamSourceLogic { chunk_frames: usize, recent_blocks: VecDeque, block_queue: VecDeque, + stats: Arc, } impl RadioParadiseStreamSourceLogic { @@ -55,6 +57,7 @@ impl RadioParadiseStreamSourceLogic { chunk_frames, recent_blocks: VecDeque::with_capacity(RECENT_BLOCKS_CACHE_SIZE), block_queue: VecDeque::new(), + stats: NodeStats::new("RadioParadiseStreamSource"), } } @@ -303,6 +306,8 @@ impl RadioParadiseStreamSourceLogic { output: &[mpsc::Sender>], segment: Arc, ) -> Result<(), AudioError> { + self.stats.record_segment_received(segment.timestamp_sec); + for (i, tx) in output.iter().enumerate() { let capacity_before = tx.capacity(); tracing::trace!( @@ -317,11 +322,23 @@ impl RadioParadiseStreamSourceLogic { let send_duration = send_start.elapsed(); if send_duration.as_millis() > 10 { + let duration_ms = send_duration.as_millis() as u64; + self.stats.record_backpressure(duration_ms); tracing::debug!( "send_to_children: Send to child {} BLOCKED for {:.3}s (backpressure triggered, timestamp={:.3}s)", i, send_duration.as_secs_f64(), segment.timestamp_sec ); } + + // Estimer la taille du segment pour les stats (frames * 2 channels * bytes_per_sample) + let segment_bytes = match &segment.segment { + pmoaudio::_AudioSegment::Chunk(chunk) => { + // Approximation: frames * 2 (stereo) * 4 bytes (i32/f32) + chunk.len() * 2 * 4 + } + _ => 0, + }; + self.stats.record_segment_sent(segment_bytes); } Ok(()) } @@ -602,6 +619,9 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { ); } + // Log des statistiques finales + tracing::info!("\n{}", self.stats.report()); + Ok(()) } }