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

debuggage des stream

Browse Source
This commit is contained in:
Eric Coissac
2025-11-14 10:43:53 +01:00
parent de84cbafbb
commit 1c2d30cbe9
44 changed files with 2018 additions and 717 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
pmoaudio-ext/src/sinks/broadcast_pacing.rs Normal file
View File

@@ -0,0 +1,108 @@
//! Shared broadcast pacing logic for streaming sinks.
//!
//! Provides intelligent backpressure based on audio timing:
//! - Detects TopZeroSync (when audio timestamp resets to 0)
//! - Drops frames that are late (audio_ts < elapsed)
//! - Paces broadcast to match audio playback rate
use std::time::Instant;
use tracing::{debug, info, warn};
/// Error returned when a frame should be skipped (too late)
#[derive(Debug)]
pub struct SkipFrame;
/// Manages broadcast pacing with TopZeroSync detection
pub struct BroadcastPacer {
/// Start time (reset on TopZeroSync)
start_time: Instant,
/// Maximum allowed lead time before sleeping (0 = no pacing)
max_lead_time: f64,
/// Label for logging (e.g., "FLAC" or "OGG")
label: String,
}
impl BroadcastPacer {
/// Create a new broadcast pacer
///
/// # Arguments
///
/// * `max_lead_time` - Maximum lead time in seconds (0 = no pacing)
/// * `label` - Label for logging
pub fn new(max_lead_time: f64, label: impl Into<String>) -> Self {
Self {
start_time: Instant::now(),
max_lead_time: max_lead_time.max(0.0),
label: label.into(),
}
}
/// Check timing and apply pacing
///
/// This function:
/// 1. Detects TopZeroSync (audio_timestamp < 0.1 after >1s) and resets timer
/// 2. Drops frames that are late (audio_ts < elapsed)
/// 3. Sleeps if too far ahead (lead_time > max_lead_time)
///
/// # Returns
///
/// - `Ok(())` if frame is on time or successfully paced
/// - `Err(SkipFrame)` if frame is too late and should be dropped
pub async fn check_and_pace(&mut self, audio_timestamp: f64) -> Result<(), SkipFrame> {
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 1. DÉTECTION TopZeroSync ║
// ║ Si le timestamp revient proche de 0, reset l'horloge ║
// ╚═══════════════════════════════════════════════════════════════╝
let elapsed_since_start = self.start_time.elapsed().as_secs_f64();
if audio_timestamp < 0.1 && elapsed_since_start > 1.0 {
self.start_time = Instant::now();
info!(
"{} broadcaster: TopZeroSync detected, resetting timer",
self.label
);
}
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 2. CALCUL DU LEAD TIME ║
// ║ lead_time > 0 : en avance (OK) ║
// ║ lead_time < 0 : en retard (SKIP) ║
// ╚═══════════════════════════════════════════════════════════════╝
let elapsed = self.start_time.elapsed().as_secs_f64();
let lead_time = audio_timestamp - elapsed;
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 3. DROP FRAMES EN RETARD (tolérance zéro) ║
// ╚═══════════════════════════════════════════════════════════════╝
if lead_time < 0.0 {
warn!(
"{}: Dropping late frame: audio_ts={:.3}s, elapsed={:.3}s, lag={:.3}s",
self.label,
audio_timestamp,
elapsed,
-lead_time
);
return Err(SkipFrame);
}
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 4. BACKPRESSURE NATURELLE - Pas de sleep ! ║
// ║ ║
// ║ Le pacing vient de : ║
// ║ - TimerBufferNode en amont (envoi régulier à 50ms/chunk) ║
// ║ - Capacité limitée du broadcast channel ║
// ║ - Client HTTP qui lit à vitesse réelle ║
// ║ ║
// ║ Pas besoin de sleep explicite qui causerait des bursts ║
// ╚═══════════════════════════════════════════════════════════════╝
// Log pour info si on est très en avance, mais on ne dort PAS
if self.max_lead_time > 0.0 && lead_time > self.max_lead_time {
debug!(
"{} broadcaster: lead_time={:.3}s > max={:.3}s (audio_ts={:.3}s, elapsed={:.3}s) - relying on natural backpressure",
self.label, lead_time, self.max_lead_time, audio_timestamp, elapsed
);
}
Ok(())
}
}

73
pmoaudio-ext/src/sinks/flac_cache_sink.rs
View File

