mirror of
https://github.com/openai/codex.git
synced 2026-10-21 09:54:56 +00:00
Add opt-in local audio devices to the voice helper (#43079)
## What changed - Add `openDevices` and `setAudioControls` to the helper protocol. Open the default microphone and speaker after runtime initialization and transport negotiation, initially muted and suppressed. - Use CPAL on macOS, GNU Linux, and MSVC Windows, with native audio dependencies confined to `codex-voice-host`. Add ALSA development dependencies to Linux CI. - Pack callback samples into preallocated, bounded queues and service devices every 5 ms. Reject incompatible callback configurations, invalidate queued audio on control transitions, and reject capture backlog after unmute using device timestamps. - Handle recoverable stream underruns and overruns without terminating the helper; terminate on device failure, active audio queue overflow, or stale queued audio. Device opening only establishes local device readiness. Capture and rendered audio references are drained locally; encoding, decoding, and peer audio are not connected by this change. ## Testing Add unit tests for mute boundaries, stale playback invalidation, sample bounds, partial-block packing and timestamps, callback size and timing limits, startup silence, queue overflow, and recoverable stream errors. GitOrigin-RevId: 108aea3d411ef6c12908fe5fb67aaf9cddf4475f
This commit is contained in:
Benjamin Carlsson
committed by
copyberry
copyberry
parent
51c97f3a67
commit
19b62211d9
@@ -82,7 +82,7 @@ jobs:
|
||||
set -euo pipefail
|
||||
if command -v apt-get >/dev/null 2>&1; then
|
||||
sudo apt-get update -y
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends build-essential pkg-config libcap-dev bubblewrap
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends build-essential pkg-config libcap-dev libasound2-dev bubblewrap
|
||||
fi
|
||||
|
||||
- uses: dtolnay/rust-toolchain@e081816240890017053eacbb1bdf337761dc5582 # 1.95.0
|
||||
@@ -297,7 +297,7 @@ jobs:
|
||||
set -euo pipefail
|
||||
if command -v apt-get >/dev/null 2>&1; then
|
||||
sudo apt-get update -y
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends pkg-config libcap-dev bubblewrap
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends pkg-config libcap-dev libasound2-dev bubblewrap
|
||||
fi
|
||||
|
||||
- uses: dtolnay/rust-toolchain@e081816240890017053eacbb1bdf337761dc5582 # 1.95.0
|
||||
|
||||
@@ -256,7 +256,7 @@ jobs:
|
||||
set -euo pipefail
|
||||
if command -v apt-get >/dev/null 2>&1; then
|
||||
sudo apt-get update -y
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends build-essential pkg-config libcap-dev
|
||||
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends build-essential pkg-config libcap-dev libasound2-dev
|
||||
fi
|
||||
- uses: dtolnay/rust-toolchain@e081816240890017053eacbb1bdf337761dc5582 # 1.95.0
|
||||
with:
|
||||
|
||||
Generated
+15
File diff suppressed because one or more lines are too long
Generated
+186
-1
@@ -305,6 +305,28 @@ version = "0.2.21"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
|
||||
|
||||
[[package]]
|
||||
name = "alsa"
|
||||
version = "0.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "812947049edcd670a82cd5c73c3661d2e58468577ba8489de58e1a73c04cbd5d"
|
||||
dependencies = [
|
||||
"alsa-sys",
|
||||
"bitflags 2.13.1",
|
||||
"cfg-if",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "alsa-sys"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ad7569085a265dd3f607ebecce7458eaab2132a84393534c95b18dcbc3f31e04"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"pkg-config",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "android_system_properties"
|
||||
version = "0.1.5"
|
||||
@@ -4970,6 +4992,8 @@ dependencies = [
|
||||
"codex-process-hardening",
|
||||
"codex-realtime-webrtc",
|
||||
"codex-utils-cargo-bin",
|
||||
"cpal",
|
||||
"crossbeam-queue",
|
||||
"futures",
|
||||
"libloading",
|
||||
"pretty_assertions",
|
||||
@@ -5359,6 +5383,50 @@ dependencies = [
|
||||
"zstd",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "coreaudio-rs"
|
||||
version = "0.14.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7d5d7dca3ebcf65a035582c9ad4385371a9d9ee6537474d2a278f4e1e475bb58"
|
||||
dependencies = [
|
||||
"bitflags 2.13.1",
|
||||
"libc",
|
||||
"objc2-audio-toolbox",
|
||||
"objc2-core-audio",
|
||||
"objc2-core-audio-types",
|
||||
"objc2-core-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cpal"
|
||||
version = "0.18.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6f02e8d0327b42d3e2e4ab2119af397344eb9fc54a34bf0ddeaa1277af8681f1"
|
||||
dependencies = [
|
||||
"alsa",
|
||||
"block2",
|
||||
"coreaudio-rs",
|
||||
"dasp_sample",
|
||||
"jni",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"mach2",
|
||||
"ndk",
|
||||
"ndk-context",
|
||||
"num-derive",
|
||||
"num-traits",
|
||||
"objc2",
|
||||
"objc2-audio-toolbox",
|
||||
"objc2-avf-audio",
|
||||
"objc2-core-audio",
|
||||
"objc2-core-audio-types",
|
||||
"objc2-core-foundation",
|
||||
"objc2-foundation",
|
||||
"web-sys",
|
||||
"windows 0.62.2",
|
||||
"windows-core 0.62.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cpufeatures"
|
||||
version = "0.2.17"
|
||||
@@ -5808,6 +5876,12 @@ dependencies = [
|
||||
"parking_lot_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dasp_sample"
|
||||
version = "0.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0c87e182de0887fd5361989c677c4e8f5000cd9491d6d563161a8f3a5519fc7f"
|
||||
|
||||
[[package]]
|
||||
name = "data-encoding"
|
||||
version = "2.10.0"
|
||||
@@ -9125,7 +9199,7 @@ dependencies = [
|
||||
"cfg-if",
|
||||
"combine",
|
||||
"jni-macros",
|
||||
"jni-sys",
|
||||
"jni-sys 0.4.1",
|
||||
"log",
|
||||
"simd_cesu8",
|
||||
"thiserror 2.0.18",
|
||||
@@ -9146,6 +9220,15 @@ dependencies = [
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jni-sys"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "41a652e1f9b6e0275df1f15b32661cf0d4b78d4d87ddec5e0c3c20f097433258"
|
||||
dependencies = [
|
||||
"jni-sys 0.4.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jni-sys"
|
||||
version = "0.4.1"
|
||||
@@ -9598,6 +9681,12 @@ dependencies = [
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mach2"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dae608c151f68243f2b000364e1f7b186d9c29845f7d2d85bd31b9ad77ad552b"
|
||||
|
||||
[[package]]
|
||||
name = "maplit"
|
||||
version = "1.0.2"
|
||||
@@ -9841,12 +9930,35 @@ dependencies = [
|
||||
"tempfile",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ndk"
|
||||
version = "0.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c3f42e7bbe13d351b6bead8286a43aac9534b82bd3cc43e47037f012ebfd62d4"
|
||||
dependencies = [
|
||||
"bitflags 2.13.1",
|
||||
"jni-sys 0.3.1",
|
||||
"log",
|
||||
"ndk-sys",
|
||||
"num_enum",
|
||||
"thiserror 1.0.69",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ndk-context"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b"
|
||||
|
||||
[[package]]
|
||||
name = "ndk-sys"
|
||||
version = "0.6.0+11769913"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ee6cda3051665f1fb8d9e08fc35c96d5a244fb1be711a03b71118828afc9a873"
|
||||
dependencies = [
|
||||
"jni-sys 0.3.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nibble_vec"
|
||||
version = "0.1.0"
|
||||
@@ -10093,6 +10205,28 @@ dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_enum"
|
||||
version = "0.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5d0bca838442ec211fa11de3a8b0e0e8f3a4522575b5c4c06ed722e005036f26"
|
||||
dependencies = [
|
||||
"num_enum_derive",
|
||||
"rustversion",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_enum_derive"
|
||||
version = "0.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "680998035259dcfcafe653688bf2aa6d3e2dc05e98be6ab46afb089dc84f1df8"
|
||||
dependencies = [
|
||||
"proc-macro-crate",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_threads"
|
||||
version = "0.1.7"
|
||||
@@ -10149,6 +10283,32 @@ dependencies = [
|
||||
"objc2-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "objc2-audio-toolbox"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6948501a91121d6399b79abaa33a8aa4ea7857fe019f341b8c23ad6e81b79b08"
|
||||
dependencies = [
|
||||
"bitflags 2.13.1",
|
||||
"libc",
|
||||
"objc2",
|
||||
"objc2-core-audio",
|
||||
"objc2-core-audio-types",
|
||||
"objc2-core-foundation",
|
||||
"objc2-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "objc2-avf-audio"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "13a380031deed8e99db00065c45937da434ca987c034e13b87e4441f9e4090be"
|
||||
dependencies = [
|
||||
"bitflags 2.13.1",
|
||||
"objc2",
|
||||
"objc2-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "objc2-cloud-kit"
|
||||
version = "0.3.2"
|
||||
@@ -10160,6 +10320,29 @@ dependencies = [
|
||||
"objc2-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "objc2-core-audio"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e1eebcea8b0dbff5f7c8504f3107c68fc061a3eb44932051c8cf8a68d969c3b2"
|
||||
dependencies = [
|
||||
"dispatch2",
|
||||
"objc2",
|
||||
"objc2-core-audio-types",
|
||||
"objc2-core-foundation",
|
||||
"objc2-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "objc2-core-audio-types"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5a89f2ec274a0cf4a32642b2991e8b351a404d290da87bb6a9a9d8632490bd1c"
|
||||
dependencies = [
|
||||
"bitflags 2.13.1",
|
||||
"objc2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "objc2-core-data"
|
||||
version = "0.3.2"
|
||||
@@ -10177,7 +10360,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536"
|
||||
dependencies = [
|
||||
"bitflags 2.13.1",
|
||||
"block2",
|
||||
"dispatch2",
|
||||
"libc",
|
||||
"objc2",
|
||||
]
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ mod message_reader;
|
||||
mod protocol;
|
||||
|
||||
pub use client::VoiceHost;
|
||||
pub use protocol::AudioControls;
|
||||
pub use protocol::MAX_FRAME_BYTES;
|
||||
pub use protocol::Message;
|
||||
pub use protocol::RUNTIME_ENVIRONMENT;
|
||||
|
||||
@@ -8,6 +8,13 @@ use serde::Serialize;
|
||||
|
||||
pub const MAX_FRAME_BYTES: usize = 128 * 1024;
|
||||
|
||||
#[derive(Debug, Deserialize, PartialEq, Serialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
pub struct AudioControls {
|
||||
pub microphone_muted: bool,
|
||||
pub speaker_suppressed: bool,
|
||||
}
|
||||
|
||||
/// SDP contains ICE credentials. Bound it at construction and never expose it in diagnostics.
|
||||
#[derive(Deserialize, PartialEq, Serialize)]
|
||||
#[serde(try_from = "String")]
|
||||
@@ -66,6 +73,10 @@ pub enum Message {
|
||||
Offer { sdp: SessionDescription },
|
||||
ApplyAnswer { sdp: SessionDescription },
|
||||
TransportReady {},
|
||||
OpenDevices {},
|
||||
DevicesOpened {},
|
||||
SetAudioControls { controls: AudioControls },
|
||||
AudioControlsApplied {},
|
||||
Close {},
|
||||
Closed {},
|
||||
}
|
||||
|
||||
@@ -20,6 +20,10 @@ rtc = "=0.20.3"
|
||||
tokio = { workspace = true, features = ["rt-multi-thread", "sync", "time"] }
|
||||
webrtc = "=0.20.3"
|
||||
|
||||
[target.'cfg(any(target_os = "macos", all(target_os = "linux", target_env = "gnu"), all(windows, target_env = "msvc")))'.dependencies]
|
||||
cpal = "=0.18.2"
|
||||
crossbeam-queue = "0.3.12"
|
||||
|
||||
[dev-dependencies]
|
||||
anyhow = { workspace = true }
|
||||
codex-install-context = { workspace = true }
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
# Private voice helper foundation
|
||||
|
||||
`codex-voice-host` establishes the inherited-pipe lifecycle for the proposed
|
||||
bundled voice process and owns WebRTC negotiation. It does not open devices or
|
||||
enable voice in the TUI. The existing CLI is unchanged.
|
||||
bundled voice process and owns WebRTC negotiation and opt-in local devices.
|
||||
It does not enable voice in the TUI. The existing CLI is unchanged.
|
||||
|
||||
Frames are a big-endian u32 length followed by at most 128 KiB of JSON. SDP is
|
||||
limited to 64 KiB and redacted in diagnostics. The
|
||||
@@ -22,6 +22,38 @@ without native audio initialization and does not establish audio readiness.
|
||||
Negotiation has a deadline; `close` tears down the peer before acknowledging exit.
|
||||
UDP and TCP peer tests use real sockets, without a backend or audio devices.
|
||||
|
||||
After native initialization and transport negotiation, `openDevices` opens the
|
||||
default microphone and speaker, initially muted and suppressed. `setAudioControls`
|
||||
invalidates old queued audio. Device errors or queue overflow end the helper.
|
||||
Startup callbacks emit silence without collecting references until worker service
|
||||
begins. After unmute, capture waits for a following callback's device timestamp
|
||||
and rejects buffers captured before that cutoff, including buffers crossing it.
|
||||
This can discard speech onset; it relies on backend timestamp estimates and does
|
||||
not establish a precise physical mute boundary.
|
||||
Both streams request roughly 10 ms callbacks within the device's supported range
|
||||
and the 8,192-frame queue budget. Unknown or incompatible ranges are rejected;
|
||||
there is no fallback to an unbounded default. Backends may deliver smaller
|
||||
callbacks than requested, so capture and rendered references pack samples into
|
||||
full 256-frame queue slots across callbacks without allocating. Each block keeps
|
||||
its oldest sample timestamp. Partial blocks are discarded on generation changes
|
||||
or rejected capture callbacks, including mute and invalid or backwards capture timestamps.
|
||||
Packing retains fewer than 256 samples (5.33 ms at 48 kHz; 32 ms at 8 kHz);
|
||||
delivery also waits for the next callback. Speaker rendering remains immediate.
|
||||
Selected callbacks must leave room for packing and the 5 ms service interval before capture becomes
|
||||
stale at one second. The queue's sample capacity must span more than that service
|
||||
interval.
|
||||
Processing lag can still overflow a queue. Before devices open, the worker blocks
|
||||
on commands instead of polling.
|
||||
`devicesOpened` confirms device opening only: capture/reference are drained locally
|
||||
until the following media stage connects encoding, decoding and the peer.
|
||||
|
||||
Cargo Linux builds require ALSA development inputs discoverable by pkg-config
|
||||
(for example `libasound2-dev` on Debian/Ubuntu). Bazel uses the declared `alsa_lib`
|
||||
source target through `alsa-sys`; macOS uses CoreAudio and Windows uses WASAPI.
|
||||
CPAL and its native link inputs belong only to the helper. Producing a closed
|
||||
Linux distribution still requires the prepared ALSA SDK/runtime/configuration
|
||||
closure; a successful source build does not establish that packaging proof.
|
||||
|
||||
Bazel stamps the binary with `STABLE_GIT_COMMIT`. Cargo builders must provide the
|
||||
same variable; an unstamped source build reports `dev` via `--build-commit` and is
|
||||
not a distributable build identity. The client/control crate has no native audio
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
//! Preallocated callback buffers pack small callbacks into full queue slots without locks.
|
||||
//! Generation changes and rejected capture buffers discard incomplete frames.
|
||||
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::AtomicU64;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
|
||||
use crossbeam_queue::ArrayQueue;
|
||||
|
||||
pub(super) const BLOCK: usize = 256;
|
||||
pub(super) const QUEUE_CAPACITY: usize = 32;
|
||||
|
||||
pub(super) struct Frame {
|
||||
pub(super) samples: [f32; BLOCK],
|
||||
pub(super) len: usize,
|
||||
pub(super) at: Instant,
|
||||
pub(super) generation: u64,
|
||||
}
|
||||
|
||||
/// Packs serialized callbacks without allocating; only complete blocks enter the queue.
|
||||
/// The oldest sample keeps its timestamp. Callers reset after rejected capture data.
|
||||
#[derive(Default)]
|
||||
pub(super) struct FramePacker {
|
||||
frame: Option<Frame>,
|
||||
}
|
||||
|
||||
impl FramePacker {
|
||||
pub(super) fn reset(&mut self) {
|
||||
self.frame = None;
|
||||
}
|
||||
|
||||
pub(super) fn push(&mut self, frame: Frame, rate: f64, queue: &ArrayQueue<Frame>) -> bool {
|
||||
if self
|
||||
.frame
|
||||
.as_ref()
|
||||
.is_some_and(|partial| partial.generation != frame.generation)
|
||||
{
|
||||
self.reset();
|
||||
}
|
||||
let mut offset = 0;
|
||||
while offset < frame.len {
|
||||
let partial = self.frame.get_or_insert_with(|| Frame {
|
||||
samples: [0.0; BLOCK],
|
||||
len: 0,
|
||||
at: frame.at + Duration::from_secs_f64(offset as f64 / rate),
|
||||
generation: frame.generation,
|
||||
});
|
||||
let count = (BLOCK - partial.len).min(frame.len - offset);
|
||||
partial.samples[partial.len..partial.len + count]
|
||||
.copy_from_slice(&frame.samples[offset..offset + count]);
|
||||
partial.len += count;
|
||||
offset += count;
|
||||
if partial.len == BLOCK
|
||||
&& let Some(full) = self.frame.take()
|
||||
&& let Err(rejected) = queue.push(full)
|
||||
{
|
||||
// Suppressed output is known silence; dropping it cannot lose audible output.
|
||||
// Never discard an active generation or a block containing real audio.
|
||||
if rejected.generation.is_multiple_of(2)
|
||||
|| rejected.samples.iter().any(|sample| *sample != 0.0)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct Buffers {
|
||||
pub(super) capture: ArrayQueue<Frame>,
|
||||
pub(super) rendered: ArrayQueue<Frame>,
|
||||
pub(super) playback: ArrayQueue<Frame>,
|
||||
pub(super) microphone: AtomicU64,
|
||||
pub(super) speaker: AtomicU64,
|
||||
pub(super) serviced: AtomicBool,
|
||||
pub(super) failed: AtomicBool,
|
||||
}
|
||||
|
||||
impl Buffers {
|
||||
pub(super) fn new() -> Self {
|
||||
Self {
|
||||
capture: ArrayQueue::new(QUEUE_CAPACITY),
|
||||
rendered: ArrayQueue::new(QUEUE_CAPACITY),
|
||||
playback: ArrayQueue::new(QUEUE_CAPACITY),
|
||||
microphone: AtomicU64::new(/*v*/ 1),
|
||||
speaker: AtomicU64::new(/*v*/ 1),
|
||||
serviced: AtomicBool::new(false),
|
||||
failed: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
|
||||
// One control worker writes each epoch. Odd epochs are disabled; every
|
||||
// transition advances the epoch so disable/re-enable cannot replay old audio.
|
||||
pub(super) fn set_disabled(epoch: &AtomicU64, disabled: bool) -> std::io::Result<()> {
|
||||
let current = epoch.load(Ordering::Acquire);
|
||||
if (current % 2 == 1) != disabled {
|
||||
epoch.store(
|
||||
current
|
||||
.checked_add(1)
|
||||
.ok_or_else(|| std::io::Error::other("audio generation exhausted"))?,
|
||||
Ordering::Release,
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Rejects capture backlog after unmute, using offsets in the device's clock.
|
||||
/// This is conservative software admission, not a guarantee of hardware clock accuracy.
|
||||
#[derive(Default)]
|
||||
pub(super) struct CaptureBoundary {
|
||||
generation: Option<u64>,
|
||||
cutoff: Option<Duration>,
|
||||
previous_capture: Option<Duration>,
|
||||
}
|
||||
|
||||
impl CaptureBoundary {
|
||||
pub(super) fn accepts(
|
||||
&mut self,
|
||||
generation: u64,
|
||||
callback: Duration,
|
||||
capture: Duration,
|
||||
) -> bool {
|
||||
if generation % 2 == 1 || self.generation != Some(generation) {
|
||||
self.generation = Some(generation);
|
||||
self.cutoff = None;
|
||||
self.previous_capture = None;
|
||||
// The current callback timestamp may have been sampled before unmute.
|
||||
// Wait for the next serialized callback before establishing a cutoff.
|
||||
return false;
|
||||
}
|
||||
let accepted = capture >= *self.cutoff.get_or_insert(callback)
|
||||
&& self
|
||||
.previous_capture
|
||||
.is_none_or(|previous| capture >= previous);
|
||||
if accepted {
|
||||
self.previous_capture = Some(capture);
|
||||
}
|
||||
accepted
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub(super) struct Playback {
|
||||
frame: Option<Frame>,
|
||||
offset: usize,
|
||||
}
|
||||
|
||||
impl Playback {
|
||||
pub(super) fn next(&mut self, buffers: &Buffers) -> f32 {
|
||||
let epoch = buffers.speaker.load(Ordering::Acquire);
|
||||
if self
|
||||
.frame
|
||||
.as_ref()
|
||||
.is_some_and(|frame| frame.generation != epoch || self.offset == frame.len)
|
||||
{
|
||||
self.frame = None;
|
||||
}
|
||||
// Bound stale-frame work even if a producer keeps writing during a mute.
|
||||
for _ in 0..buffers.playback.capacity() {
|
||||
if self.frame.is_some() {
|
||||
break;
|
||||
}
|
||||
let Some(frame) = buffers.playback.pop() else {
|
||||
break;
|
||||
};
|
||||
if epoch.is_multiple_of(2)
|
||||
&& frame.generation == epoch
|
||||
&& frame.len > 0
|
||||
&& frame.len <= BLOCK
|
||||
{
|
||||
self.frame = Some(frame);
|
||||
self.offset = 0;
|
||||
}
|
||||
}
|
||||
let Some(frame) = &self.frame else {
|
||||
return 0.0;
|
||||
};
|
||||
let sample = frame.samples[self.offset];
|
||||
self.offset += 1;
|
||||
if epoch % 2 == 1 || !sample.is_finite() {
|
||||
0.0
|
||||
} else {
|
||||
sample.clamp(-1.0, 1.0)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "device_buffers_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,212 @@
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn unmute_rejects_backlog_until_capture_reaches_the_following_callback() {
|
||||
let mut boundary = CaptureBoundary::default();
|
||||
// Times are offsets in the same device clock. The first callback's timestamp
|
||||
// can precede unmute, so even capture after that timestamp must be rejected.
|
||||
for (generation, callback, capture, expected) in [
|
||||
(1, 20, 10, false), // muted
|
||||
(2, 21, 15, false), // first observed unmute
|
||||
(2, 25, 22, false), // next callback establishes cutoff 25
|
||||
(2, 26, 24, false), // another old/crossing buffer
|
||||
(2, 27, 25, true),
|
||||
(2, 28, 26, true), // same generation must not move the cutoff
|
||||
(2, 29, 25, false), // backwards device capture is a discontinuity
|
||||
(2, 30, 26, true), // forward capture can resume without moving the cutoff
|
||||
(4, 30, 29, false), // mute/unmute happened without a callback
|
||||
(4, 34, 31, false),
|
||||
(4, 35, 33, false),
|
||||
(4, 36, 34, true),
|
||||
(5, 37, 35, false), // observed mute
|
||||
(6, 38, 36, false),
|
||||
(6, 40, 40, true),
|
||||
(6, 101, 100, true),
|
||||
(6, 102, 90, false),
|
||||
(6, 103, 95, false), // rejection must not lower the last accepted timestamp
|
||||
(6, 104, 101, true),
|
||||
] {
|
||||
assert_eq!(
|
||||
boundary.accepts(
|
||||
generation,
|
||||
Duration::from_millis(callback),
|
||||
Duration::from_millis(capture),
|
||||
),
|
||||
expected,
|
||||
"generation={generation}, callback={callback}, capture={capture}",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn suppression_discards_partial_and_queued_previous_generations() {
|
||||
let buffers = Buffers::new();
|
||||
let mut playback = Playback::default();
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
let generation = buffers.speaker.load(Ordering::Acquire);
|
||||
for _ in 0..2 {
|
||||
assert!(
|
||||
buffers
|
||||
.playback
|
||||
.push(Frame {
|
||||
samples: [0.5; BLOCK],
|
||||
len: BLOCK,
|
||||
at: Instant::now(),
|
||||
generation
|
||||
})
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
assert_eq!(playback.next(&buffers), 0.5);
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ true).unwrap();
|
||||
assert_eq!(playback.next(&buffers), 0.0);
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
assert_eq!(playback.next(&buffers), 0.0);
|
||||
assert!(buffers.playback.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn output_is_finite_and_bounded_and_underflow_is_silence() {
|
||||
let buffers = Buffers::new();
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
let generation = buffers.speaker.load(Ordering::Acquire);
|
||||
let mut samples = [0.0; BLOCK];
|
||||
samples[..4].copy_from_slice(&[f32::NAN, f32::INFINITY, -2.0, 2.0]);
|
||||
assert!(
|
||||
buffers
|
||||
.playback
|
||||
.push(Frame {
|
||||
samples,
|
||||
len: 4,
|
||||
at: Instant::now(),
|
||||
generation
|
||||
})
|
||||
.is_ok()
|
||||
);
|
||||
let mut playback = Playback::default();
|
||||
let output: Vec<_> = (0..5).map(|_| playback.next(&buffers)).collect();
|
||||
assert_eq!(output, [0.0, 0.0, -1.0, 1.0, 0.0]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repeated_controls_preserve_epoch_but_transitions_invalidate_capture() {
|
||||
let buffers = Buffers::new();
|
||||
Buffers::set_disabled(&buffers.microphone, /*disabled*/ false).unwrap();
|
||||
let first = buffers.microphone.load(Ordering::Acquire);
|
||||
Buffers::set_disabled(&buffers.microphone, /*disabled*/ false).unwrap();
|
||||
assert_eq!(buffers.microphone.load(Ordering::Acquire), first);
|
||||
Buffers::set_disabled(&buffers.microphone, /*disabled*/ true).unwrap();
|
||||
Buffers::set_disabled(&buffers.microphone, /*disabled*/ false).unwrap();
|
||||
assert_ne!(buffers.microphone.load(Ordering::Acquire), first);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tiny_callbacks_share_slots_and_preserve_the_oldest_timestamp() {
|
||||
let buffers = Buffers::new();
|
||||
let start = Instant::now();
|
||||
for queue in [&buffers.capture, &buffers.rendered] {
|
||||
let mut packer = FramePacker::default();
|
||||
// Five milliseconds at 384 kHz arrives as 120 sixteen-frame callbacks.
|
||||
for callback in 0..120 {
|
||||
assert!(packer.push(
|
||||
Frame {
|
||||
samples: [callback as f32; BLOCK],
|
||||
len: 16,
|
||||
at: start + Duration::from_secs_f64((callback * 16) as f64 / 384_000.0),
|
||||
generation: 2,
|
||||
},
|
||||
/*rate*/ 384_000.0,
|
||||
queue,
|
||||
));
|
||||
}
|
||||
assert_eq!(queue.len(), 7);
|
||||
for block in 0..7 {
|
||||
let frame = queue.pop().unwrap();
|
||||
assert_eq!(
|
||||
(frame.samples, frame.len, frame.at, frame.generation),
|
||||
(
|
||||
std::array::from_fn(|sample| (block * 16 + sample / 16) as f32),
|
||||
BLOCK,
|
||||
start + Duration::from_secs_f64((block * BLOCK) as f64 / 384_000.0),
|
||||
2,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn packing_drops_partial_audio_on_generation_changes_and_capture_rejection() {
|
||||
let queue = ArrayQueue::new(QUEUE_CAPACITY);
|
||||
let mut packer = FramePacker::default();
|
||||
let start = Instant::now();
|
||||
for (generation, sample) in [(2, 0.1), (4, 0.2), (4, 0.3)] {
|
||||
assert!(packer.push(
|
||||
Frame {
|
||||
samples: [sample; BLOCK],
|
||||
len: BLOCK / 2,
|
||||
at: start,
|
||||
generation,
|
||||
},
|
||||
/*rate*/ 48_000.0,
|
||||
&queue,
|
||||
));
|
||||
}
|
||||
let frame = queue.pop().unwrap();
|
||||
assert_eq!(
|
||||
(frame.samples, frame.len, frame.at, frame.generation),
|
||||
(
|
||||
std::array::from_fn(|i| if i < BLOCK / 2 { 0.2 } else { 0.3 }),
|
||||
BLOCK,
|
||||
start,
|
||||
4
|
||||
),
|
||||
);
|
||||
for sample in [0.4, 0.5] {
|
||||
assert!(packer.push(
|
||||
Frame {
|
||||
samples: [sample; BLOCK],
|
||||
len: BLOCK / 2,
|
||||
at: start,
|
||||
generation: 4
|
||||
},
|
||||
/*rate*/ 48_000.0,
|
||||
&queue,
|
||||
));
|
||||
packer.reset();
|
||||
}
|
||||
assert!(queue.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn packing_preserves_a_remainder_timestamp_across_uneven_callbacks() {
|
||||
let queue = ArrayQueue::new(QUEUE_CAPACITY);
|
||||
let mut packer = FramePacker::default();
|
||||
let start = Instant::now();
|
||||
let second = start + Duration::from_millis(10);
|
||||
for (len, sample, at) in [(100, 0.1, start), (200, 0.2, second), (212, 0.3, second)] {
|
||||
assert!(packer.push(
|
||||
Frame {
|
||||
samples: [sample; BLOCK],
|
||||
len,
|
||||
at,
|
||||
generation: 2
|
||||
},
|
||||
/*rate*/ 48_000.0,
|
||||
&queue,
|
||||
));
|
||||
}
|
||||
let first = queue.pop().unwrap();
|
||||
let remainder = queue.pop().unwrap();
|
||||
assert_eq!(
|
||||
(first.samples, first.at, remainder.samples, remainder.at),
|
||||
(
|
||||
std::array::from_fn(|i| if i < 100 { 0.1 } else { 0.2 }),
|
||||
start,
|
||||
std::array::from_fn(|i| if i < 44 { 0.2 } else { 0.3 }),
|
||||
second + Duration::from_secs_f64(156.0 / 48_000.0),
|
||||
),
|
||||
);
|
||||
assert!(queue.is_empty());
|
||||
}
|
||||
@@ -0,0 +1,327 @@
|
||||
//! Owns local device streams on the helper worker. Callbacks allocate no buffers and take no locks.
|
||||
//! Small callbacks share full queue slots; processing lag still fails the session closed.
|
||||
//! Capture and actual rendered output carry device timing.
|
||||
//! References start with worker service; unmute rejects earlier device capture buffers.
|
||||
|
||||
#[path = "device_buffers.rs"]
|
||||
mod buffers;
|
||||
|
||||
use std::io;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
|
||||
use cpal::FromSample;
|
||||
use cpal::Sample;
|
||||
use cpal::SampleFormat;
|
||||
use cpal::SizedSample;
|
||||
use cpal::traits::DeviceTrait;
|
||||
use cpal::traits::HostTrait;
|
||||
use cpal::traits::StreamTrait;
|
||||
|
||||
use buffers::BLOCK;
|
||||
use buffers::Buffers;
|
||||
use buffers::CaptureBoundary;
|
||||
use buffers::Frame;
|
||||
use buffers::FramePacker;
|
||||
use buffers::Playback;
|
||||
use buffers::QUEUE_CAPACITY;
|
||||
|
||||
const MAX_CAPTURE_AGE: Duration = Duration::from_secs(/*secs*/ 1);
|
||||
|
||||
pub(super) struct Devices {
|
||||
_input: cpal::Stream,
|
||||
_output: cpal::Stream,
|
||||
buffers: Arc<Buffers>,
|
||||
}
|
||||
|
||||
macro_rules! stream {
|
||||
($format:expr, $build:ident, $($arg:expr),+) => {
|
||||
match $format {
|
||||
SampleFormat::I8 => $build::<i8>($($arg),+),
|
||||
SampleFormat::I16 => $build::<i16>($($arg),+),
|
||||
SampleFormat::I32 => $build::<i32>($($arg),+),
|
||||
SampleFormat::I64 => $build::<i64>($($arg),+),
|
||||
SampleFormat::U8 => $build::<u8>($($arg),+),
|
||||
SampleFormat::U16 => $build::<u16>($($arg),+),
|
||||
SampleFormat::U32 => $build::<u32>($($arg),+),
|
||||
SampleFormat::U64 => $build::<u64>($($arg),+),
|
||||
SampleFormat::F32 => $build::<f32>($($arg),+),
|
||||
SampleFormat::F64 => $build::<f64>($($arg),+),
|
||||
_ => Err(cpal::Error::from(cpal::ErrorKind::UnsupportedConfig)),
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
impl Devices {
|
||||
pub(super) fn open() -> io::Result<Self> {
|
||||
let host = cpal::default_host();
|
||||
let input = host
|
||||
.default_input_device()
|
||||
.ok_or_else(|| io::Error::other("microphone unavailable"))?;
|
||||
let output = host
|
||||
.default_output_device()
|
||||
.ok_or_else(|| io::Error::other("speaker unavailable"))?;
|
||||
let input_config = input
|
||||
.default_input_config()
|
||||
.map_err(|_| io::Error::other("microphone configuration unavailable"))?;
|
||||
let output_config = output
|
||||
.default_output_config()
|
||||
.map_err(|_| io::Error::other("speaker configuration unavailable"))?;
|
||||
for config in [&input_config, &output_config] {
|
||||
if config.channels() == 0
|
||||
|| config.channels() > 32
|
||||
|| !(8_000..=384_000).contains(&config.sample_rate())
|
||||
{
|
||||
return Err(io::Error::other("unsupported audio device configuration"));
|
||||
}
|
||||
}
|
||||
let input_stream_config = bounded_stream_config(&input_config)?;
|
||||
let output_stream_config = bounded_stream_config(&output_config)?;
|
||||
let buffers = Arc::new(Buffers::new());
|
||||
let input = stream!(
|
||||
input_config.sample_format(),
|
||||
build_input,
|
||||
&input,
|
||||
&input_stream_config,
|
||||
buffers.clone()
|
||||
)
|
||||
.map_err(|_| io::Error::other("failed to open microphone"))?;
|
||||
let output = stream!(
|
||||
output_config.sample_format(),
|
||||
build_output,
|
||||
&output,
|
||||
&output_stream_config,
|
||||
buffers.clone()
|
||||
)
|
||||
.map_err(|_| io::Error::other("failed to open speaker"))?;
|
||||
output
|
||||
.play()
|
||||
.map_err(|_| io::Error::other("failed to start speaker"))?;
|
||||
input
|
||||
.play()
|
||||
.map_err(|_| io::Error::other("failed to start microphone"))?;
|
||||
Ok(Self {
|
||||
_input: input,
|
||||
_output: output,
|
||||
buffers,
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn set_controls(
|
||||
&self,
|
||||
controls: codex_realtime_webrtc::AudioControls,
|
||||
) -> io::Result<()> {
|
||||
Buffers::set_disabled(&self.buffers.microphone, controls.microphone_muted)?;
|
||||
Buffers::set_disabled(&self.buffers.speaker, controls.speaker_suppressed)
|
||||
}
|
||||
|
||||
pub(super) fn service(&self) -> io::Result<()> {
|
||||
// Some backends start callbacks during stream construction, before open returns.
|
||||
self.buffers.serviced.store(true, Ordering::Release);
|
||||
// Device-only stage: consume locally until the media worker is attached.
|
||||
// Reading timestamps here also rejects a stalled callback before readiness is reused.
|
||||
for queue in [&self.buffers.capture, &self.buffers.rendered] {
|
||||
for _ in 0..queue.capacity() {
|
||||
let Some(frame) = queue.pop() else {
|
||||
break;
|
||||
};
|
||||
if frame.at.elapsed() > MAX_CAPTURE_AGE {
|
||||
return Err(io::Error::other("audio device processing fell behind"));
|
||||
}
|
||||
}
|
||||
}
|
||||
if self.buffers.failed.load(Ordering::Acquire) {
|
||||
return Err(io::Error::other("audio device failed"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn bounded_stream_config(
|
||||
supported: &cpal::SupportedStreamConfig,
|
||||
) -> io::Result<cpal::StreamConfig> {
|
||||
let cpal::SupportedBufferSize::Range { min, max } = *supported.buffer_size() else {
|
||||
return Err(io::Error::other("audio callback size range unavailable"));
|
||||
};
|
||||
let mut config = supported.config();
|
||||
// Leave room for a partial block and smaller callbacks arriving before service.
|
||||
let service_frames = (f64::from(config.sample_rate)
|
||||
* super::DEVICE_SERVICE_INTERVAL.as_secs_f64())
|
||||
.ceil() as usize;
|
||||
let callback_capacity = (BLOCK * QUEUE_CAPACITY).saturating_sub(service_frames + BLOCK);
|
||||
let min = min.max(1);
|
||||
let max = max.min(callback_capacity as u32);
|
||||
if min > max {
|
||||
return Err(io::Error::other("unsupported audio callback size range"));
|
||||
}
|
||||
// Aim for 10 ms without consuming the queue's service headroom.
|
||||
// Do not fall back to the backend's potentially much larger default buffer.
|
||||
let frames = (config.sample_rate / 100).clamp(min, max);
|
||||
let callback_duration =
|
||||
Duration::from_secs_f64(f64::from(frames) / f64::from(config.sample_rate));
|
||||
// Backends may deliver smaller callbacks than requested. Packing makes queue
|
||||
// coverage depend on samples, while an incomplete block adds bounded latency.
|
||||
let packing_duration = Duration::from_secs_f64(BLOCK as f64 / f64::from(config.sample_rate));
|
||||
let queue_duration = packing_duration * QUEUE_CAPACITY as u32;
|
||||
if callback_duration + packing_duration + super::DEVICE_SERVICE_INTERVAL >= MAX_CAPTURE_AGE
|
||||
|| queue_duration <= super::DEVICE_SERVICE_INTERVAL
|
||||
{
|
||||
return Err(io::Error::other("unsupported audio callback timing"));
|
||||
}
|
||||
config.buffer_size = cpal::BufferSize::Fixed(frames);
|
||||
Ok(config)
|
||||
}
|
||||
|
||||
// CPAL reports recoverable underruns/overruns through the same callback as device loss.
|
||||
fn handle_stream_error(buffers: &Buffers, error: cpal::Error) {
|
||||
if error.kind() != cpal::ErrorKind::Xrun {
|
||||
buffers.failed.store(true, Ordering::Release);
|
||||
}
|
||||
}
|
||||
|
||||
fn build_input<T>(
|
||||
device: &cpal::Device,
|
||||
config: &cpal::StreamConfig,
|
||||
buffers: Arc<Buffers>,
|
||||
) -> Result<cpal::Stream, cpal::Error>
|
||||
where
|
||||
T: SizedSample,
|
||||
f32: FromSample<T>,
|
||||
{
|
||||
let channels = usize::from(config.channels);
|
||||
let rate = f64::from(config.sample_rate);
|
||||
let failure = buffers.clone();
|
||||
let mut origin = None;
|
||||
let mut boundary = CaptureBoundary::default();
|
||||
let mut capture = FramePacker::default();
|
||||
device.build_input_stream(
|
||||
*config,
|
||||
move |data: &[T], info: &cpal::InputCallbackInfo| {
|
||||
let generation = buffers.microphone.load(Ordering::Acquire);
|
||||
if generation % 2 == 1 {
|
||||
capture.reset();
|
||||
return;
|
||||
}
|
||||
let timestamp = info.timestamp();
|
||||
let origin = origin.get_or_insert(timestamp.capture);
|
||||
let (Some(callback), Some(captured)) = (
|
||||
timestamp.callback.checked_duration_since(*origin),
|
||||
timestamp.capture.checked_duration_since(*origin),
|
||||
) else {
|
||||
capture.reset();
|
||||
return;
|
||||
};
|
||||
if !boundary.accepts(generation, callback, captured) {
|
||||
capture.reset();
|
||||
return;
|
||||
}
|
||||
let start = Instant::now()
|
||||
.checked_sub(
|
||||
timestamp
|
||||
.callback
|
||||
.checked_duration_since(timestamp.capture)
|
||||
.unwrap_or_default(),
|
||||
)
|
||||
.unwrap_or_else(Instant::now);
|
||||
for (index, chunk) in data.chunks(BLOCK * channels).enumerate() {
|
||||
let mut frame = Frame {
|
||||
samples: [0.0; BLOCK],
|
||||
len: chunk.len() / channels,
|
||||
at: start + Duration::from_secs_f64((index * BLOCK) as f64 / rate),
|
||||
generation,
|
||||
};
|
||||
for (output, input) in frame.samples.iter_mut().zip(chunk.chunks_exact(channels)) {
|
||||
*output = input
|
||||
.iter()
|
||||
.map(|sample| f32::from_sample(*sample))
|
||||
.sum::<f32>()
|
||||
/ channels as f32;
|
||||
if !output.is_finite() {
|
||||
buffers.failed.store(true, Ordering::Release);
|
||||
return;
|
||||
}
|
||||
}
|
||||
if !capture.push(frame, rate, &buffers.capture) {
|
||||
buffers.failed.store(true, Ordering::Release);
|
||||
return;
|
||||
}
|
||||
}
|
||||
},
|
||||
move |error| handle_stream_error(&failure, error),
|
||||
/*timeout*/ None,
|
||||
)
|
||||
}
|
||||
|
||||
fn build_output<T>(
|
||||
device: &cpal::Device,
|
||||
config: &cpal::StreamConfig,
|
||||
buffers: Arc<Buffers>,
|
||||
) -> Result<cpal::Stream, cpal::Error>
|
||||
where
|
||||
T: SizedSample + FromSample<f32>,
|
||||
f32: FromSample<T>,
|
||||
{
|
||||
let channels = usize::from(config.channels);
|
||||
let rate = f64::from(config.sample_rate);
|
||||
let failure = buffers.clone();
|
||||
let mut output = OutputState::default();
|
||||
device.build_output_stream(
|
||||
*config,
|
||||
move |data: &mut [T], info: &cpal::OutputCallbackInfo| {
|
||||
let timestamp = info.timestamp();
|
||||
let start = Instant::now()
|
||||
+ timestamp
|
||||
.playback
|
||||
.checked_duration_since(timestamp.callback)
|
||||
.unwrap_or_default();
|
||||
render_output(data, channels, rate, start, &buffers, &mut output);
|
||||
},
|
||||
move |error| handle_stream_error(&failure, error),
|
||||
/*timeout*/ None,
|
||||
)
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct OutputState {
|
||||
playback: Playback,
|
||||
reference: FramePacker,
|
||||
}
|
||||
|
||||
fn render_output<T>(
|
||||
data: &mut [T],
|
||||
channels: usize,
|
||||
rate: f64,
|
||||
start: Instant,
|
||||
buffers: &Buffers,
|
||||
output: &mut OutputState,
|
||||
) where
|
||||
T: SizedSample + FromSample<f32>,
|
||||
f32: FromSample<T>,
|
||||
{
|
||||
if !buffers.serviced.load(Ordering::Acquire) {
|
||||
data.fill(T::from_sample(0.0));
|
||||
return;
|
||||
}
|
||||
for (index, chunk) in data.chunks_mut(BLOCK * channels).enumerate() {
|
||||
let mut reference = Frame {
|
||||
samples: [0.0; BLOCK],
|
||||
len: chunk.len() / channels,
|
||||
at: start + Duration::from_secs_f64((index * BLOCK) as f64 / rate),
|
||||
generation: buffers.speaker.load(Ordering::Acquire),
|
||||
};
|
||||
for (frame, sample) in chunk.chunks_mut(channels).zip(&mut reference.samples) {
|
||||
let rendered = T::from_sample(output.playback.next(buffers));
|
||||
frame.fill(rendered);
|
||||
*sample = f32::from_sample(rendered);
|
||||
}
|
||||
if !output.reference.push(reference, rate, &buffers.rendered) {
|
||||
buffers.failed.store(true, Ordering::Release);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "devices_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,271 @@
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn startup_output_waits_for_service_but_later_overflow_still_fails() {
|
||||
let buffers = Buffers::new();
|
||||
let mut state = OutputState::default();
|
||||
let mut output = [1.0_f32; BLOCK * 2];
|
||||
let start = Instant::now();
|
||||
for _ in 0..QUEUE_CAPACITY + 1 {
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 2,
|
||||
/*rate*/ 384_000.0,
|
||||
start,
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
assert_eq!(output, [0.0; BLOCK * 2]);
|
||||
}
|
||||
assert!(buffers.rendered.is_empty());
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 2,
|
||||
/*rate*/ 384_000.0,
|
||||
start,
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
let reference = buffers.rendered.pop().unwrap();
|
||||
assert_eq!(
|
||||
(
|
||||
reference.samples,
|
||||
reference.len,
|
||||
reference.at,
|
||||
reference.generation
|
||||
),
|
||||
(
|
||||
[0.0; BLOCK],
|
||||
BLOCK,
|
||||
start,
|
||||
buffers.speaker.load(Ordering::Acquire)
|
||||
),
|
||||
);
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
for _ in 0..QUEUE_CAPACITY {
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 2,
|
||||
/*rate*/ 384_000.0,
|
||||
start,
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
}
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 2,
|
||||
/*rate*/ 384_000.0,
|
||||
start,
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
assert!(buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn callback_configuration_fits_supported_range_and_actual_queue() {
|
||||
for (rate, min, max, frames) in [
|
||||
(96_000, 0, u32::MAX, 960),
|
||||
(96_000, 1, 16, 16),
|
||||
(384_000, 1, 64, 64),
|
||||
(8_000, 7_680, 8_192, 7_680),
|
||||
(384_000, 1, 16, 16),
|
||||
(96_000, 1, 15, 15),
|
||||
(48_000, 1, 128, 128),
|
||||
(48_000, 2_048, 16_384, 2_048),
|
||||
(384_000, 6_016, 16_384, 6_016),
|
||||
] {
|
||||
let supported = cpal::SupportedStreamConfig::new(
|
||||
/*channels*/ 2,
|
||||
rate,
|
||||
cpal::SupportedBufferSize::Range { min, max },
|
||||
cpal::SampleFormat::F32,
|
||||
);
|
||||
let config = bounded_stream_config(&supported).unwrap();
|
||||
assert_eq!(
|
||||
config,
|
||||
cpal::StreamConfig {
|
||||
channels: 2,
|
||||
sample_rate: rate,
|
||||
buffer_size: cpal::BufferSize::Fixed(frames),
|
||||
}
|
||||
);
|
||||
let buffers = Buffers::new();
|
||||
for queue in [&buffers.capture, &buffers.rendered] {
|
||||
for offset in (0..frames as usize).step_by(BLOCK) {
|
||||
assert!(
|
||||
queue
|
||||
.push(Frame {
|
||||
samples: [0.0; BLOCK],
|
||||
len: (frames as usize - offset).min(BLOCK),
|
||||
at: Instant::now(),
|
||||
generation: 0,
|
||||
})
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn callback_configuration_reserves_room_for_partial_and_following_callbacks() {
|
||||
for min in [6_016, 6_017, 8_192] {
|
||||
let supported = cpal::SupportedStreamConfig::new(
|
||||
/*channels*/ 1,
|
||||
384_000,
|
||||
cpal::SupportedBufferSize::Range { min, max: 16_384 },
|
||||
cpal::SampleFormat::F32,
|
||||
);
|
||||
let config = bounded_stream_config(&supported);
|
||||
if min != 6_016 {
|
||||
assert!(config.is_err(), "minimum callback size {min}");
|
||||
continue;
|
||||
}
|
||||
let cpal::BufferSize::Fixed(frames) = config.unwrap().buffer_size else {
|
||||
panic!("callback size must be bounded");
|
||||
};
|
||||
let buffers = Buffers::new();
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
let mut output = OutputState::default();
|
||||
let start = Instant::now();
|
||||
// An existing partial block, a full callback, then 5 ms of 128-frame callbacks.
|
||||
for count in [BLOCK - 1, frames as usize].into_iter().chain([128; 15]) {
|
||||
render_output(
|
||||
&mut vec![0.0_f32; count],
|
||||
/*channels*/ 1,
|
||||
/*rate*/ 384_000.0,
|
||||
start,
|
||||
&buffers,
|
||||
&mut output,
|
||||
);
|
||||
}
|
||||
assert_eq!(buffers.rendered.len(), QUEUE_CAPACITY - 1);
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unsupported_callback_ranges_do_not_fall_back_to_default() {
|
||||
for range in [
|
||||
cpal::SupportedBufferSize::Unknown,
|
||||
cpal::SupportedBufferSize::Range { min: 0, max: 0 },
|
||||
cpal::SupportedBufferSize::Range {
|
||||
min: 1_024,
|
||||
max: 512,
|
||||
},
|
||||
cpal::SupportedBufferSize::Range {
|
||||
min: 8_193,
|
||||
max: 16_384,
|
||||
},
|
||||
] {
|
||||
let supported = cpal::SupportedStreamConfig::new(
|
||||
/*channels*/ 2,
|
||||
96_000,
|
||||
range,
|
||||
cpal::SampleFormat::F32,
|
||||
);
|
||||
assert!(bounded_stream_config(&supported).is_err(), "{range:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn callback_timing_must_fit_service_and_stale_limits() {
|
||||
for (rate, min, max) in [(8_000, 8_192, 8_192), (8_000, 7_704, 8_192)] {
|
||||
let supported = cpal::SupportedStreamConfig::new(
|
||||
/*channels*/ 2,
|
||||
rate,
|
||||
cpal::SupportedBufferSize::Range { min, max },
|
||||
cpal::SampleFormat::F32,
|
||||
);
|
||||
assert!(
|
||||
bounded_stream_config(&supported).is_err(),
|
||||
"rate={rate}, range={min}..={max}",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tiny_output_callbacks_pack_references_without_delaying_rendering() {
|
||||
let buffers = Buffers::new();
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
let generation = buffers.speaker.load(Ordering::Acquire);
|
||||
let start = Instant::now();
|
||||
for _ in 0..8 {
|
||||
assert!(
|
||||
buffers
|
||||
.playback
|
||||
.push(Frame {
|
||||
samples: [0.5; BLOCK],
|
||||
len: BLOCK,
|
||||
at: start,
|
||||
generation,
|
||||
})
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
let mut state = OutputState::default();
|
||||
for callback in 0..120 {
|
||||
let mut output = [0.0_f32; 32];
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 2,
|
||||
/*rate*/ 384_000.0,
|
||||
start + Duration::from_secs_f64((callback * 16) as f64 / 384_000.0),
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
assert_eq!(output, [0.5; 32]);
|
||||
}
|
||||
assert_eq!(buffers.rendered.len(), 7);
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn suppressed_output_does_not_fail_when_reference_queue_is_full() {
|
||||
let buffers = Buffers::new();
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
let mut state = OutputState::default();
|
||||
let mut output = [1.0_f32; BLOCK];
|
||||
for _ in 0..QUEUE_CAPACITY * 3 {
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 1,
|
||||
/*rate*/ 384_000.0,
|
||||
Instant::now(),
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
assert_eq!(output, [0.0; BLOCK]);
|
||||
}
|
||||
assert_eq!(buffers.rendered.len(), QUEUE_CAPACITY);
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 1,
|
||||
/*rate*/ 384_000.0,
|
||||
Instant::now(),
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
assert!(buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_xruns_recover_without_clearing_device_failure() {
|
||||
let buffers = Buffers::new();
|
||||
handle_stream_error(&buffers, cpal::ErrorKind::Xrun.into());
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
handle_stream_error(&buffers, cpal::ErrorKind::DeviceNotAvailable.into());
|
||||
assert!(buffers.failed.load(Ordering::Acquire));
|
||||
handle_stream_error(&buffers, cpal::ErrorKind::Xrun.into());
|
||||
assert!(buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
//! Keep non-native helper targets buildable without linking a host audio backend.
|
||||
|
||||
use std::io;
|
||||
|
||||
pub(super) struct Devices;
|
||||
|
||||
impl Devices {
|
||||
pub(super) fn open() -> io::Result<Self> {
|
||||
Err(io::Error::other(
|
||||
"audio devices unavailable for this helper target",
|
||||
))
|
||||
}
|
||||
|
||||
pub(super) fn set_controls(&self, _: codex_realtime_webrtc::AudioControls) -> io::Result<()> {
|
||||
Self::open().map(|_| ())
|
||||
}
|
||||
|
||||
pub(super) fn service(&self) -> io::Result<()> {
|
||||
Self::open().map(|_| ())
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,14 @@
|
||||
//! Same-build helper lifecycle, private runtime initialization and owned transport. No devices yet.
|
||||
//! Same-build helper lifecycle with privately owned runtime, transport and opt-in local devices.
|
||||
|
||||
#[cfg_attr(
|
||||
not(any(
|
||||
target_os = "macos",
|
||||
all(target_os = "linux", target_env = "gnu"),
|
||||
all(windows, target_env = "msvc")
|
||||
)),
|
||||
path = "devices_unavailable.rs"
|
||||
)]
|
||||
mod devices;
|
||||
mod runtime;
|
||||
mod transport;
|
||||
mod transport_runtime;
|
||||
@@ -13,6 +22,8 @@ use codex_realtime_webrtc::Message;
|
||||
use codex_realtime_webrtc::encode_frame;
|
||||
use codex_realtime_webrtc::read_message;
|
||||
|
||||
const DEVICE_SERVICE_INTERVAL: Duration = Duration::from_millis(/*millis*/ 5);
|
||||
|
||||
const BUILD_COMMIT: &str = match option_env!("STABLE_GIT_COMMIT") {
|
||||
Some(commit) => commit,
|
||||
None => "dev",
|
||||
@@ -80,8 +91,28 @@ fn run(
|
||||
let executor = tokio::runtime::Runtime::new()?;
|
||||
let mut transport = None;
|
||||
let mut answered = false;
|
||||
let mut devices: Option<devices::Devices> = None;
|
||||
loop {
|
||||
let reply = match receiver.recv() {
|
||||
let message = if let Some(devices) = &devices {
|
||||
devices.service()?;
|
||||
receiver.recv_timeout(DEVICE_SERVICE_INTERVAL)
|
||||
} else {
|
||||
receiver
|
||||
.recv()
|
||||
.map_err(|_| mpsc::RecvTimeoutError::Disconnected)
|
||||
};
|
||||
let reply = match message {
|
||||
Ok(Message::OpenDevices {}) if devices.is_none() && runtime.is_some() && answered => {
|
||||
devices = Some(devices::Devices::open()?);
|
||||
Message::DevicesOpened {}
|
||||
}
|
||||
Ok(Message::SetAudioControls { controls }) => {
|
||||
devices
|
||||
.as_ref()
|
||||
.ok_or_else(|| io::Error::other("audio devices not open"))?
|
||||
.set_controls(controls)?;
|
||||
Message::AudioControlsApplied {}
|
||||
}
|
||||
Ok(Message::StartTransport {}) if transport.is_none() => {
|
||||
let peer = start_transport(&executor)?;
|
||||
let sdp = executor.block_on(peer.offer()).map_err(io::Error::other)?;
|
||||
@@ -108,13 +139,15 @@ fn run(
|
||||
Message::RuntimeReady {}
|
||||
}
|
||||
Ok(Message::Close {}) => {
|
||||
let _ = devices.take();
|
||||
if let Some(mut peer) = transport.take() {
|
||||
executor.block_on(peer.close()).map_err(io::Error::other)?;
|
||||
}
|
||||
output.write_all(&encode_frame(&Message::Closed {})?)?;
|
||||
return output.flush();
|
||||
}
|
||||
Err(_) => return Ok(()),
|
||||
Err(mpsc::RecvTimeoutError::Timeout) => continue,
|
||||
Err(mpsc::RecvTimeoutError::Disconnected) => return Ok(()),
|
||||
Ok(
|
||||
Message::Hello { .. }
|
||||
| Message::Ready {}
|
||||
@@ -123,6 +156,9 @@ fn run(
|
||||
| Message::ApplyAnswer { .. }
|
||||
| Message::Offer { .. }
|
||||
| Message::TransportReady {}
|
||||
| Message::OpenDevices {}
|
||||
| Message::DevicesOpened {}
|
||||
| Message::AudioControlsApplied {}
|
||||
| Message::Closed {},
|
||||
) => return Err(io::Error::other("invalid voice control sequence")),
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user