TheLanc3/rustdesk
1
0
Fork 0
mirror of https://github.com/rustdesk/rustdesk.git synced 2026-09-18 18:31:02 +03:00
Code Issues Packages Projects Releases Wiki Activity

Fix/audio stream continuity (#16095)

Browse Source 
* fix(audio): add streaming resampler

* fix(audio): preserve stream resampling state

* fix(audio): keep playback callback nonblocking

* fix(audio): decouple capture conversion from dasp

* fix(audio): support stateful samplerate backend

* refactor(audio): isolate stream callback state

* refactor(audio): group capture output options

* fix(audio): clear stale playback state after startup failure

Reset non-Linux playback state when stream startup fails to prevent
new-format audio from using the previous stream or resampler.

Add regression tests for failed format changes and successful playback.

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): honor capture resampler selection and reuse buffers

Use the selected resampling backend for fixed-frame capture.
Convert samples directly into the input queue and
reuse the PCM frame buffer.

Add tests for anti-aliasing, thread transfer, and
partial-frame draining.

Signed-off-by: fufesou <linlong1266@gmail.com>

* refact: reduce diffs

Signed-off-by: fufesou <linlong1266@gmail.com>

* test(audio): check resampler output count and passband energy

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): reset incompatible Linux playback state on
  startup failure

Preserve compatible output streams when replacement
  startup fails.
Clear state when no compatible stream exists and cover
  both paths in tests.

Signed-off-by: fufesou <linlong1266@gmail.com>

* perf(audio): reuse PCM buffers in the capture pipeline

- Reuse capture framing, resampling, and channel conversion buffers
- Deliver borrowed packets and write Sinc output into reusable storage
- Add allocation and output-equivalence regression tests

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): smooth buffer discard discontinuities

Signal receiver PCM discards and fade from the current playback output when the callback reaches the new timeline.

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): add missing Cargo.toml

Signed-off-by: fufesou <linlong1266@gmail.com>

* perf(audio): move capture encoding off the CPAL callback

Move Opus encoding and service delivery to a dedicated worker.
Use a preallocated bounded PCM queue with explicit loss reporting.
Add tests for callback allocations and queue saturation.

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): smooth capture gaps and report losses during backlog

Signed-off-by: fufesou <linlong1266@gmail.com>

* feat(audio): report capture queue high-water mark

Track peak queued PCM packets and log the approximate
queued audio duration alongside capture loss statistics.

Signed-off-by: fufesou <linlong1266@gmail.com>

* refact(audio): reduce diffs

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): avoid blocking capture on encoder queue contention

Use preallocated queues with try_lock in the capture callback.
Count and drop the current packet on contention, preserving
drop-oldest behavior on overflow.

Add regressions for paused workers, buffer reuse, and sequence wrap.

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix: add the missing files

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): isolate zero-gate state per encoder

Signed-off-by: fufesou <linlong1266@gmail.com>

* refact: reduce diffs

Signed-off-by: fufesou <linlong1266@gmail.com>

* refact(audio): simple refactor

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): avoid waiting on playback callback locks

Use one PCM try_lock attempt and preserve queued samples during contention. Replace readiness locking with per-stream atomic status and report callback errors from the receiving thread.

Cover callback progress, retained audio, recovery, and poisoned-buffer handling.

* fix(audio): restart capture after processing errors

Stop further processing until the service recreates the stream.
Document the guard as defensive recovery for an unconfirmed failure.
Group capture and resampler submodules under their parent directories.

Signed-off-by: fufesou <linlong1266@gmail.com>

* audio: report capture queue contention drops separately

- Add contention_dropped to loss reports while preserving total drop counts
- Document packet rejection on contention even when buffers are available
- Extend existing contention and saturation test assertions

Signed-off-by: fufesou <linlong1266@gmail.com>

* refact unit tests

Signed-off-by: fufesou <linlong1266@gmail.com>

---------

Signed-off-by: fufesou <linlong1266@gmail.com>
This commit is contained in:
fufesou
2026-09-10 16:00:58 +08:00
committed by GitHub
parent 978e2e28b9
commit c4221469d8
20 changed files with 2957 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

511
src/server/audio_service.rs
View File