@@ -93,10 +93,16 @@ impl NodeLogic for FlacCacheSinkLogic {
loop {
// Attendre le premier chunk audio pour cette track
tracing::debug!("FlacCacheSink: Waiting for first audio chunk (track_number={})", track_number);
let (first_segment, track_metadata) = if let Some(metadata) = next_track_metadata.take() {
tracing::debug!(
"FlacCacheSink: Waiting for first audio chunk (track_number={})",
track_number
);
let (first_segment, track_metadata) = if let Some(metadata) = next_track_metadata.take()
{
// On a déjà reçu le TrackBoundary en Phase 3 de la track précédente
tracing::debug!("FlacCacheSink: Using TrackBoundary metadata from previous track's Phase 3");
tracing::debug!(
"FlacCacheSink: Using TrackBoundary metadata from previous track's Phase 3"
);
// Attendre juste le premier chunk
match wait_for_first_audio_chunk(&mut rx, &stop_token).await {
Ok(chunk) => {
@@ -296,18 +302,30 @@ impl NodeLogic for FlacCacheSinkLogic {
url
}
Err(e) if e.is_transient() => {
tracing::debug!("FlacCacheSink: Transient error getting cover URL for pk {}: {}", pk, e);
tracing::debug!(
"FlacCacheSink: Transient error getting cover URL for pk {}: {}",
pk,
e
);
None
}
Err(e) => {
tracing::warn!("FlacCacheSink: Cannot obtain cover URL for audio asset {}: {}", pk, e);
tracing::warn!(
"FlacCacheSink: Cannot obtain cover URL for audio asset {}: {}",
pk,
e
);
None
}
};
if let Some(cover_url) = url {
tracing::debug!("FlacCacheSink: Attempting to cache cover from URL: {}", cover_url);
match self.covers
tracing::debug!(
"FlacCacheSink: Attempting to cache cover from URL: {}",
cover_url
);
match self
.covers
.add_from_url(&cover_url, self.collection.as_deref())
.await
{
@@ -323,7 +341,11 @@ impl NodeLogic for FlacCacheSinkLogic {
}
}
Err(e) => {
tracing::warn!("FlacCacheSink: Failed to cache cover for audio asset {}: {}", pk, e);
tracing::warn!(
"FlacCacheSink: Failed to cache cover for audio asset {}: {}",
pk,
e
);
}
}
} else {
@@ -339,20 +361,27 @@ impl NodeLogic for FlacCacheSinkLogic {
playlist_handle.push(pk.clone()).await.map_err(|e| {
AudioError::ProcessingError(format!("Failed to add to playlist: {}", e))
})?;
tracing::info!("FlacCacheSink: Successfully pushed to playlist in {:?}", push_start.elapsed());
tracing::info!(
"FlacCacheSink: Successfully pushed to playlist in {:?}",
push_start.elapsed()
);
}
// Si EndOfStream a été reçu pendant le prebuffer, on a déjà tout traité
// Il faut juste attendre que le pump se termine et retourner
if end_of_stream_received {
tracing::debug!("FlacCacheSink: EndOfStream was received during prebuffer, track complete");
tracing::debug!(
"FlacCacheSink: EndOfStream was received during prebuffer, track complete"
);
drop(pump_handle);
track_number += 1;
continue; // Passer à la track suivante (qui n'arrivera pas car EndOfStream)
}
// Phase 3: Continuer à dispatcher jusqu'au TrackBoundary
tracing::debug!("FlacCacheSink: Continuing dispatch until TrackBoundary (pump runs in background)");
tracing::debug!(
"FlacCacheSink: Continuing dispatch until TrackBoundary (pump runs in background)"
);
let mut track_tx = track_tx_opt; // track_tx_opt contient Some(track_tx) car end_of_stream_received est false
let mut pump_handle = Some(pump_handle);
let mut pump_closed = false;
@@ -384,7 +413,9 @@ impl NodeLogic for FlacCacheSinkLogic {
if tx.send(segment).await.is_err() {
// Le pump a fermé son channel - cela peut arriver si le fichier
// était déjà en cache (add_from_reader retourne immédiatement)
tracing::debug!("FlacCacheSink: pump closed track_tx, checking pump status");
tracing::debug!(
"FlacCacheSink: pump closed track_tx, checking pump status"
);
drop(track_tx.take());
// Attendre que le pump se termine et vérifier le résultat
@@ -397,13 +428,21 @@ impl NodeLogic for FlacCacheSinkLogic {
}
Ok(Err(e)) => {
// Le pump a rencontré une erreur
tracing::error!("FlacCacheSink: pump died with error: {}", e);
tracing::error!(
"FlacCacheSink: pump died with error: {}",
e
);
return Err(e);
}
Err(e) => {
// Le pump task a paniqué
tracing::error!("FlacCacheSink: pump task panicked: {}", e);
return Err(AudioError::ProcessingError("Pump task panicked".to_string()));
tracing::error!(
"FlacCacheSink: pump task panicked: {}",
e
);
return Err(AudioError::ProcessingError(
"Pump task panicked".to_string(),
));
}
}
}
@@ -534,7 +573,9 @@ async fn wait_for_first_audio_chunk(
_AudioSegment::Sync(marker) => match &**marker {
SyncMarker::TrackBoundary { .. } => {
// On ne devrait pas recevoir de TrackBoundary ici car on l'a déjà
tracing::warn!("FlacCacheSink: Unexpected TrackBoundary while waiting for first chunk");
tracing::warn!(
"FlacCacheSink: Unexpected TrackBoundary while waiting for first chunk"
);
continue;
}
SyncMarker::EndOfStream => {

4
pmoaudio-ext/src/sinks/flac_frame_utils.rs
View File

@@ -374,8 +374,8 @@ mod tests {
// Real-world example: first frame at 0, false positive at 7
let data = vec![
0xFF, 0xF8, 0xC9, 0xA8, // Valid frame header at position 0
0x00, 0x8D, 0x4C,
0xFF, 0xFE, 0x00, 0x00, // False positive at position 7 (0xFE has reserved bit set)
0x00, 0x8D, 0x4C, 0xFF, 0xFE, 0x00,
0x00, // False positive at position 7 (0xFE has reserved bit set)
];
// Position 0 should be valid

9
pmoaudio-ext/src/sinks/mod.rs
View File

@@ -10,6 +10,9 @@ mod flac_cache_sink;
#[cfg(feature = "cache-sink")]
pub use flac_cache_sink::{FlacCacheSink, FlacCacheSinkStats, TrackStats};
#[cfg(feature = "http-stream")]
mod broadcast_pacing;
#[cfg(feature = "http-stream")]
mod flac_frame_utils;
@@ -17,10 +20,12 @@ mod flac_frame_utils;
mod streaming_flac_sink;
#[cfg(feature = "http-stream")]
pub use streaming_flac_sink::{StreamingFlacSink, StreamHandle, MetadataSnapshot, FlacClientStream, IcyClientStream};
pub use streaming_flac_sink::{
FlacClientStream, IcyClientStream, MetadataSnapshot, StreamHandle, StreamingFlacSink,
};
#[cfg(feature = "http-stream")]
mod streaming_ogg_flac_sink;
#[cfg(feature = "http-stream")]
pub use streaming_ogg_flac_sink::{StreamingOggFlacSink, OggFlacStreamHandle, OggFlacClientStream};
pub use streaming_ogg_flac_sink::{OggFlacClientStream, OggFlacStreamHandle, StreamingOggFlacSink};

213
pmoaudio-ext/src/sinks/streaming_flac_sink.rs
View File

@@ -62,7 +62,7 @@ use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use super::flac_frame_utils;
use super::{broadcast_pacing::BroadcastPacer, flac_frame_utils};
use async_trait::async_trait;
use bytes::Bytes;
use pmoaudio::{
@@ -81,16 +81,29 @@ use tracing::{debug, error, info, trace, warn};
/// Standard value used by most streaming servers.
const DEFAULT_ICY_METAINT: usize = 16000;
/// Broadcast channel capacity for FLAC bytes.
/// Set to 128 to provide ~10 seconds of buffer for network jitter.
/// With TimerNode pacing the stream to real-time, this is sufficient
/// while keeping metadata synchronized (larger buffers cause metadata drift).
const BROADCAST_CAPACITY: usize = 128;
/// Maximum lead time for HTTP broadcast pacing (in seconds).
/// Default maximum lead time for HTTP broadcast pacing (in seconds).
/// The broadcaster will sleep if it's ahead of real-time by more than this amount.
/// This is much smaller than the pipeline TimerNode's 3.0s to provide tighter control.
const BROADCAST_MAX_LEAD_TIME: f64 = 0.5;
const DEFAULT_BROADCAST_MAX_LEAD_TIME: f64 = 0.5;
/// Calculate broadcast channel capacity based on max_lead_time.
///
/// Estimates the number of items needed to buffer max_lead_time seconds of audio.
/// Assumes ~20 items per second (50ms per chunk).
///
/// # Arguments
///
/// * `max_lead_time` - Maximum lead time in seconds
///
/// # Returns
///
/// Broadcast channel capacity (minimum 100 items)
fn calculate_broadcast_capacity(max_lead_time: f64) -> usize {
// Estimation: ~20 items/second (chunks de 50ms en moyenne)
// Pour 10s: 200 items
let estimated_items_per_second = 20.0;
let capacity = (max_lead_time * estimated_items_per_second) as usize;
capacity.max(100) // Minimum 100 items
}
/// PCM chunk with audio data and timestamp for precise pacing.
#[derive(Debug)]
@@ -190,7 +203,11 @@ impl StreamHandle {
/// Subscribe to the FLAC stream with custom ICY metadata interval.
pub fn subscribe_icy_with_interval(&self, metaint: usize) -> IcyClientStream {
let count = self.active_clients.fetch_add(1, Ordering::SeqCst);
debug!("New ICY client subscribed (total: {}, metaint: {})", count + 1, metaint);
debug!(
"New ICY client subscribed (total: {}, metaint: {})",
count + 1,
metaint
);
IcyClientStream {
rx: self.flac_broadcast.subscribe(),
@@ -254,7 +271,10 @@ impl AsyncRead for FlacClientStream {
if let Some(header) = header_opt {
self.buffer.extend(header.iter());
info!("Sending cached FLAC header to new client ({} bytes)", header.len());
info!(
"Sending cached FLAC header to new client ({} bytes)",
header.len()
);
self.state = FlacStreamState::Streaming;
continue; // Now copy header to output buffer
} else {
@@ -413,7 +433,10 @@ impl AsyncRead for IcyClientStream {
if let Some(header) = header_opt {
self.buffer.extend(header.iter());
info!("Sending cached FLAC header to new ICY client ({} bytes)", header.len());
info!(
"Sending cached FLAC header to new ICY client ({} bytes)",
header.len()
);
self.state = FlacStreamState::Streaming;
continue; // Now copy header to output buffer
} else {
@@ -533,6 +556,7 @@ struct StreamingFlacSinkLogic {
flac_header: Arc<RwLock<Option<Bytes>>>,
encoder_state: Option<EncoderState>,
sample_rate: Option<u32>,
broadcast_max_lead_time: f64,
}
impl StreamingFlacSinkLogic {
@@ -542,12 +566,16 @@ impl StreamingFlacSinkLogic {
return Ok(()); // Already initialized
}
info!("Initializing FLAC encoder with sample rate: {} Hz", sample_rate);
info!(
"Initializing FLAC encoder with sample rate: {} Hz",
sample_rate
);
// Take the PCM receiver (we only initialize once)
let pcm_rx = self.pcm_rx.take().ok_or_else(|| {
AudioError::ProcessingError("PCM receiver already consumed".into())
})?;
let pcm_rx = self
.pcm_rx
.take()
.ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?;
// Create shared timestamp for pacing
let current_timestamp = Arc::new(RwLock::new(0.0f64));
@@ -565,15 +593,26 @@ impl StreamingFlacSinkLogic {
// Start the FLAC encoder
let flac_stream = encode_flac_stream(pcm_reader, pcm_format, self.encoder_options.clone())
.await
.map_err(|e| AudioError::ProcessingError(format!("Failed to start FLAC encoder: {}", e)))?;
.map_err(|e| {
AudioError::ProcessingError(format!("Failed to start FLAC encoder: {}", e))
})?;
info!("FLAC encoder initialized successfully");
// Spawn broadcaster task with timestamp for pacing
let flac_broadcast = self.flac_broadcast.clone();
let flac_header = self.flac_header.clone();
let max_lead = self.broadcast_max_lead_time;
let broadcaster_task = tokio::spawn(async move {
if let Err(e) = broadcast_flac_stream(flac_stream, flac_broadcast, flac_header, current_timestamp).await {
if let Err(e) = broadcast_flac_stream(
flac_stream,
flac_broadcast,
flac_header,
current_timestamp,
max_lead,
)
.await
{
error!("Broadcaster task error: {}", e);
}
});
@@ -682,10 +721,19 @@ impl NodeLogic for StreamingFlacSinkLogic {
bytes: pcm_bytes,
timestamp_sec: seg.timestamp_sec,
};
let send_start = std::time::Instant::now();
if let Err(e) = self.pcm_tx.send(pcm_chunk).await {
warn!("Failed to send PCM data to encoder: {}", e);
break;
}
let send_duration = send_start.elapsed();
if send_duration.as_millis() >= 50 {
debug!(
"StreamingFlacSink: pcm_tx send blocked for {:.3}s (ts={:.3}s)",
send_duration.as_secs_f64(),
seg.timestamp_sec
);
}
}
_AudioSegment::Sync(marker) => {
@@ -728,7 +776,6 @@ impl NodeLogic for StreamingFlacSinkLogic {
}
}
/// Broadcaster task: reads FLAC bytes from encoder and broadcasts to all clients.
/// Implements precise real-time pacing based on audio timestamps.
/// Ensures data is sent at FLAC frame boundaries to prevent sync errors in strict decoders like FFPlay.
@@ -737,8 +784,12 @@ async fn broadcast_flac_stream(
broadcast_tx: broadcast::Sender<Bytes>,
header_cache: Arc<RwLock<Option<Bytes>>>,
current_timestamp: Arc<RwLock<f64>>,
broadcast_max_lead_time: f64,
) -> Result<(), AudioError> {
info!("Broadcaster task started with FLAC frame boundary detection");
info!(
"Broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)",
broadcast_max_lead_time
);
// Use larger read buffer (16KB) to reduce syscalls and accumulator for frame boundary detection
// The accumulator is necessary to ensure we only send complete FLAC frames
@@ -746,9 +797,17 @@ async fn broadcast_flac_stream(
let mut accumulator = Vec::with_capacity(32768); // Pre-allocate to reduce reallocations
let mut total_bytes = 0u64;
let mut header_captured = false;
let start_time = std::time::Instant::now();
let mut pacer = BroadcastPacer::new(broadcast_max_lead_time, "FLAC");
let mut stats_last_log = std::time::Instant::now();
// Timing instrumentation for burst detection
let mut last_broadcast_time = std::time::Instant::now();
let mut broadcast_count = 0u64;
let mut total_read_time = 0.0f64;
let mut read_count = 0u64;
loop {
let read_start = std::time::Instant::now();
match flac_stream.read(&mut read_buffer).await {
Ok(0) => {
// EOF - send any remaining data
@@ -760,14 +819,38 @@ async fn broadcast_flac_stream(
break;
}
Ok(n) => {
let read_duration = read_start.elapsed().as_secs_f64();
read_count += 1;
total_read_time += read_duration;
if read_duration > 0.01 {
debug!(
"FLAC: flac_stream.read() took {:.3}s for {} bytes (avg: {:.3}s over {} reads)",
read_duration,
n,
total_read_time / read_count as f64,
read_count
);
}
total_bytes += n as u64;
if total_bytes % 100000 == 0 || total_bytes < 10000 {
trace!("Read {} bytes from FLAC encoder (total: {})", n, total_bytes);
trace!(
"Read {} bytes from FLAC encoder (total: {})",
n,
total_bytes
);
}
// Append to accumulator
accumulator.extend_from_slice(&read_buffer[..n]);
trace!(
"FLAC: accumulator now {} bytes after reading {} bytes",
accumulator.len(),
n
);
// Find where to split: position of last sync code (start of last incomplete frame)
// Everything before this position contains only complete frames
let boundary = flac_frame_utils::find_complete_frames_boundary(&accumulator);
@@ -781,18 +864,33 @@ async fn broadcast_flac_stream(
// Only broadcast if we have at least one complete frame (1KB minimum to avoid excessive small sends)
if boundary >= 1024 {
// Precise pacing based on audio timestamp
// ╔═══════════════════════════════════════════════════════════════╗
// ║ BACKPRESSURE INTELLIGENTE BASÉE SUR LE TIMING ║
// ║ ║
// ║ BroadcastPacer gère : ║
// ║ 1. Détection TopZeroSync (audio_ts < 0.1) ║
// ║ 2. Drop des chunks en retard (audio_ts < elapsed) ║
// ║ 3. Pacing pour contrôler le débit (max_lead_time) ║
// ║ ║
// ║ Cela crée la backpressure vers TimerBufferNode tout en ║
// ║ permettant de dropper les chunks vraiment périmés. ║
// ╚═══════════════════════════════════════════════════════════════╝
let audio_timestamp = *current_timestamp.read().await;
let elapsed = start_time.elapsed().as_secs_f64();
let lead_time = audio_timestamp - elapsed;
if lead_time > BROADCAST_MAX_LEAD_TIME {
let sleep_duration = lead_time - BROADCAST_MAX_LEAD_TIME;
if stats_last_log.elapsed() >= Duration::from_secs(1) {
debug!(
"Broadcaster pacing: sleeping {:.3}s (audio_ts={:.3}s, elapsed={:.3}s, lead={:.3}s)",
sleep_duration, audio_timestamp, elapsed, lead_time
"Broadcaster pacing snapshot: audio_ts={:.3}s buffer_bytes={}",
audio_timestamp,
accumulator.len()
);
tokio::time::sleep(tokio::time::Duration::from_secs_f64(sleep_duration)).await;
stats_last_log = std::time::Instant::now();
}
// Check timing et apply pacing (skip si en retard)
if pacer.check_and_pace(audio_timestamp).await.is_err() {
// Chunk en retard : vider l'accumulator et continuer
accumulator.clear();
continue;
}
// Split at boundary to avoid copying - extract prefix, keep suffix
@@ -800,6 +898,31 @@ async fn broadcast_flac_stream(
let to_send = std::mem::replace(&mut accumulator, remaining);
let bytes = Bytes::from(to_send);
// Measure broadcast interval for burst detection
let broadcast_interval = last_broadcast_time.elapsed().as_secs_f64();
last_broadcast_time = std::time::Instant::now();
broadcast_count += 1;
// Log if interval is unusual (too short = burst, too long = stall)
if broadcast_interval < 0.01 || broadcast_interval > 0.1 {
debug!(
"FLAC: broadcast interval {:.3}s ({}ms) - size={} bytes (count={})",
broadcast_interval,
(broadcast_interval * 1000.0) as u32,
bytes.len(),
broadcast_count
);
}
// Periodic stats
if broadcast_count % 100 == 0 {
debug!(
"FLAC: {} broadcasts sent, accumulator={} bytes remaining",
broadcast_count,
accumulator.len()
);
}
// Capture first chunk as header if it contains "fLaC"
if !header_captured && bytes.len() >= 4 && &bytes[0..4] == b"fLaC" {
*header_cache.write().await = Some(bytes.clone());
@@ -812,7 +935,11 @@ async fn broadcast_flac_stream(
// No receivers, but that's okay - clients may not be connected yet
trace!("No active receivers for FLAC broadcast: {}", e);
} else if num_receivers > 0 {
trace!("Broadcasted {} bytes to {} receivers", bytes.len(), num_receivers);
trace!(
"Broadcasted {} bytes to {} receivers",
bytes.len(),
num_receivers
);
}
}
}
@@ -857,9 +984,19 @@ impl StreamingFlacSink {
/// A tuple of `(sink, handle)` where:
/// - `sink` is added to the audio pipeline
/// - `handle` is used by HTTP handlers to serve streams
pub fn new(
pub fn new(encoder_options: EncoderOptions, bits_per_sample: u8) -> (Self, StreamHandle) {
Self::with_max_broadcast_lead(
encoder_options,
bits_per_sample,
DEFAULT_BROADCAST_MAX_LEAD_TIME,
)
}
/// Create a sink with a custom broadcast pacing limit.
pub fn with_max_broadcast_lead(
encoder_options: EncoderOptions,
bits_per_sample: u8,
broadcast_max_lead_time: f64,
) -> (Self, StreamHandle) {
// Validate bit depth
if ![16, 24, 32].contains(&bits_per_sample) {
@@ -872,8 +1009,15 @@ impl StreamingFlacSink {
// Shared metadata
let metadata = Arc::new(RwLock::new(MetadataSnapshot::default()));
// Calculate broadcast capacity based on max_lead_time
let broadcast_capacity = calculate_broadcast_capacity(broadcast_max_lead_time);
info!(
"StreamingFlacSink: using broadcast capacity of {} items (max_lead_time={:.1}s)",
broadcast_capacity, broadcast_max_lead_time
);
// Broadcast channel for FLAC bytes
let (flac_broadcast, _) = broadcast::channel(BROADCAST_CAPACITY);
let (flac_broadcast, _) = broadcast::channel(broadcast_capacity);
// FLAC header cache
let flac_header = Arc::new(RwLock::new(None));
@@ -900,6 +1044,7 @@ impl StreamingFlacSink {
flac_header,
encoder_state: None,
sample_rate: None,
broadcast_max_lead_time: broadcast_max_lead_time.max(0.0),
};
let sink = Self {

268
pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs
View File

@@ -53,7 +53,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use super::flac_frame_utils;
use super::{broadcast_pacing::BroadcastPacer, flac_frame_utils};
use async_trait::async_trait;
use bytes::Bytes;
use pmoaudio::{
@@ -68,13 +68,18 @@ use tokio::sync::{broadcast, mpsc, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, trace, warn};
/// Broadcast channel capacity for OGG-FLAC bytes.
/// Same as StreamingFlacSink for consistency.
const BROADCAST_CAPACITY: usize = 128;
/// Default maximum lead time for HTTP broadcast pacing (in seconds).
const DEFAULT_BROADCAST_MAX_LEAD_TIME: f64 = 0.0;
/// Maximum lead time for HTTP broadcast pacing (in seconds).
/// The broadcaster will sleep if it's ahead of real-time by more than this amount.
const BROADCAST_MAX_LEAD_TIME: f64 = 0.5;
/// Calculate broadcast channel capacity based on max lead time.
///
/// Estimate: ~20 OGG pages per second (assuming 50ms chunks).
/// Minimum capacity: 100 items for buffering even with 0 lead time.
fn calculate_broadcast_capacity(max_lead_time: f64) -> usize {
let estimated_items_per_second = 20.0;
let capacity = (max_lead_time * estimated_items_per_second) as usize;
capacity.max(100) // Minimum 100 items
}
/// PCM chunk with audio data and timestamp for precise pacing.
#[derive(Debug)]
@@ -167,7 +172,10 @@ impl AsyncRead for OggFlacClientStream {
if let Some(header) = header_opt {
self.buffer.extend(header.iter());
info!("Sending cached OGG-FLAC header to new client ({} bytes)", header.len());
info!(
"Sending cached OGG-FLAC header to new client ({} bytes)",
header.len()
);
self.state = OggFlacStreamState::Streaming;
continue; // Now copy header to output buffer
} else {
@@ -203,7 +211,7 @@ impl AsyncRead for OggFlacClientStream {
// Schedule a wakeup after a small delay to avoid busy-loop polling.
let waker = cx.waker().clone();
tokio::spawn(async move {
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
tokio::time::sleep(tokio::time::Duration::from_micros(100)).await;
waker.wake();
});
return Poll::Pending;
@@ -248,6 +256,7 @@ struct StreamingOggFlacSinkLogic {
ogg_header: Arc<RwLock<Option<Bytes>>>,
encoder_state: Option<EncoderState>,
sample_rate: Option<u32>,
broadcast_max_lead_time: f64,
}
impl StreamingOggFlacSinkLogic {
@@ -257,12 +266,16 @@ impl StreamingOggFlacSinkLogic {
return Ok(()); // Already initialized
}
info!("Initializing OGG-FLAC encoder with sample rate: {} Hz", sample_rate);
info!(
"Initializing OGG-FLAC encoder with sample rate: {} Hz",
sample_rate
);
// Take the PCM receiver (we only initialize once)
let pcm_rx = self.pcm_rx.take().ok_or_else(|| {
AudioError::ProcessingError("PCM receiver already consumed".into())
})?;
let pcm_rx = self
.pcm_rx
.take()
.ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?;
// Create shared timestamp for pacing
let current_timestamp = Arc::new(RwLock::new(0.0f64));
@@ -280,15 +293,26 @@ impl StreamingOggFlacSinkLogic {
// Start the FLAC encoder
let flac_stream = encode_flac_stream(pcm_reader, pcm_format, self.encoder_options.clone())
.await
.map_err(|e| AudioError::ProcessingError(format!("Failed to start FLAC encoder: {}", e)))?;
.map_err(|e| {
AudioError::ProcessingError(format!("Failed to start FLAC encoder: {}", e))
})?;
info!("OGG-FLAC encoder initialized successfully");
// Spawn OGG wrapper + broadcaster task with timestamp for pacing
let ogg_broadcast = self.ogg_broadcast.clone();
let ogg_header = self.ogg_header.clone();
let max_lead = self.broadcast_max_lead_time;
let broadcaster_task = tokio::spawn(async move {
if let Err(e) = broadcast_ogg_flac_stream(flac_stream, ogg_broadcast, ogg_header, current_timestamp).await {
if let Err(e) = broadcast_ogg_flac_stream(
flac_stream,
ogg_broadcast,
ogg_header,
current_timestamp,
max_lead,
)
.await
{
error!("OGG broadcaster task error: {}", e);
}
});
@@ -464,6 +488,19 @@ impl StreamingOggFlacSink {
pub fn new(
encoder_options: EncoderOptions,
bits_per_sample: u8,
) -> (Self, OggFlacStreamHandle) {
Self::with_max_broadcast_lead(
encoder_options,
bits_per_sample,
DEFAULT_BROADCAST_MAX_LEAD_TIME,
)
}
/// Create a sink with a custom broadcast pacing limit.
pub fn with_max_broadcast_lead(
encoder_options: EncoderOptions,
bits_per_sample: u8,
broadcast_max_lead_time: f64,
) -> (Self, OggFlacStreamHandle) {
// Validate bit depth
if ![16, 24, 32].contains(&bits_per_sample) {
@@ -477,7 +514,14 @@ impl StreamingOggFlacSink {
let metadata = Arc::new(RwLock::new(MetadataSnapshot::default()));
// Broadcast channel for OGG-FLAC bytes
let (ogg_broadcast, _) = broadcast::channel(BROADCAST_CAPACITY);
// Capacity calculated from max_lead_time to ensure enough buffering
let broadcast_capacity = calculate_broadcast_capacity(broadcast_max_lead_time);
tracing::debug!(
"OGG-FLAC broadcast capacity: {} items (for {:.1}s max lead time)",
broadcast_capacity,
broadcast_max_lead_time
);
let (ogg_broadcast, _) = broadcast::channel(broadcast_capacity);
// OGG-FLAC header cache
let ogg_header = Arc::new(RwLock::new(None));
@@ -504,6 +548,7 @@ impl StreamingOggFlacSink {
ogg_header,
encoder_state: None,
sample_rate: None,
broadcast_max_lead_time: broadcast_max_lead_time.max(0.0),
};
let sink = Self {
@@ -709,17 +754,27 @@ async fn broadcast_ogg_flac_stream(
broadcast_tx: broadcast::Sender<Bytes>,
header_cache: Arc<RwLock<Option<Bytes>>>,
current_timestamp: Arc<RwLock<f64>>,
broadcast_max_lead_time: f64,
) -> Result<(), AudioError> {
info!("OGG-FLAC broadcaster task started with FLAC frame boundary detection");
info!(
"OGG-FLAC broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)",
broadcast_max_lead_time
);
let stream_serial = rand::random::<u32>();
let mut ogg_writer = OggPageWriter::new(stream_serial);
let mut total_ogg_bytes = 0u64;
let mut header_captured = false;
let start_time = std::time::Instant::now();
let mut pacer = BroadcastPacer::new(broadcast_max_lead_time, "OGG");
let mut last_granule_update_time = 0.0f64;
// Timing instrumentation for burst detection
let mut last_broadcast_time = std::time::Instant::now();
let mut broadcast_count = 0u64;
let mut total_read_time = 0.0f64;
let mut read_count = 0u64;
// Step 1: Read FLAC header (fLaC + metadata blocks)
let flac_header = read_flac_header(&mut flac_stream).await?;
info!("Read FLAC header: {} bytes", flac_header.len());
@@ -731,7 +786,10 @@ async fn broadcast_ogg_flac_stream(
// Step 2: Create OGG-FLAC identification packet (BOS)
// Format according to https://xiph.org/flac/ogg_mapping.html
let ogg_flac_id = create_ogg_flac_identification(&flac_header)?;
info!("Created OGG-FLAC identification packet: {} bytes", ogg_flac_id.len());
info!(
"Created OGG-FLAC identification packet: {} bytes",
ogg_flac_id.len()
);
let bos_page = ogg_writer.create_page(&ogg_flac_id, true, false, false);
let bos_bytes = Bytes::from(bos_page);
@@ -747,7 +805,10 @@ async fn broadcast_ogg_flac_stream(
cached_header.extend_from_slice(&comment_bytes);
*header_cache.write().await = Some(Bytes::from(cached_header));
header_captured = true;
info!("OGG-FLAC header cached ({} bytes: BOS + Vorbis Comment)", bos_bytes.len() + comment_bytes.len());
info!(
"OGG-FLAC header cached ({} bytes: BOS + Vorbis Comment)",
bos_bytes.len() + comment_bytes.len()
);
// Broadcast header
let _ = broadcast_tx.send(bos_bytes);
@@ -761,6 +822,7 @@ async fn broadcast_ogg_flac_stream(
let mut flac_accumulator = Vec::with_capacity(32768);
loop {
let read_start = std::time::Instant::now();
match flac_stream.read(&mut read_buffer).await {
Ok(0) => {
// EOF - create final page with EOS flag and any remaining data
@@ -769,7 +831,10 @@ async fn broadcast_ogg_flac_stream(
let eos_bytes = Bytes::from(eos_page);
total_ogg_bytes += eos_bytes.len() as u64;
let _ = broadcast_tx.send(eos_bytes);
info!("Sent final EOS page with {} bytes of data", flac_accumulator.len());
info!(
"Sent final EOS page with {} bytes of data",
flac_accumulator.len()
);
} else {
// Send empty EOS page
let eos_page = ogg_writer.create_page(&[], false, true, false);
@@ -779,13 +844,36 @@ async fn broadcast_ogg_flac_stream(
info!("Sent empty EOS page");
}
info!("OGG-FLAC stream ended, total OGG bytes: {}", total_ogg_bytes);
info!(
"OGG-FLAC stream ended, total OGG bytes: {}",
total_ogg_bytes
);
break;
}
Ok(n) => {
let read_duration = read_start.elapsed().as_secs_f64();
read_count += 1;
total_read_time += read_duration;
if read_duration > 0.01 {
debug!(
"OGG: flac_stream.read() took {:.3}s for {} bytes (avg: {:.3}s over {} reads)",
read_duration,
n,
total_read_time / read_count as f64,
read_count
);
}
// Append to accumulator
flac_accumulator.extend_from_slice(&read_buffer[..n]);
trace!(
"OGG: accumulator now {} bytes after reading {} bytes",
flac_accumulator.len(),
n
);
// Process complete FLAC frames one at a time
// OGG-FLAC spec requires: "Each audio data packet contains one complete FLAC frame"
loop {
@@ -804,7 +892,9 @@ async fn broadcast_ogg_flac_stream(
if byte1 == 0xFF && byte2 >= 0xF8 && byte2 <= 0xFE {
// Validate frame header with CRC-8 to avoid false positives
if flac_frame_utils::validate_frame_header_crc(&flac_accumulator, i) {
if let Some(samples) = flac_frame_utils::parse_flac_block_size(&flac_accumulator, i) {
if let Some(samples) =
flac_frame_utils::parse_flac_block_size(&flac_accumulator, i)
{
sync_data.push((i, samples));
}
}
@@ -823,26 +913,34 @@ async fn broadcast_ogg_flac_stream(
// Verify first frame starts at position 0 (otherwise we have garbage data)
if first_frame_start != 0 {
warn!("OGG-FLAC: Skipping {} bytes of garbage data before first frame", first_frame_start);
warn!(
"OGG-FLAC: Skipping {} bytes of garbage data before first frame",
first_frame_start
);
flac_accumulator.drain(0..first_frame_start);
continue;
}
// Extract just the first frame
let first_frame: Vec<u8> = flac_accumulator.drain(0..second_frame_start).collect();
let first_frame: Vec<u8> =
flac_accumulator.drain(0..second_frame_start).collect();
// Precise pacing based on audio timestamp
// ╔═══════════════════════════════════════════════════════════════╗
// ║ BACKPRESSURE INTELLIGENTE BASÉE SUR LE TIMING ║
// ║ ║
// ║ BroadcastPacer gère : ║
// ║ 1. Détection TopZeroSync (audio_ts < 0.1) ║
// ║ 2. Drop des chunks en retard (audio_ts < elapsed) ║
// ║ 3. Pacing pour contrôler le débit (max_lead_time) ║
// ║ ║
// ║ Cela crée la backpressure vers TimerBufferNode tout en ║
// ║ permettant de dropper les chunks vraiment périmés. ║
// ╚═══════════════════════════════════════════════════════════════╝
let audio_timestamp = *current_timestamp.read().await;
let elapsed = start_time.elapsed().as_secs_f64();
let lead_time = audio_timestamp - elapsed;
if lead_time > BROADCAST_MAX_LEAD_TIME {
let sleep_duration = lead_time - BROADCAST_MAX_LEAD_TIME;
debug!(
"OGG broadcaster pacing: sleeping {:.3}s (audio_ts={:.3}s, elapsed={:.3}s, lead={:.3}s)",
sleep_duration, audio_timestamp, elapsed, lead_time
);
tokio::time::sleep(tokio::time::Duration::from_secs_f64(sleep_duration)).await;
// Check timing et apply pacing (skip si en retard)
if pacer.check_and_pace(audio_timestamp).await.is_err() {
continue; // Skip ce chunk (trop en retard)
}
// Update granule position (cumulative sample count)
@@ -853,10 +951,41 @@ async fn broadcast_ogg_flac_stream(
let ogg_bytes = Bytes::from(ogg_page);
total_ogg_bytes += ogg_bytes.len() as u64;
if let Err(e) = broadcast_tx.send(ogg_bytes.clone()) {
trace!("No active receivers for OGG-FLAC broadcast: {}", e);
} else {
trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead)", first_frame.len(), first_frame_samples, ogg_bytes.len());
// Measure broadcast interval for burst detection
let broadcast_interval = last_broadcast_time.elapsed().as_secs_f64();
last_broadcast_time = std::time::Instant::now();
broadcast_count += 1;
// Log if interval is unusual (too short = burst, too long = stall)
if broadcast_interval < 0.01 || broadcast_interval > 0.1 {
debug!(
"OGG: broadcast interval {:.3}s ({}ms) - frame_size={} bytes, samples={} (count={})",
broadcast_interval,
(broadcast_interval * 1000.0) as u32,
first_frame.len(),
first_frame_samples,
broadcast_count
);
}
// Periodic stats
if broadcast_count % 100 == 0 {
debug!(
"OGG: {} broadcasts sent, avg_interval={:.3}s, accumulator={} bytes",
broadcast_count,
last_broadcast_time.elapsed().as_secs_f64() / broadcast_count as f64,
flac_accumulator.len()
);
}
// Envoyer au broadcast
match broadcast_tx.send(ogg_bytes.clone()) {
Ok(n) => {
trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers", first_frame.len(), first_frame_samples, ogg_bytes.len(), n);
}
Err(e) => {
trace!("No active receivers for OGG-FLAC broadcast: {}", e);
}
}
}
}
@@ -897,13 +1026,17 @@ fn extract_sample_rate_from_streaminfo(flac_header: &[u8]) -> Result<u32, AudioE
// First metadata block should be STREAMINFO (type 0)
let block_type = flac_header[4] & 0x7F;
if block_type != 0 {
return Err(AudioError::ProcessingError("First block is not STREAMINFO".into()));
return Err(AudioError::ProcessingError(
"First block is not STREAMINFO".into(),
));
}
// STREAMINFO data starts at offset 8 (after magic + block header)
// Sample rate is at offset 10-12 of STREAMINFO data (bytes 18-20 of header)
if flac_header.len() < 21 {
return Err(AudioError::ProcessingError("STREAMINFO block truncated".into()));
return Err(AudioError::ProcessingError(
"STREAMINFO block truncated".into(),
));
}
// Sample rate: 20 bits starting at byte 10 of STREAMINFO
@@ -917,7 +1050,9 @@ fn extract_sample_rate_from_streaminfo(flac_header: &[u8]) -> Result<u32, AudioE
let sample_rate = (byte10 << 12) | (byte11 << 4) | (byte12 >> 4);
if sample_rate == 0 {
return Err(AudioError::ProcessingError("Invalid sample rate (0)".into()));
return Err(AudioError::ProcessingError(
"Invalid sample rate (0)".into(),
));
}
Ok(sample_rate)
@@ -929,12 +1064,15 @@ async fn read_flac_header(stream: &mut FlacEncodedStream) -> Result<Vec<u8>, Aud
let mut buffer = [0u8; 4];
// Read "fLaC" magic
stream.read_exact(&mut buffer).await.map_err(|e| {
AudioError::ProcessingError(format!("Failed to read FLAC magic: {}", e))
})?;
stream
.read_exact(&mut buffer)
.await
.map_err(|e| AudioError::ProcessingError(format!("Failed to read FLAC magic: {}", e)))?;
if &buffer != b"fLaC" {
return Err(AudioError::ProcessingError("Invalid FLAC stream: missing fLaC magic".into()));
return Err(AudioError::ProcessingError(
"Invalid FLAC stream: missing fLaC magic".into(),
));
}
header.extend_from_slice(&buffer);
@@ -948,7 +1086,8 @@ async fn read_flac_header(stream: &mut FlacEncodedStream) -> Result<Vec<u8>, Aud
})?;
let is_last = (block_header[0] & 0x80) != 0;
let block_length = u32::from_be_bytes([0, block_header[1], block_header[2], block_header[3]]) as usize;
let block_length =
u32::from_be_bytes([0, block_header[1], block_header[2], block_header[3]]) as usize;
header.extend_from_slice(&block_header);
@@ -984,11 +1123,14 @@ fn create_ogg_flac_identification(flac_header: &[u8]) -> Result<Vec<u8>, AudioEr
let first_block_type = flac_header[4] & 0x7F; // Remove last-metadata-block flag
if first_block_type != 0 {
return Err(AudioError::ProcessingError("First FLAC metadata block is not STREAMINFO".into()));
return Err(AudioError::ProcessingError(
"First FLAC metadata block is not STREAMINFO".into(),
));
}
// Extract block length (3 bytes big-endian after type byte)
let block_length = u32::from_be_bytes([0, flac_header[5], flac_header[6], flac_header[7]]) as usize;
let block_length =
u32::from_be_bytes([0, flac_header[5], flac_header[6], flac_header[7]]) as usize;
info!("STREAMINFO block_length = {} bytes", block_length);
@@ -1007,18 +1149,21 @@ fn create_ogg_flac_identification(flac_header: &[u8]) -> Result<Vec<u8>, AudioEr
// Extract just the STREAMINFO block (type + length + data)
let streaminfo = &flac_header[4..4 + streaminfo_size];
info!("Extracted STREAMINFO: {} bytes (type+length+data)", streaminfo.len());
info!(
"Extracted STREAMINFO: {} bytes (type+length+data)",
streaminfo.len()
);
let mut packet = Vec::new();
// OGG-FLAC identification header
packet.push(0x7F); // Byte 0: 0x7F
packet.extend_from_slice(b"FLAC"); // Bytes 1-4: "FLAC"
packet.push(0x01); // Byte 5: Major version
packet.push(0x00); // Byte 6: Minor version
packet.push(0x7F); // Byte 0: 0x7F
packet.extend_from_slice(b"FLAC"); // Bytes 1-4: "FLAC"
packet.push(0x01); // Byte 5: Major version
packet.push(0x00); // Byte 6: Minor version
packet.extend_from_slice(&1u16.to_be_bytes()); // Bytes 7-8: 1 header packet (Vorbis Comment)
packet.extend_from_slice(b"fLaC"); // Bytes 9-12: Native FLAC signature
packet.extend_from_slice(streaminfo); // Bytes 13+: STREAMINFO block only
packet.extend_from_slice(b"fLaC"); // Bytes 9-12: Native FLAC signature
packet.extend_from_slice(streaminfo); // Bytes 13+: STREAMINFO block only
Ok(packet)
}
@@ -1075,7 +1220,13 @@ impl OggPageWriter {
self.granule_position += samples;
}
fn create_page(&mut self, packet_data: &[u8], is_bos: bool, is_eos: bool, is_continuation: bool) -> Vec<u8> {
fn create_page(
&mut self,
packet_data: &[u8],
is_bos: bool,
is_eos: bool,
is_continuation: bool,
) -> Vec<u8> {
use std::io::Write;
let mut segments = Vec::new();
@@ -1117,7 +1268,8 @@ impl OggPageWriter {
page.write_all(&[header_type]).unwrap();
// Granule position
page.write_all(&self.granule_position.to_le_bytes()).unwrap();
page.write_all(&self.granule_position.to_le_bytes())
.unwrap();
// Stream serial number
page.write_all(&self.stream_serial.to_le_bytes()).unwrap();