mirror of
https://github.com/openai/codex.git
synced 2026-10-21 04:01:19 +00:00
Send processed microphone audio over RTP in voice-host (#43090)
## Why `voice-host` previously drained captured audio locally without sending it to the peer. Connect capture to the outgoing media track while preserving mute boundaries and limiting stale audio. ## What changed - Resample capture and render references to 48 kHz with Rubato, apply Sonora echo cancellation, noise suppression, and gain control, and encode capture as 20 ms mono Opus packets. - Attach an Opus track to the WebRTC peer and advance RTP timestamps across mute gaps without introducing sequence-number gaps. - Prioritize queued controls and shutdown between sends. Clear pending audio and processing history on microphone mute transitions, reject delayed pre-unmute buffers, and reset partial history after capture gaps. - Size bounded capture and render queues for device rates, validate callback timing, enforce a 500 ms processing deadline, and time out stalled sends after 100 ms. Receive/decode and TUI integration remain unimplemented. The bundled Opus encoder build requires CMake and a C compiler. ## Testing Add synthetic-audio tests for Opus decoding at 44.1 and 48 kHz input rates, echo reduction, capture gaps, stale audio, queue capacity, and control priority. A loopback WebRTC test verifies RTP delivery, decoded signal, mute isolation, and timestamp and sequence continuity across mute gaps. GitOrigin-RevId: 38b8438577cfadb8124f8aeb426dffbab4845964
This commit is contained in:
Benjamin Carlsson
committed by
copyberry
copyberry
parent
0683368584
commit
7dc7c7a756
Generated
+16
File diff suppressed because one or more lines are too long
Generated
+162
@@ -844,6 +844,39 @@ version = "1.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
|
||||
|
||||
[[package]]
|
||||
name = "audio-codec-algorithms"
|
||||
version = "0.8.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1254ebf6529f3763c491acfb5ab6e960809e3c75a38584cc664f8c5667fb7107"
|
||||
|
||||
[[package]]
|
||||
name = "audioadapter"
|
||||
version = "5.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d1292ef9edf681b7426ed089004b021a42896492f58bd0c909ad44e5faec9ac4"
|
||||
|
||||
[[package]]
|
||||
name = "audioadapter-buffers"
|
||||
version = "5.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "46289a81a3bfa26d0f8b2415f8ca9decde9a66308a150568af5c102381b3c8ce"
|
||||
dependencies = [
|
||||
"audioadapter",
|
||||
"audioadapter-sample",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "audioadapter-sample"
|
||||
version = "5.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5258faecf4edbe35ec483bda9b1f7438e6b391891819269c6b5461aeaa0e8d7d"
|
||||
dependencies = [
|
||||
"audio-codec-algorithms",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "autocfg"
|
||||
version = "1.5.0"
|
||||
@@ -4996,8 +5029,12 @@ dependencies = [
|
||||
"crossbeam-queue",
|
||||
"futures",
|
||||
"libloading",
|
||||
"opus",
|
||||
"pretty_assertions",
|
||||
"rand 0.9.3",
|
||||
"rtc",
|
||||
"rubato",
|
||||
"sonora",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
"webrtc",
|
||||
@@ -10720,6 +10757,24 @@ version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
|
||||
|
||||
[[package]]
|
||||
name = "opus"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "33718946cc77d4032911d4efe03a66dbcbfbd2bb16c3da06aaeadcc637c32216"
|
||||
dependencies = [
|
||||
"opusic-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "opusic-sys"
|
||||
version = "0.7.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c9d1ecdf206421bc74343ab3bb2f30ad2abbfee41fa341f7181fecbaf957769a"
|
||||
dependencies = [
|
||||
"cmake",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ordered-float"
|
||||
version = "4.6.0"
|
||||
@@ -12873,6 +12928,20 @@ dependencies = [
|
||||
"sansio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rubato"
|
||||
version = "5.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a7cb1ffaf8738df50aab642a7f6465df81c6ba9e2818268053487165298114be"
|
||||
dependencies = [
|
||||
"audioadapter",
|
||||
"audioadapter-buffers",
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
"visibility",
|
||||
"windowfunctions",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "runfiles"
|
||||
version = "0.1.0"
|
||||
@@ -13931,6 +14000,79 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sonora"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2e1ce442ab5cb392c4067b8f2cced20be4cf12e479f6dfbf44acbe9b14ac9250"
|
||||
dependencies = [
|
||||
"sonora-aec3",
|
||||
"sonora-agc2",
|
||||
"sonora-common-audio",
|
||||
"sonora-ns",
|
||||
"sonora-simd",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sonora-aec3"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "50280e31256d09f7838aff6715ed4b7640e514dc1092a1e14013c01e552380db"
|
||||
dependencies = [
|
||||
"sonora-common-audio",
|
||||
"sonora-fft",
|
||||
"sonora-simd",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sonora-agc2"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9a93a9fa8373a33e61a7fb441d8a9e725ac6024079d2c77a845cf6abe1790b9f"
|
||||
dependencies = [
|
||||
"bytemuck",
|
||||
"derive_more 2.1.1",
|
||||
"sonora-common-audio",
|
||||
"sonora-fft",
|
||||
"sonora-simd",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sonora-common-audio"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6635bc61b24807ceac93ad0807b1292b6b0620caa6cd360d3be89fc616b833cc"
|
||||
dependencies = [
|
||||
"derive_more 2.1.1",
|
||||
"sonora-simd",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sonora-fft"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "27e272c70f34363afb5382dd1818c05f8f63ef6785635736a14e9ccbbb5ba012"
|
||||
|
||||
[[package]]
|
||||
name = "sonora-ns"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d511feef19302c716d245c4555087d161ba91c178735ea11107dcce0d0b7a86f"
|
||||
dependencies = [
|
||||
"sonora-fft",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sonora-simd"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a7cf0804e4752a5562262e143014f53043080ead4cd5a98f9278d3ceaf32d868"
|
||||
dependencies = [
|
||||
"cpufeatures 0.3.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sorted_vector_map"
|
||||
version = "0.2.1"
|
||||
@@ -15956,6 +16098,17 @@ version = "0.9.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
|
||||
|
||||
[[package]]
|
||||
name = "visibility"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d674d135b4a8c1d7e813e2f8d1c9a58308aee4a680323066025e53132218bd91"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "vsimd"
|
||||
version = "0.8.0"
|
||||
@@ -16481,6 +16634,15 @@ version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
|
||||
|
||||
[[package]]
|
||||
name = "windowfunctions"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "90628d739333b7c5d2ee0b70210b97b8cddc38440c682c96fd9e2c24c2db5f3a"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows"
|
||||
version = "0.58.0"
|
||||
|
||||
@@ -16,6 +16,7 @@ codex-process-hardening = { workspace = true }
|
||||
codex-realtime-webrtc = { workspace = true }
|
||||
futures = { workspace = true }
|
||||
libloading = "=0.8.9"
|
||||
rand = { workspace = true }
|
||||
rtc = "=0.20.3"
|
||||
tokio = { workspace = true, features = ["rt-multi-thread", "sync", "time"] }
|
||||
webrtc = "=0.20.3"
|
||||
@@ -23,6 +24,9 @@ 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"
|
||||
opus = "=0.4.0"
|
||||
rubato = { version = "=5.0.0", default-features = false }
|
||||
sonora = { version = "=0.2.0", default-features = false }
|
||||
|
||||
[dev-dependencies]
|
||||
anyhow = { workspace = true }
|
||||
|
||||
@@ -44,8 +44,14 @@ stale at one second. The queue's sample capacity must span more than that servic
|
||||
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.
|
||||
`devicesOpened` confirms device opening only. Capture now uses Rubato resampling,
|
||||
Sonora echo/noise/gain processing and 20 ms Opus encoding before sending RTP.
|
||||
Mute resets retained capture history and rejects delayed pre-unmute buffers.
|
||||
The receive/decode pipeline and TUI connection remain subsequent stages.
|
||||
|
||||
The capture encoder uses `opus 0.4.0` and its bundled `opusic-sys` build, which
|
||||
requires CMake and a C compiler. This is separate from the runtime's decoder Opus
|
||||
copy; final symbol binding and packaging validation must cover both copies.
|
||||
|
||||
Cargo Linux builds require ALSA development inputs discoverable by pkg-config
|
||||
(for example `libasound2-dev` on Debian/Ubuntu). Bazel uses the declared `alsa_lib`
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
//! A single Opus RTP track. Elapsed mute time advances its clock without inventing packet loss.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
|
||||
use rtc::media::Sample;
|
||||
use rtc::rtp_transceiver::rtp_sender::RTCRtpCodec;
|
||||
use rtc::rtp_transceiver::rtp_sender::RTCRtpCodecParameters;
|
||||
use rtc::rtp_transceiver::rtp_sender::RTCRtpCodingParameters;
|
||||
use rtc::rtp_transceiver::rtp_sender::RTCRtpEncodingParameters;
|
||||
use rtc::rtp_transceiver::rtp_sender::RtpCodecKind;
|
||||
use webrtc::media_stream::MediaStreamTrack;
|
||||
use webrtc::media_stream::track_local::static_sample::TrackLocalStaticSample;
|
||||
use webrtc::peer_connection::MediaEngine;
|
||||
|
||||
pub(crate) const SEND_TIMEOUT: Duration = Duration::from_millis(/*millis*/ 100);
|
||||
|
||||
pub(crate) struct EncodedAudio {
|
||||
pub(crate) data: Vec<u8>,
|
||||
pub(crate) at: Instant,
|
||||
}
|
||||
|
||||
pub(crate) struct AudioTrack {
|
||||
pub(crate) track: Arc<TrackLocalStaticSample>,
|
||||
ssrc: u32,
|
||||
end: Option<Instant>,
|
||||
}
|
||||
|
||||
impl AudioTrack {
|
||||
pub(crate) fn new() -> Result<(MediaEngine, Self), &'static str> {
|
||||
let ssrc = rand::random();
|
||||
let codec = RTCRtpCodec {
|
||||
mime_type: "audio/opus".into(),
|
||||
clock_rate: 48_000,
|
||||
channels: 2,
|
||||
sdp_fmtp_line: "minptime=10;useinbandfec=1".into(),
|
||||
rtcp_feedback: vec![],
|
||||
};
|
||||
let mut media = MediaEngine::default();
|
||||
media
|
||||
.register_codec(
|
||||
RTCRtpCodecParameters {
|
||||
rtp_codec: codec.clone(),
|
||||
payload_type: 111,
|
||||
},
|
||||
RtpCodecKind::Audio,
|
||||
)
|
||||
.map_err(|_| "failed to register voice codec")?;
|
||||
let track = TrackLocalStaticSample::new(MediaStreamTrack::new(
|
||||
"realtime".into(),
|
||||
format!("audio-{ssrc}"),
|
||||
"microphone".into(),
|
||||
RtpCodecKind::Audio,
|
||||
vec![RTCRtpEncodingParameters {
|
||||
rtp_coding_parameters: RTCRtpCodingParameters {
|
||||
ssrc: Some(ssrc),
|
||||
..Default::default()
|
||||
},
|
||||
codec,
|
||||
active: true,
|
||||
..Default::default()
|
||||
}],
|
||||
))
|
||||
.map_err(|_| "failed to create voice track")?;
|
||||
Ok((
|
||||
media,
|
||||
Self {
|
||||
track: Arc::new(track),
|
||||
ssrc,
|
||||
end: None,
|
||||
},
|
||||
))
|
||||
}
|
||||
|
||||
pub(crate) async fn send(&mut self, frame: EncodedAudio) -> Result<(), &'static str> {
|
||||
tokio::time::timeout(SEND_TIMEOUT, async {
|
||||
let mut gap = self
|
||||
.end
|
||||
.map(|end| frame.at.saturating_duration_since(end))
|
||||
.unwrap_or_default();
|
||||
while !gap.is_zero() {
|
||||
let duration = gap.min(Duration::from_secs(/*secs*/ 3600));
|
||||
self.track
|
||||
.write_sample(
|
||||
self.ssrc,
|
||||
/*payload_type*/ 111,
|
||||
&Sample {
|
||||
duration,
|
||||
..Default::default()
|
||||
},
|
||||
&[],
|
||||
)
|
||||
.await
|
||||
.map_err(|_| "failed to advance voice clock")?;
|
||||
gap -= duration;
|
||||
}
|
||||
let duration = Duration::from_millis(/*millis*/ 20);
|
||||
self.track
|
||||
.write_sample(
|
||||
self.ssrc,
|
||||
/*payload_type*/ 111,
|
||||
&Sample {
|
||||
data: frame.data.into(),
|
||||
duration,
|
||||
..Default::default()
|
||||
},
|
||||
&[],
|
||||
)
|
||||
.await
|
||||
.map_err(|_| "failed to send voice audio")?;
|
||||
self.end = Some(self.end.unwrap_or(frame.at).max(frame.at) + duration);
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
.map_err(|_| "voice audio sender stalled")?
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "audio_track_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,17 @@
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[tokio::test]
|
||||
async fn timestamp_jitter_does_not_accumulate_but_mute_advances_the_clock() {
|
||||
let (_, mut track) = AudioTrack::new().unwrap();
|
||||
let start = Instant::now();
|
||||
// Empty samples exercise the real packetizer without requiring a bound network peer.
|
||||
for tick in 0..500 {
|
||||
let at = start + Duration::from_millis(tick * 20 + tick % 2);
|
||||
track.send(EncodedAudio { data: vec![], at }).await.unwrap();
|
||||
}
|
||||
assert_eq!(track.end, Some(start + Duration::from_millis(10_001)));
|
||||
let at = start + Duration::from_secs(30);
|
||||
track.send(EncodedAudio { data: vec![], at }).await.unwrap();
|
||||
assert_eq!(track.end, Some(at + Duration::from_millis(20)));
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
//! Pumps bounded device queues through capture processing and the outgoing audio track.
|
||||
//! Devices retain real stream ownership; all queued and partial capture obey mute generations.
|
||||
//! Retain at most one encoded batch and return between packets so controls can take priority.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::io;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Instant;
|
||||
|
||||
use super::MAX_CAPTURE_AGE;
|
||||
use super::buffers::Buffers;
|
||||
use super::processing;
|
||||
|
||||
pub(super) struct CaptureWorker {
|
||||
pub(super) buffers: Arc<Buffers>,
|
||||
pub(super) processor: processing::Processor,
|
||||
pub(super) pending: VecDeque<crate::audio_track::EncodedAudio>,
|
||||
}
|
||||
|
||||
impl CaptureWorker {
|
||||
pub(super) fn set_controls(
|
||||
&mut self,
|
||||
controls: codex_realtime_webrtc::AudioControls,
|
||||
) -> io::Result<()> {
|
||||
let previous = self.buffers.microphone.load(Ordering::Acquire);
|
||||
Buffers::set_disabled(&self.buffers.microphone, controls.microphone_muted)?;
|
||||
if previous != self.buffers.microphone.load(Ordering::Acquire) {
|
||||
self.pending.clear();
|
||||
self.processor.reset().map_err(io::Error::other)?;
|
||||
}
|
||||
Buffers::set_disabled(&self.buffers.speaker, controls.speaker_suppressed)
|
||||
}
|
||||
|
||||
pub(super) async fn service(
|
||||
&mut self,
|
||||
audio: &mut crate::audio_track::AudioTrack,
|
||||
now: impl Fn() -> Instant,
|
||||
) -> io::Result<usize> {
|
||||
// Some backends start callbacks during stream construction, before open returns.
|
||||
self.buffers.serviced.store(true, Ordering::Release);
|
||||
let generation = self.buffers.microphone.load(Ordering::Acquire);
|
||||
for _ in 0..self.buffers.rendered.capacity() {
|
||||
let Some(frame) = self.buffers.rendered.pop() else {
|
||||
break;
|
||||
};
|
||||
if frame.at.elapsed() > MAX_CAPTURE_AGE {
|
||||
return Err(io::Error::other("audio rendering fell behind"));
|
||||
}
|
||||
self.processor.render(&frame).map_err(io::Error::other)?;
|
||||
}
|
||||
if self.pending.is_empty() {
|
||||
for _ in 0..self.buffers.capture.capacity() {
|
||||
let Some(frame) = self.buffers.capture.pop() else {
|
||||
break;
|
||||
};
|
||||
if generation.is_multiple_of(2) && frame.generation == generation {
|
||||
let frames = self.processor.capture(&frame, &now);
|
||||
self.pending.extend(frames.map_err(io::Error::other)?);
|
||||
}
|
||||
}
|
||||
}
|
||||
if self.buffers.failed.load(Ordering::Acquire) {
|
||||
return Err(io::Error::other("audio device failed"));
|
||||
}
|
||||
let Some(frame) = self.pending.pop_front() else {
|
||||
return Ok(0);
|
||||
};
|
||||
if now().saturating_duration_since(frame.at) > processing::MAX_PROCESSING_DELAY {
|
||||
return Err(io::Error::other("voice processing fell behind"));
|
||||
}
|
||||
audio.send(frame).await.map_err(io::Error::other)?;
|
||||
Ok(1)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "capture_worker_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,300 @@
|
||||
use super::super::buffers::BLOCK;
|
||||
use super::super::buffers::Frame;
|
||||
use super::*;
|
||||
use crate::audio_track::AudioTrack;
|
||||
use crate::transport::Transport;
|
||||
use codex_realtime_webrtc::AudioControls;
|
||||
use futures::future::BoxFuture;
|
||||
use pretty_assertions::assert_eq;
|
||||
use rtc::interceptor::Interceptor;
|
||||
use rtc::interceptor::NoopInterceptor;
|
||||
use rtc::interceptor::Packet;
|
||||
use rtc::interceptor::StreamInfo;
|
||||
use rtc::interceptor::TaggedPacket;
|
||||
use rtc::interceptor::interceptor;
|
||||
use rtc::sansio;
|
||||
use rtc::shared::error::Error;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::Notify;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::timeout;
|
||||
use webrtc::peer_connection::PeerConnection;
|
||||
use webrtc::peer_connection::PeerConnectionBuilder;
|
||||
use webrtc::peer_connection::PeerConnectionEventHandler;
|
||||
use webrtc::peer_connection::RTCIceGatheringState;
|
||||
use webrtc::peer_connection::RTCSessionDescription;
|
||||
|
||||
// Consume remote RTP before the upstream track queue: receipt proves the packet
|
||||
// traversed negotiation, encryption, the loopback socket, and remote decryption.
|
||||
#[derive(Interceptor)]
|
||||
struct RemoteRtp {
|
||||
#[next]
|
||||
next: NoopInterceptor,
|
||||
sender: mpsc::Sender<rtc::rtp::packet::Packet>,
|
||||
}
|
||||
|
||||
#[interceptor]
|
||||
impl RemoteRtp {
|
||||
#[overrides]
|
||||
fn handle_read(&mut self, message: TaggedPacket) -> Result<(), Self::Error> {
|
||||
if let Packet::Rtp(packet) = message.message {
|
||||
self.sender.try_send(packet).unwrap();
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct Gathered(Arc<Notify>);
|
||||
|
||||
impl PeerConnectionEventHandler for Gathered {
|
||||
fn on_ice_gathering_state_change<'a, 'async_trait>(
|
||||
&'a self,
|
||||
state: RTCIceGatheringState,
|
||||
) -> BoxFuture<'async_trait, ()>
|
||||
where
|
||||
'a: 'async_trait,
|
||||
Self: 'async_trait,
|
||||
{
|
||||
Box::pin(async move {
|
||||
if state == RTCIceGatheringState::Complete {
|
||||
self.0.notify_one();
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Synthetic callback output enters the same bounded queue as the real cpal
|
||||
// callback. No hardware availability or physical microphone behavior is claimed.
|
||||
fn enqueue(worker: &CaptureWorker, samples: usize, generation: u64, at: Instant) {
|
||||
for offset in (0..samples).step_by(BLOCK) {
|
||||
let mut frame = Frame {
|
||||
samples: [0.0; BLOCK],
|
||||
len: BLOCK.min(samples - offset),
|
||||
at: at + Duration::from_secs_f64(offset as f64 / 48_000.0),
|
||||
generation,
|
||||
};
|
||||
for (index, sample) in frame.samples.iter_mut().enumerate() {
|
||||
*sample =
|
||||
(std::f32::consts::TAU * 440.0 * (offset + index) as f32 / 48_000.0).sin() * 0.25;
|
||||
}
|
||||
assert!(worker.buffers.capture.push(frame).is_ok());
|
||||
}
|
||||
}
|
||||
|
||||
async fn receive_capture(
|
||||
worker: &mut CaptureWorker,
|
||||
local: &mut Transport,
|
||||
received: &mut mpsc::Receiver<rtc::rtp::packet::Packet>,
|
||||
now: Instant,
|
||||
) -> Vec<rtc::rtp::packet::Packet> {
|
||||
let mut count = 0;
|
||||
loop {
|
||||
let sent = worker.service(&mut local.audio, || now).await.unwrap();
|
||||
if sent == 0 {
|
||||
break;
|
||||
}
|
||||
assert_eq!(sent, 1);
|
||||
count += sent;
|
||||
}
|
||||
assert!(count > 0, "fresh capture must produce audio");
|
||||
let mut packets = Vec::new();
|
||||
let mut decoder = opus::Decoder::new(/*sample_rate*/ 48_000, opus::Channels::Mono).unwrap();
|
||||
let mut energy = 0.0_f32;
|
||||
for _ in 0..count {
|
||||
let packet = received.recv().await.unwrap();
|
||||
assert_eq!(packet.header.payload_type, 111);
|
||||
let mut decoded = [0.0; 960];
|
||||
assert_eq!(
|
||||
decoder
|
||||
.decode_float(&packet.payload, &mut decoded, /*fec*/ false)
|
||||
.unwrap(),
|
||||
960
|
||||
);
|
||||
assert!(decoded.iter().all(|sample| sample.is_finite()));
|
||||
energy += decoded.iter().map(|sample| sample * sample).sum::<f32>();
|
||||
packets.push(packet);
|
||||
}
|
||||
assert!(
|
||||
energy > 0.01,
|
||||
"decoded microphone signal must not be silence"
|
||||
);
|
||||
packets
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn capture_reaches_remote_rtp_and_mute_discards_queued_and_partial_audio() {
|
||||
timeout(Duration::from_secs(/*secs*/ 30), async {
|
||||
let (sender, mut received) = mpsc::channel(/*buffer*/ 32);
|
||||
let gathered = Arc::new(Notify::new());
|
||||
let (media, _) = AudioTrack::new().unwrap();
|
||||
let mut settings = webrtc::peer_connection::SettingEngine::default();
|
||||
settings.set_lite(/*lite*/ true);
|
||||
let remote = PeerConnectionBuilder::new()
|
||||
.with_media_engine(media)
|
||||
.with_setting_engine(settings)
|
||||
.with_handler(Arc::new(Gathered(gathered.clone())))
|
||||
.with_interceptor_registry(rtc::interceptor::Registry::from(RemoteRtp {
|
||||
next: NoopInterceptor::new(),
|
||||
sender,
|
||||
}))
|
||||
.with_udp_addrs(vec!["0.0.0.0:0"])
|
||||
.build()
|
||||
.await
|
||||
.unwrap();
|
||||
let mut local = Transport::new().await.unwrap();
|
||||
remote
|
||||
.set_remote_description(
|
||||
RTCSessionDescription::offer(local.offer().await.unwrap()).unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let answer = remote.create_answer(/*options*/ None).await.unwrap();
|
||||
remote.set_local_description(answer).await.unwrap();
|
||||
gathered.notified().await;
|
||||
local
|
||||
.apply_answer(remote.local_description().await.unwrap().sdp)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut worker = CaptureWorker {
|
||||
buffers: Arc::new(Buffers::new(
|
||||
/*input_rate*/ 48_000, /*output_rate*/ 48_000,
|
||||
)),
|
||||
pending: Default::default(),
|
||||
processor: processing::Processor::new(
|
||||
/*input_rate*/ 48_000, /*output_rate*/ 44_100,
|
||||
)
|
||||
.unwrap(),
|
||||
};
|
||||
let unmuted = || AudioControls {
|
||||
microphone_muted: false,
|
||||
speaker_suppressed: false,
|
||||
};
|
||||
let muted = || AudioControls {
|
||||
microphone_muted: true,
|
||||
speaker_suppressed: true,
|
||||
};
|
||||
worker.set_controls(unmuted()).unwrap();
|
||||
let generation = worker.buffers.microphone.load(Ordering::Acquire);
|
||||
// Freeze processing time so this transport test does not benchmark the CI worker.
|
||||
// Processor tests separately exercise the real backlog cutoff.
|
||||
let start = Instant::now();
|
||||
let mut now = start;
|
||||
// Feed a distinct speaker tone through the real render resampler and APM
|
||||
// before capture; this exercises a different output rate without hardware.
|
||||
for offset in (0..2_646).step_by(BLOCK) {
|
||||
let mut frame = Frame {
|
||||
samples: [0.0; BLOCK],
|
||||
len: BLOCK.min(2_646 - offset),
|
||||
at: start + Duration::from_secs_f64(offset as f64 / 44_100.0),
|
||||
generation: 0,
|
||||
};
|
||||
for (index, sample) in frame.samples.iter_mut().enumerate() {
|
||||
*sample = (std::f32::consts::TAU * 220.0 * (offset + index) as f32 / 44_100.0)
|
||||
.sin()
|
||||
* 0.1;
|
||||
}
|
||||
assert!(worker.buffers.rendered.push(frame).is_ok());
|
||||
}
|
||||
enqueue(&worker, /*samples*/ 2_880, generation, now);
|
||||
let before = receive_capture(&mut worker, &mut local, &mut received, now).await;
|
||||
assert!(worker.buffers.rendered.is_empty());
|
||||
|
||||
// Start with a clean processor, then retain less than one Opus packet.
|
||||
// If either resampler or partial encoder history survives the next mute,
|
||||
// the equally short fresh segment below would emit audio too early.
|
||||
worker.set_controls(muted()).unwrap();
|
||||
worker.set_controls(unmuted()).unwrap();
|
||||
let old_generation = worker.buffers.microphone.load(Ordering::Acquire);
|
||||
now = Instant::now();
|
||||
enqueue(&worker, /*samples*/ 720, old_generation, now);
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 0);
|
||||
enqueue(&worker, /*samples*/ 2_880, old_generation, now);
|
||||
worker.set_controls(muted()).unwrap();
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 0);
|
||||
|
||||
// A callback that began before mute may enqueue its old generation only
|
||||
// after unmute; a newly started callback may still carry an old device
|
||||
// timestamp. Exercise both independently through the actual service loop.
|
||||
let pre_unmute = Instant::now() - Duration::from_millis(/*millis*/ 100);
|
||||
worker.set_controls(unmuted()).unwrap();
|
||||
let generation = worker.buffers.microphone.load(Ordering::Acquire);
|
||||
let resumed = Instant::now() + Duration::from_secs(/*secs*/ 1);
|
||||
now = resumed;
|
||||
enqueue(&worker, /*samples*/ 2_880, old_generation, now);
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 0);
|
||||
enqueue(&worker, /*samples*/ 2_880, generation, pre_unmute);
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 0);
|
||||
enqueue(&worker, /*samples*/ 720, generation, now);
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 0);
|
||||
|
||||
now += Duration::from_millis(/*millis*/ 15);
|
||||
enqueue(&worker, /*samples*/ 2_880, generation, now);
|
||||
let after = receive_capture(&mut worker, &mut local, &mut received, now).await;
|
||||
let clock_gap = after[0]
|
||||
.header
|
||||
.timestamp
|
||||
.wrapping_sub(before[0].header.timestamp);
|
||||
let expected_gap = (resumed.duration_since(start).as_secs_f64() * 48_000.0) as u32;
|
||||
// Allow 1 ms for sample rounding without imposing a wall-clock throughput requirement.
|
||||
assert!(
|
||||
clock_gap.abs_diff(expected_gap) <= 48,
|
||||
"received RTP clock gap: {clock_gap}, expected {expected_gap}"
|
||||
);
|
||||
assert_eq!(
|
||||
after[0].header.sequence_number,
|
||||
before
|
||||
.last()
|
||||
.unwrap()
|
||||
.header
|
||||
.sequence_number
|
||||
.wrapping_add(1)
|
||||
);
|
||||
assert!(received.try_recv().is_err());
|
||||
|
||||
// A mute received during a send must discard the rest of that encoded
|
||||
// batch before the next service call, including after a quick unmute.
|
||||
worker.set_controls(muted()).unwrap();
|
||||
worker.set_controls(unmuted()).unwrap();
|
||||
let generation = worker.buffers.microphone.load(Ordering::Acquire);
|
||||
now = Instant::now() + Duration::from_secs(/*secs*/ 1);
|
||||
enqueue(&worker, /*samples*/ 2_880, generation, now);
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 1);
|
||||
received.recv().await.unwrap();
|
||||
assert!(!worker.pending.is_empty());
|
||||
let pending = worker.pending.len();
|
||||
worker
|
||||
.set_controls(AudioControls {
|
||||
microphone_muted: false,
|
||||
speaker_suppressed: true,
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(worker.pending.len(), pending);
|
||||
worker.set_controls(muted()).unwrap();
|
||||
worker.set_controls(unmuted()).unwrap();
|
||||
assert!(worker.pending.is_empty());
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 0);
|
||||
assert!(received.try_recv().is_err());
|
||||
|
||||
// Waiting behind a slow send cannot extend the capture freshness limit.
|
||||
let generation = worker.buffers.microphone.load(Ordering::Acquire);
|
||||
now = Instant::now() + Duration::from_secs(/*secs*/ 1);
|
||||
enqueue(&worker, /*samples*/ 2_880, generation, now);
|
||||
assert_eq!(worker.service(&mut local.audio, || now).await.unwrap(), 1);
|
||||
received.recv().await.unwrap();
|
||||
let stale = now + Duration::from_secs(/*secs*/ 1);
|
||||
assert_eq!(
|
||||
worker
|
||||
.service(&mut local.audio, || stale)
|
||||
.await
|
||||
.unwrap_err()
|
||||
.to_string(),
|
||||
"voice processing fell behind"
|
||||
);
|
||||
assert!(received.try_recv().is_err());
|
||||
local.close().await.unwrap();
|
||||
remote.close().await.unwrap();
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
//! Preallocated callback buffers pack small callbacks into full queue slots without locks.
|
||||
//! Generation changes and rejected capture buffers discard incomplete frames.
|
||||
//! Generation changes, rejected capture buffers and capture gaps discard incomplete frames.
|
||||
//! Capture/render capacity covers the worker deadline; playback and callback limits stay fixed.
|
||||
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::AtomicU64;
|
||||
@@ -11,6 +12,9 @@ use crossbeam_queue::ArrayQueue;
|
||||
|
||||
pub(super) const BLOCK: usize = 256;
|
||||
pub(super) const QUEUE_CAPACITY: usize = 32;
|
||||
pub(super) const MAX_CALLBACK_FRAMES: usize = BLOCK * QUEUE_CAPACITY;
|
||||
// Allow ordinary timestamp jitter consistently at both capture buffering boundaries.
|
||||
pub(super) const CAPTURE_GAP_TOLERANCE: Duration = Duration::from_millis(/*millis*/ 20);
|
||||
|
||||
pub(super) struct Frame {
|
||||
pub(super) samples: [f32; BLOCK],
|
||||
@@ -31,6 +35,17 @@ impl FramePacker {
|
||||
self.frame = None;
|
||||
}
|
||||
|
||||
// Capture checks continuity before packing can hide a gap inside an old partial block.
|
||||
// Render callbacks retain their separate existing packing behavior.
|
||||
pub(super) fn discard_capture_gap(&mut self, start: Instant, rate: f64) {
|
||||
if self.frame.as_ref().is_some_and(|partial| {
|
||||
let end = partial.at + Duration::from_secs_f64(partial.len as f64 / rate);
|
||||
start.saturating_duration_since(end) > CAPTURE_GAP_TOLERANCE
|
||||
}) {
|
||||
self.reset();
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn push(&mut self, frame: Frame, rate: f64, queue: &ArrayQueue<Frame>) -> bool {
|
||||
if self
|
||||
.frame
|
||||
@@ -80,10 +95,19 @@ pub(super) struct Buffers {
|
||||
}
|
||||
|
||||
impl Buffers {
|
||||
pub(super) fn new() -> Self {
|
||||
pub(super) fn new(input_rate: u32, output_rate: u32) -> Self {
|
||||
// Rates are validated before opening devices. A pending batch can defer
|
||||
// capture consumption until its freshness deadline plus one in-flight send.
|
||||
let budget = super::processing::MAX_PROCESSING_DELAY
|
||||
+ crate::audio_track::SEND_TIMEOUT
|
||||
+ 2 * crate::DEVICE_SERVICE_INTERVAL;
|
||||
let capacity = |rate: u32| {
|
||||
let samples = (u128::from(rate) * budget.as_nanos()).div_ceil(1_000_000_000);
|
||||
(samples as usize + MAX_CALLBACK_FRAMES + BLOCK).div_ceil(BLOCK)
|
||||
};
|
||||
Self {
|
||||
capture: ArrayQueue::new(QUEUE_CAPACITY),
|
||||
rendered: ArrayQueue::new(QUEUE_CAPACITY),
|
||||
capture: ArrayQueue::new(capacity(input_rate)),
|
||||
rendered: ArrayQueue::new(capacity(output_rate)),
|
||||
playback: ArrayQueue::new(QUEUE_CAPACITY),
|
||||
microphone: AtomicU64::new(/*v*/ 1),
|
||||
speaker: AtomicU64::new(/*v*/ 1),
|
||||
|
||||
@@ -41,7 +41,7 @@ fn unmute_rejects_backlog_until_capture_reaches_the_following_callback() {
|
||||
|
||||
#[test]
|
||||
fn suppression_discards_partial_and_queued_previous_generations() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000);
|
||||
let mut playback = Playback::default();
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
let generation = buffers.speaker.load(Ordering::Acquire);
|
||||
@@ -68,7 +68,7 @@ fn suppression_discards_partial_and_queued_previous_generations() {
|
||||
|
||||
#[test]
|
||||
fn output_is_finite_and_bounded_and_underflow_is_silence() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000);
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
let generation = buffers.speaker.load(Ordering::Acquire);
|
||||
let mut samples = [0.0; BLOCK];
|
||||
@@ -91,7 +91,7 @@ fn output_is_finite_and_bounded_and_underflow_is_silence() {
|
||||
|
||||
#[test]
|
||||
fn repeated_controls_preserve_epoch_but_transitions_invalidate_capture() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000);
|
||||
Buffers::set_disabled(&buffers.microphone, /*disabled*/ false).unwrap();
|
||||
let first = buffers.microphone.load(Ordering::Acquire);
|
||||
Buffers::set_disabled(&buffers.microphone, /*disabled*/ false).unwrap();
|
||||
@@ -103,7 +103,7 @@ fn repeated_controls_preserve_epoch_but_transitions_invalidate_capture() {
|
||||
|
||||
#[test]
|
||||
fn tiny_callbacks_share_slots_and_preserve_the_oldest_timestamp() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000);
|
||||
let start = Instant::now();
|
||||
for queue in [&buffers.capture, &buffers.rendered] {
|
||||
let mut packer = FramePacker::default();
|
||||
@@ -210,3 +210,43 @@ fn packing_preserves_a_remainder_timestamp_across_uneven_callbacks() {
|
||||
);
|
||||
assert!(queue.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capture_queue_retains_high_rate_audio_during_bounded_service_pause() {
|
||||
let buffers = Buffers::new(/*input_rate*/ 384_000, /*output_rate*/ 48_000);
|
||||
let mut packer = FramePacker::default();
|
||||
let start = Instant::now();
|
||||
// 630 ms covers a pending batch's 500 ms deadline, an in-flight 100 ms
|
||||
// send and callback/service margins. Consumption is deliberately paused.
|
||||
let samples = 384_000 * 63 / 100;
|
||||
for offset in (0..samples).step_by(17) {
|
||||
assert!(packer.push(
|
||||
Frame {
|
||||
samples: [0.25; BLOCK],
|
||||
len: (samples - offset).min(17),
|
||||
at: start + Duration::from_secs_f64(offset as f64 / 384_000.0),
|
||||
generation: 2,
|
||||
},
|
||||
/*rate*/ 384_000.0,
|
||||
&buffers.capture,
|
||||
));
|
||||
}
|
||||
assert_eq!(buffers.capture.len(), samples / BLOCK);
|
||||
// Enlarging the queue must not turn overflow into unbounded accumulation.
|
||||
for _ in buffers.capture.len()..=buffers.capture.capacity() {
|
||||
let available = buffers.capture.len() < buffers.capture.capacity();
|
||||
assert_eq!(
|
||||
packer.push(
|
||||
Frame {
|
||||
samples: [0.25; BLOCK],
|
||||
len: BLOCK,
|
||||
at: start,
|
||||
generation: 2,
|
||||
},
|
||||
/*rate*/ 384_000.0,
|
||||
&buffers.capture,
|
||||
),
|
||||
available,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,10 @@
|
||||
|
||||
#[path = "device_buffers.rs"]
|
||||
mod buffers;
|
||||
#[path = "capture_worker.rs"]
|
||||
mod capture_worker;
|
||||
#[path = "processing.rs"]
|
||||
mod processing;
|
||||
|
||||
use std::io;
|
||||
use std::sync::Arc;
|
||||
@@ -25,6 +29,7 @@ use buffers::Buffers;
|
||||
use buffers::CaptureBoundary;
|
||||
use buffers::Frame;
|
||||
use buffers::FramePacker;
|
||||
use buffers::MAX_CALLBACK_FRAMES;
|
||||
use buffers::Playback;
|
||||
use buffers::QUEUE_CAPACITY;
|
||||
|
||||
@@ -33,7 +38,7 @@ const MAX_CAPTURE_AGE: Duration = Duration::from_secs(/*secs*/ 1);
|
||||
pub(super) struct Devices {
|
||||
_input: cpal::Stream,
|
||||
_output: cpal::Stream,
|
||||
buffers: Arc<Buffers>,
|
||||
worker: capture_worker::CaptureWorker,
|
||||
}
|
||||
|
||||
macro_rules! stream {
|
||||
@@ -79,7 +84,22 @@ impl Devices {
|
||||
}
|
||||
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 buffers = Arc::new(Buffers::new(
|
||||
input_config.sample_rate(),
|
||||
output_config.sample_rate(),
|
||||
));
|
||||
let processor =
|
||||
processing::Processor::new(input_config.sample_rate(), output_config.sample_rate())
|
||||
.map_err(io::Error::other)?;
|
||||
let (cpal::BufferSize::Fixed(input_frames), cpal::BufferSize::Fixed(output_frames)) = (
|
||||
input_stream_config.buffer_size,
|
||||
output_stream_config.buffer_size,
|
||||
) else {
|
||||
return Err(io::Error::other("audio callback size unavailable"));
|
||||
};
|
||||
processor
|
||||
.validate_callback_timing(input_frames, output_frames)
|
||||
.map_err(io::Error::other)?;
|
||||
let input = stream!(
|
||||
input_config.sample_format(),
|
||||
build_input,
|
||||
@@ -105,37 +125,26 @@ impl Devices {
|
||||
Ok(Self {
|
||||
_input: input,
|
||||
_output: output,
|
||||
buffers,
|
||||
worker: capture_worker::CaptureWorker {
|
||||
buffers,
|
||||
processor,
|
||||
pending: Default::default(),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn set_controls(
|
||||
&self,
|
||||
&mut 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)
|
||||
self.worker.set_controls(controls)
|
||||
}
|
||||
|
||||
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(())
|
||||
pub(super) async fn service(
|
||||
&mut self,
|
||||
audio: &mut crate::audio_track::AudioTrack,
|
||||
) -> io::Result<usize> {
|
||||
self.worker.service(audio, Instant::now).await
|
||||
}
|
||||
}
|
||||
|
||||
@@ -146,13 +155,8 @@ fn bounded_stream_config(
|
||||
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);
|
||||
let max = max.min(MAX_CALLBACK_FRAMES as u32);
|
||||
if min > max {
|
||||
return Err(io::Error::other("unsupported audio callback size range"));
|
||||
}
|
||||
@@ -225,6 +229,7 @@ where
|
||||
.unwrap_or_default(),
|
||||
)
|
||||
.unwrap_or_else(Instant::now);
|
||||
capture.discard_capture_gap(start, rate);
|
||||
for (index, chunk) in data.chunks(BLOCK * channels).enumerate() {
|
||||
let mut frame = Frame {
|
||||
samples: [0.0; BLOCK],
|
||||
|
||||
@@ -3,7 +3,7 @@ use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn startup_output_waits_for_service_but_later_overflow_still_fails() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 384_000);
|
||||
let mut state = OutputState::default();
|
||||
let mut output = [1.0_f32; BLOCK * 2];
|
||||
let start = Instant::now();
|
||||
@@ -46,7 +46,7 @@ fn startup_output_waits_for_service_but_later_overflow_still_fails() {
|
||||
),
|
||||
);
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
for _ in 0..QUEUE_CAPACITY {
|
||||
for _ in 0..buffers.rendered.capacity() {
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 2,
|
||||
@@ -96,7 +96,7 @@ fn callback_configuration_fits_supported_range_and_actual_queue() {
|
||||
buffer_size: cpal::BufferSize::Fixed(frames),
|
||||
}
|
||||
);
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(rate, rate);
|
||||
for queue in [&buffers.capture, &buffers.rendered] {
|
||||
for offset in (0..frames as usize).step_by(BLOCK) {
|
||||
assert!(
|
||||
@@ -115,23 +115,20 @@ fn callback_configuration_fits_supported_range_and_actual_queue() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn callback_configuration_reserves_room_for_partial_and_following_callbacks() {
|
||||
for min in [6_016, 6_017, 8_192] {
|
||||
fn callback_configuration_fits_dynamic_queue_with_following_callbacks() {
|
||||
for min in [6_016, 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 {
|
||||
let cpal::BufferSize::Fixed(frames) =
|
||||
bounded_stream_config(&supported).unwrap().buffer_size
|
||||
else {
|
||||
panic!("callback size must be bounded");
|
||||
};
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 384_000);
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
let mut output = OutputState::default();
|
||||
let start = Instant::now();
|
||||
@@ -146,7 +143,11 @@ fn callback_configuration_reserves_room_for_partial_and_following_callbacks() {
|
||||
&mut output,
|
||||
);
|
||||
}
|
||||
assert_eq!(buffers.rendered.len(), QUEUE_CAPACITY - 1);
|
||||
assert_eq!(
|
||||
buffers.rendered.len(),
|
||||
(BLOCK - 1 + frames as usize + 15 * 128) / BLOCK,
|
||||
);
|
||||
assert!(buffers.rendered.len() < buffers.rendered.capacity());
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
}
|
||||
@@ -191,9 +192,41 @@ fn callback_timing_must_fit_service_and_stale_limits() {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn callback_pair_must_fit_the_processing_deadline() {
|
||||
for (input_rate, input_frames, output_rate, output_frames, accepted) in [
|
||||
(8_000, 7_680, 48_000, 480, false),
|
||||
(8_000, 2_000, 8_000, 2_000, false),
|
||||
(48_000, 480, 48_000, 480, true),
|
||||
(44_100, 441, 48_000, 480, true),
|
||||
(8_000, 1_600, 48_000, 480, true),
|
||||
] {
|
||||
for (rate, frames) in [(input_rate, input_frames), (output_rate, output_frames)] {
|
||||
let supported = cpal::SupportedStreamConfig::new(
|
||||
/*channels*/ 1,
|
||||
rate,
|
||||
cpal::SupportedBufferSize::Range {
|
||||
min: frames,
|
||||
max: frames,
|
||||
},
|
||||
cpal::SampleFormat::F32,
|
||||
);
|
||||
assert!(bounded_stream_config(&supported).is_ok());
|
||||
}
|
||||
let processor = processing::Processor::new(input_rate, output_rate).unwrap();
|
||||
assert_eq!(
|
||||
processor
|
||||
.validate_callback_timing(input_frames, output_frames)
|
||||
.is_ok(),
|
||||
accepted,
|
||||
"input={input_rate}/{input_frames}, output={output_rate}/{output_frames}",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tiny_output_callbacks_pack_references_without_delaying_rendering() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 384_000);
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
let generation = buffers.speaker.load(Ordering::Acquire);
|
||||
@@ -228,13 +261,37 @@ fn tiny_output_callbacks_pack_references_without_delaying_rendering() {
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rendered_queue_retains_high_rate_callbacks_during_bounded_service_pause() {
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 384_000);
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
let mut state = OutputState::default();
|
||||
let start = Instant::now();
|
||||
let samples = 384_000 * 63 / 100;
|
||||
for offset in (0..samples).step_by(17) {
|
||||
let mut output = [1.0_f32; 34];
|
||||
let len = (samples - offset).min(17) * 2;
|
||||
render_output(
|
||||
&mut output[..len],
|
||||
/*channels*/ 2,
|
||||
/*rate*/ 384_000.0,
|
||||
start + Duration::from_secs_f64(offset as f64 / 384_000.0),
|
||||
&buffers,
|
||||
&mut state,
|
||||
);
|
||||
assert_eq!(&output[..len], &[0.0; 34][..len]);
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
}
|
||||
assert_eq!(buffers.rendered.len(), samples / BLOCK);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn suppressed_output_does_not_fail_when_reference_queue_is_full() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 384_000);
|
||||
buffers.serviced.store(true, Ordering::Release);
|
||||
let mut state = OutputState::default();
|
||||
let mut output = [1.0_f32; BLOCK];
|
||||
for _ in 0..QUEUE_CAPACITY * 3 {
|
||||
for _ in 0..buffers.rendered.capacity() * 3 {
|
||||
render_output(
|
||||
&mut output,
|
||||
/*channels*/ 1,
|
||||
@@ -245,7 +302,7 @@ fn suppressed_output_does_not_fail_when_reference_queue_is_full() {
|
||||
);
|
||||
assert_eq!(output, [0.0; BLOCK]);
|
||||
}
|
||||
assert_eq!(buffers.rendered.len(), QUEUE_CAPACITY);
|
||||
assert_eq!(buffers.rendered.len(), buffers.rendered.capacity());
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
Buffers::set_disabled(&buffers.speaker, /*disabled*/ false).unwrap();
|
||||
render_output(
|
||||
@@ -261,7 +318,7 @@ fn suppressed_output_does_not_fail_when_reference_queue_is_full() {
|
||||
|
||||
#[test]
|
||||
fn stream_xruns_recover_without_clearing_device_failure() {
|
||||
let buffers = Buffers::new();
|
||||
let buffers = Buffers::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000);
|
||||
handle_stream_error(&buffers, cpal::ErrorKind::Xrun.into());
|
||||
assert!(!buffers.failed.load(Ordering::Acquire));
|
||||
handle_stream_error(&buffers, cpal::ErrorKind::DeviceNotAvailable.into());
|
||||
|
||||
@@ -15,7 +15,10 @@ impl Devices {
|
||||
Self::open().map(|_| ())
|
||||
}
|
||||
|
||||
pub(super) fn service(&self) -> io::Result<()> {
|
||||
Self::open().map(|_| ())
|
||||
pub(super) async fn service(
|
||||
&mut self,
|
||||
_: &mut crate::audio_track::AudioTrack,
|
||||
) -> io::Result<usize> {
|
||||
Self::open().map(|_| 0)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,16 @@
|
||||
//! Same-build helper lifecycle with privately owned runtime, transport and opt-in local devices.
|
||||
//! Queued privacy controls take priority over starting another capture batch.
|
||||
|
||||
// Non-native helper targets negotiate transport but have no device backend to send audio.
|
||||
#[cfg_attr(
|
||||
not(any(
|
||||
target_os = "macos",
|
||||
all(target_os = "linux", target_env = "gnu"),
|
||||
all(windows, target_env = "msvc")
|
||||
)),
|
||||
allow(dead_code)
|
||||
)]
|
||||
mod audio_track;
|
||||
#[cfg_attr(
|
||||
not(any(
|
||||
target_os = "macos",
|
||||
@@ -89,13 +100,19 @@ fn run(
|
||||
output.flush()?;
|
||||
let mut runtime = None;
|
||||
let executor = tokio::runtime::Runtime::new()?;
|
||||
let mut transport = None;
|
||||
let mut transport: Option<transport::Transport> = None;
|
||||
let mut answered = false;
|
||||
let mut devices: Option<devices::Devices> = None;
|
||||
loop {
|
||||
let message = if let Some(devices) = &devices {
|
||||
devices.service()?;
|
||||
receiver.recv_timeout(DEVICE_SERVICE_INTERVAL)
|
||||
let message = if let Some(devices) = &mut devices {
|
||||
let peer = transport
|
||||
.as_mut()
|
||||
.ok_or_else(|| io::Error::other("voice peer not started"))?;
|
||||
wait_for_control(&receiver, || {
|
||||
executor.block_on(devices.service(&mut peer.audio))?;
|
||||
Ok(())
|
||||
})?
|
||||
.ok_or(mpsc::RecvTimeoutError::Disconnected)
|
||||
} else {
|
||||
receiver
|
||||
.recv()
|
||||
@@ -108,7 +125,7 @@ fn run(
|
||||
}
|
||||
Ok(Message::SetAudioControls { controls }) => {
|
||||
devices
|
||||
.as_ref()
|
||||
.as_mut()
|
||||
.ok_or_else(|| io::Error::other("audio devices not open"))?
|
||||
.set_controls(controls)?;
|
||||
Message::AudioControlsApplied {}
|
||||
@@ -167,6 +184,21 @@ fn run(
|
||||
}
|
||||
}
|
||||
|
||||
// Pending privacy controls and shutdown always precede the next capture batch.
|
||||
// A command arriving after service begins cannot retract an already in-flight packet.
|
||||
fn wait_for_control(
|
||||
receiver: &mpsc::Receiver<Message>,
|
||||
mut service: impl FnMut() -> io::Result<()>,
|
||||
) -> io::Result<Option<Message>> {
|
||||
loop {
|
||||
match receiver.recv_timeout(DEVICE_SERVICE_INTERVAL) {
|
||||
Ok(message) => return Ok(Some(message)),
|
||||
Err(mpsc::RecvTimeoutError::Disconnected) => return Ok(None),
|
||||
Err(mpsc::RecvTimeoutError::Timeout) => service()?,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "main_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
//! Verify the real parent-loss watchdog while the startup worker is deterministically blocked.
|
||||
//! Verify control-first capture servicing and the real parent-loss watchdog during blocked startup.
|
||||
|
||||
use super::*;
|
||||
use std::process::Stdio;
|
||||
@@ -12,6 +12,48 @@ use tokio::time::timeout;
|
||||
const STARTUP_BLOCKED: &[u8] = b"voice startup blocked\n";
|
||||
const CHILD_ENV: &str = "CODEX_VOICE_WATCHDOG_TEST_CHILD";
|
||||
|
||||
#[test]
|
||||
fn queued_privacy_controls_and_close_precede_capture_service() {
|
||||
for message in [
|
||||
Message::SetAudioControls {
|
||||
controls: codex_realtime_webrtc::AudioControls {
|
||||
microphone_muted: true,
|
||||
speaker_suppressed: true,
|
||||
},
|
||||
},
|
||||
Message::Close {},
|
||||
] {
|
||||
let expected = encode_frame(&message).unwrap();
|
||||
let (sender, receiver) = mpsc::sync_channel(/*bound*/ 1);
|
||||
sender.send(message).unwrap();
|
||||
let next = wait_for_control(&receiver, || panic!("capture preceded queued control"))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(encode_frame(&next).unwrap(), expected);
|
||||
drop(sender);
|
||||
assert!(
|
||||
wait_for_control(&receiver, || panic!("capture after parent loss"))
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn idle_service_rechecks_controls_before_another_batch() {
|
||||
let (sender, receiver) = mpsc::sync_channel(/*bound*/ 1);
|
||||
let mut services = 0;
|
||||
let next = wait_for_control(&receiver, || {
|
||||
services += 1;
|
||||
sender.send(Message::Close {}).unwrap();
|
||||
Ok(())
|
||||
})
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(next, Message::Close {});
|
||||
assert_eq!(services, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore = "subprocess fixture for the parent-loss watchdog test"]
|
||||
fn blocked_startup_fixture() {
|
||||
|
||||
@@ -0,0 +1,251 @@
|
||||
//! Streaming device-rate conversion, echo/noise/gain processing, and 20 ms Opus encoding.
|
||||
//! Mute transitions and capture gaps reset history; delayed pre-unmute buffers are discarded.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
|
||||
use rubato::Resampler;
|
||||
use rubato::audioadapter_buffers::direct::InterleavedSlice;
|
||||
use sonora::AudioProcessing;
|
||||
use sonora::StreamConfig;
|
||||
use sonora::config::AdaptiveDigital;
|
||||
use sonora::config::EchoCanceller;
|
||||
use sonora::config::GainController2;
|
||||
use sonora::config::NoiseSuppression;
|
||||
|
||||
use super::buffers::CAPTURE_GAP_TOLERANCE;
|
||||
use super::buffers::Frame;
|
||||
use crate::audio_track::EncodedAudio;
|
||||
|
||||
type Result<T> = std::result::Result<T, &'static str>;
|
||||
const BLOCK: usize = 480;
|
||||
pub(super) const MAX_PROCESSING_DELAY: Duration = Duration::from_millis(/*millis*/ 500);
|
||||
|
||||
struct Converter {
|
||||
resampler: rubato::Async<f32>,
|
||||
input: VecDeque<f32>,
|
||||
output: VecDeque<f32>,
|
||||
scratch: Vec<f32>,
|
||||
rate: u32,
|
||||
end: Option<Instant>,
|
||||
}
|
||||
|
||||
impl Converter {
|
||||
fn new(rate: u32) -> Result<Self> {
|
||||
let resampler = rubato::Async::new_sinc(
|
||||
48_000.0 / f64::from(rate),
|
||||
/*max_resample_ratio_relative*/ 1.0,
|
||||
&rubato::SincInterpolationParameters::default(),
|
||||
rate.div_ceil(100) as usize,
|
||||
/*nbr_channels*/ 1,
|
||||
rubato::FixedAsync::Input,
|
||||
)
|
||||
.map_err(|_| "failed to create voice resampler")?;
|
||||
let scratch = vec![0.0; resampler.output_frames_max()];
|
||||
Ok(Self {
|
||||
resampler,
|
||||
input: VecDeque::new(),
|
||||
output: VecDeque::new(),
|
||||
scratch,
|
||||
rate,
|
||||
end: None,
|
||||
})
|
||||
}
|
||||
|
||||
fn push(&mut self, frame: &Frame) -> Result<()> {
|
||||
if frame.len > frame.samples.len()
|
||||
|| self.input.len() + frame.len > self.rate as usize
|
||||
|| self.output.len() > 48_000
|
||||
{
|
||||
return Err("voice resampler backlog exceeded");
|
||||
}
|
||||
self.end =
|
||||
Some(frame.at + Duration::from_secs_f64(frame.len as f64 / f64::from(self.rate)));
|
||||
self.input.extend(&frame.samples[..frame.len]);
|
||||
while self.input.len() >= self.resampler.input_frames_next() {
|
||||
let samples = self.input.make_contiguous();
|
||||
let input = InterleavedSlice::new(samples, /*channels*/ 1, samples.len())
|
||||
.map_err(|_| "invalid voice resampler input")?;
|
||||
let capacity = self.scratch.len();
|
||||
let mut output =
|
||||
InterleavedSlice::new_mut(&mut self.scratch, /*channels*/ 1, capacity)
|
||||
.map_err(|_| "invalid voice resampler output")?;
|
||||
let (consumed, produced) = self
|
||||
.resampler
|
||||
.process_into_buffer(&input, &mut output, /*indexing*/ None)
|
||||
.map_err(|_| "failed to resample voice audio")?;
|
||||
self.input.drain(..consumed);
|
||||
self.output.extend(&self.scratch[..produced]);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn next(&mut self) -> Option<(Instant, [f32; BLOCK])> {
|
||||
if self.output.len() < BLOCK {
|
||||
return None;
|
||||
}
|
||||
let delay = self.input.len() as f64 / f64::from(self.rate)
|
||||
+ (self.output.len() + self.resampler.output_delay()) as f64 / 48_000.0;
|
||||
let end = self.end?;
|
||||
let at = end
|
||||
.checked_sub(Duration::from_secs_f64(delay))
|
||||
.unwrap_or(end);
|
||||
let mut output = [0.0; BLOCK];
|
||||
for (out, sample) in output.iter_mut().zip(self.output.drain(..BLOCK)) {
|
||||
*out = sample;
|
||||
}
|
||||
Some((at, output))
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct Processor {
|
||||
capture: Converter,
|
||||
render: Converter,
|
||||
apm: AudioProcessing,
|
||||
encoder: opus::Encoder,
|
||||
pending: Vec<f32>,
|
||||
at: Instant,
|
||||
cutoff: Instant,
|
||||
render_delay: i64,
|
||||
}
|
||||
|
||||
impl Processor {
|
||||
pub(super) fn new(input_rate: u32, output_rate: u32) -> Result<Self> {
|
||||
Ok(Self {
|
||||
capture: Converter::new(input_rate)?,
|
||||
render: Converter::new(output_rate)?,
|
||||
apm: AudioProcessing::builder()
|
||||
.config(sonora::Config {
|
||||
echo_canceller: Some(EchoCanceller::default()),
|
||||
noise_suppression: Some(NoiseSuppression::default()),
|
||||
gain_controller2: Some(GainController2 {
|
||||
adaptive_digital: Some(AdaptiveDigital::default()),
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
})
|
||||
.capture_config(StreamConfig::new(
|
||||
/*sample_rate_hz*/ 48_000, /*num_channels*/ 1,
|
||||
))
|
||||
.render_config(StreamConfig::new(
|
||||
/*sample_rate_hz*/ 48_000, /*num_channels*/ 1,
|
||||
))
|
||||
.build(),
|
||||
encoder: opus::Encoder::new(
|
||||
/*sample_rate*/ 48_000,
|
||||
opus::Channels::Mono,
|
||||
opus::Application::Voip,
|
||||
)
|
||||
.map_err(|_| "failed to create voice encoder")?,
|
||||
pending: Vec::with_capacity(960),
|
||||
at: Instant::now(),
|
||||
cutoff: Instant::now(),
|
||||
render_delay: 0,
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn reset(&mut self) -> Result<()> {
|
||||
*self = Self::new(self.capture.rate, self.render.rate)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn validate_callback_timing(
|
||||
&self,
|
||||
input_frames: u32,
|
||||
output_frames: u32,
|
||||
) -> Result<()> {
|
||||
let mut delay = super::super::DEVICE_SERVICE_INTERVAL;
|
||||
for (converter, frames) in [(&self.capture, input_frames), (&self.render, output_frames)] {
|
||||
// Include callback and queue packing latency, plus the resampler's
|
||||
// actual delay and one processing block, for both device directions.
|
||||
delay += Duration::from_secs_f64(
|
||||
(f64::from(frames) + super::buffers::BLOCK as f64) / f64::from(converter.rate)
|
||||
+ (converter.resampler.output_delay() + BLOCK) as f64 / 48_000.0,
|
||||
);
|
||||
}
|
||||
if delay >= MAX_PROCESSING_DELAY {
|
||||
return Err("audio callback timing exceeds processing deadline");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn render(&mut self, frame: &Frame) -> Result<()> {
|
||||
self.render.push(frame)?;
|
||||
while let Some((at, input)) = self.render.next() {
|
||||
let mut output = [0.0; BLOCK];
|
||||
self.apm
|
||||
.process_render_f32(&[&input], &mut [&mut output])
|
||||
.map_err(|_| "voice echo reference failed")?;
|
||||
let now = Instant::now();
|
||||
self.render_delay = if at > now {
|
||||
at.duration_since(now).as_millis() as i64
|
||||
} else {
|
||||
-(now.duration_since(at).as_millis() as i64)
|
||||
};
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn capture(
|
||||
&mut self,
|
||||
frame: &Frame,
|
||||
now: impl Fn() -> Instant,
|
||||
) -> Result<Vec<EncodedAudio>> {
|
||||
if frame.at < self.cutoff {
|
||||
return Ok(vec![]);
|
||||
}
|
||||
if self
|
||||
.capture
|
||||
.end
|
||||
.is_some_and(|end| frame.at.saturating_duration_since(end) > CAPTURE_GAP_TOLERANCE)
|
||||
{
|
||||
// Drop resampler, APM and half-packet history without advancing the
|
||||
// unmute boundary: this frame may already have waited in the queue.
|
||||
let cutoff = self.cutoff;
|
||||
self.reset()?;
|
||||
self.cutoff = cutoff;
|
||||
}
|
||||
self.capture.push(frame)?;
|
||||
let mut encoded = Vec::new();
|
||||
while let Some((at, input)) = self.capture.next() {
|
||||
let capture_age = now().saturating_duration_since(at);
|
||||
if capture_age > MAX_PROCESSING_DELAY {
|
||||
return Err("voice processing fell behind");
|
||||
}
|
||||
let delay = (capture_age.as_millis() as i64 + self.render_delay).max(0);
|
||||
if delay > MAX_PROCESSING_DELAY.as_millis() as i64 {
|
||||
return Err("voice processing fell behind");
|
||||
}
|
||||
self.apm
|
||||
.set_stream_delay_ms(delay as i32)
|
||||
.map_err(|_| "invalid voice echo delay")?;
|
||||
let mut output = [0.0; BLOCK];
|
||||
self.apm
|
||||
.process_capture_f32(&[&input], &mut [&mut output])
|
||||
.map_err(|_| "voice capture processing failed")?;
|
||||
if output.iter().any(|sample| !sample.is_finite()) {
|
||||
return Err("invalid processed voice audio");
|
||||
}
|
||||
if self.pending.is_empty() {
|
||||
self.at = at;
|
||||
}
|
||||
self.pending.extend(output);
|
||||
if self.pending.len() == 960 {
|
||||
let mut data = vec![0; 1275];
|
||||
let len = self
|
||||
.encoder
|
||||
.encode_float(&self.pending, &mut data)
|
||||
.map_err(|_| "failed to encode voice audio")?;
|
||||
data.truncate(len);
|
||||
encoded.push(EncodedAudio { data, at: self.at });
|
||||
self.pending.clear();
|
||||
}
|
||||
}
|
||||
Ok(encoded)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "processing_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,285 @@
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn rendered_reference_changes_decoded_echo_capture() {
|
||||
let mut processors = [
|
||||
Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 44_100).unwrap(),
|
||||
Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 44_100).unwrap(),
|
||||
];
|
||||
for processor in &mut processors {
|
||||
// Isolate echo cancellation from independent noise/gain adaptation.
|
||||
processor.apm.apply_config(sonora::Config {
|
||||
echo_canceller: Some(EchoCanceller::default()),
|
||||
..Default::default()
|
||||
});
|
||||
}
|
||||
let start = Instant::now();
|
||||
let signal = |seconds: f32| {
|
||||
[311.0, 719.0, 1_423.0]
|
||||
.into_iter()
|
||||
.map(|frequency| (std::f32::consts::TAU * frequency * seconds).sin() * 0.1)
|
||||
.sum::<f32>()
|
||||
};
|
||||
let mut packets = [Vec::new(), Vec::new()];
|
||||
// Two seconds of deterministic samples allow AEC startup, without sleeping.
|
||||
for block in 0..200 {
|
||||
for offset in [0, 256] {
|
||||
let mut frame = Frame {
|
||||
samples: [0.0; 256],
|
||||
len: 256.min(441 - offset),
|
||||
at: start - Duration::from_secs(/*secs*/ 10)
|
||||
+ Duration::from_secs_f64((block * 441 + offset) as f64 / 44_100.0),
|
||||
generation: 0,
|
||||
};
|
||||
for (index, sample) in frame.samples[..frame.len].iter_mut().enumerate() {
|
||||
*sample = signal((block * 441 + offset + index) as f32 / 44_100.0);
|
||||
}
|
||||
processors[0].render(&frame).unwrap();
|
||||
}
|
||||
// Keep delay metadata identical: only consuming reference samples may
|
||||
// explain the difference in the decoded capture signal.
|
||||
processors[1].render_delay = processors[0].render_delay;
|
||||
for offset in [0, 256] {
|
||||
let mut frame = Frame {
|
||||
samples: [0.0; 256],
|
||||
len: 256.min(480 - offset),
|
||||
at: start + Duration::from_secs_f64((block * 480 + offset) as f64 / 48_000.0),
|
||||
generation: 2,
|
||||
};
|
||||
for (index, sample) in frame.samples[..frame.len].iter_mut().enumerate() {
|
||||
*sample = signal((block * 480 + offset + index) as f32 / 48_000.0 - 0.05);
|
||||
}
|
||||
for (processor, packets) in processors.iter_mut().zip(&mut packets) {
|
||||
packets.extend(processor.capture(&frame, || frame.at).unwrap());
|
||||
}
|
||||
}
|
||||
}
|
||||
let decoded = packets.map(|packets| {
|
||||
let mut decoder = opus::Decoder::new(/*sample_rate*/ 48_000, opus::Channels::Mono).unwrap();
|
||||
let mut decoded = Vec::new();
|
||||
for packet in packets {
|
||||
let mut samples = [0.0; 960];
|
||||
assert_eq!(
|
||||
decoder
|
||||
.decode_float(&packet.data, &mut samples, /*fec*/ false)
|
||||
.unwrap(),
|
||||
960
|
||||
);
|
||||
assert!(samples.iter().all(|sample| sample.is_finite()));
|
||||
decoded.extend(samples);
|
||||
}
|
||||
decoded
|
||||
});
|
||||
assert_eq!(decoded[0].len(), decoded[1].len());
|
||||
assert!(decoded[0].len() > 48_000);
|
||||
let energy = decoded.map(|samples| {
|
||||
samples[48_000..]
|
||||
.iter()
|
||||
.map(|sample| sample * sample)
|
||||
.sum::<f32>()
|
||||
});
|
||||
assert!(
|
||||
energy[1] > 1.0,
|
||||
"control capture must contain the echo signal"
|
||||
);
|
||||
assert!(
|
||||
energy[0] < energy[1] * 0.5,
|
||||
"reference/control echo energy: {energy:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn real_encoder_produces_twenty_millisecond_opus_packets() {
|
||||
for rate in [44_100, 48_000] {
|
||||
let mut processor = Processor::new(rate, /*output_rate*/ 48_000).unwrap();
|
||||
let mut packets = Vec::new();
|
||||
for block in 0..40 {
|
||||
let mut frame = Frame {
|
||||
samples: [0.0; 256],
|
||||
len: 256,
|
||||
at: Instant::now(),
|
||||
generation: 2,
|
||||
};
|
||||
for (index, sample) in frame.samples.iter_mut().enumerate() {
|
||||
*sample = (std::f32::consts::TAU * 440.0 * (block as usize * 256 + index) as f32
|
||||
/ rate as f32)
|
||||
.sin()
|
||||
* 0.1;
|
||||
}
|
||||
packets.extend(processor.capture(&frame, || frame.at).unwrap());
|
||||
}
|
||||
assert!(!packets.is_empty());
|
||||
let mut decoder = opus::Decoder::new(/*sample_rate*/ 48_000, opus::Channels::Mono).unwrap();
|
||||
let mut energy = 0.0_f32;
|
||||
for packet in packets {
|
||||
let mut output = [0.0; 960];
|
||||
assert_eq!(
|
||||
decoder
|
||||
.decode_float(&packet.data, &mut output, /*fec*/ false)
|
||||
.unwrap(),
|
||||
960
|
||||
);
|
||||
assert!(output.iter().all(|sample| sample.is_finite()));
|
||||
energy += output.iter().map(|sample| sample * sample).sum::<f32>();
|
||||
}
|
||||
assert!(
|
||||
energy > 0.01,
|
||||
"decoded capture must contain signal at {rate} Hz"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unmute_reset_rejects_delayed_pre_unmute_audio_and_partial_history() {
|
||||
let mut processor = Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000).unwrap();
|
||||
let frame = Frame {
|
||||
samples: [0.25; 256],
|
||||
len: 256,
|
||||
at: Instant::now(),
|
||||
generation: 2,
|
||||
};
|
||||
assert!(processor.capture(&frame, || frame.at).unwrap().is_empty());
|
||||
assert_eq!(
|
||||
(0..10).find_map(|_| processor
|
||||
.capture(&frame, || frame.at + Duration::from_secs(1))
|
||||
.err()),
|
||||
Some("voice processing fell behind")
|
||||
);
|
||||
processor.reset().unwrap();
|
||||
for _ in 0..10 {
|
||||
assert!(processor.capture(&frame, || frame.at).unwrap().is_empty());
|
||||
}
|
||||
assert!(processor.pending.is_empty());
|
||||
assert!(processor.capture.input.is_empty());
|
||||
assert!(processor.capture.output.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capture_gap_discards_old_partial_packet_and_preserves_unmute_cutoff() {
|
||||
let mut processor = Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000).unwrap();
|
||||
let mut fresh = Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000).unwrap();
|
||||
let cutoff = processor.cutoff;
|
||||
let start = Instant::now();
|
||||
let mut frame = Frame {
|
||||
samples: [0.25; 256],
|
||||
len: 240,
|
||||
at: start,
|
||||
generation: 2,
|
||||
};
|
||||
for index in 0..5 {
|
||||
frame.at = start + Duration::from_millis(index * 5);
|
||||
assert!(processor.capture(&frame, || frame.at).unwrap().is_empty());
|
||||
}
|
||||
assert!(!processor.pending.is_empty());
|
||||
assert!(!processor.capture.input.is_empty());
|
||||
|
||||
let resumed = start + Duration::from_millis(/*millis*/ 100);
|
||||
frame.samples.fill(0.0);
|
||||
let mut packets = Vec::new();
|
||||
let mut expected = Vec::new();
|
||||
for index in 0..12 {
|
||||
frame.at = resumed + Duration::from_millis(index * 5);
|
||||
let emitted = processor.capture(&frame, || frame.at).unwrap();
|
||||
if index < 2 {
|
||||
assert!(emitted.is_empty());
|
||||
}
|
||||
packets.extend(emitted.into_iter().map(|packet| (packet.at, packet.data)));
|
||||
expected.extend(
|
||||
fresh
|
||||
.capture(&frame, || frame.at)
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.map(|packet| (packet.at, packet.data)),
|
||||
);
|
||||
}
|
||||
assert!(!packets.is_empty());
|
||||
assert_eq!(packets, expected);
|
||||
assert_eq!(processor.cutoff, cutoff);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ordinary_capture_timestamp_jitter_keeps_partial_audio() {
|
||||
for jitter_ms in [-2_i64, 0, 2] {
|
||||
let mut processor = Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000).unwrap();
|
||||
let start = Instant::now();
|
||||
let mut frame = Frame {
|
||||
samples: [0.1; 256],
|
||||
len: 240,
|
||||
at: start,
|
||||
generation: 2,
|
||||
};
|
||||
for index in 0..5 {
|
||||
frame.at = start + Duration::from_millis(index * 5);
|
||||
assert!(processor.capture(&frame, || frame.at).unwrap().is_empty());
|
||||
}
|
||||
frame.at = start + Duration::from_millis((25 + jitter_ms) as u64);
|
||||
assert_eq!(processor.capture(&frame, || frame.at).unwrap().len(), 1);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn callback_gap_cannot_hide_in_a_packed_partial_block() {
|
||||
let mut processor = Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000).unwrap();
|
||||
let mut packer = super::super::buffers::FramePacker::default();
|
||||
let queue = crossbeam_queue::ArrayQueue::new(/*capacity*/ 8);
|
||||
let start = Instant::now();
|
||||
// Leave both an incomplete callback block and a processed half-packet.
|
||||
for index in 0..6 {
|
||||
let at = start + Duration::from_millis(index * 5);
|
||||
packer.discard_capture_gap(at, /*rate*/ 48_000.0);
|
||||
assert!(packer.push(
|
||||
Frame {
|
||||
samples: [0.25; 256],
|
||||
len: 240,
|
||||
at,
|
||||
generation: 2
|
||||
},
|
||||
/*rate*/ 48_000.0,
|
||||
&queue
|
||||
));
|
||||
while let Some(frame) = queue.pop() {
|
||||
assert!(processor.capture(&frame, || frame.at).unwrap().is_empty());
|
||||
}
|
||||
}
|
||||
assert!(!processor.pending.is_empty());
|
||||
let resumed = start + Duration::from_millis(/*millis*/ 100);
|
||||
packer.discard_capture_gap(resumed, /*rate*/ 48_000.0);
|
||||
assert!(packer.push(
|
||||
Frame {
|
||||
samples: [0.0; 256],
|
||||
len: 256,
|
||||
at: resumed,
|
||||
generation: 2
|
||||
},
|
||||
/*rate*/ 48_000.0,
|
||||
&queue
|
||||
));
|
||||
let frame = queue.pop().unwrap();
|
||||
assert_eq!(
|
||||
(frame.samples, frame.len, frame.at, frame.generation),
|
||||
([0.0; 256], 256, resumed, 2)
|
||||
);
|
||||
assert!(processor.capture(&frame, || resumed).unwrap().is_empty());
|
||||
assert!(processor.pending.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stale_echo_reference_cannot_cancel_capture_age() {
|
||||
for render_delay in [0, -750] {
|
||||
let mut processor = Processor::new(/*input_rate*/ 48_000, /*output_rate*/ 48_000).unwrap();
|
||||
processor.render_delay = render_delay;
|
||||
let start = Instant::now();
|
||||
let now = start + Duration::from_millis(/*millis*/ 750);
|
||||
let failure = (0..8).find_map(|index| {
|
||||
let frame = Frame {
|
||||
samples: [0.25; 256],
|
||||
len: 256,
|
||||
at: start + Duration::from_secs_f64((index * 256) as f64 / 48_000.0),
|
||||
generation: 2,
|
||||
};
|
||||
processor.capture(&frame, || now).err()
|
||||
});
|
||||
assert_eq!(failure, Some("voice processing fell behind"));
|
||||
}
|
||||
}
|
||||
@@ -60,6 +60,7 @@ impl PeerConnectionEventHandler for Events {
|
||||
}
|
||||
|
||||
pub(crate) struct Transport {
|
||||
pub(crate) audio: crate::audio_track::AudioTrack,
|
||||
connection: Arc<dyn PeerConnection>,
|
||||
gathered: Arc<Notify>,
|
||||
observer: JoinHandle<()>,
|
||||
@@ -72,6 +73,7 @@ impl Transport {
|
||||
}
|
||||
|
||||
async fn with_runtime(runtime: Arc<dyn webrtc::runtime::Runtime>) -> Result<Self> {
|
||||
let (media, audio) = crate::audio_track::AudioTrack::new()?;
|
||||
let gathered = Arc::new(Notify::new());
|
||||
// Upstream defaults exhaust checks after 1.4s, including checks sent before
|
||||
// a TCP connection exists. Keep probing throughout our negotiation deadline.
|
||||
@@ -82,6 +84,7 @@ impl Transport {
|
||||
settings.set_ice_connection_attempts(Some(check_interval), Some(attempts));
|
||||
let connection: Arc<dyn PeerConnection> = Arc::new(
|
||||
PeerConnectionBuilder::new()
|
||||
.with_media_engine(media)
|
||||
.with_runtime(runtime)
|
||||
.with_setting_engine(settings)
|
||||
.with_handler(Arc::new(Events(gathered.clone())))
|
||||
@@ -93,6 +96,10 @@ impl Transport {
|
||||
.await
|
||||
.map_err(|_| "failed to create voice peer")?,
|
||||
);
|
||||
if connection.add_track(audio.track.clone()).await.is_err() {
|
||||
let _ = timeout(Duration::from_secs(/*secs*/ 2), connection.close()).await;
|
||||
return Err("failed to attach voice track");
|
||||
}
|
||||
let channel = match connection
|
||||
.create_data_channel("oai-events", /*options*/ None)
|
||||
.await
|
||||
@@ -121,6 +128,7 @@ impl Transport {
|
||||
sender.send_replace(false);
|
||||
});
|
||||
Ok(Self {
|
||||
audio,
|
||||
connection,
|
||||
gathered,
|
||||
observer,
|
||||
|
||||
@@ -55,7 +55,9 @@ async fn check_negotiation(runtime: Arc<dyn webrtc::runtime::Runtime>) {
|
||||
let gathered = Arc::new(Notify::new());
|
||||
let mut settings = webrtc::peer_connection::SettingEngine::default();
|
||||
settings.set_lite(/*lite*/ true);
|
||||
let (media, _) = crate::audio_track::AudioTrack::new().unwrap();
|
||||
let builder = PeerConnectionBuilder::new()
|
||||
.with_media_engine(media)
|
||||
.with_setting_engine(settings)
|
||||
.with_handler(Arc::new(RemoteEvents(sender, gathered.clone())));
|
||||
let remote = if tcp {
|
||||
|
||||
@@ -27,6 +27,7 @@ use tokio::sync::Notify;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::timeout;
|
||||
use webrtc::data_channel::DataChannel;
|
||||
use webrtc::peer_connection::MediaEngine;
|
||||
use webrtc::peer_connection::PeerConnection;
|
||||
use webrtc::peer_connection::PeerConnectionBuilder;
|
||||
use webrtc::peer_connection::PeerConnectionEventHandler;
|
||||
@@ -307,7 +308,10 @@ async fn installed_client_negotiates_and_closes_over_udp_and_tcp() -> Result<()>
|
||||
let gathered = Arc::new(Notify::new());
|
||||
let mut settings = SettingEngine::default();
|
||||
settings.set_lite(/*lite*/ true);
|
||||
let mut media = MediaEngine::default();
|
||||
media.register_default_codecs()?;
|
||||
let builder = PeerConnectionBuilder::new()
|
||||
.with_media_engine(media)
|
||||
.with_setting_engine(settings)
|
||||
.with_handler(Arc::new(RemoteEvents {
|
||||
channels: sender,
|
||||
|
||||
Reference in New Issue
Block a user