eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Ou encore un peu de factorisation dans les nœuds.

Browse Source
This commit is contained in:
Eric Coissac
2025-11-17 21:43:57 +01:00
parent 9bc0c2544d
commit 9be6835ddc
12 changed files with 194 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -3035,6 +3035,7 @@ dependencies = [
"bytemuck",
"cpal",
"futures-util",
"once_cell",
"paste",
"pmoflac",
"pmometadata",

9
pmoaudio-ext/src/nodes/track_boundary_cover_node.rs
View File

@@ -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::<Self>();
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(())

47
pmoaudio-ext/src/sources/playlist_source.rs
View File

@@ -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::<Self>();
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<Arc<AudioSegment>>],
@@ -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?;
}
}

1
pmoaudio/Cargo.toml
View File

@@ -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"

9
pmoaudio/src/nodes/converter_nodes.rs
View File

@@ -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::<Self>(), &output, output_segment).await?;
}
Ok(())

23
pmoaudio/src/nodes/file_source.rs
View File

@@ -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::<Self>(), &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::<Self>(), &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::<Self>(), &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::<Self>(), &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::<Self>(), &output, eos).await?;
// Attendre la fin du décodage
stream

27
pmoaudio/src/nodes/http_source.rs
View File

@@ -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<mpsc::Sender<Arc<AudioSegment>>>,
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::<Self>(),
&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::<Self>(), &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::<Self>(), &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::<Self>(), &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::<Self>(), &output, eos).await?;
// Attendre la fin du décodage
stream

9
pmoaudio/src/nodes/resampling_node.rs
View File

@@ -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::<Self>(), &output, output_segment).await?;
}
Ok(())

19
pmoaudio/src/nodes/timer_buffer_node.rs
View File

@@ -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::<Self>(), 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::<Self>(),
&output,
segment.clone(),
)
.await?;
}
_AudioSegment::Chunk(chunk) => {

59
pmoaudio/src/nodes/timer_node.rs
View File

@@ -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::<Self>(),
&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::<Self>(),
&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?;
}
}
}

82
pmoaudio/src/pipeline.rs
View File

@@ -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<AudioSegment>` et la
/// conversion de l'erreur `mpsc::error::SendError` en `AudioError::ChildDied`.
static FIRST_AUDIO_CHUNK_TRACKER: Lazy<Mutex<HashSet<usize>>> =
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<Arc<AudioSegment>>],
segment: &Arc<AudioSegment>,
) {
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<Arc<AudioSegment>>],
segment: Arc<AudioSegment>,
) -> 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<F>(
node_name: &'static str,
outputs: &[mpsc::Sender<Arc<AudioSegment>>],
segment: Arc<AudioSegment>,
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 :

67
pmoparadise/src/radio_paradise_stream_source.rs
View File

@@ -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<Arc<AudioSegment>>],
segment: Arc<AudioSegment>,
) -> 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::<Self>(),
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::<Self>(), &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)