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

Nouveau broadcast stratégie

Browse Source
This commit is contained in:
Eric Coissac
2025-11-16 22:02:00 +01:00
parent 66416dafa8
commit 765070c4b0
4 changed files with 251 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
pmoaudio-ext/src/sinks/broadcast_pacing.rs
View File

@@ -40,85 +40,20 @@ impl BroadcastPacer {
} }
} }
/// Check timing and apply pacing /// Check timing and apply pacing - NO-OP VERSION
/// ///
/// This function: /// Pacing is now handled entirely by the expiration-based system in
/// 1. Detects TopZeroSync (audio_timestamp < 0.1 after >1s) and marks pending reset /// TimedBroadcast. This method is kept for backward compatibility
/// 2. On next chunk, resets timer with elapsed=0 guarantee /// but always returns Ok(()).
/// 3. Drops frames that are late (audio_ts < elapsed)
/// 4. Sleeps if too far ahead (lead_time > max_lead_time)
/// ///
/// # Returns /// # Returns
/// ///
/// - `Ok(())` if frame is on time or successfully paced /// - Always returns `Ok(())`
/// - `Err(SkipFrame)` if frame is too late and should be dropped
pub async fn check_and_pace(&mut self, audio_timestamp: f64) -> Result<(), SkipFrame> { pub async fn check_and_pace(&mut self, audio_timestamp: f64) -> Result<(), SkipFrame> {
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 1. DÉTECTION timestamp proche de 0 → marquer reset ║
// ║ Quand timestamp < 0.1s, c'est un nouveau morceau ║
// ╚═══════════════════════════════════════════════════════════════╝
if audio_timestamp < 0.1 && !self.pending_reset {
trace!( trace!(
"{} broadcaster: Timestamp near zero detected, will reset timer on next chunk", "{} broadcaster: check_and_pace called with audio_ts={:.3}s (no-op - pacing handled by TimedBroadcast)",
self.label
);
self.pending_reset = true;
}
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 2. RESET TIMER si pending ║
// ║ Le reset se fait AVANT le calcul d'elapsed pour garantir ║
// ║ elapsed=0 pour le premier chunk du nouveau morceau ║
// ╚═══════════════════════════════════════════════════════════════╝
let elapsed = if self.pending_reset {
self.start_time = Instant::now();
self.pending_reset = false;
trace!(
"{} broadcaster: Timer reset at audio_ts={:.3}s",
self.label, audio_timestamp self.label, audio_timestamp
); );
0.0 // Garantit elapsed=0 pour ce chunk
} else {
self.start_time.elapsed().as_secs_f64()
};
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 3. CALCUL DU LEAD TIME ║
// ║ lead_time > 0 : en avance (OK) ║
// ║ lead_time < 0 : en retard (SKIP) ║
// ╚═══════════════════════════════════════════════════════════════╝
let lead_time = audio_timestamp - elapsed;
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 4. DROP FRAMES EN RETARD ║
// ╚═══════════════════════════════════════════════════════════════╝
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);
}
// ╔═══════════════════════════════════════════════════════════════╗
// ║ 5. 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 {
trace!(
"{} 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(()) Ok(())
} }
} }

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