@@ -15,7 +15,9 @@
use super::*;
#[cfg(not(any(target_os = "linux", target_os = "android")))]
use hbb_common::anyhow::anyhow;
use magnum_opus::{Application::*, Channels::*, Encoder};
#[cfg(any(target_os = "linux", target_os = "android"))]
use magnum_opus::Application::LowDelay;
use magnum_opus::{Channels::*, Encoder};
use std::sync::atomic::{AtomicBool, Ordering};
pub const NAME: &'static str = "audio";
@@ -97,10 +99,11 @@ mod pa_impl {
RESTARTING.store(false, Ordering::SeqCst);
#[cfg(target_os = "linux")]
let mut stream = crate::ipc::connect(1000, "_pa").await?;
unsafe {
AUDIO_ZERO_COUNT = 0;
}
let mut encoder = Encoder::new(crate::platform::PA_SAMPLE_RATE, Stereo, LowDelay)?;
let mut encoder = AudioEncoder::new(Encoder::new(
crate::platform::PA_SAMPLE_RATE,
Stereo,
LowDelay,
)?);
#[cfg(target_os = "linux")]
allow_err!(
stream
@@ -172,8 +175,11 @@ pub fn is_screen_capture_kit_available() -> bool {
}
#[cfg(not(any(target_os = "linux", target_os = "android")))]
#[path = "audio_capture_error.rs"]
mod audio_capture;
#[cfg(not(any(target_os = "linux", target_os = "android")))]
mod audio_capture_error;
#[cfg(not(any(target_os = "linux", target_os = "android")))]
mod audio_capture_queue;
#[cfg(not(any(target_os = "linux", target_os = "android")))]
mod cpal_impl {
@@ -182,14 +188,15 @@ mod cpal_impl {
use super::*;
use cpal::{
traits::{DeviceTrait, HostTrait, StreamTrait},
BufferSize, Device, Host, InputCallbackInfo, StreamConfig, SupportedStreamConfig,
Device, Host, InputCallbackInfo, SupportedStreamConfig,
};
lazy_static::lazy_static! {
static ref HOST: Host = cpal::default_host();
static ref INPUT_BUFFER: Arc<Mutex<std::collections::VecDeque<f32>>> = Default::default();
}
const AUDIO_PACKETS_PER_SECOND: usize = 100;
#[cfg(feature = "screencapturekit")]
lazy_static::lazy_static! {
static ref HOST_SCREEN_CAPTURE_KIT: Result<Host, cpal::HostUnavailable> = cpal::host_from_id(cpal::HostId::ScreenCaptureKit);
@@ -197,7 +204,20 @@ mod cpal_impl {
#[derive(Default)]
pub struct State {
stream: Option<(Box<dyn StreamTrait>, Arc<Message>, CaptureErrorHandler)>,
stream: Option<ActiveCaptureStream>,
}
struct ActiveCaptureStream {
stream: Option<Box<dyn StreamTrait>>,
format: Arc<Message>,
_encoder_worker: audio_capture_queue::CaptureEncoderWorker,
errors: CaptureErrorHandler,
}
impl Drop for ActiveCaptureStream {
fn drop(&mut self) {
self.stream.take();
}
}
impl super::service::Reset for State {
@@ -215,8 +235,8 @@ mod cpal_impl {
}
_ => {}
}
if let Some((_, format, _)) = &state.stream {
sp.send_shared(format.clone());
if let Some(stream) = &state.stream {
sp.send_shared(stream.format.clone());
#[cfg(target_os = "macos")]
log::info!("Audio capture stream recreated; replacement format sent");
}
@@ -232,8 +252,8 @@ mod cpal_impl {
}
_ => {}
}
if let Some((_, format, _)) = &state.stream {
sps.send_shared(format.clone());
if let Some(stream) = &state.stream {
sps.send_shared(stream.format.clone());
}
Ok(())
})?;
@@ -241,10 +261,10 @@ mod cpal_impl {
}
pub fn run(sp: EmptyExtraFieldService, state: &mut State) -> ResultType<()> {
if let Some((_, _, errors)) = &state.stream {
if errors.needs_restart() {
// Recreate on the service thread, outside the backend's error callback.
log::warn!("Recreating interrupted audio capture stream");
if let Some(stream) = &state.stream {
if stream.errors.needs_restart() {
// Recreate on the service thread, outside the capture callbacks.
log::warn!("Recreating audio capture stream after an error");
super::restart();
}
}
@@ -255,29 +275,89 @@ mod cpal_impl {
}
}
fn send(
data: Vec<f32>,
sample_rate0: u32,
sample_rate: u32,
#[derive(Clone, Copy)]
struct CaptureFrameProcessorConfig {
input_rate: u32,
output_rate: u32,
device_channel: u16,
encode_channel: u16,
encoder: &mut Encoder,
sp: &GenericService,
) {
let mut data = data;
if sample_rate0 != sample_rate {
data = crate::common::audio_resample(&data, sample_rate0, sample_rate, device_channel);
}
struct CaptureFrameProcessor {
config: CaptureFrameProcessorConfig,
resampler: Option<crate::audio_resampler::FixedFrameAudioResampler>,
sender: audio_capture_queue::CapturePcmSender,
rechannel_buffer: Vec<f32>,
}
struct CaptureStreamOutput {
sender: audio_capture_queue::CapturePcmSender,
sample_rate: u32,
encode_channel: magnum_opus::Channels,
}
impl CaptureFrameProcessor {
fn new(
config: CaptureFrameProcessorConfig,
sender: audio_capture_queue::CapturePcmSender,
) -> ResultType<Self> {
let resampler = if config.input_rate == config.output_rate {
None
} else {
let output_frames = config.output_rate as usize / AUDIO_PACKETS_PER_SECOND;
Some(crate::audio_resampler::FixedFrameAudioResampler::new(
crate::audio_resampler::AudioResamplerConfig {
input_rate: config.input_rate,
output_rate: config.output_rate,
channels: config.device_channel,
},
output_frames,
)?)
};
Ok(Self {
config,
resampler,
sender,
rechannel_buffer: Vec::with_capacity(
capture_packet_layout(config.output_rate, config.encode_channel)?.1,
),
})
}
if device_channel != encode_channel {
data = crate::common::audio_rechannel(
data,
sample_rate,
sample_rate,
device_channel,
encode_channel,
fn process(&mut self, data: &[f32]) -> ResultType<()> {
let config = self.config;
let sender = &mut self.sender;
let rechannel_buffer = &mut self.rechannel_buffer;
let mut send_packet = |packet: &[f32]| {
let packet =
audio_capture::rechannel(packet, config.device_channel, rechannel_buffer);
sender.submit(packet);
};
if let Some(resampler) = self.resampler.as_mut() {
resampler.process_with(data, send_packet).with_context(|| {
format!(
"Failed to resample captured audio from {} Hz to {} Hz",
config.input_rate, config.output_rate
)
})?;
} else {
send_packet(data);
}
Ok(())
}
}
fn capture_packet_layout(sample_rate: u32, channels: u16) -> ResultType<(usize, usize)> {
if sample_rate < AUDIO_PACKETS_PER_SECOND as u32 || channels == 0 {
bail!("Invalid audio capture layout: sample_rate={sample_rate}, channels={channels}");
}
let frames = sample_rate as usize / AUDIO_PACKETS_PER_SECOND;
let samples = frames.checked_mul(channels as usize).with_context(|| {
format!(
"Audio capture frame size overflow: sample_rate={sample_rate}, channels={channels}"
)
}
send_f32(&data, encoder, sp);
})?;
Ok((frames, samples))
}
#[cfg(feature = "screencapturekit")]
@@ -367,9 +447,7 @@ mod cpal_impl {
Ok((device, format))
}
fn play(
sp: &GenericService,
) -> ResultType<(Box<dyn StreamTrait>, Arc<Message>, CaptureErrorHandler)> {
fn play(sp: &GenericService) -> ResultType<ActiveCaptureStream> {
use cpal::SampleFormat::*;
let (device, config) = get_device()?;
let sp = sp.clone();
@@ -387,109 +465,274 @@ mod cpal_impl {
48000
};
let ch = if config.channels() > 1 { Stereo } else { Mono };
let max_channels = config.channels().max(ch as u16);
let (_, max_packet_samples) = capture_packet_layout(sample_rate, max_channels)?;
let encoder_config = audio_capture_queue::CaptureEncoderConfig {
sample_rate,
encode_channel: ch,
max_packet_samples,
};
let (sender, encoder_worker) =
audio_capture_queue::start_capture_encoder(encoder_config, sp)?;
let output = CaptureStreamOutput {
sender,
sample_rate,
encode_channel: ch,
};
let (stream, errors) = match config.sample_format() {
I8 => build_input_stream::<i8>(device, &config, sp, sample_rate, ch)?,
I16 => build_input_stream::<i16>(device, &config, sp, sample_rate, ch)?,
I32 => build_input_stream::<i32>(device, &config, sp, sample_rate, ch)?,
I64 => build_input_stream::<i64>(device, &config, sp, sample_rate, ch)?,
U8 => build_input_stream::<u8>(device, &config, sp, sample_rate, ch)?,
U16 => build_input_stream::<u16>(device, &config, sp, sample_rate, ch)?,
U32 => build_input_stream::<u32>(device, &config, sp, sample_rate, ch)?,
U64 => build_input_stream::<u64>(device, &config, sp, sample_rate, ch)?,
F32 => build_input_stream::<f32>(device, &config, sp, sample_rate, ch)?,
F64 => build_input_stream::<f64>(device, &config, sp, sample_rate, ch)?,
I8 => build_input_stream::<i8>(device, &config, output)?,
I16 => build_input_stream::<i16>(device, &config, output)?,
I32 => build_input_stream::<i32>(device, &config, output)?,
I64 => build_input_stream::<i64>(device, &config, output)?,
U8 => build_input_stream::<u8>(device, &config, output)?,
U16 => build_input_stream::<u16>(device, &config, output)?,
U32 => build_input_stream::<u32>(device, &config, output)?,
U64 => build_input_stream::<u64>(device, &config, output)?,
F32 => build_input_stream::<f32>(device, &config, output)?,
F64 => build_input_stream::<f64>(device, &config, output)?,
f => bail!("unsupported audio format: {:?}", f),
};
stream.play()?;
#[cfg(target_os = "macos")]
log::info!("Audio capture start call succeeded");
Ok((
Box::new(stream),
Arc::new(create_format_msg(sample_rate, ch as _)),
Ok(ActiveCaptureStream {
stream: Some(Box::new(stream)),
format: Arc::new(create_format_msg(sample_rate, ch as _)),
_encoder_worker: encoder_worker,
errors,
))
})
}
fn convert_input_samples<T>(data: &[T]) -> impl Iterator<Item = f32> + '_
where
T: cpal::SizedSample,
f32: cpal::FromSample<T>,
{
data.iter()
.map(|sample| <f32 as cpal::FromSample<T>>::from_sample_(*sample))
}
#[cfg(target_os = "macos")]
fn log_capture_startup<T>(
data: &[T],
received_samples: bool,
received_signal: bool,
) -> (bool, bool)
where
T: cpal::SizedSample,
f32: cpal::FromSample<T>,
{
// Starting capture does not guarantee sample delivery or audible data.
if !received_samples && !data.is_empty() {
log::info!(
"Audio capture received first PCM block: {} samples",
data.len()
);
}
let has_signal = received_signal
|| convert_input_samples(data).any(|sample| sample.is_finite() && sample != 0.0);
if !received_signal && has_signal {
log::info!("Audio capture received first nonzero PCM");
}
(received_samples || !data.is_empty(), has_signal)
}
fn build_input_stream<T>(
device: cpal::Device,
config: &cpal::SupportedStreamConfig,
sp: GenericService,
sample_rate: u32,
encode_channel: magnum_opus::Channels,
output: CaptureStreamOutput,
) -> ResultType<(cpal::Stream, CaptureErrorHandler)>
where
T: cpal::SizedSample + dasp::sample::ToSample<f32>,
T: cpal::SizedSample,
f32: cpal::FromSample<T>,
{
let errors = CaptureErrorHandler::default();
let callback_errors = errors.clone();
let err_fn = move |err| callback_errors.handle(err);
let processor_errors = errors.clone();
#[cfg(target_os = "macos")]
let (mut received_samples, mut received_signal) = (false, false);
let sample_rate_0 = config.sample_rate().0;
log::debug!("Audio sample rate : {}", sample_rate);
unsafe {
AUDIO_ZERO_COUNT = 0;
}
log::debug!("Audio sample rate : {}", output.sample_rate);
let device_channel = config.channels();
let mut encoder = Encoder::new(sample_rate, encode_channel, LowDelay)?;
// https://www.opus-codec.org/docs/html_api/group__opusencoder.html#gace941e4ef26ed844879fde342ffbe546
// https://chromium.googlesource.com/chromium/deps/opus/+/1.1.1/include/opus.h
// Do not set `frame_size = sample_rate as usize / 100;`
// Because we find `sample_rate as usize / 100` will cause encoder error in `encoder.encode_vec_float()` sometimes.
// https://github.com/xiph/opus/blob/2554a89e02c7fc30a980b4f7e635ceae1ecba5d6/src/opus_encoder.c#L725
let frame_size = sample_rate_0 as usize / 100; // 10 ms
let encode_len = frame_size * encode_channel as usize;
let rechannel_len = encode_len * device_channel as usize / encode_channel as usize;
INPUT_BUFFER.lock().unwrap().clear();
let timeout = None;
let stream_config = StreamConfig {
channels: device_channel,
sample_rate: config.sample_rate(),
buffer_size: BufferSize::Default,
let (_, capture_frame_samples) = capture_packet_layout(sample_rate_0, device_channel)?;
let mut frame = audio_capture::CaptureFrameBuffer::new(capture_frame_samples)?;
let processor_config = CaptureFrameProcessorConfig {
input_rate: sample_rate_0,
output_rate: output.sample_rate,
device_channel,
encode_channel: output.encode_channel as _,
};
let mut processor = CaptureFrameProcessor::new(processor_config, output.sender)?;
let timeout = None;
let stream = device.build_input_stream(
&stream_config,
&config.config(),
move |data: &[T], _: &InputCallbackInfo| {
let buffer: Vec<f32> = data.iter().map(|s| T::to_sample(*s)).collect();
if processor_errors.needs_restart() {
return;
}
#[cfg(target_os = "macos")]
{
// Starting capture does not guarantee sample delivery or audible data.
if !received_samples && !buffer.is_empty() {
received_samples = true;
log::info!(
"Audio capture received first PCM block: {} samples",
buffer.len()
);
}
if !received_signal
&& buffer
.iter()
.any(|sample| sample.is_finite() && *sample != 0.0)
{
received_signal = true;
log::info!("Audio capture received first nonzero PCM");
}
}
let mut lock = INPUT_BUFFER.lock().unwrap();
lock.extend(buffer);
while lock.len() >= rechannel_len {
let frame: Vec<f32> = lock.drain(0..rechannel_len).collect();
send(
frame,
sample_rate_0,
sample_rate,
device_channel,
encode_channel as _,
&mut encoder,
&sp,
);
(received_samples, received_signal) =
log_capture_startup(data, received_samples, received_signal);
}
frame.process(convert_input_samples(data), |frame| {
processor_errors.process_frame(|| processor.process(frame));
});
},
err_fn,
timeout,
)?;
Ok((stream, errors))
}
#[cfg(test)]
mod tests {
use super::super::audio_capture_queue::{
new_pcm_handoff, start_capture_encoder, CaptureEncoderConfig,
};
use super::{
capture_packet_layout, convert_input_samples, CaptureFrameProcessor,
CaptureFrameProcessorConfig,
};
use crate::audio_resampler::allocation_tests::assert_no_allocations;
use crate::server::EmptyExtraFieldService;
use magnum_opus::Channels::{Mono, Stereo};
const INVALID_CAPTURE_RATE: u32 = 99;
const RATE_24_KHZ: u32 = 24_000;
const RATE_44_1_KHZ: u32 = 44_100;
const RATE_48_KHZ: u32 = 48_000;
const MONO_CHANNELS: u16 = 1;
const NEGATIVE_FULL_SCALE_LIMIT: f32 = -0.99;
const POSITIVE_FULL_SCALE_LIMIT: f32 = 0.99;
const STEREO_CHANNELS: u16 = 2;
const SURROUND_CHANNELS: u16 = 6;
const ZERO_CHANNELS: u16 = 0;
#[test]
fn capture_sample_conversion_uses_cpal_traits() {
let input = [i16::MIN, 0, i16::MAX];
let output: Vec<_> = convert_input_samples(&input).collect();
assert_eq!(output.len(), input.len());
assert!(output[0] <= NEGATIVE_FULL_SCALE_LIMIT);
assert_eq!(output[1], 0.0);
assert!(output[2] >= POSITIVE_FULL_SCALE_LIMIT);
}
#[test]
fn capture_packet_layout_validates_rate_and_channels() {
let expected_frames = RATE_48_KHZ as usize / super::AUDIO_PACKETS_PER_SECOND;
assert_eq!(
capture_packet_layout(RATE_48_KHZ, STEREO_CHANNELS).unwrap(),
(expected_frames, expected_frames * STEREO_CHANNELS as usize)
);
assert!(capture_packet_layout(INVALID_CAPTURE_RATE, MONO_CHANNELS).is_err());
assert!(capture_packet_layout(RATE_48_KHZ, ZERO_CHANNELS).is_err());
}
#[test]
fn capture_callback_pipeline_does_not_allocate_after_warmup() {
for (input_rate, output_rate, device_channel, encode_channel) in [
(RATE_48_KHZ, RATE_48_KHZ, MONO_CHANNELS, MONO_CHANNELS),
(RATE_48_KHZ, RATE_48_KHZ, STEREO_CHANNELS, STEREO_CHANNELS),
(RATE_44_1_KHZ, RATE_24_KHZ, STEREO_CHANNELS, STEREO_CHANNELS),
(RATE_48_KHZ, RATE_48_KHZ, SURROUND_CHANNELS, STEREO_CHANNELS),
] {
assert_capture_processor_does_not_allocate(CaptureFrameProcessorConfig {
input_rate,
output_rate,
device_channel,
encode_channel,
});
}
}
#[test]
fn capture_pcm_handoff_reuses_buffers_and_accounts_for_loss() {
const QUEUE_CAPACITY: usize = 2;
const PACKET_SAMPLES: usize = 4;
const FIRST: [f32; PACKET_SAMPLES] = [1.0; PACKET_SAMPLES];
const SECOND: [f32; PACKET_SAMPLES] = [2.0; PACKET_SAMPLES];
const THIRD: [f32; PACKET_SAMPLES] = [3.0; PACKET_SAMPLES];
const OVERSIZED_SAMPLES: usize = PACKET_SAMPLES + 1;
const OVERSIZED: [f32; OVERSIZED_SAMPLES] = [1.0; OVERSIZED_SAMPLES];
let (mut sender, receiver) = new_pcm_handoff(QUEUE_CAPACITY, PACKET_SAMPLES).unwrap();
sender.set_wake_thread(std::thread::current()).unwrap();
assert_no_allocations(|| {
sender.submit(&FIRST);
sender.submit(&SECOND);
sender.submit(&THIRD);
});
let loss = receiver.take_loss();
assert_eq!(loss.dropped, 1);
assert_eq!(loss.oversized, 0);
assert_eq!(loss.recycle_failures, 0);
let second = receiver.pop().unwrap();
let third = receiver.pop().unwrap();
assert_eq!(second, SECOND);
assert_eq!(third, THIRD);
receiver.recycle(second);
receiver.recycle(third);
assert!(receiver.is_empty());
assert_no_allocations(|| sender.submit(&OVERSIZED));
let loss = receiver.take_loss();
assert_eq!(loss.dropped, 0);
assert_eq!(loss.oversized, 1);
assert_eq!(loss.recycle_failures, 0);
assert!(receiver.is_empty());
}
#[test]
fn capture_pcm_handoff_rejects_invalid_layouts() {
assert!(new_pcm_handoff(0, 1).is_err());
assert!(new_pcm_handoff(1, 0).is_err());
}
fn assert_capture_processor_does_not_allocate(config: CaptureFrameProcessorConfig) {
const INPUT_LEVEL: f32 = 0.25;
const TEST_SERVICE_NAME: &str = "audio-allocation-test";
let service = EmptyExtraFieldService::new(TEST_SERVICE_NAME.to_owned(), true).sp;
let encode_channel = if config.encode_channel == MONO_CHANNELS {
Mono
} else {
Stereo
};
let encoder_config = CaptureEncoderConfig {
sample_rate: config.output_rate,
encode_channel,
max_packet_samples: config.output_rate as usize / super::AUDIO_PACKETS_PER_SECOND
* config.device_channel.max(config.encode_channel) as usize,
};
let (sender, worker) = start_capture_encoder(encoder_config, service).unwrap();
let mut processor = CaptureFrameProcessor::new(config, sender).unwrap();
let errors = super::CaptureErrorHandler::default();
let input = vec![
INPUT_LEVEL;
config.input_rate as usize / super::AUDIO_PACKETS_PER_SECOND
* config.device_channel as usize
];
let mut frame_buffer =
super::audio_capture::CaptureFrameBuffer::new(input.len()).unwrap();
frame_buffer.process(convert_input_samples(&input), |frame| {
errors.process_frame(|| processor.process(frame));
});
assert_no_allocations(|| {
frame_buffer.process(convert_input_samples(&input), |frame| {
errors.process_frame(|| processor.process(frame));
});
});
assert!(!errors.needs_restart());
drop(processor);
drop(worker);
}
}
}
fn create_format_msg(sample_rate: u32, channels: u16) -> Message {
@@ -505,28 +748,43 @@ fn create_format_msg(sample_rate: u32, channels: u16) -> Message {
msg
}
// use AUDIO_ZERO_COUNT for the Noise(Zero) Gate Attack Time
// Use a per-encoder counter for the Noise(Zero) Gate Attack Time.
// every audio data length is set to 480
// MAX_AUDIO_ZERO_COUNT=800 is similar as Gate Attack Time 3~5s(Linux) || 6~8s(Windows)
const MAX_AUDIO_ZERO_COUNT: u16 = 800;
static mut AUDIO_ZERO_COUNT: u16 = 0;
fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
if data.iter().filter(|x| **x != 0.).next().is_some() {
unsafe {
AUDIO_ZERO_COUNT = 0;
struct AudioEncoder {
encoder: Encoder,
zero_count: u16,
}
impl AudioEncoder {
fn new(encoder: Encoder) -> Self {
Self {
encoder,
zero_count: 0,
}
} else {
unsafe {
if AUDIO_ZERO_COUNT > MAX_AUDIO_ZERO_COUNT {
if AUDIO_ZERO_COUNT == MAX_AUDIO_ZERO_COUNT + 1 {
log::debug!("Audio Zero Gate Attack");
AUDIO_ZERO_COUNT += 1;
}
return;
}
fn should_encode(&mut self, data: &[f32]) -> bool {
if data.iter().filter(|x| **x != 0.).next().is_some() {
self.zero_count = 0;
} else if self.zero_count > MAX_AUDIO_ZERO_COUNT {
if self.zero_count == MAX_AUDIO_ZERO_COUNT + 1 {
log::debug!("Audio Zero Gate Attack");
self.zero_count += 1;
}
AUDIO_ZERO_COUNT += 1;
return false;
} else {
self.zero_count += 1;
}
true
}
}
fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) {
if !encoder.should_encode(data) {
return;
}
#[cfg(target_os = "android")]
{
@@ -539,6 +797,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
let n = input_size / BATCH_SIZE;
for i in 0..n {
match encoder
.encoder
.encode_vec_float(&data[i * BATCH_SIZE..(i + 1) * BATCH_SIZE], BATCH_SIZE)
{
Ok(data) => {
@@ -549,7 +808,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
});
sp.send(msg_out);
}
Err(_) => {}
Err(error) => log::warn!("Failed to encode audio frame: {error:?}"),
}
}
} else {
@@ -559,7 +818,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
}
#[cfg(not(target_os = "android"))]
match encoder.encode_vec_float(data, data.len() * 6) {
match encoder.encoder.encode_vec_float(data, data.len() * 6) {
Ok(data) => {
let mut msg_out = Message::new();
msg_out.set_audio_frame(AudioFrame {
@@ -568,6 +827,6 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
});
sp.send(msg_out);
}
Err(_) => {}
Err(error) => log::warn!("Failed to encode audio frame: {error:?}"),
}
}

144
src/server/audio_service/audio_capture.rs Normal file
View File

@@ -0,0 +1,144 @@
use hbb_common::anyhow::{bail, Result};
const STEREO_CHANNELS: usize = 2;
pub(super) struct CaptureFrameBuffer {
samples: Vec<f32>,
filled: usize,
}
impl CaptureFrameBuffer {
pub(super) fn new(samples: usize) -> Result<Self> {
if samples == 0 {
bail!("Audio capture frame must contain at least one sample");
}
Ok(Self {
samples: vec![0.0; samples],
filled: 0,
})
}
pub(super) fn process(
&mut self,
input: impl Iterator<Item = f32>,
mut on_frame: impl FnMut(&[f32]),
) {
for sample in input {
self.samples[self.filled] = sample;
self.filled += 1;
if self.filled == self.samples.len() {
self.filled = 0;
on_frame(&self.samples);
}
}
}
}
pub(super) fn rechannel<'a>(
input: &'a [f32],
channels: u16,
output: &'a mut Vec<f32>,
) -> &'a [f32] {
let input = if channels > STEREO_CHANNELS as u16 {
&input[..input.len() / channels as usize * channels as usize]
} else {
input
};
output.clear();
match channels {
3 => rechannel_frame::<3>(input, output),
4 => rechannel_frame::<4>(input, output),
5 => rechannel_frame::<5>(input, output),
6 => rechannel_frame::<6>(input, output),
7 => rechannel_frame::<7>(input, output),
8 => rechannel_frame::<8>(input, output),
// Preserve the existing passthrough for mono/stereo and unsupported layouts.
_ => return input,
}
output
}
fn rechannel_frame<const CHANNELS: usize>(input: &[f32], output: &mut Vec<f32>) {
use fon::{
chan::{Ch32, Channel},
Frame,
};
for samples in input.chunks_exact(CHANNELS) {
let mut frame = Frame::<Ch32, CHANNELS>::default();
for (channel, sample) in frame.channels_mut().iter_mut().zip(samples) {
*channel = (*sample).into();
}
// Match the same-rate Stream::pipe conversion before fon's SinkTo conversion.
let stereo = frame.to::<Ch32, CHANNELS>().to::<Ch32, STEREO_CHANNELS>();
output.extend(stereo.channels().iter().map(|channel| channel.to_f32()));
}
}
#[cfg(test)]
mod tests {
use super::{rechannel, CaptureFrameBuffer};
use crate::audio_resampler::allocation_tests::assert_no_allocations;
#[test]
fn capture_channel_conversion_reuses_storage_and_preserves_mapping() {
const SAMPLE_RATE: u32 = 48_000;
const FRAMES: usize = 7;
const STEREO_CHANNELS: u16 = 2;
const MIN_SAMPLE: f32 = -1.25;
const SAMPLE_STEP: f32 = 0.2;
for channels in [1, 2, 3, 4, 5, 6, 7, 8, 16] {
let input: Vec<_> = (0..FRAMES * channels as usize + 1)
.map(|sample| MIN_SAMPLE + sample as f32 * SAMPLE_STEP)
.collect();
let expected = crate::common::audio_rechannel(
input.clone(),
SAMPLE_RATE,
SAMPLE_RATE,
channels,
channels.min(STEREO_CHANNELS),
);
let mut output = Vec::with_capacity(FRAMES * STEREO_CHANNELS as usize);
assert_no_allocations(|| {
let actual = rechannel(&input, channels, &mut output);
assert_eq!(
actual, expected,
"channel mapping changed for {channels} channels"
);
});
}
}
#[test]
fn capture_framing_retains_partial_input_without_allocating() {
const FRAME_SAMPLES: usize = 6;
const CALLBACK_SIZES: [usize; 7] = [0, 1, 17, 2, 257, 492, 5];
let input: Vec<_> = (0..CALLBACK_SIZES.iter().sum::<usize>())
.map(|sample| sample as f32)
.collect();
let mut buffer = CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap();
let mut input_position = 0;
let mut output_position = 0;
assert!(CaptureFrameBuffer::new(0).is_err());
assert_no_allocations(|| {
for samples in CALLBACK_SIZES {
let end = input_position + samples;
buffer.process(input[input_position..end].iter().copied(), |frame| {
assert_eq!(
frame,
&input[output_position..output_position + FRAME_SAMPLES]
);
output_position += FRAME_SAMPLES;
});
input_position = end;
assert_eq!(
output_position,
input_position / FRAME_SAMPLES * FRAME_SAMPLES
);
}
});
assert_eq!(output_position, input.len());
}
}

128
src/server/audio_service/audio_capture_encoder.rs Normal file
View File

@@ -0,0 +1,128 @@
use super::super::AudioEncoder;
use super::{
send_f32, CaptureEncoderConfig, CaptureEncoderContext, CapturePcmReceiver, CapturePcmStats,
CAPTURE_PCM_QUEUE_PACKETS,
};
use hbb_common::log;
use magnum_opus::Channels;
use std::{
sync::atomic::Ordering,
time::{Duration, Instant},
};
const CAPTURE_DECLICK_MS: usize = 5;
const CAPTURE_PACKET_MS: usize = 10;
const MILLISECONDS_PER_SECOND: usize = 1_000;
const MAX_ENCODE_CHANNELS: usize = Channels::Stereo as usize;
const CAPTURE_STATS_LOG_INTERVAL: Duration = Duration::from_secs(5);
struct CaptureEncoderState {
channels: usize,
expected_sequence: usize,
fade_frames: usize,
last_frame: [f32; MAX_ENCODE_CHANNELS],
reporter: CaptureStatsReporter,
}
impl CaptureEncoderState {
fn new(sample_rate: u32, channels: Channels) -> Self {
// The encoder has already validated the supported Opus rate and channel count.
let fade_frames = sample_rate as usize * CAPTURE_DECLICK_MS / MILLISECONDS_PER_SECOND;
Self {
channels: channels as usize,
expected_sequence: 0,
fade_frames,
last_frame: [0.0; MAX_ENCODE_CHANNELS],
reporter: CaptureStatsReporter::new(),
}
}
fn next_packet(&mut self, receiver: &CapturePcmReceiver) -> Option<Vec<f32>> {
self.reporter.pending.add(receiver.take_stats());
self.reporter.report(false);
let (sequence, mut packet) = receiver.pop_packet()?;
self.smooth_packet(sequence, &mut packet);
Some(packet)
}
fn smooth_packet(&mut self, sequence: usize, packet: &mut [f32]) {
if sequence != self.expected_sequence {
// Capture packets contain 10 ms of PCM, so the transition fits in this packet.
for (index, frame) in packet
.chunks_exact_mut(self.channels)
.take(self.fade_frames)
.enumerate()
{
let weight = (index + 1) as f32 / self.fade_frames as f32;
for (channel, sample) in frame.iter_mut().enumerate() {
*sample = self.last_frame[channel] * (1.0 - weight) + *sample * weight;
}
}
}
self.expected_sequence = sequence.wrapping_add(1);
if let Some(frame) = packet.chunks_exact(self.channels).next_back() {
self.last_frame[..self.channels].copy_from_slice(frame);
}
}
}
struct CaptureStatsReporter {
pending: CapturePcmStats,
last_report: Instant,
}
impl CaptureStatsReporter {
fn new() -> Self {
Self {
pending: Default::default(),
last_report: Instant::now(),
}
}
fn report(&mut self, force: bool) {
if self.pending.is_empty() {
return;
}
if !force && self.last_report.elapsed() < CAPTURE_STATS_LOG_INTERVAL {
return;
}
let stats = std::mem::take(&mut self.pending);
log::debug!(
"Audio capture PCM handoff stats: observed_max_queued_packets={}, approx_queued_audio_ms={}, capacity_packets={}",
stats.max_queued_packets,
stats.max_queued_packets.saturating_mul(CAPTURE_PACKET_MS),
CAPTURE_PCM_QUEUE_PACKETS
);
if !stats.loss.is_empty() {
log::warn!(
"Audio capture PCM handoff loss: dropped={}, contention_dropped={}, oversized={}, recycle_failures={}",
stats.loss.dropped,
stats.loss.contention_dropped,
stats.loss.oversized,
stats.loss.recycle_failures
);
}
self.last_report = Instant::now();
}
}
pub(super) fn run_capture_encoder(context: CaptureEncoderContext, config: CaptureEncoderConfig) {
let mut encoder = AudioEncoder::new(context.encoder);
let mut state = CaptureEncoderState::new(config.sample_rate, config.encode_channel);
loop {
while let Some(packet) = state.next_packet(&context.receiver) {
send_f32(&packet, &mut encoder, &context.service);
context.receiver.recycle(packet);
}
if context.stop.load(Ordering::Acquire) && context.receiver.is_empty() {
state.reporter.pending.add(context.receiver.take_stats());
state.reporter.report(true);
return;
}
std::thread::park_timeout(CAPTURE_STATS_LOG_INTERVAL);
}
}
#[cfg(test)]
#[path = "audio_capture_encoder_tests.rs"]
mod tests;

186
src/server/audio_service/audio_capture_encoder_tests.rs Normal file
View File

@@ -0,0 +1,186 @@
use super::super::{new_pcm_handoff, CAPTURE_PCM_QUEUE_PACKETS};
use super::*;
use crate::audio_resampler::allocation_tests::assert_no_allocations;
use magnum_opus::{Application::LowDelay, Decoder, Encoder};
const SAMPLE_RATE: u32 = 48_000;
const PACKETS_PER_SECOND: usize = 100;
const PACKET_COUNT: usize = 120;
const PAUSE_PACKET: usize = 50;
const DROPPED_PACKETS: usize = 6;
const SIGNAL_FREQUENCY: f64 = 97.0;
const SIGNAL_AMPLITUDE: f64 = 0.5;
const ACTIVE_LEVEL: f32 = 0.8;
const SAMPLE_TOLERANCE: f32 = 0.000001;
const MAX_ENCODE_BYTES_PER_SAMPLE: usize = 6;
fn signal_packet(index: usize, channels: usize) -> Vec<f32> {
let frames = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
(0..frames)
.flat_map(|frame| {
let phase = std::f64::consts::TAU * SIGNAL_FREQUENCY * (index * frames + frame) as f64
/ f64::from(SAMPLE_RATE);
(0..channels)
.map(move |channel| (SIGNAL_AMPLITUDE * (phase + channel as f64).sin()) as f32)
})
.collect()
}
fn encoded_audio(dropped: usize, channels: Channels) -> Vec<f32> {
let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels as usize;
let (mut sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, samples).unwrap();
let mut state = CaptureEncoderState::new(SAMPLE_RATE, channels);
let mut encoder = Encoder::new(SAMPLE_RATE, channels, LowDelay).unwrap();
let mut decoder = Decoder::new(SAMPLE_RATE, channels).unwrap();
let mut decoded = vec![0.0; samples];
let mut output = Vec::new();
let resume_packet = PAUSE_PACKET + CAPTURE_PCM_QUEUE_PACKETS + dropped - 1;
for index in 0..PACKET_COUNT {
sender.submit(&signal_packet(index, channels as usize));
if (PAUSE_PACKET..resume_packet).contains(&index) {
continue;
}
if index == resume_packet {
let stats = receiver.take_stats();
assert_eq!(stats.max_queued_packets, CAPTURE_PCM_QUEUE_PACKETS);
assert_eq!(stats.loss.dropped, dropped);
assert_eq!(stats.loss.contention_dropped, 0);
assert_eq!(stats.loss.oversized, 0);
assert_eq!(stats.loss.recycle_failures, 0);
}
while let Some(packet) = state.next_packet(&receiver) {
let encoded = encoder
.encode_vec_float(&packet, samples * MAX_ENCODE_BYTES_PER_SAMPLE)
.unwrap();
let frames = decoder.decode_float(&encoded, &mut decoded, false).unwrap();
assert_eq!(frames * channels as usize, samples);
output.extend_from_slice(&decoded);
receiver.recycle(packet);
}
}
assert_eq!(output.len(), (PACKET_COUNT - dropped) * samples);
output
}
fn maximum_join_step(pcm: &[f32], channels: usize) -> f32 {
let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels;
pcm[(PAUSE_PACKET - 1) * samples..(PAUSE_PACKET + 2) * samples]
.windows(channels + 1)
.map(|window| (window[channels] - window[0]).abs())
.fold(0.0, f32::max)
}
#[test]
fn capture_overflow_is_smoothed_before_encoding() {
const MAX_STEP_RATIO: f32 = 2.0;
for channels in [Channels::Mono, Channels::Stereo] {
let clean = encoded_audio(0, channels);
let overflow = encoded_audio(DROPPED_PACKETS, channels);
let clean_step = maximum_join_step(&clean, channels as usize);
let overflow_step = maximum_join_step(&overflow, channels as usize);
assert!(clean_step > 0.0);
assert!(
overflow_step < clean_step * MAX_STEP_RATIO,
"capture discard introduced a sharp join: clean={clean_step}, overflow={overflow_step}"
);
}
}
#[test]
fn rejected_packets_mark_the_gap_after_already_queued_audio() {
const CAPACITY: usize = 3;
let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap();
let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono);
let active = vec![ACTIVE_LEVEL; samples];
let opposite = vec![-ACTIVE_LEVEL; samples];
sender.submit(&active);
let packet = state.next_packet(&receiver).unwrap();
assert_eq!(packet, active);
receiver.recycle(packet);
sender.submit(&active);
sender.submit(&active);
sender.submit(&vec![ACTIVE_LEVEL; samples + 1]);
sender.submit(&opposite);
assert_eq!(receiver.take_loss().oversized, 1);
assert_no_allocations(|| {
for _ in 0..CAPACITY - 1 {
let packet = state.next_packet(&receiver).unwrap();
assert_eq!(packet, active);
receiver.recycle(packet);
}
let packet = state.next_packet(&receiver).unwrap();
let expected = ACTIVE_LEVEL * (1.0 - 2.0 / state.fade_frames as f32);
assert!((packet[0] - expected).abs() < SAMPLE_TOLERANCE);
assert_eq!(&packet[samples / 2..], &opposite[samples / 2..]);
receiver.recycle(packet);
});
}
#[test]
fn gaps_and_sequence_wrap_preserve_channel_history() {
const RATE_8_KHZ: u32 = 8_000;
for (rate, channels) in [
(RATE_8_KHZ, Channels::Mono),
(RATE_8_KHZ, Channels::Stereo),
(SAMPLE_RATE, Channels::Mono),
(SAMPLE_RATE, Channels::Stereo),
] {
let channels_count = channels as usize;
let mut state = CaptureEncoderState::new(rate, channels);
let samples = rate as usize / PACKETS_PER_SECOND * channels_count;
let active: Vec<_> = [ACTIVE_LEVEL, -ACTIVE_LEVEL][..channels_count]
.iter()
.copied()
.cycle()
.take(samples)
.collect();
let mut first = active.clone();
state.smooth_packet(0, &mut first);
assert_eq!(first, active);
let mut opposite: Vec<_> = active.iter().map(|sample| -sample).collect();
let mut resumed = active.clone();
assert_no_allocations(|| {
state.smooth_packet(2, &mut opposite);
state.smooth_packet(4, &mut resumed);
});
for channel in 0..channels_count {
let expected = active[channel] * (1.0 - 2.0 / state.fade_frames as f32);
assert!((opposite[channel] - expected).abs() < SAMPLE_TOLERANCE);
let previous = opposite[samples - channels_count + channel];
let expected = previous + (active[channel] - previous) / state.fade_frames as f32;
assert!((resumed[channel] - expected).abs() < SAMPLE_TOLERANCE);
}
assert_eq!(&resumed[samples / 2..], &active[samples / 2..]);
state.expected_sequence = usize::MAX;
let mut last = active.clone();
let mut wrapped: Vec<_> = active.iter().map(|sample| -sample).collect();
let expected = wrapped.clone();
assert_no_allocations(|| {
state.smooth_packet(usize::MAX, &mut last);
state.smooth_packet(0, &mut wrapped);
});
assert_eq!(last, active);
assert_eq!(wrapped, expected);
}
}
#[test]
fn capture_loss_is_reported_while_packets_remain_queued() {
const CAPACITY: usize = 2;
let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap();
let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono);
let before_report = Instant::now() - CAPTURE_STATS_LOG_INTERVAL;
state.reporter.last_report = before_report;
let packet = vec![ACTIVE_LEVEL; samples];
for _ in 0..=CAPACITY {
sender.submit(&packet);
}
let packet = state.next_packet(&receiver).unwrap();
assert!(!receiver.is_empty());
assert!(state.reporter.last_report > before_report);
assert!(state.reporter.pending.is_empty());
assert!(receiver.take_loss().is_empty());
receiver.recycle(packet);
}

54
src/server/audio_capture_error.rs → src/server/audio_service/audio_capture_error.rs
View File

@@ -26,6 +26,20 @@ impl CaptureErrorHandler {
}
}
pub(super) fn process_frame(&self, process: impl FnOnce() -> hbb_common::ResultType<()>) {
// Defensive recovery: persistent processing failures with valid capture input
// have not been reproduced. Stop using a failed processor until stream replacement.
if self.needs_restart() {
return;
}
if let Err(error) = process() {
self.interrupted.store(true, Ordering::Relaxed);
hbb_common::log::error!(
"Failed to process captured audio frame; requesting stream restart: {error:#}"
);
}
}
pub(super) fn needs_restart(&self) -> bool {
self.interrupted.load(Ordering::Relaxed)
}
@@ -72,4 +86,44 @@ mod tests {
});
assert!(!errors.needs_restart());
}
#[test]
fn processing_failure_skips_remaining_frames_until_stream_replacement() {
const FRAME_SAMPLES: usize = 2;
const FRAMES_PER_CALLBACK: usize = 3;
const CALLBACK_COUNT: usize = 2;
const FAILURE_CALL: usize = 2;
let errors = CaptureErrorHandler::default();
let callback_errors = errors.clone();
let mut framer =
super::super::audio_capture::CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap();
let input = [0.0; FRAME_SAMPLES * FRAMES_PER_CALLBACK];
let mut processed = 0;
let mut failures = 0;
// Inject an error to test recovery; this is not a valid-input backend failure reproduction.
for _ in 0..CALLBACK_COUNT {
framer.process(input.iter().copied(), |_| {
callback_errors.process_frame(|| {
processed += 1;
if processed >= FAILURE_CALL {
failures += 1;
hbb_common::anyhow::bail!("Injected capture processing failure");
}
Ok(())
});
});
}
assert_eq!(processed, FAILURE_CALL);
assert_eq!(failures, 1);
assert!(errors.needs_restart());
let replacement = CaptureErrorHandler::default();
replacement.process_frame(|| {
processed += 1;
Ok(())
});
assert_eq!(processed, FAILURE_CALL + 1);
assert!(!replacement.needs_restart());
}
}

289
src/server/audio_service/audio_capture_queue.rs Normal file
View File

@@ -0,0 +1,289 @@
use super::{send_f32, GenericService};
use hbb_common::{
anyhow::{bail, Context, Result},
log,
};
use magnum_opus::{Application::LowDelay, Channels, Encoder};
use std::{
collections::VecDeque,
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc, Mutex, OnceLock, TryLockError,
},
thread::{JoinHandle, Thread},
};
#[path = "audio_capture_encoder.rs"]
mod encoder;
const CAPTURE_PCM_QUEUE_PACKETS: usize = 10;
const CAPTURE_ENCODER_THREAD_NAME: &str = "audio-encoder";
#[derive(Debug, Default, PartialEq, Eq)]
pub(super) struct CapturePcmLoss {
pub(super) dropped: usize,
pub(super) contention_dropped: usize,
pub(super) oversized: usize,
pub(super) recycle_failures: usize,
}
impl CapturePcmLoss {
pub(super) fn is_empty(&self) -> bool {
self.dropped == 0 && self.oversized == 0 && self.recycle_failures == 0
}
pub(super) fn add(&mut self, other: Self) {
self.dropped += other.dropped;
self.contention_dropped += other.contention_dropped;
self.oversized += other.oversized;
self.recycle_failures += other.recycle_failures;
}
}
#[derive(Debug, Default, PartialEq, Eq)]
pub(super) struct CapturePcmStats {
pub(super) loss: CapturePcmLoss,
pub(super) max_queued_packets: usize,
}
impl CapturePcmStats {
pub(super) fn is_empty(&self) -> bool {
self.loss.is_empty() && self.max_queued_packets == 0
}
pub(super) fn add(&mut self, other: Self) {
self.loss.add(other.loss);
self.max_queued_packets = self.max_queued_packets.max(other.max_queued_packets);
}
}
struct CapturePcmHandoff {
buffers: Mutex<CapturePcmBuffers>,
wake_thread: OnceLock<Thread>,
other_dropped: AtomicUsize,
contention_dropped: AtomicUsize,
oversized: AtomicUsize,
recycle_failures: AtomicUsize,
max_queued_packets: AtomicUsize,
max_samples: usize,
}
struct CapturePcmBuffers {
available: Vec<Vec<f32>>,
ready: VecDeque<(usize, Vec<f32>)>,
}
pub(super) struct CapturePcmSender {
handoff: Arc<CapturePcmHandoff>,
sequence: usize,
}
pub(super) struct CapturePcmReceiver {
handoff: Arc<CapturePcmHandoff>,
}
pub(super) struct CaptureEncoderConfig {
pub(super) sample_rate: u32,
pub(super) encode_channel: Channels,
pub(super) max_packet_samples: usize,
}
pub(super) struct CaptureEncoderWorker {
stop: Arc<AtomicBool>,
handle: Option<JoinHandle<()>>,
}
impl Drop for CaptureEncoderWorker {
fn drop(&mut self) {
self.stop.store(true, Ordering::Release);
if let Some(handle) = self.handle.take() {
handle.thread().unpark();
if let Err(error) = handle.join() {
log::error!("Failed to join audio encoder thread: {error:?}");
}
}
}
}
struct CaptureEncoderContext {
receiver: CapturePcmReceiver,
encoder: Encoder,
service: GenericService,
stop: Arc<AtomicBool>,
}
pub(super) fn new_pcm_handoff(
capacity: usize,
max_samples: usize,
) -> Result<(CapturePcmSender, CapturePcmReceiver)> {
if capacity == 0 || max_samples == 0 {
bail!("Audio capture PCM handoff requires nonzero capacity and packet size");
}
let handoff = Arc::new(CapturePcmHandoff {
buffers: Mutex::new(CapturePcmBuffers {
available: Vec::with_capacity(capacity),
ready: VecDeque::with_capacity(capacity),
}),
wake_thread: OnceLock::new(),
other_dropped: AtomicUsize::new(0),
contention_dropped: AtomicUsize::new(0),
oversized: AtomicUsize::new(0),
recycle_failures: AtomicUsize::new(0),
max_queued_packets: AtomicUsize::new(0),
max_samples,
});
{
// Initialize the mutex on this thread, including on platforms with lazy allocation.
let mut buffers = handoff.buffers.lock().unwrap();
for _ in 0..capacity {
buffers.available.push(Vec::with_capacity(max_samples));
}
}
Ok((
CapturePcmSender {
handoff: handoff.clone(),
sequence: 0,
},
CapturePcmReceiver { handoff },
))
}
impl CapturePcmSender {
pub(super) fn set_wake_thread(&self, thread: Thread) -> Result<()> {
if self.handoff.wake_thread.set(thread).is_err() {
bail!("Audio capture PCM wake thread is already configured");
}
Ok(())
}
pub(super) fn submit(&mut self, input: &[f32]) {
let sequence = self.sequence;
self.sequence = self.sequence.wrapping_add(1);
if input.len() > self.handoff.max_samples {
self.handoff.oversized.fetch_add(1, Ordering::Relaxed);
self.wake();
return;
}
// Do not wait for a descheduled worker. Contention rejects the current
// packet even if buffers are available; this is separate from drop-oldest
// when the buffer pool is exhausted.
let mut buffers = match self.handoff.buffers.try_lock() {
Ok(buffers) => buffers,
Err(TryLockError::WouldBlock) => {
self.handoff
.contention_dropped
.fetch_add(1, Ordering::Relaxed);
self.wake();
return;
}
Err(TryLockError::Poisoned(error)) => {
self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
log::error!("Audio capture PCM handoff is poisoned: {error}");
self.wake();
return;
}
};
if let Some(mut buffer) = self.take_buffer(&mut buffers) {
buffer.clear();
buffer.extend_from_slice(input);
buffers.ready.push_back((sequence, buffer));
self.handoff
.max_queued_packets
.fetch_max(buffers.ready.len(), Ordering::Relaxed);
} else {
self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
}
drop(buffers);
self.wake();
}
fn take_buffer(&self, buffers: &mut CapturePcmBuffers) -> Option<Vec<f32>> {
buffers.available.pop().or_else(|| {
let buffer = buffers.ready.pop_front().map(|(_, buffer)| buffer);
if buffer.is_some() {
self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
}
buffer
})
}
fn wake(&self) {
if let Some(thread) = self.handoff.wake_thread.get() {
thread.unpark();
}
}
}
impl CapturePcmReceiver {
#[cfg(test)]
pub(super) fn pop(&self) -> Option<Vec<f32>> {
self.pop_packet().map(|(_, buffer)| buffer)
}
fn pop_packet(&self) -> Option<(usize, Vec<f32>)> {
self.handoff.buffers.lock().unwrap().ready.pop_front()
}
pub(super) fn recycle(&self, mut buffer: Vec<f32>) {
buffer.clear();
let mut buffers = self.handoff.buffers.lock().unwrap();
if buffers.available.len() == buffers.available.capacity() {
self.handoff
.recycle_failures
.fetch_add(1, Ordering::Relaxed);
} else {
buffers.available.push(buffer);
}
}
pub(super) fn is_empty(&self) -> bool {
self.handoff.buffers.lock().unwrap().ready.is_empty()
}
pub(super) fn take_loss(&self) -> CapturePcmLoss {
let contention_dropped = self.handoff.contention_dropped.swap(0, Ordering::Relaxed);
CapturePcmLoss {
dropped: self.handoff.other_dropped.swap(0, Ordering::Relaxed) + contention_dropped,
contention_dropped,
oversized: self.handoff.oversized.swap(0, Ordering::Relaxed),
recycle_failures: self.handoff.recycle_failures.swap(0, Ordering::Relaxed),
}
}
pub(super) fn take_stats(&self) -> CapturePcmStats {
CapturePcmStats {
loss: self.take_loss(),
max_queued_packets: self.handoff.max_queued_packets.swap(0, Ordering::Relaxed),
}
}
}
pub(super) fn start_capture_encoder(
config: CaptureEncoderConfig,
service: GenericService,
) -> Result<(CapturePcmSender, CaptureEncoderWorker)> {
let (sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, config.max_packet_samples)?;
let encoder = Encoder::new(config.sample_rate, config.encode_channel, LowDelay)?;
let stop = Arc::new(AtomicBool::new(false));
let context = CaptureEncoderContext {
receiver,
encoder,
service,
stop: stop.clone(),
};
let handle = std::thread::Builder::new()
.name(CAPTURE_ENCODER_THREAD_NAME.to_owned())
.spawn(move || encoder::run_capture_encoder(context, config))
.with_context(|| "Failed to start audio encoder thread")?;
let wake_thread = handle.thread().clone();
let worker = CaptureEncoderWorker {
stop,
handle: Some(handle),
};
sender.set_wake_thread(wake_thread)?;
Ok((sender, worker))
}
#[cfg(test)]
#[path = "audio_capture_queue_tests.rs"]
mod tests;

169
src/server/audio_service/audio_capture_queue_tests.rs Normal file
View File

@@ -0,0 +1,169 @@
use super::*;
use crate::audio_resampler::allocation_tests::assert_no_allocations;
use std::{sync::mpsc, time::Duration};
const PACKET_SAMPLES: usize = 4;
const TEST_TIMEOUT: Duration = Duration::from_secs(2);
const CONCURRENT_PACKETS: usize = 10_000;
#[derive(Clone, Copy)]
enum PausePoint {
Recycle,
Consume,
}
struct WorkerPause {
entered: mpsc::Sender<()>,
resume: mpsc::Receiver<()>,
}
fn paused_worker(
receiver: CapturePcmReceiver,
point: PausePoint,
pause: WorkerPause,
) -> CapturePcmReceiver {
let recycled = match point {
PausePoint::Recycle => Some(receiver.pop_packet().unwrap().1),
PausePoint::Consume => None,
};
let consumed = {
let mut buffers = receiver.handoff.buffers.lock().unwrap();
let consumed = match recycled {
Some(mut packet) => {
packet.clear();
buffers.available.push(packet);
None
}
None => Some(buffers.ready.pop_front().unwrap().1),
};
pause.entered.send(()).unwrap();
pause.resume.recv().unwrap();
consumed
};
if let Some(packet) = consumed {
receiver.recycle(packet);
}
receiver
}
fn assert_callback_progress(point: PausePoint) {
let (mut sender, receiver) =
new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, PACKET_SAMPLES).unwrap();
for sequence in 0..CAPTURE_PCM_QUEUE_PACKETS {
sender.submit(&[sequence as f32; PACKET_SAMPLES]);
}
let (entered_tx, entered_rx) = mpsc::channel();
let (resume_tx, resume_rx) = mpsc::channel();
let pause = WorkerPause {
entered: entered_tx,
resume: resume_rx,
};
let worker = std::thread::spawn(move || paused_worker(receiver, point, pause));
sender.set_wake_thread(worker.thread().clone()).unwrap();
entered_rx.recv_timeout(TEST_TIMEOUT).unwrap();
let (completed_tx, completed_rx) = mpsc::channel();
let callback = std::thread::spawn(move || {
let packet = [CAPTURE_PCM_QUEUE_PACKETS as f32; PACKET_SAMPLES];
assert_no_allocations(|| sender.submit(&packet));
completed_tx.send(()).unwrap();
sender
});
let completed = completed_rx.recv_timeout(TEST_TIMEOUT);
// Release and join both threads before asserting progress, including on failure.
resume_tx.send(()).unwrap();
let receiver = worker.join().unwrap();
let mut sender = callback.join().unwrap();
assert!(completed.is_ok(), "callback waited for the encoder worker");
assert_eq!(
receiver.take_loss(),
CapturePcmLoss {
dropped: 1,
contention_dropped: 1,
..Default::default()
}
);
assert_packets_after_contention(&mut sender, &receiver);
}
fn assert_packets_after_contention(sender: &mut CapturePcmSender, receiver: &CapturePcmReceiver) {
let next_sequence = CAPTURE_PCM_QUEUE_PACKETS + 1;
sender.submit(&[next_sequence as f32; PACKET_SAMPLES]);
for expected in (1..CAPTURE_PCM_QUEUE_PACKETS).chain(std::iter::once(next_sequence)) {
let (sequence, packet) = receiver.pop_packet().unwrap();
assert_eq!(sequence, expected);
assert_eq!(packet, [expected as f32; PACKET_SAMPLES]);
receiver.recycle(packet);
}
assert!(receiver.is_empty());
assert!(receiver.take_loss().is_empty());
assert_pool_restored(receiver, CAPTURE_PCM_QUEUE_PACKETS);
}
fn assert_pool_restored(receiver: &CapturePcmReceiver, capacity: usize) {
let buffers = receiver.handoff.buffers.lock().unwrap();
assert_eq!(buffers.available.len(), capacity);
assert!(buffers
.available
.iter()
.all(|buffer| buffer.capacity() >= PACKET_SAMPLES));
}
#[test]
fn callback_finishes_while_worker_recycles_a_buffer() {
assert_callback_progress(PausePoint::Recycle);
}
#[test]
fn callback_finishes_while_worker_releases_a_ready_packet() {
assert_callback_progress(PausePoint::Consume);
}
fn consume_concurrently(
receiver: CapturePcmReceiver,
finished: Arc<AtomicBool>,
first_sequence: usize,
) -> (CapturePcmReceiver, usize) {
let mut received = 0;
let mut next_ordinal = 0;
loop {
if let Some((sequence, packet)) = receiver.pop_packet() {
let ordinal = sequence.wrapping_sub(first_sequence);
assert!(ordinal >= next_ordinal && ordinal < CONCURRENT_PACKETS);
assert_eq!(packet, [ordinal as f32; PACKET_SAMPLES]);
next_ordinal = ordinal + 1;
received += 1;
receiver.recycle(packet);
} else if finished.load(Ordering::Acquire) && receiver.is_empty() {
return (receiver, received);
} else {
std::thread::yield_now();
}
}
}
#[test]
fn concurrent_handoff_preserves_order_buffers_and_loss_counts_across_sequence_wrap() {
const FIRST_SEQUENCE: usize = usize::MAX - CONCURRENT_PACKETS / 2;
for capacity in [1, 2, CAPTURE_PCM_QUEUE_PACKETS] {
let (mut sender, receiver) = new_pcm_handoff(capacity, PACKET_SAMPLES).unwrap();
sender.sequence = FIRST_SEQUENCE;
let finished = Arc::new(AtomicBool::new(false));
let worker_finished = finished.clone();
let worker = std::thread::spawn(move || {
consume_concurrently(receiver, worker_finished, FIRST_SEQUENCE)
});
assert_no_allocations(|| {
for ordinal in 0..CONCURRENT_PACKETS {
sender.submit(&[ordinal as f32; PACKET_SAMPLES]);
}
});
finished.store(true, Ordering::Release);
let (receiver, received) = worker.join().unwrap();
let stats = receiver.take_stats();
assert_eq!(received + stats.loss.dropped, CONCURRENT_PACKETS);
assert_eq!(stats.loss.oversized, 0);
assert_eq!(stats.loss.recycle_failures, 0);
assert!(stats.max_queued_packets <= capacity);
assert_pool_restored(&receiver, capacity);
}
}