From 215b097f4b0018ab543cb6123edd8d10cf6da49c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Nov 2025 10:27:35 +0000 Subject: [PATCH] Add detailed tracing for backpressure investigation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Investigation revealed the root cause of premature streaming termination: 1. MPSC Channel Size Issue: - DEFAULT_CHANNEL_SIZE = 16 chunks × 50ms = 800ms capacity - TimerNode max_lead_time = 3.0 seconds - The channel fills up in 0.8s while TimerNode wants 3s buffer - This creates stop-and-go pattern instead of smooth backpressure 2. Channel Closure Issue: - When RadioParadiseStreamSource::process() returns, the Node automatically closes output channels - TimerNode receives EOF and terminates immediately - Remaining chunks in MPSC buffer (up to 16) are never sent to sink Added comprehensive tracing: - RadioParadiseStreamSource: Track backpressure blocking, chunk counts, decode timing - TimerNode: Log all pacing decisions, sleep durations, lead time - Both use trace! for high-frequency events, debug! for blocking Next steps: - Option A: Increase channel size to match max_lead_time (60 chunks for 3s @ 50ms) - Option B: Wait for channels to drain before closing (use tx.closed().await) - Option C: Both A and B for optimal behavior The previous "wait for playback duration" fix is a valid workaround but doesn't address the architectural issue. --- pmoaudio/src/nodes/timer_node.rs | 21 ++++++++++++--- .../src/radio_paradise_stream_source.rs | 27 ++++++++++++++++++- 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/pmoaudio/src/nodes/timer_node.rs b/pmoaudio/src/nodes/timer_node.rs index b960e040..6e872907 100644 --- a/pmoaudio/src/nodes/timer_node.rs +++ b/pmoaudio/src/nodes/timer_node.rs @@ -153,18 +153,26 @@ impl NodeLogic for TimerNodeLogic { let elapsed = start.elapsed().as_secs_f64(); let lead_time = chunk_timestamp - elapsed; + tracing::trace!( + "TimerNodeLogic: chunk received (ts={:.3}s, elapsed={:.3}s, lead_time={:.3}s, max_lead={:.1}s)", + chunk_timestamp, elapsed, lead_time, self.max_lead_time_sec + ); + 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: SLEEPING {:.3}s (lead_time={:.3}s > max={:.1}s)", + tracing::debug!( + "TimerNodeLogic: SLEEPING {:.3}s (lead_time={:.3}s > max={:.1}s, chunk_ts={:.3}s)", sleep_duration, lead_time, - self.max_lead_time_sec + self.max_lead_time_sec, + chunk_timestamp ); tokio::select! { - _ = tokio::time::sleep(Duration::from_secs_f64(sleep_duration)) => {} + _ = tokio::time::sleep(Duration::from_secs_f64(sleep_duration)) => { + tracing::trace!("TimerNodeLogic: woke up from sleep"); + } _ = stop_token.cancelled() => { tracing::debug!("TimerNodeLogic cancelled during sleep"); break; @@ -178,6 +186,11 @@ impl NodeLogic for TimerNodeLogic { chunk_timestamp, elapsed ); + } else { + tracing::trace!( + "TimerNodeLogic: chunk on time (lead_time={:.3}s within tolerance)", + lead_time + ); } } else { // Pas encore de TopZeroSync reçu, passthrough sans pacing diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index c1db0154..f569b5d0 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -175,11 +175,16 @@ impl RadioParadiseStreamSourceLogic { let mut pending: Vec = Vec::with_capacity(chunk_byte_len * 2); // Traiter les chunks audio + let mut chunk_count = 0; loop { // Vérifier stop_token if stop_token.is_cancelled() { // Retourner le timestamp actuel et start_instant si on est interrompu let current_timestamp = total_samples as f64 / sample_rate as f64; + tracing::debug!( + "Block decode cancelled: sent {} chunks, {:.2}s duration", + chunk_count, current_timestamp + ); return Ok((current_timestamp, start_instant)); } @@ -189,6 +194,10 @@ impl RadioParadiseStreamSourceLogic { .map_err(|e| AudioError::ProcessingError(format!("Read error: {}", e)))?; if read == 0 { + tracing::debug!( + "FLAC decode EOF reached: sent {} chunks, {:.2}s total", + chunk_count, total_samples as f64 / sample_rate as f64 + ); break; // EOF } pending.extend_from_slice(&read_buf[..read]); @@ -247,6 +256,7 @@ impl RadioParadiseStreamSourceLogic { *order += 1; total_samples += chunk_len; + chunk_count += 1; } // Retourner le timestamp du dernier chunk (durée totale du bloc) et l'instant de début @@ -262,10 +272,25 @@ impl RadioParadiseStreamSourceLogic { output: &[mpsc::Sender>], segment: Arc, ) -> Result<(), AudioError> { - for tx in output { + for (i, tx) in output.iter().enumerate() { + let capacity_before = tx.capacity(); + tracing::trace!( + "send_to_children: Sending to child {} (channel capacity={}, timestamp={:.3}s)", + i, capacity_before, segment.timestamp_sec + ); + + let send_start = std::time::Instant::now(); tx.send(segment.clone()) .await .map_err(|_| AudioError::ChildDied)?; + let send_duration = send_start.elapsed(); + + if send_duration.as_millis() > 10 { + 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 + ); + } } Ok(()) }