@@ -116,6 +116,8 @@ struct PcmChunk {
bytes: Vec<u8>, bytes: Vec<u8>,
/// Timestamp in seconds (from AudioSegment) /// Timestamp in seconds (from AudioSegment)
timestamp_sec: f64, timestamp_sec: f64,
/// Duration in seconds of this PCM chunk (samples / sample_rate)
duration_sec: f64,
} }
/// Snapshot of track metadata at a point in time. /// Snapshot of track metadata at a point in time.
@@ -259,6 +261,12 @@ enum FlacStreamState {
} }
/// Pure FLAC client stream (implements AsyncRead). /// Pure FLAC client stream (implements AsyncRead).
///
/// Each read pulls bytes out of a [`timed_broadcast`] receiver.
/// If the receiver reports [`TryRecvError::Lagged`] it means the underlying
/// queue expired packets before the client consumed them; we log the skip
/// and immediately keep draining so that a late client can resynchronise
/// with the latest epoch instead of stalling forever.
pub struct FlacClientStream { pub struct FlacClientStream {
rx: timed_broadcast::Receiver<Bytes>, rx: timed_broadcast::Receiver<Bytes>,
buffer: VecDeque<u8>, buffer: VecDeque<u8>,
@@ -369,6 +377,9 @@ impl Drop for FlacClientStream {
/// ///
/// This stream injects ICY metadata blocks at regular intervals, /// This stream injects ICY metadata blocks at regular intervals,
/// allowing clients to display "Now Playing" information. /// allowing clients to display "Now Playing" information.
/// As with [`FlacClientStream`], hitting [`TryRecvError::Lagged`]
/// simply indicates that the timed broadcast discarded a stale chunk;
/// the client resumes with fresh data to avoid wedging the HTTP response.
pub struct IcyClientStream { pub struct IcyClientStream {
rx: timed_broadcast::Receiver<Bytes>, rx: timed_broadcast::Receiver<Bytes>,
metadata: Arc<RwLock<MetadataSnapshot>>, metadata: Arc<RwLock<MetadataSnapshot>>,
@@ -614,11 +625,13 @@ impl StreamingFlacSinkLogic {
.take() .take()
.ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?; .ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?;
// Create shared timestamp for pacing // Create shared timestamp and duration for pacing
let current_timestamp = Arc::new(RwLock::new(0.0f64)); let current_timestamp = Arc::new(RwLock::new(0.0f64));
let current_duration = Arc::new(RwLock::new(0.0f64));
// Create ByteStreamReader for the encoder // Create ByteStreamReader for the encoder
let pcm_reader = ByteStreamReader::new(pcm_rx, current_timestamp.clone()); let pcm_reader =
ByteStreamReader::new(pcm_rx, current_timestamp.clone(), current_duration.clone());
// Create PCM format // Create PCM format
let pcm_format = PcmFormat { let pcm_format = PcmFormat {
@@ -636,7 +649,7 @@ impl StreamingFlacSinkLogic {
debug!("FLAC encoder initialized successfully"); debug!("FLAC encoder initialized successfully");
// Spawn broadcaster task with timestamp for pacing // Spawn broadcaster task with timestamp and duration for pacing
let flac_broadcast = self.flac_broadcast.clone(); let flac_broadcast = self.flac_broadcast.clone();
let flac_header = self.flac_header.clone(); let flac_header = self.flac_header.clone();
let max_lead = self.broadcast_max_lead_time; let max_lead = self.broadcast_max_lead_time;
@@ -646,7 +659,9 @@ impl StreamingFlacSinkLogic {
flac_broadcast, flac_broadcast,
flac_header, flac_header,
current_timestamp, current_timestamp,
current_duration,
max_lead, max_lead,
sample_rate,
) )
.await .await
{ {
@@ -746,17 +761,25 @@ impl NodeLogic for StreamingFlacSinkLogic {
// Convert chunk to PCM bytes // Convert chunk to PCM bytes
let pcm_bytes = chunk_to_pcm_bytes(&chunk, self.bits_per_sample)?; let pcm_bytes = chunk_to_pcm_bytes(&chunk, self.bits_per_sample)?;
// Calculate exact duration from samples and sample rate
let sample_rate = self
.sample_rate
.expect("sample_rate should be initialized");
let duration_sec = chunk.len() as f64 / sample_rate as f64;
trace!( trace!(
"Sending PCM chunk: {} bytes, {} samples @ {:.2}s", "Sending PCM chunk: {} bytes, {} samples @ {:.2}s (duration={:.3}s)",
pcm_bytes.len(), pcm_bytes.len(),
chunk.len(), chunk.len(),
seg.timestamp_sec seg.timestamp_sec,
duration_sec
); );
// Send to FLAC encoder with timestamp // Send to FLAC encoder with timestamp and duration
let pcm_chunk = PcmChunk { let pcm_chunk = PcmChunk {
bytes: pcm_bytes, bytes: pcm_bytes,
timestamp_sec: seg.timestamp_sec, timestamp_sec: seg.timestamp_sec,
duration_sec,
}; };
let send_start = std::time::Instant::now(); let send_start = std::time::Instant::now();
if let Err(e) = self.pcm_tx.send(pcm_chunk).await { if let Err(e) = self.pcm_tx.send(pcm_chunk).await {
@@ -786,11 +809,6 @@ impl NodeLogic for StreamingFlacSinkLogic {
break; break;
} }
SyncMarker::TopZeroSync => {
self.flac_broadcast.mark_top_zero();
trace!("TopZeroSync propagated to FLAC broadcast");
}
_ => { _ => {
trace!("Received other sync marker"); trace!("Received other sync marker");
} }
@@ -826,7 +844,9 @@ async fn broadcast_flac_stream(
broadcast_tx: timed_broadcast::Sender<Bytes>, broadcast_tx: timed_broadcast::Sender<Bytes>,
header_cache: Arc<RwLock<Option<Bytes>>>, header_cache: Arc<RwLock<Option<Bytes>>>,
current_timestamp: Arc<RwLock<f64>>, current_timestamp: Arc<RwLock<f64>>,
current_duration: Arc<RwLock<f64>>,
broadcast_max_lead_time: f64, broadcast_max_lead_time: f64,
sample_rate: u32,
) -> Result<(), AudioError> { ) -> Result<(), AudioError> {
trace!( trace!(
"Broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)", "Broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)",
@@ -847,6 +867,8 @@ async fn broadcast_flac_stream(
let mut broadcast_count = 0u64; let mut broadcast_count = 0u64;
let mut total_read_time = 0.0f64; let mut total_read_time = 0.0f64;
let mut read_count = 0u64; let mut read_count = 0u64;
let mut encoded_samples = 0u64;
let sample_rate_f64 = sample_rate as f64;
loop { loop {
let read_start = std::time::Instant::now(); let read_start = std::time::Instant::now();
@@ -856,7 +878,12 @@ async fn broadcast_flac_stream(
if !accumulator.is_empty() { if !accumulator.is_empty() {
let bytes = Bytes::from(std::mem::take(&mut accumulator)); let bytes = Bytes::from(std::mem::take(&mut accumulator));
let audio_ts = *current_timestamp.read().await; let audio_ts = *current_timestamp.read().await;
if broadcast_tx.send(bytes.clone(), audio_ts).await.is_err() { let segment_dur = *current_duration.read().await;
if broadcast_tx
.send(bytes.clone(), audio_ts, segment_dur)
.await
.is_err()
{
trace!("Broadcast closed before sending final FLAC data"); trace!("Broadcast closed before sending final FLAC data");
break; break;
} }
@@ -897,19 +924,20 @@ async fn broadcast_flac_stream(
n n
); );
// Find where to split: position of last sync code (start of last incomplete frame) // Locate complete frames and total samples they represent
// Everything before this position contains only complete frames let (boundary, total_samples) =
let boundary = flac_frame_utils::find_complete_frames_boundary(&accumulator); flac_frame_utils::find_complete_frames_with_samples(&accumulator);
trace!( trace!(
"Buffer state: accumulator={} bytes, boundary={} bytes, will_send={}", "Buffer state: accumulator={} bytes, boundary={} bytes, total_samples={}, will_send={}",
accumulator.len(), accumulator.len(),
boundary, boundary,
boundary >= 1024 total_samples,
boundary >= 1024 && total_samples > 0
); );
// Only broadcast if we have at least one complete frame (1KB minimum to avoid excessive small sends) // Only broadcast if we have at least one complete frame (keep 1KB minimum to avoid tiny sends)
if boundary >= 1024 { if boundary >= 1024 && total_samples > 0 {
// ╔═══════════════════════════════════════════════════════════════╗ // ╔═══════════════════════════════════════════════════════════════╗
// ║ BACKPRESSURE INTELLIGENTE BASÉE SUR LE TIMING ║ // ║ BACKPRESSURE INTELLIGENTE BASÉE SUR LE TIMING ║
// ║ ║ // ║ ║
@@ -921,13 +949,17 @@ async fn broadcast_flac_stream(
// ║ Cela crée la backpressure vers TimerBufferNode tout en ║ // ║ Cela crée la backpressure vers TimerBufferNode tout en ║
// ║ permettant de dropper les chunks vraiment périmés. ║ // ║ permettant de dropper les chunks vraiment périmés. ║
// ╚═══════════════════════════════════════════════════════════════╝ // ╚═══════════════════════════════════════════════════════════════╝
let audio_timestamp = *current_timestamp.read().await; let frame_start_samples = encoded_samples;
encoded_samples = encoded_samples.saturating_add(total_samples);
let audio_timestamp = frame_start_samples as f64 / sample_rate_f64;
let segment_duration = total_samples as f64 / sample_rate_f64;
if stats_last_log.elapsed() >= Duration::from_secs(1) { if stats_last_log.elapsed() >= Duration::from_secs(1) {
trace!( trace!(
"Broadcaster pacing snapshot: audio_ts={:.3}s buffer_bytes={}", "Broadcaster pacing snapshot: audio_ts={:.3}s buffer_bytes={} samples={} ",
audio_timestamp, audio_timestamp,
accumulator.len() accumulator.len(),
total_samples
); );
stats_last_log = std::time::Instant::now(); stats_last_log = std::time::Instant::now();
} }
@@ -939,6 +971,13 @@ async fn broadcast_flac_stream(
continue; continue;
} }
if let Ok(mut ts) = current_timestamp.try_write() {
*ts = audio_timestamp;
}
if let Ok(mut dur) = current_duration.try_write() {
*dur = segment_duration;
}
// Split at boundary to avoid copying - extract prefix, keep suffix // Split at boundary to avoid copying - extract prefix, keep suffix
let remaining = accumulator.split_off(boundary); let remaining = accumulator.split_off(boundary);
let to_send = std::mem::replace(&mut accumulator, remaining); let to_send = std::mem::replace(&mut accumulator, remaining);
@@ -973,17 +1012,24 @@ async fn broadcast_flac_stream(
if !header_captured && bytes.len() >= 4 && &bytes[0..4] == b"fLaC" { if !header_captured && bytes.len() >= 4 && &bytes[0..4] == b"fLaC" {
*header_cache.write().await = Some(bytes.clone()); *header_cache.write().await = Some(bytes.clone());
header_captured = true; header_captured = true;
trace!("FLAC header captured ({} bytes)", bytes.len()); trace!("FLAC header captured ({} bytes), not broadcasting", bytes.len());
// Skip broadcasting the header: clients prepend it locally on subscribe
continue;
} }
let num_receivers = broadcast_tx.receiver_count(); let num_receivers = broadcast_tx.receiver_count();
match broadcast_tx.send(bytes.clone(), audio_timestamp).await { match broadcast_tx
.send(bytes.clone(), audio_timestamp, segment_duration)
.await
{
Ok(_) => { Ok(_) => {
if num_receivers > 0 { if num_receivers > 0 {
trace!( trace!(
"Broadcasted {} bytes to {} receivers", "Broadcasted {} bytes to {} receivers (ts={:.3}s, dur={:.3}s)",
bytes.len(), bytes.len(),
num_receivers num_receivers,
audio_timestamp,
segment_duration
); );
} }
} }
@@ -1237,15 +1283,22 @@ struct ByteStreamReader {
finished: bool, finished: bool,
/// Shared timestamp for broadcaster pacing /// Shared timestamp for broadcaster pacing
current_timestamp: Arc<RwLock<f64>>, current_timestamp: Arc<RwLock<f64>>,
/// Shared duration for broadcaster pacing
current_duration: Arc<RwLock<f64>>,
} }
impl ByteStreamReader { impl ByteStreamReader {
fn new(rx: mpsc::Receiver<PcmChunk>, current_timestamp: Arc<RwLock<f64>>) -> Self { fn new(
rx: mpsc::Receiver<PcmChunk>,
current_timestamp: Arc<RwLock<f64>>,
current_duration: Arc<RwLock<f64>>,
) -> Self {
Self { Self {
rx, rx,
buffer: VecDeque::new(), buffer: VecDeque::new(),
finished: false, finished: false,
current_timestamp, current_timestamp,
current_duration,
} }
} }
} }
@@ -1278,10 +1331,13 @@ impl AsyncRead for ByteStreamReader {
if chunk.bytes.is_empty() { if chunk.bytes.is_empty() {
continue; continue;
} }
// Update shared timestamp for broadcaster pacing // Update shared timestamp and duration for broadcaster pacing
if let Ok(mut ts) = self.current_timestamp.try_write() { if let Ok(mut ts) = self.current_timestamp.try_write() {
*ts = chunk.timestamp_sec; *ts = chunk.timestamp_sec;
} }
if let Ok(mut dur) = self.current_duration.try_write() {
*dur = chunk.duration_sec;
}
self.buffer.extend(chunk.bytes); self.buffer.extend(chunk.bytes);
} }
Poll::Ready(None) => { Poll::Ready(None) => {

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

@@ -92,6 +92,8 @@ struct PcmChunk {
bytes: Vec<u8>, bytes: Vec<u8>,
/// Timestamp in seconds (from AudioSegment) /// Timestamp in seconds (from AudioSegment)
timestamp_sec: f64, timestamp_sec: f64,
/// Duration in seconds of this PCM chunk (samples / sample_rate)
duration_sec: f64,
} }
/// Snapshot of track metadata (reuse from streaming_flac_sink) /// Snapshot of track metadata (reuse from streaming_flac_sink)
@@ -301,11 +303,13 @@ impl StreamingOggFlacSinkLogic {
.take() .take()
.ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?; .ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?;
// Create shared timestamp for pacing // Create shared timestamp and duration for pacing
let current_timestamp = Arc::new(RwLock::new(0.0f64)); let current_timestamp = Arc::new(RwLock::new(0.0f64));
let current_duration = Arc::new(RwLock::new(0.0f64));
// Create ByteStreamReader for the encoder // Create ByteStreamReader for the encoder
let pcm_reader = ByteStreamReader::new(pcm_rx, current_timestamp.clone()); let pcm_reader =
ByteStreamReader::new(pcm_rx, current_timestamp.clone(), current_duration.clone());
// Create PCM format // Create PCM format
let pcm_format = PcmFormat { let pcm_format = PcmFormat {
@@ -323,7 +327,7 @@ impl StreamingOggFlacSinkLogic {
debug!("OGG-FLAC encoder initialized successfully"); debug!("OGG-FLAC encoder initialized successfully");
// Spawn OGG wrapper + broadcaster task with timestamp for pacing // Spawn OGG wrapper + broadcaster task with timestamp and duration for pacing
let ogg_broadcast = self.ogg_broadcast.clone(); let ogg_broadcast = self.ogg_broadcast.clone();
let ogg_header = self.ogg_header.clone(); let ogg_header = self.ogg_header.clone();
let max_lead = self.broadcast_max_lead_time; let max_lead = self.broadcast_max_lead_time;
@@ -333,6 +337,7 @@ impl StreamingOggFlacSinkLogic {
ogg_broadcast, ogg_broadcast,
ogg_header, ogg_header,
current_timestamp, current_timestamp,
current_duration,
max_lead, max_lead,
) )
.await .await
@@ -432,17 +437,23 @@ impl NodeLogic for StreamingOggFlacSinkLogic {
// Convert chunk to PCM bytes // Convert chunk to PCM bytes
let pcm_bytes = chunk_to_pcm_bytes(&chunk, self.bits_per_sample)?; let pcm_bytes = chunk_to_pcm_bytes(&chunk, self.bits_per_sample)?;
// Calculate exact duration from samples and sample rate
let sample_rate = self.sample_rate.expect("sample_rate should be initialized");
let duration_sec = chunk.len() as f64 / sample_rate as f64;
trace!( trace!(
"Sending PCM chunk: {} bytes, {} samples @ {:.2}s", "Sending PCM chunk: {} bytes, {} samples @ {:.2}s (duration={:.3}s)",
pcm_bytes.len(), pcm_bytes.len(),
chunk.len(), chunk.len(),
seg.timestamp_sec seg.timestamp_sec,
duration_sec
); );
// Send to FLAC encoder with timestamp // Send to FLAC encoder with timestamp and duration
let pcm_chunk = PcmChunk { let pcm_chunk = PcmChunk {
bytes: pcm_bytes, bytes: pcm_bytes,
timestamp_sec: seg.timestamp_sec, timestamp_sec: seg.timestamp_sec,
duration_sec,
}; };
if let Err(e) = self.pcm_tx.send(pcm_chunk).await { if let Err(e) = self.pcm_tx.send(pcm_chunk).await {
warn!("Failed to send PCM data to encoder: {}", e); warn!("Failed to send PCM data to encoder: {}", e);
@@ -626,15 +637,22 @@ struct ByteStreamReader {
finished: bool, finished: bool,
/// Shared timestamp for broadcaster pacing /// Shared timestamp for broadcaster pacing
current_timestamp: Arc<RwLock<f64>>, current_timestamp: Arc<RwLock<f64>>,
/// Shared duration for broadcaster pacing
current_duration: Arc<RwLock<f64>>,
} }
impl ByteStreamReader { impl ByteStreamReader {
fn new(rx: mpsc::Receiver<PcmChunk>, current_timestamp: Arc<RwLock<f64>>) -> Self { fn new(
rx: mpsc::Receiver<PcmChunk>,
current_timestamp: Arc<RwLock<f64>>,
current_duration: Arc<RwLock<f64>>,
) -> Self {
Self { Self {
rx, rx,
buffer: VecDeque::new(), buffer: VecDeque::new(),
finished: false, finished: false,
current_timestamp, current_timestamp,
current_duration,
} }
} }
} }
@@ -667,10 +685,13 @@ impl AsyncRead for ByteStreamReader {
if chunk.bytes.is_empty() { if chunk.bytes.is_empty() {
continue; continue;
} }
// Update shared timestamp for broadcaster pacing // Update shared timestamp and duration for broadcaster pacing
if let Ok(mut ts) = self.current_timestamp.try_write() { if let Ok(mut ts) = self.current_timestamp.try_write() {
*ts = chunk.timestamp_sec; *ts = chunk.timestamp_sec;
} }
if let Ok(mut dur) = self.current_duration.try_write() {
*dur = chunk.duration_sec;
}
self.buffer.extend(chunk.bytes); self.buffer.extend(chunk.bytes);
} }
Poll::Ready(None) => { Poll::Ready(None) => {
@@ -784,6 +805,7 @@ async fn broadcast_ogg_flac_stream(
broadcast_tx: timed_broadcast::Sender<Bytes>, broadcast_tx: timed_broadcast::Sender<Bytes>,
header_cache: Arc<RwLock<Option<Bytes>>>, header_cache: Arc<RwLock<Option<Bytes>>>,
current_timestamp: Arc<RwLock<f64>>, current_timestamp: Arc<RwLock<f64>>,
current_duration: Arc<RwLock<f64>>,
broadcast_max_lead_time: f64, broadcast_max_lead_time: f64,
) -> Result<(), AudioError> { ) -> Result<(), AudioError> {
trace!( trace!(
@@ -840,13 +862,21 @@ async fn broadcast_ogg_flac_stream(
bos_bytes.len() + comment_bytes.len() bos_bytes.len() + comment_bytes.len()
); );
// Broadcast header // Broadcast header (BOS and comment are metadata, not audio, so duration=0.0)
if broadcast_tx.send(bos_bytes.clone(), 0.0).await.is_err() { if broadcast_tx
.send(bos_bytes.clone(), 0.0, 0.0)
.await
.is_err()
{
trace!("No receivers for BOS page, terminating broadcast"); trace!("No receivers for BOS page, terminating broadcast");
return Ok(()); return Ok(());
} }
total_ogg_bytes += comment_bytes.len() as u64; total_ogg_bytes += comment_bytes.len() as u64;
if broadcast_tx.send(comment_bytes.clone(), 0.0).await.is_err() { if broadcast_tx
.send(comment_bytes.clone(), 0.0, 0.0)
.await
.is_err()
{
trace!("No receivers for comment page, terminating broadcast"); trace!("No receivers for comment page, terminating broadcast");
return Ok(()); return Ok(());
} }
@@ -867,7 +897,12 @@ async fn broadcast_ogg_flac_stream(
let eos_bytes = Bytes::from(eos_page); let eos_bytes = Bytes::from(eos_page);
total_ogg_bytes += eos_bytes.len() as u64; total_ogg_bytes += eos_bytes.len() as u64;
let eos_ts = *current_timestamp.read().await; let eos_ts = *current_timestamp.read().await;
if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() { let eos_dur = *current_duration.read().await;
if broadcast_tx
.send(eos_bytes.clone(), eos_ts, eos_dur)
.await
.is_err()
{
trace!("Broadcast closed before sending final EOS page"); trace!("Broadcast closed before sending final EOS page");
break; break;
} }
@@ -876,12 +911,16 @@ async fn broadcast_ogg_flac_stream(
flac_accumulator.len() flac_accumulator.len()
); );
} else { } else {
// Send empty EOS page // Send empty EOS page (metadata page, duration=0.0)
let eos_page = ogg_writer.create_page(&[], false, true, false); let eos_page = ogg_writer.create_page(&[], false, true, false);
let eos_bytes = Bytes::from(eos_page); let eos_bytes = Bytes::from(eos_page);
total_ogg_bytes += eos_bytes.len() as u64; total_ogg_bytes += eos_bytes.len() as u64;
let eos_ts = *current_timestamp.read().await; let eos_ts = *current_timestamp.read().await;
if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() { if broadcast_tx
.send(eos_bytes.clone(), eos_ts, 0.0)
.await
.is_err()
{
trace!("Broadcast closed before sending empty EOS page"); trace!("Broadcast closed before sending empty EOS page");
break; break;
} }
@@ -1023,9 +1062,13 @@ async fn broadcast_ogg_flac_stream(
} }
// Envoyer au broadcast // Envoyer au broadcast
match broadcast_tx.send(ogg_bytes.clone(), audio_timestamp).await { let segment_duration = *current_duration.read().await;
match broadcast_tx
.send(ogg_bytes.clone(), audio_timestamp, segment_duration)
.await
{
Ok(n) => { 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); trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers (ts={:.3}s, dur={:.3}s)", first_frame.len(), first_frame_samples, ogg_bytes.len(), n, audio_timestamp, segment_duration);
} }
Err(_) => { Err(_) => {
trace!("No active receivers for OGG-FLAC broadcast, terminating loop"); trace!("No active receivers for OGG-FLAC broadcast, terminating loop");

124
pmoaudio-ext/src/sinks/timed_broadcast.rs
View File

@@ -15,7 +15,7 @@ use std::{
}; };
use tokio::sync::Notify; use tokio::sync::Notify;
use tracing::{trace, warn}; use tracing::{trace, info, warn};
/// Paquet diffusé contenant la charge utile + méta timing. /// Paquet diffusé contenant la charge utile + méta timing.
#[derive(Clone)] #[derive(Clone)]
@@ -40,8 +40,17 @@ impl<T> fmt::Debug for TimedPacket<T> {
/// Erreur remontée par `Receiver::try_recv`. /// Erreur remontée par `Receiver::try_recv`.
#[derive(Debug)] #[derive(Debug)]
pub enum TryRecvError { pub enum TryRecvError {
/// Aucun paquet n'est disponible pour le moment.
Empty, Empty,
/// Le receiver est en retard : le champ contient combien de paquets ont expiré
/// ou ont déjà été consommés par les autres abonnés.
///
/// Ce cas survient lorsque `purge_expired()` avance `head_seq` et que ce
/// `Receiver` réclamait encore l'un des numéros supprimés. Le client doit
/// donc ignorer les données perdues et se resynchroniser sur les paquets
/// courants.
Lagged(u64), Lagged(u64),
/// Le channel est fermé et plus aucun paquet n'est disponible.
Closed, Closed,
} }
@@ -71,7 +80,10 @@ struct State<T> {
closed: bool, closed: bool,
epoch: u64, epoch: u64,
epoch_start: Instant, epoch_start: Instant,
last_segment_end: Option<Instant>,
cursors: Vec<Weak<ReceiverCursor>>, cursors: Vec<Weak<ReceiverCursor>>,
initialized: bool,
saw_positive_timestamp: bool,
} }
impl<T> State<T> { impl<T> State<T> {
@@ -83,14 +95,19 @@ impl<T> State<T> {
closed: false, closed: false,
epoch: 0, epoch: 0,
epoch_start, epoch_start,
last_segment_end: None,
cursors: Vec::new(), cursors: Vec::new(),
initialized: false,
saw_positive_timestamp: false,
} }
} }
fn purge_expired(&mut self, now: Instant) -> bool { fn purge_expired(&mut self) -> bool {
let mut purged = 0u64; let mut purged = 0u64;
while let Some(entry) = self.buffer.front() { while let Some(entry) = self.buffer.front() {
if entry.expires_at <= now { if entry.expires_at <= Instant::now() {
let delta=Instant::now()-entry.expires_at;
info!("TimedBroadcast: purging expired packet (epoch={},delta={})",entry.epoch,delta.as_millis());
self.buffer.pop_front(); self.buffer.pop_front();
self.head_seq += 1; self.head_seq += 1;
purged += 1; purged += 1;
@@ -220,7 +237,11 @@ impl<T> Clone for Sender<T> {
impl<T> Sender<T> { impl<T> Sender<T> {
/// Diffuse un paquet. Bloque si la capacité est atteinte avec des paquets non périmés. /// Diffuse un paquet. Bloque si la capacité est atteinte avec des paquets non périmés.
pub async fn send(&self, payload: T, audio_timestamp: f64) -> Result<usize, SendError<T>> ///
/// Le TTL de chaque paquet est calculé à partir du `epoch_start` courant et du
/// `audio_timestamp` fournis, ce qui signifie qu’un receiver en retard finira
/// par recevoir un [`TryRecvError::Lagged`] lorsque `expires_at` est dépassé.
pub async fn send(&self, payload: T, audio_timestamp: f64, segment_duration: f64) -> Result<usize, SendError<T>>
where where
T: Clone, T: Clone,
{ {
@@ -228,7 +249,6 @@ impl<T> Sender<T> {
loop { loop {
let mut wait_deadline = None; let mut wait_deadline = None;
{ {
let now = Instant::now();
let mut state = self let mut state = self
.inner .inner
.state .state
@@ -239,17 +259,64 @@ impl<T> Sender<T> {
return Err(SendError(payload.expect("payload already consumed"))); return Err(SendError(payload.expect("payload already consumed")));
} }
if state.purge_expired(now) { // Détecter si c'est un TopZero
self.inner.space_notify.notify_waiters(); let is_top_zero = audio_timestamp == 0.0;
// Gérer l'initialisation ET le TopZero ensemble
if !state.initialized {
let now = Instant::now();
if is_top_zero {
// Premier paquet = TopZero → epoch commence à 0
state.epoch_start = now;
state.epoch = 0;
info!("TimedBroadcast: initialized with TopZero (epoch=0)");
} else {
// Premier paquet avec ts > 0 → calculer epoch_start rétroactif
let offset = Duration::from_secs_f64(audio_timestamp);
state.epoch_start = now.checked_sub(offset).unwrap_or(now);
state.epoch = 0;
info!(
"TimedBroadcast: initialized with ts={:.3}s (epoch=0)",
audio_timestamp
);
}
state.initialized = true;
state.saw_positive_timestamp = !is_top_zero;
} else if is_top_zero {
// TopZero sur un channel déjà initialisé
let allow_reset = state.saw_positive_timestamp;
if allow_reset {
let now = Instant::now();
state.epoch_start = state
.last_segment_end
.map(|end| end.max(now))
.unwrap_or(now);
state.epoch = state.epoch.wrapping_add(1);
state.saw_positive_timestamp = false; // Reset pour le prochain cycle
info!(
"TimedBroadcast: TopZero detected, new epoch={} (had_last_segment={})",
state.epoch,
state.last_segment_end.is_some()
);
} else {
warn!(
"TimedBroadcast: Ignoring duplicate TopZero (epoch={})",
state.epoch
);
}
} else if audio_timestamp > 0.0 {
state.saw_positive_timestamp = true;
} }
if state.prune_consumed() { let consumed = state.prune_consumed();
let expired = state.purge_expired();
if consumed || expired {
self.inner.space_notify.notify_waiters(); self.inner.space_notify.notify_waiters();
} }
if state.buffer.len() < self.inner.capacity { if state.buffer.len() < self.inner.capacity {
let audio_offset = Duration::from_secs_f64(audio_timestamp.max(0.0)); // Le paquet expire à la fin de son segment audio
let expires_at = state.epoch_start + audio_offset; let expires_at = state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration);
let entry = Entry { let entry = Entry {
seq: state.next_seq, seq: state.next_seq,
expires_at, expires_at,
@@ -259,6 +326,10 @@ impl<T> Sender<T> {
}; };
state.next_seq += 1; state.next_seq += 1;
state.buffer.push_back(entry); state.buffer.push_back(entry);
// Stocker la fin de ce segment pour la continuité temporelle
state.last_segment_end = Some(expires_at);
let receivers = self.inner.receiver_count.load(Ordering::SeqCst); let receivers = self.inner.receiver_count.load(Ordering::SeqCst);
drop(state); drop(state);
self.inner.data_notify.notify_waiters(); self.inner.data_notify.notify_waiters();
@@ -304,21 +375,15 @@ impl<T> Sender<T> {
} }
} }
/// Marque un TopZero : incrémente l'epoch pour les paquets suivants. /// Marque un TopZero : DEPRECATED - no-op pour compatibilité.
/// ///
/// Reset le timer epoch_start sans effacer le buffer. Les paquets /// Le vrai TopZero est maintenant détecté automatiquement dans send()
/// du morceau précédent continueront à être distribués naturellement. /// quand audio_timestamp == 0.0. Cette méthode est conservée pour
/// Cela évite de perdre les dernières frames FLAC à la transition entre morceaux. /// compatibilité avec le code existant mais ne fait rien.
pub fn mark_top_zero(&self) { pub fn mark_top_zero(&self) {
let mut state = self trace!("TimedBroadcast: mark_top_zero() called but ignored (auto-detection active)");
.inner // No-op - TopZero est maintenant détecté automatiquement dans send()
.state // quand audio_timestamp == 0.0
.lock()
.expect("timed broadcast mutex poisoned");
state.epoch = state.epoch.wrapping_add(1);
state.epoch_start = Instant::now();
// Ne PAS effacer le buffer - laisser les paquets du morceau précédent
// se vider naturellement pour éviter de perdre les dernières frames
} }
/// Nombre actuel de receivers abonnés. /// Nombre actuel de receivers abonnés.
@@ -341,6 +406,10 @@ impl<T> Drop for Sender<T> {
} }
/// Receiver côté consommateur. /// Receiver côté consommateur.
///
/// Chaque receiver garde son propre curseur `next_seq`. Si le producteur
/// recycle un paquet via `purge_expired()` avant que ce curseur ne l’ait lu,
/// la prochaine tentative de lecture retournera [`TryRecvError::Lagged`].
pub struct Receiver<T> { pub struct Receiver<T> {
inner: Arc<Inner<T>>, inner: Arc<Inner<T>>,
next_seq: u64, next_seq: u64,
@@ -366,8 +435,7 @@ where
return Err(TryRecvError::Closed); return Err(TryRecvError::Closed);
} }
let now = Instant::now(); if state.purge_expired() {
if state.purge_expired(now) {
self.inner.space_notify.notify_waiters(); self.inner.space_notify.notify_waiters();
} }
@@ -401,6 +469,12 @@ where
} }
/// Version synchrone utilisée dans `poll_read`. /// Version synchrone utilisée dans `poll_read`.
///
/// # Erreurs
///
/// * [`TryRecvError::Lagged`] — des paquets ont expiré avant d'être consommés.
/// * [`TryRecvError::Empty`] — la file est vide pour l'instant.
/// * [`TryRecvError::Closed`] — plus aucun paquet n'arrivera.
pub fn try_recv(&mut self) -> Result<TimedPacket<T>, TryRecvError> { pub fn try_recv(&mut self) -> Result<TimedPacket<T>, TryRecvError> {
self.poll_entry() self.poll_entry()
} }