diff --git a/Cargo.lock b/Cargo.lock index 319c9689..0742b58d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3035,6 +3035,7 @@ dependencies = [ "bytemuck", "cpal", "futures-util", + "once_cell", "paste", "pmoflac", "pmometadata", diff --git a/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs b/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs index b157101f..d22f7332 100644 --- a/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs +++ b/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use pmoaudio::{ nodes::{AudioError, DEFAULT_CHANNEL_SIZE}, - pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle}, + pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic, PipelineHandle}, AudioSegment, TypeRequirement, TypedAudioNode, }; use pmocovers::Cache as CoverCache; @@ -140,6 +140,7 @@ impl NodeLogic for TrackBoundaryCoverLogic { "TrackBoundaryCoverNode requires an upstream input channel".into(), ) })?; + let node_name = std::any::type_name::(); loop { let segment = select! { @@ -159,11 +160,7 @@ impl NodeLogic for TrackBoundaryCoverLogic { self.ensure_cover_pk(Arc::clone(metadata)).await; } - for tx in &output { - if tx.send(segment.clone()).await.is_err() { - return Err(AudioError::ChildDied); - } - } + send_to_children(node_name, &output, segment).await?; } Ok(()) diff --git a/pmoaudio-ext/src/sources/playlist_source.rs b/pmoaudio-ext/src/sources/playlist_source.rs index 27b801d5..caff6d64 100644 --- a/pmoaudio-ext/src/sources/playlist_source.rs +++ b/pmoaudio-ext/src/sources/playlist_source.rs @@ -111,7 +111,7 @@ use pmoaudio::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, - pipeline::{AudioPipelineNode, Node, NodeLogic}, + pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioChunkData, AudioSegment, I24, }; @@ -175,16 +175,7 @@ impl NodeLogic for PlaylistSourceLogic { 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)?; - } - }; - } + let node_name = std::any::type_name::(); let mut first_track = true; @@ -193,7 +184,7 @@ impl NodeLogic for PlaylistSourceLogic { if stop_token.is_cancelled() { tracing::info!("PlaylistSourceLogic: stop requested, emitting EndOfStream"); let eos = AudioSegment::new_end_of_stream(0, 0.0); - send_to_children!(eos); + send_to_children(node_name, &output, eos).await?; break; } @@ -202,7 +193,7 @@ impl NodeLogic for PlaylistSourceLogic { _ = stop_token.cancelled() => { tracing::info!("PlaylistSourceLogic: stop cancelled during pop"); let eos = AudioSegment::new_end_of_stream(0, 0.0); - send_to_children!(eos); + send_to_children(node_name, &output, eos).await?; break; } result = self.playlist_handle.pop() => { @@ -236,7 +227,7 @@ impl NodeLogic for PlaylistSourceLogic { 0.0, format!("Playlist error: {}", e) ); - send_to_children!(error_marker); + send_to_children(node_name, &output, error_marker).await?; continue; } } @@ -250,7 +241,7 @@ impl NodeLogic for PlaylistSourceLogic { tracing::warn!("PlaylistSourceLogic: failed to get metadata: {}", e); let error_marker = AudioSegment::new_error(0, 0.0, format!("Failed to get metadata: {}", e)); - send_to_children!(error_marker); + send_to_children(node_name, &output, error_marker).await?; continue; } }; @@ -279,7 +270,7 @@ impl NodeLogic for PlaylistSourceLogic { tracing::debug!("PlaylistSourceLogic: emitting TrackBoundary"); let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata); - send_to_children!(boundary); + send_to_children(node_name, &output, boundary).await?; // Obtenir le chemin du fichier let file_path = match track.file_path() { @@ -288,7 +279,7 @@ impl NodeLogic for PlaylistSourceLogic { tracing::warn!("PlaylistSourceLogic: failed to get file path: {}", e); let error_marker = AudioSegment::new_error(0, 0.0, format!("Failed to get file path: {}", e)); - send_to_children!(error_marker); + send_to_children(node_name, &output, error_marker).await?; continue; } }; @@ -303,6 +294,7 @@ impl NodeLogic for PlaylistSourceLogic { first_track = false; match decode_and_emit_track( + node_name, &file_path, self.chunk_frames, &output, @@ -334,7 +326,7 @@ impl NodeLogic for PlaylistSourceLogic { tracing::error!("PlaylistSourceLogic: error decoding track: {}", e); let error_marker = AudioSegment::new_error(0, 0.0, format!("Decode error: {}", e)); - send_to_children!(error_marker); + send_to_children(node_name, &output, error_marker).await?; // Continue vers la piste suivante } } @@ -356,6 +348,7 @@ impl NodeLogic for PlaylistSourceLogic { /// Gère le cache progressif : si EOF est atteint et que le download est toujours en cours, /// attend et réessaie au lieu de terminer immédiatement. async fn decode_and_emit_track( + node_name: &'static str, path: &PathBuf, chunk_frames: usize, output: &[mpsc::Sender>], @@ -491,18 +484,10 @@ async fn decode_and_emit_track( if emit_top_zero && total_frames == 0 { tracing::debug!("decode_and_emit_track: emitting TopZeroSync (first chunk)"); let top_zero = AudioSegment::new_top_zero_sync(); - for tx in output { - tx.send(top_zero.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children(node_name, output, top_zero).await?; } - for tx in output { - tx.send(segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children(node_name, output, segment).await?; chunk_index += 1; total_frames += frames_to_emit as u64; @@ -517,11 +502,7 @@ async fn decode_and_emit_track( let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64; let segment = bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?; - for tx in output { - tx.send(segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children(node_name, output, segment).await?; } } diff --git a/pmoaudio/Cargo.toml b/pmoaudio/Cargo.toml index 7d569e02..720d58ce 100755 --- a/pmoaudio/Cargo.toml +++ b/pmoaudio/Cargo.toml @@ -20,6 +20,7 @@ bytemuck = "1.24.0" reqwest = { version = "0.12", features = ["stream"] } tracing = "0.1" cpal = "0.15" +once_cell = "1.20" [dev-dependencies] tokio-test = "0.4" diff --git a/pmoaudio/src/nodes/converter_nodes.rs b/pmoaudio/src/nodes/converter_nodes.rs index 402feefe..952cf3ae 100755 --- a/pmoaudio/src/nodes/converter_nodes.rs +++ b/pmoaudio/src/nodes/converter_nodes.rs @@ -15,7 +15,7 @@ use crate::{ nodes::AudioError, - pipeline::{Node, NodeLogic}, + pipeline::{send_to_children, Node, NodeLogic}, AudioChunk, AudioPipelineNode, AudioSegment, }; use std::sync::Arc; @@ -100,12 +100,7 @@ where segment }; - // Envoyer à tous les enfants - for tx in &output { - tx.send(output_segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children(std::any::type_name::(), &output, output_segment).await?; } Ok(()) diff --git a/pmoaudio/src/nodes/file_source.rs b/pmoaudio/src/nodes/file_source.rs index c0bddeb9..ccd7bf07 100755 --- a/pmoaudio/src/nodes/file_source.rs +++ b/pmoaudio/src/nodes/file_source.rs @@ -1,6 +1,6 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, - pipeline::{AudioPipelineNode, Node, NodeLogic}, + pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioChunkData, AudioSegment, I24, }; @@ -47,17 +47,6 @@ impl NodeLogic for FileSourceLogic { 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)?; - } - }; - } - // Ouvrir le fichier let file = File::open(&self.path) .await @@ -82,7 +71,7 @@ impl NodeLogic for FileSourceLogic { // Émettre TopZeroSync let top_zero = AudioSegment::new_top_zero_sync(); - send_to_children!(top_zero); + send_to_children(std::any::type_name::(), &output, top_zero).await?; // Extraire et émettre les métadonnées du fichier if let Ok(file_metadata) = AudioFileMetadata::from_file(&self.path) { @@ -110,7 +99,7 @@ impl NodeLogic for FileSourceLogic { 0.0, Arc::new(tokio::sync::RwLock::new(metadata)), ); - send_to_children!(track_boundary); + send_to_children(std::any::type_name::(), &output, track_boundary).await?; } // Préparer la lecture des chunks audio @@ -168,7 +157,7 @@ impl NodeLogic for FileSourceLogic { timestamp_sec, )?; - send_to_children!(segment); + send_to_children(std::any::type_name::(), &output, segment).await?; chunk_index += 1; total_frames += frames_to_emit as u64; @@ -183,7 +172,7 @@ impl NodeLogic for FileSourceLogic { let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64; let segment = bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?; - send_to_children!(segment); + send_to_children(std::any::type_name::(), &output, segment).await?; total_frames += frames as u64; chunk_index += 1; } @@ -192,7 +181,7 @@ impl NodeLogic for FileSourceLogic { // Émettre EndOfStream let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64; let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp); - send_to_children!(eos); + send_to_children(std::any::type_name::(), &output, eos).await?; // Attendre la fin du décodage stream diff --git a/pmoaudio/src/nodes/http_source.rs b/pmoaudio/src/nodes/http_source.rs index 1ec55a1c..ae703caf 100755 --- a/pmoaudio/src/nodes/http_source.rs +++ b/pmoaudio/src/nodes/http_source.rs @@ -1,6 +1,6 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, - pipeline::{Node, NodeLogic}, + pipeline::{send_to_children, Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioChunkData, AudioPipelineNode, AudioSegment, I24, }; @@ -127,16 +127,6 @@ impl NodeLogic for HttpSourceLogic { output: Vec>>, stop_token: CancellationToken, ) -> Result<(), AudioError> { - macro_rules! send_to_children { - ($segment:expr) => { - for tx in &output { - tx.send($segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } - }; - } - // Effectuer la requête HTTP let response = reqwest::get(&self.url).await.map_err(|e| { AudioError::ProcessingError(format!("HTTP request failed for {}: {}", self.url, e)) @@ -179,12 +169,17 @@ impl NodeLogic for HttpSourceLogic { }; // Émettre TopZeroSync - send_to_children!(AudioSegment::new_top_zero_sync()); + send_to_children( + std::any::type_name::(), + &output, + AudioSegment::new_top_zero_sync(), + ) + .await?; // Émettre TrackBoundary avec les métadonnées HTTP let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata))); - send_to_children!(track_boundary); + send_to_children(std::any::type_name::(), &output, track_boundary).await?; // Préparer la lecture des chunks audio let frame_bytes = stream_info.bytes_per_sample() * stream_info.channels as usize; @@ -242,7 +237,7 @@ impl NodeLogic for HttpSourceLogic { timestamp_sec, )?; - send_to_children!(segment); + send_to_children(std::any::type_name::(), &output, segment).await?; chunk_index += 1; total_frames += frames_to_emit as u64; @@ -255,7 +250,7 @@ impl NodeLogic for HttpSourceLogic { let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64; let segment = bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?; - send_to_children!(segment); + send_to_children(std::any::type_name::(), &output, segment).await?; total_frames += frames as u64; chunk_index += 1; } @@ -264,7 +259,7 @@ impl NodeLogic for HttpSourceLogic { // Émettre EndOfStream let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64; let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp); - send_to_children!(eos); + send_to_children(std::any::type_name::(), &output, eos).await?; // Attendre la fin du décodage stream diff --git a/pmoaudio/src/nodes/resampling_node.rs b/pmoaudio/src/nodes/resampling_node.rs index 397ebaa3..7cbac24b 100644 --- a/pmoaudio/src/nodes/resampling_node.rs +++ b/pmoaudio/src/nodes/resampling_node.rs @@ -31,7 +31,7 @@ use crate::{ dsp::resampling::{build_resampler, resampling, Resampler}, nodes::{AudioError, TypedAudioNode}, - pipeline::{AudioPipelineNode, Node, NodeLogic}, + pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioChunkData, AudioSegment, BitDepth, I24, }; @@ -172,12 +172,7 @@ impl NodeLogic for ResamplingLogic { segment }; - // Envoyer à tous les enfants - for tx in &output { - tx.send(output_segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children(std::any::type_name::(), &output, output_segment).await?; } Ok(()) diff --git a/pmoaudio/src/nodes/timer_buffer_node.rs b/pmoaudio/src/nodes/timer_buffer_node.rs index 2468733c..69379441 100644 --- a/pmoaudio/src/nodes/timer_buffer_node.rs +++ b/pmoaudio/src/nodes/timer_buffer_node.rs @@ -52,7 +52,7 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE, DEFAULT_CHUNK_DURATION_MS}, - pipeline::{AudioPipelineNode, Node, NodeLogic}, + pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, AudioSegment, SyncMarker, _AudioSegment, }; @@ -135,11 +135,7 @@ impl TimerBufferNodeLogic { self.buffer.len() ); - for tx in output { - tx.send(segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children(std::any::type_name::(), output, segment).await?; } Ok(()) } @@ -228,11 +224,12 @@ impl NodeLogic for TimerBufferNodeLogic { } // Propager le marker immédiatement - for tx in &output { - tx.send(segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children( + std::any::type_name::(), + &output, + segment.clone(), + ) + .await?; } _AudioSegment::Chunk(chunk) => { diff --git a/pmoaudio/src/nodes/timer_node.rs b/pmoaudio/src/nodes/timer_node.rs index 67038319..bc005153 100644 --- a/pmoaudio/src/nodes/timer_node.rs +++ b/pmoaudio/src/nodes/timer_node.rs @@ -53,7 +53,7 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE}, - pipeline::{AudioPipelineNode, Node, NodeLogic}, + pipeline::{send_to_children_with_timing, AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, AudioSegment, SyncMarker, _AudioSegment, }; @@ -132,27 +132,6 @@ impl NodeLogic for TimerNodeLogic { output.len() ); - // Macro helper pour envoyer à tous les enfants - macro_rules! send_to_children { - ($segment:expr) => { - for (idx, tx) in output.iter().enumerate() { - let send_start = Instant::now(); - tx.send($segment.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - let send_duration = send_start.elapsed(); - if send_duration.as_millis() >= 50 { - tracing::debug!( - "TimerNode: send to child {} blocked for {:.3}s (segment ts={:.3}s)", - idx, - send_duration.as_secs_f64(), - $segment.timestamp_sec - ); - } - } - }; - } - loop { let segment = tokio::select! { _ = stop_token.cancelled() => { @@ -184,7 +163,23 @@ impl NodeLogic for TimerNodeLogic { // Autres markers: passthrough transparent } } - send_to_children!(segment); + let segment_ts = segment.timestamp_sec; + send_to_children_with_timing( + std::any::type_name::(), + &output, + segment.clone(), + |idx, send_duration, _capacity_before| { + if send_duration.as_millis() >= 50 { + tracing::debug!( + "TimerNode: send to child {} blocked for {:.3}s (segment ts={:.3}s)", + idx, + send_duration.as_secs_f64(), + segment_ts + ); + } + }, + ) + .await?; } _AudioSegment::Chunk(_) => { @@ -263,7 +258,23 @@ impl NodeLogic for TimerNodeLogic { tracing::warn!("TimerNodeLogic: NO TIMER SET - passthrough without pacing! (ts={:.3}s)", segment.timestamp_sec); } - send_to_children!(segment); + let segment_ts = segment.timestamp_sec; + send_to_children_with_timing( + std::any::type_name::(), + &output, + segment.clone(), + |idx, send_duration, _capacity_before| { + if send_duration.as_millis() >= 50 { + tracing::debug!( + "TimerNode: send to child {} blocked for {:.3}s (segment ts={:.3}s)", + idx, + send_duration.as_secs_f64(), + segment_ts + ); + } + }, + ) + .await?; } } } diff --git a/pmoaudio/src/pipeline.rs b/pmoaudio/src/pipeline.rs index 3610463f..e5f539f3 100755 --- a/pmoaudio/src/pipeline.rs +++ b/pmoaudio/src/pipeline.rs @@ -41,7 +41,10 @@ //! ``` use crate::{nodes::AudioError, AudioSegment}; -use std::sync::Arc; +use once_cell::sync::Lazy; +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use tokio::sync::mpsc; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -304,6 +307,83 @@ pub trait NodeLogic: Send + 'static { } } +/// Envoie un segment à l'ensemble des enfants d'un nœud. +/// +/// Cette fonction gère la logique de clonage d'`Arc` et la +/// conversion de l'erreur `mpsc::error::SendError` en `AudioError::ChildDied`. +static FIRST_AUDIO_CHUNK_TRACKER: Lazy>> = + Lazy::new(|| Mutex::new(HashSet::new())); + +const FIRST_CHUNK_EPSILON: f64 = 1e-6; + +fn record_first_audio_chunk_timestamp( + node_name: &'static str, + outputs: &[mpsc::Sender>], + segment: &Arc, +) { + if outputs.is_empty() || !segment.is_audio_chunk() { + return; + } + + let mut tracker = FIRST_AUDIO_CHUNK_TRACKER + .lock() + .expect("invariant tracker mutex poisoned"); + let key = outputs.as_ptr() as usize; + if tracker.contains(&key) { + return; + } + + if segment.timestamp_sec.abs() > FIRST_CHUNK_EPSILON { + tracing::warn!( + "First audio chunk emitted by {node_name} started at {:.6}s (order={}), expected 0s", + segment.timestamp_sec, + segment.order, + node_name = node_name, + ); + } + + tracker.insert(key); +} + +pub async fn send_to_children( + node_name: &'static str, + outputs: &[mpsc::Sender>], + segment: Arc, +) -> Result<(), AudioError> { + record_first_audio_chunk_timestamp(node_name, outputs, &segment); + for tx in outputs { + tx.send(segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + Ok(()) +} + +/// Variante de [`send_to_children`] qui expose le temps passé à envoyer à chaque enfant. +/// +/// Utile pour les nœuds qui souhaitent instrumenter les blocages éventuels lors +/// de l'envoi (ex: TimerNode). +pub async fn send_to_children_with_timing( + node_name: &'static str, + outputs: &[mpsc::Sender>], + segment: Arc, + mut inspector: F, +) -> Result<(), AudioError> +where + F: FnMut(usize, Duration, usize), +{ + record_first_audio_chunk_timestamp(node_name, outputs, &segment); + for (idx, tx) in outputs.iter().enumerate() { + let capacity_before = tx.capacity(); + let send_start = Instant::now(); + tx.send(segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + inspector(idx, send_start.elapsed(), capacity_before); + } + Ok(()) +} + /// Handle pour contrôler un pipeline en cours d'exécution /// /// Retourné par la méthode `start()`, ce handle permet de : diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index 337e2ab8..b046b41f 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -11,7 +11,7 @@ use crate::{ use futures_util::StreamExt; use pmoaudio::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, - pipeline::{Node, NodeLogic}, + pipeline::{send_to_children, send_to_children_with_timing, Node, NodeLogic}, type_constraints::TypeRequirement, AudioPipelineNode, AudioSegment, SyncMarker, I24, }; @@ -395,45 +395,42 @@ impl RadioParadiseStreamSourceLogic { output: &[mpsc::Sender>], segment: Arc, ) -> Result<(), AudioError> { - self.stats.record_segment_received(segment.timestamp_sec); + let segment_ts = segment.timestamp_sec; + self.stats.record_segment_received(segment_ts); - 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 segment_bytes = match &segment.segment { + pmoaudio::_AudioSegment::Chunk(chunk) => chunk.len() * 2 * 4, + _ => 0, + }; - 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 { - let duration_ms = send_duration.as_millis() as u64; - self.stats.record_backpressure(duration_ms); + send_to_children_with_timing( + std::any::type_name::(), + output, + segment, + |i, send_duration, capacity_before| { tracing::trace!( - "send_to_children: Send to child {} BLOCKED for {:.3}s (channel capacity before send={}, timestamp={:.3}s)", + "send_to_children: Sending to child {} (channel capacity={}, timestamp={:.3}s)", i, - send_duration.as_secs_f64(), capacity_before, - segment.timestamp_sec + segment_ts ); - } - // 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 + if send_duration.as_millis() > 10 { + let duration_ms = send_duration.as_millis() as u64; + self.stats.record_backpressure(duration_ms); + tracing::trace!( + "send_to_children: Send to child {} BLOCKED for {:.3}s (channel capacity before send={}, timestamp={:.3}s)", + i, + send_duration.as_secs_f64(), + capacity_before, + segment_ts + ); } - _ => 0, - }; - self.stats.record_segment_sent(segment_bytes); - } + + self.stats.record_segment_sent(segment_bytes); + }, + ) + .await?; Ok(()) } } @@ -705,11 +702,7 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { output.len() ); let eos = AudioSegment::new_end_of_stream(order, last_timestamp); - for tx in &output { - tx.send(eos.clone()) - .await - .map_err(|_| AudioError::ChildDied)?; - } + send_to_children(std::any::type_name::(), &output, eos).await?; // IMPORTANT: Attendre que tous les channels soient fermés par les enfants // Cela garantit que tous les chunks (y compris ceux en attente dans les buffers MPSC)