Add detailed tracing for backpressure investigation

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.
This commit is contained in:
Claude
2025-11-12 10:27:35 +00:00
parent b3f22d1b61
commit 215b097f4b
2 changed files with 43 additions and 5 deletions

View File

@@ -175,11 +175,16 @@ impl RadioParadiseStreamSourceLogic {
let mut pending: Vec<u8> = 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<Arc<AudioSegment>>],
segment: Arc<AudioSegment>,
) -> 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(())
}