diff --git a/docker/pipewire/10-virtual.conf b/docker/pipewire/10-virtual.conf index 665253ef7..4b3dfddcc 100644 --- a/docker/pipewire/10-virtual.conf +++ b/docker/pipewire/10-virtual.conf @@ -1,9 +1,10 @@ # Declarative virtual audio devices for the headless StreamLib container. # -# StreamLib's Linux audio is cpal -> ALSA (packages/audio), and the container -# bridges ALSA -> PipeWire via `pipewire-alsa`. This drop-in declares a virtual -# null sink (plus its recordable `.monitor` source) at PipeWire startup so the -# devices exist with no hardware, no `/dev/snd`, and no imperative `pactl` race. +# StreamLib reaches this daemon directly: its backend chain dlopen's +# libpipewire, falling back to libasound and then to the null backend. This +# drop-in declares a virtual null sink (plus its recordable `.monitor` source) +# at PipeWire startup so the devices exist with no hardware, no `/dev/snd`, and +# no imperative `pactl` race. # # Loaded from /etc/pipewire/pipewire.conf.d/ — keys are UNQUOTED, string values # ARE quoted; a syntax error silently drops the section. diff --git a/runtime/streamlib-engine/src/core/context/audio_clock.rs b/runtime/streamlib-engine/src/core/context/audio_clock.rs index 9e87a1780..62349d827 100644 --- a/runtime/streamlib-engine/src/core/context/audio_clock.rs +++ b/runtime/streamlib-engine/src/core/context/audio_clock.rs @@ -50,7 +50,11 @@ impl AudioClockConfig { /// Context passed to audio clock tick callbacks. #[derive(Debug, Clone, Copy)] pub struct AudioTickContext { - /// Machine monotonic timestamp in nanoseconds, the epoch a frame timestamp carries. + /// Machine monotonic timestamp in nanoseconds, the epoch a frame timestamp + /// carries. It is the wake time rather than the expiration the tick was + /// scheduled for, and a catch-up burst may hand every tick the same one, so + /// a block's instant derives from an anchor plus samples already delivered + /// — never from one tick's stamp. pub timestamp_ns: i64, /// Number of samples to produce this tick (per channel). pub samples_needed: usize, diff --git a/runtime/streamlib-engine/src/core/mod.rs b/runtime/streamlib-engine/src/core/mod.rs index 416115ba1..2bb84b628 100644 --- a/runtime/streamlib-engine/src/core/mod.rs +++ b/runtime/streamlib-engine/src/core/mod.rs @@ -40,7 +40,6 @@ pub mod processors; pub mod pubsub; pub mod rhi; pub mod runtime; -pub mod sync; pub mod texture; pub mod utils; // Linux-only: winit is a Linux-target engine dependency, and the window seam @@ -60,7 +59,6 @@ pub use graph_snapshot::*; pub use processors::*; pub use rhi::{GlContext, GlTextureBinding, NativeTextureHandle, RhiBackend, gl_constants}; pub use runtime::*; -pub use sync::*; pub use texture::*; pub use utils::*; diff --git a/runtime/streamlib-engine/src/core/observability/perception.rs b/runtime/streamlib-engine/src/core/observability/perception.rs index 579650969..1e27a7f9b 100644 --- a/runtime/streamlib-engine/src/core/observability/perception.rs +++ b/runtime/streamlib-engine/src/core/observability/perception.rs @@ -37,19 +37,6 @@ pub struct SampledFrame { pub timestamp_ns: i64, } -/// A sampled audio buffer for AI perception. -#[derive(Debug, Clone)] -pub struct SampledAudio { - /// Audio samples (f32 interleaved). - pub samples: Vec, - /// Sample rate in Hz. - pub sample_rate: u32, - /// Number of channels. - pub channels: u32, - /// Duration in milliseconds. - pub duration_ms: u32, -} - /// Current status of a processor. #[derive(Debug, Clone)] pub struct ProcessorStatus { @@ -68,9 +55,6 @@ pub trait AgentPerception: Send + Sync { /// Sample a video frame from a processor's output. fn sample_video(&self, id: &ProcessorId, config: SampleConfig) -> Option; - /// Sample audio from a processor's output. - fn sample_audio(&self, id: &ProcessorId, duration_ms: u32) -> Option; - /// Get current status of a processor. fn processor_status(&self, id: &ProcessorId) -> Option; diff --git a/runtime/streamlib-engine/src/core/sync.rs b/runtime/streamlib-engine/src/core/sync.rs deleted file mode 100644 index f0249b4bf..000000000 --- a/runtime/streamlib-engine/src/core/sync.rs +++ /dev/null @@ -1,104 +0,0 @@ -// Copyright (c) 2025 Jonathan Fontanez -// SPDX-License-Identifier: BUSL-1.1 - -pub const DEFAULT_SYNC_TOLERANCE_MS: f64 = 16.6; - -/// Action to take when audio and video streams are out of sync. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SyncAction { - NoAction, - DropVideoFrame, - DuplicateVideoFrame, -} - -#[inline] -pub fn timestamp_delta_ms(timestamp_a_ns: i64, timestamp_b_ns: i64) -> f64 { - let delta_ns = (timestamp_a_ns - timestamp_b_ns).abs(); - delta_ns as f64 / 1_000_000.0 -} - -#[inline] -pub fn are_synchronized(timestamp_a_ns: i64, timestamp_b_ns: i64, tolerance_ms: f64) -> bool { - timestamp_delta_ms(timestamp_a_ns, timestamp_b_ns) <= tolerance_ms -} - -#[inline] -pub fn sync_action( - video_timestamp_ns: i64, - audio_timestamp_ns: i64, - tolerance_ms: f64, -) -> SyncAction { - let drift_ns = video_timestamp_ns - audio_timestamp_ns; - let drift_ms = drift_ns as f64 / 1_000_000.0; - - if drift_ms.abs() <= tolerance_ms { - SyncAction::NoAction - } else if drift_ms > 0.0 { - SyncAction::DropVideoFrame - } else { - SyncAction::DuplicateVideoFrame - } -} - -#[inline] -pub fn sync_statistics( - video_timestamp_ns: i64, - audio_timestamp_ns: i64, - tolerance_ms: f64, -) -> (f64, bool) { - let drift_ns = video_timestamp_ns - audio_timestamp_ns; - let drift_ms = drift_ns as f64 / 1_000_000.0; - let is_synced = drift_ms.abs() <= tolerance_ms; - (drift_ms, is_synced) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_timestamp_delta() { - assert_eq!(timestamp_delta_ms(1_000_000_000, 1_000_000_000), 0.0); - assert_eq!(timestamp_delta_ms(1_000_000_000, 1_001_000_000), 1.0); - assert_eq!(timestamp_delta_ms(1_001_000_000, 1_000_000_000), 1.0); - let delta = timestamp_delta_ms(1_000_000_000, 1_016_600_000); - assert!((delta - 16.6).abs() < 0.01); - } - - #[test] - fn test_are_synchronized() { - assert!(are_synchronized(1_000_000_000, 1_010_000_000, 20.0)); - assert!(are_synchronized(1_000_000_000, 1_020_000_000, 20.0)); - assert!(!are_synchronized(1_000_000_000, 1_030_000_000, 20.0)); - } - - #[test] - fn test_sync_action() { - // NoAction within tolerance. - assert_eq!( - sync_action(1_000_000_000, 1_000_000_000, DEFAULT_SYNC_TOLERANCE_MS), - SyncAction::NoAction - ); - // Video ahead → drop. - assert_eq!( - sync_action(1_050_000_000, 1_000_000_000, DEFAULT_SYNC_TOLERANCE_MS), - SyncAction::DropVideoFrame - ); - // Video behind → duplicate. - assert_eq!( - sync_action(950_000_000, 1_000_000_000, DEFAULT_SYNC_TOLERANCE_MS), - SyncAction::DuplicateVideoFrame - ); - } - - #[test] - fn test_sync_statistics() { - let (drift_ms, is_synced) = - sync_statistics(1_020_000_000, 1_000_000_000, DEFAULT_SYNC_TOLERANCE_MS); - assert!((drift_ms - 20.0).abs() < 0.1); - assert!(!is_synced); - - let (_drift_ms, is_synced) = sync_statistics(1_020_000_000, 1_000_000_000, 25.0); - assert!(is_synced); - } -} diff --git a/runtime/streamlib-engine/src/iceoryx2/delivery_profile.rs b/runtime/streamlib-engine/src/iceoryx2/delivery_profile.rs index bc96c47b7..dde30ef30 100644 --- a/runtime/streamlib-engine/src/iceoryx2/delivery_profile.rs +++ b/runtime/streamlib-engine/src/iceoryx2/delivery_profile.rs @@ -37,12 +37,12 @@ pub enum DeliveryProfile { /// stale sample has no value once a fresher one exists. Latest, /// FIFO with a bounded backlog: read next in order, evict + count the - /// oldest under sustained overrun, deeper ring. Sample streams — audio, - /// encoded frames — where order matters but the producer must never block. + /// oldest under sustained overrun, deeper ring. Sample streams — encoded + /// frames — where order matters but the producer must never block. EverySample, /// Lossless FIFO: read next in order, the producer blocks rather than - /// drop, deeper ring. File writers, muxers, loggers where every sample - /// must be delivered. + /// drop, deeper ring. Audio, file writers, muxers, loggers where every + /// sample must be delivered. /// /// What this configures is the *publisher's* policy, and that is as far as /// it reaches today: the consumer's host mailbox diff --git a/runtime/streamlib-engine/src/iceoryx2/node.rs b/runtime/streamlib-engine/src/iceoryx2/node.rs index 6ed0907d5..65aece51d 100644 --- a/runtime/streamlib-engine/src/iceoryx2/node.rs +++ b/runtime/streamlib-engine/src/iceoryx2/node.rs @@ -104,11 +104,10 @@ impl Iceoryx2Node { /// [`crate::iceoryx2::delivery_profile_for_input_port`]. /// /// `enable_safe_overflow` derives from that same profile's overflow policy. - /// When `true` (the realtime default — `Overflow::DropOldest`), the subscriber - /// buffer auto-evicts the oldest sample on overflow and the publisher's - /// `send()` never blocks. When `false` (`Overflow::Block`, the `lossless` - /// profile), the producer blocks until the consumer drains a slot — reserve - /// for muxers / file writers that need every sample in order. + /// When `true` (`Overflow::DropOldest`), the subscriber buffer auto-evicts + /// the oldest sample on overflow and the publisher's `send()` never blocks. + /// When `false` (`Overflow::Block`, the `lossless` profile), the producer + /// blocks until the consumer drains a slot. pub fn open_or_create_service( &self, service_name: &str, diff --git a/runtime/streamlib-engine/src/iceoryx2/overflow.rs b/runtime/streamlib-engine/src/iceoryx2/overflow.rs index 8c9382f2a..210acb40c 100644 --- a/runtime/streamlib-engine/src/iceoryx2/overflow.rs +++ b/runtime/streamlib-engine/src/iceoryx2/overflow.rs @@ -20,16 +20,14 @@ use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum Overflow { - /// Buffer evicts oldest sample to make room; publisher never blocks. - /// The realtime-media default — producer represents real-world - /// time advancing and must not be made to wait. Glitches on the - /// consumer side beat freezing the whole pipeline. + /// Buffer evicts oldest sample to make room; publisher never blocks — a + /// producer that represents real-world time advancing must not be made to + /// wait, and glitches on the consumer side beat freezing the whole pipeline. #[default] DropOldest, - /// Producer blocks until the consumer drains a slot. Use only when - /// every sample must be delivered in order — file writers, muxers, - /// loggers — and the consumer's mailbox `buffer_size` is sized for - /// the expected hiccup envelope. + /// Producer blocks until the consumer drains a slot. Use only where every + /// sample must be delivered in order and the consumer's mailbox + /// `buffer_size` is sized for the expected hiccup envelope. Block, } diff --git a/runtime/streamlib-engine/src/lib.rs b/runtime/streamlib-engine/src/lib.rs index 19976b222..27dc0cf35 100644 --- a/runtime/streamlib-engine/src/lib.rs +++ b/runtime/streamlib-engine/src/lib.rs @@ -102,7 +102,6 @@ pub use core::{ ConnectionDefinition, // Processor traits (mode-specific) ContinuousProcessor, - DEFAULT_SYNC_TOLERANCE_MS, Error, GlContext, GlTextureBinding, @@ -128,13 +127,11 @@ pub use core::{ TexturePoolDescriptor, TextureUsages, TimeContext, - are_synchronized, gl_constants, // Port marker traits and helpers for compile-time safe connections input, media_clock::MediaClock, output, - timestamp_delta_ms, }; // GPU Backends - Metal and Vulkan @@ -288,7 +285,6 @@ pub mod sdk { pub use crate::core::prelude; pub use crate::core::rhi; pub use crate::core::runtime; - pub use crate::core::sync; pub use crate::core::texture; pub use crate::core::utils; diff --git a/runtime/streamlib-engine/src/linux/audio_clock.rs b/runtime/streamlib-engine/src/linux/audio_clock.rs index 40f733539..d82759e87 100644 --- a/runtime/streamlib-engine/src/linux/audio_clock.rs +++ b/runtime/streamlib-engine/src/linux/audio_clock.rs @@ -272,14 +272,6 @@ fn run_timerfd_loop( return Err(Error::Runtime(format!("timerfd read failed: {}", err))); } - // FIXME(audio-backend): one read stamps every tick in a catch-up burst, so - // N blocks of samples claim the same instant, and the value is the wake - // time rather than the expiration the timerfd was programmed for. Both - // follow from the timer's own absolute schedule. Deferred because a - // free-running clock's tick time is meaningless until a device paces it: - // the audio backend is OPEN (docs/plan/ARCHITECTURE.md §Media I/O), and it - // also owns the capture path's discarded driver stamp - // (cpal `InputCallbackInfo::timestamp().capture`). let timestamp_ns = MediaClock::now().as_nanos() as i64; if expirations > 1 { diff --git a/sdk/streamlib-sdk/src/lib.rs b/sdk/streamlib-sdk/src/lib.rs index 9565f71e8..146bd699b 100644 --- a/sdk/streamlib-sdk/src/lib.rs +++ b/sdk/streamlib-sdk/src/lib.rs @@ -94,7 +94,6 @@ pub mod sdk { #[cfg(target_os = "linux")] pub use streamlib_engine::core::processor_owned_window; - pub use streamlib_engine::core::sync; pub use streamlib_engine::core::texture; #[cfg(target_os = "linux")] pub use streamlib_engine::core::window_event_pump;