Add node statistics tracking + reduce MPSC buffer to 8 chunks

pmoparadise/examples/stream_block.rs

@@ -212,12 +212,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     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);

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;
 

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<String>) -> Arc<Self> {
+        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 }
+        )
+    }
+}

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<EventId>,
     block_queue: VecDeque<EventId>,
+    stats: Arc<NodeStats>,
 }
 
 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<Arc<AudioSegment>>],
         segment: Arc<AudioSegment>,
     ) -> 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(())
     }
 }