diff --git a/Cargo.lock b/Cargo.lock index 9a6cf9a5a..096b75685 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1717,7 +1717,7 @@ dependencies = [ [[package]] name = "cpal" version = "0.15.3" -source = "git+https://github.com/rustdesk-org/cpal?branch=osx-screencapturekit#69ad2578adc9200093fc81cdfbdad63dbc4274f9" +source = "git+https://github.com/rustdesk-org/cpal?branch=osx-screencapturekit#96d4da121b7d949677ac5b6887413a9185fd7f39" dependencies = [ "alsa", "cidre", diff --git a/src/client.rs b/src/client.rs index 8138fed8b..3710d2639 100644 --- a/src/client.rs +++ b/src/client.rs @@ -95,7 +95,10 @@ pub use super::lang::*; #[cfg(not(target_os = "linux"))] mod audio_playback; +#[cfg(target_os = "windows")] +mod audio_playback_recovery; #[cfg(all(test, not(target_os = "linux")))] +#[path = "client/tests/audio_state_tests.rs"] mod audio_state_tests; pub mod file_trait; pub mod helper; @@ -2083,6 +2086,8 @@ pub struct AudioHandler { device_channel: u16, #[cfg(not(target_os = "linux"))] playback_status: Arc, + #[cfg(target_os = "windows")] + playback_recovery: audio_playback_recovery::PlaybackRecovery, } #[cfg(not(target_os = "linux"))] @@ -2390,22 +2395,53 @@ impl AudioHandler { /// Handle audio format and create an audio decoder. pub fn handle_format(&mut self, f: AudioFormat) { + self.handle_format_with_start(f, Self::start_audio); + } + + fn handle_format_with_start( + &mut self, + f: AudioFormat, + start: impl FnOnce(&mut Self, AudioFormat) -> ResultType<()>, + ) { if !is_supported_audio_channel_count(f.channels) { log::error!("Unsupported audio channel count: {}", f.channels); return; } match AudioDecoder::new(f.sample_rate, if f.channels > 1 { Stereo } else { Mono }) { Ok(d) => { + #[cfg(target_os = "windows")] + let playback_failed = self.cancel_pending_playback(); #[cfg(target_os = "linux")] let keep_existing_stream = self.simple.is_some() && self.sample_rate.0 == f.sample_rate && u32::from(self.channels) == f.channels; #[cfg(not(target_os = "linux"))] - let keep_existing_stream = false; + let keep_existing_stream = self.audio_stream.is_some() + && self.sample_rate.0 == f.sample_rate + && u32::from(self.channels) == f.channels; let buffer = vec![0.; f.sample_rate as usize * f.channels as usize]; + #[cfg(not(target_os = "linux"))] + let mut previous = std::mem::take(self); + #[cfg(target_os = "windows")] + self.prepare_playback(&f); self.audio_decoder = Some((d, buffer)); self.channels = f.channels as _; - let result = self.start_audio(f); + let result = start(self, f); + #[cfg(target_os = "windows")] + let keep_existing_stream = keep_existing_stream + && !playback_failed + && !previous.playback_recovery.report_pending(); + #[cfg(not(target_os = "linux"))] + if result.is_err() && keep_existing_stream { + // The restarted capture has new Opus history even when output startup fails. + previous.audio_decoder = self.audio_decoder.take(); + *self = previous; + self.handle_audio_start_result(result, true); + return; + } + #[cfg(target_os = "windows")] + self.finish_playback_replacement(result, keep_existing_stream.then_some(previous)); + #[cfg(not(target_os = "windows"))] self.handle_audio_start_result(result, keep_existing_stream); } Err(err) => { @@ -2493,6 +2529,9 @@ impl AudioHandler { device: &Device, ) -> ResultType<()> { self.device_channel = config.channels; + #[cfg(target_os = "windows")] + let err_fn = self.playback_recovery.new_error_callback(); + #[cfg(not(target_os = "windows"))] let err_fn = move |err| { // too many errors, will improve later log::trace!("an error occurred on stream: {}", err); @@ -4065,7 +4104,11 @@ pub fn start_audio_thread() -> MediaSender { std::thread::spawn(move || { let mut audio_handler = AudioHandler::default(); loop { - if let Ok(data) = audio_receiver.recv() { + #[cfg(target_os = "windows")] + let received = audio_handler.receive_audio(&audio_receiver); + #[cfg(not(target_os = "windows"))] + let received = audio_receiver.recv(); + if let Ok(data) = received { match data { MediaData::AudioFrame(af) => { audio_handler.handle_frame(*af); diff --git a/src/client/audio_playback_recovery.rs b/src/client/audio_playback_recovery.rs new file mode 100644 index 000000000..8d3755f84 --- /dev/null +++ b/src/client/audio_playback_recovery.rs @@ -0,0 +1,184 @@ +use super::{AudioDecoder, AudioFormat, AudioHandler, MediaData, Mono, Stereo}; +use cpal::StreamError; +use crossbeam_queue::SegQueue; +use hbb_common::{log, tokio::time::Instant, ResultType}; +use std::{ + sync::{atomic::Ordering, mpsc, Arc}, + time::Duration, +}; + +const RECOVERY_INTERVAL: Duration = Duration::from_secs(1); +pub(super) const STARTUP_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(5); +// The pinned WASAPI backend reports this warning but keeps its worker running. +const PRIORITY_WARNING_PREFIX: &str = "SetThreadPriority failed: "; + +#[path = "audio_playback_startup.rs"] +mod startup; + +#[derive(Default)] +pub(super) struct PlaybackRecovery { + pub(super) errors: Arc>, + format: Option, + pub(super) retry_at: Option, + restart_not_before: Option, + awaiting_callback: bool, + startup_deadline: Option, + pending_output: Option>, +} + +impl PlaybackRecovery { + pub(super) fn new_error_callback(&mut self) -> impl FnMut(StreamError) + Send + 'static { + self.errors = Default::default(); + let errors = self.errors.clone(); + move |error| errors.push(error) + } + + pub(super) fn report_pending(&self) -> bool { + let mut failed = false; + while let Some(error) = self.errors.pop() { + if matches!(&error, StreamError::BackendSpecific { err } + if err.description.starts_with(PRIORITY_WARNING_PREFIX)) + { + log::warn!("Audio playback nonterminal priority warning: {error}"); + } else { + log::error!("Audio playback stream failed: {error}"); + failed = true; + } + } + failed + } +} + +impl AudioHandler { + fn clear_playback_stream(&mut self) { + // Dropping CPAL may join its worker; run this on the owner, not its callback. + self.audio_stream = None; + self.playback_recovery.report_pending(); + self.playback_status.report_errors(); + let recovery = std::mem::take(self).playback_recovery; + self.playback_recovery.format = recovery.format; + self.playback_recovery.retry_at = recovery.retry_at; + self.playback_recovery.restart_not_before = recovery.restart_not_before; + } + + pub(super) fn prepare_playback(&mut self, format: &AudioFormat) { + self.clear_playback_stream(); + self.playback_recovery.format = Some(format.clone()); + self.playback_recovery.retry_at = None; + self.playback_recovery.restart_not_before = None; + } + + pub(super) fn finish_playback_start(&mut self, result: ResultType<()>) { + let now = Instant::now(); + let retry_at = now + RECOVERY_INTERVAL; + self.playback_recovery.restart_not_before = Some(retry_at); + match result { + Ok(()) => { + self.playback_recovery.retry_at = None; + self.playback_recovery.awaiting_callback = true; + self.playback_recovery.startup_deadline = Some(now + STARTUP_CONFIRMATION_TIMEOUT); + log::info!("Audio playback stream opened; waiting for output callback"); + } + Err(error) => { + self.clear_playback_stream(); + self.playback_recovery.retry_at = Some(retry_at); + log::error!( + "Audio playback start failed: {error:#}; retrying in {RECOVERY_INTERVAL:?}" + ); + } + } + } + + fn playback_start_timed_out(&mut self, now: Instant) -> bool { + if !self.playback_recovery.awaiting_callback + || self.playback_status.ready.load(Ordering::Acquire) + || !self + .playback_recovery + .startup_deadline + .is_some_and(|due| now >= due) + { + return false; + } + self.playback_recovery.awaiting_callback = false; + self.playback_recovery.startup_deadline = None; + log::error!("Audio playback start timed out waiting for output callback"); + true + } + + fn restart_playback(&mut self, format: AudioFormat) -> ResultType<()> { + let channels = if format.channels > 1 { Stereo } else { Mono }; + let decoder = AudioDecoder::new(format.sample_rate, channels)?; + let buffer = vec![0.; format.sample_rate as usize * format.channels as usize]; + let channel_count = format.channels as _; + self.start_audio(format)?; + self.channels = channel_count; + self.audio_decoder = Some((decoder, buffer)); + Ok(()) + } + + pub(super) fn recover_playback_with( + &mut self, + now: Instant, + restart: impl FnOnce(&mut Self, AudioFormat) -> ResultType<()>, + ) { + let failed = self.resolve_pending_playback(now).unwrap_or_else(|| { + self.playback_recovery.report_pending() || self.playback_start_timed_out(now) + }); + if failed { + self.clear_playback_stream(); + self.playback_recovery.retry_at = Some( + self.playback_recovery + .restart_not_before + .map_or(now, |due| due.max(now)), + ); + } + if self.playback_recovery.awaiting_callback + && self.playback_status.ready.load(Ordering::Acquire) + { + self.playback_recovery.awaiting_callback = false; + self.playback_recovery.startup_deadline = None; + log::info!("Audio playback output callback started"); + } + if !self + .playback_recovery + .retry_at + .is_some_and(|due| now >= due) + { + return; + } + let Some(format) = self.playback_recovery.format.clone() else { + return; + }; + log::info!("Recreating audio playback on the current default output device"); + let result = restart(self, format); + self.finish_playback_start(result); + } + + pub(super) fn receive_audio( + &mut self, + receiver: &mpsc::Receiver, + ) -> Result { + receive_with_recovery(receiver, RECOVERY_INTERVAL, || { + self.recover_playback_with(Instant::now(), Self::restart_playback); + }) + } +} + +pub(super) fn receive_with_recovery( + receiver: &mpsc::Receiver, + interval: Duration, + mut recover: impl FnMut(), +) -> Result { + loop { + match receiver.recv_timeout(interval) { + Ok(data) => { + if !matches!(data, MediaData::AudioFormat(_)) { + recover(); + } + return Ok(data); + } + Err(mpsc::RecvTimeoutError::Timeout) => recover(), + Err(mpsc::RecvTimeoutError::Disconnected) => return Err(mpsc::RecvError), + } + } +} diff --git a/src/client/audio_playback_startup.rs b/src/client/audio_playback_startup.rs new file mode 100644 index 000000000..e789cc462 --- /dev/null +++ b/src/client/audio_playback_startup.rs @@ -0,0 +1,63 @@ +use super::{AudioHandler, Instant, Ordering, ResultType}; +use hbb_common::log; + +impl AudioHandler { + pub(in crate::client) fn cancel_pending_playback(&mut self) -> bool { + // Format messages bypass recovery; retain a usable candidate before superseding it. + let failed = self + .resolve_pending_playback(Instant::now()) + .unwrap_or(false); + if let Some(mut pending) = self.playback_recovery.pending_output.take() { + pending.audio_stream = None; + pending.playback_recovery.report_pending(); + pending.playback_status.report_errors(); + } + failed + } + + pub(in crate::client) fn finish_playback_replacement( + &mut self, + result: ResultType<()>, + previous: Option, + ) { + self.finish_playback_start(result); + let Some(mut previous) = previous else { + return; + }; + previous.audio_decoder = self.audio_decoder.take(); + let candidate = std::mem::replace(self, previous); + self.playback_recovery.pending_output = Some(Box::new(candidate)); + log::info!("Audio playback replacement pending; continuing on the compatible output"); + self.recover_playback_with(Instant::now(), Self::restart_playback); + } + + pub(super) fn resolve_pending_playback(&mut self, now: Instant) -> Option { + let mut candidate = self.playback_recovery.pending_output.take()?; + let candidate_failed = + candidate.playback_recovery.report_pending() || candidate.playback_start_timed_out(now); + let previous_failed = + self.playback_recovery.report_pending() || self.playback_start_timed_out(now); + if candidate_failed { + self.playback_recovery.restart_not_before = + candidate.playback_recovery.restart_not_before; + candidate.audio_stream = None; + candidate.playback_recovery.report_pending(); + candidate.playback_status.report_errors(); + if !previous_failed { + log::error!("Audio playback replacement failed before startup confirmation; keeping the existing compatible stream"); + } + return Some(previous_failed); + } + if candidate.playback_status.ready.load(Ordering::Acquire) || previous_failed { + candidate.audio_decoder = self.audio_decoder.take(); + self.audio_stream = None; + self.playback_recovery.report_pending(); + self.playback_status.report_errors(); + *self = *candidate; + return Some(false); + } + self.playback_recovery.pending_output = Some(candidate); + // A second active-queue read could discard a healthy pending candidate. + Some(false) + } +} diff --git a/src/client/audio_state_tests.rs b/src/client/audio_state_tests.rs deleted file mode 100644 index 3c2924f5e..000000000 --- a/src/client/audio_state_tests.rs +++ /dev/null @@ -1,113 +0,0 @@ -use super::{create_audio_resampler, AudioDecoder, AudioFrame, AudioHandler, Stereo}; -use cpal::traits::StreamTrait; -use hbb_common::anyhow::anyhow; -use magnum_opus::{Application::LowDelay, Encoder}; -use ringbuf::{ring_buffer::RbBase, Rb}; -use std::sync::{ - atomic::{AtomicBool, Ordering}, - Arc, -}; - -const INPUT_RATE: u32 = 24_000; -const OUTPUT_RATE: u32 = 48_000; -const CHANNELS: u16 = 2; -const PACKETS_PER_SECOND: usize = 100; -const MAX_PACKET_BYTES: usize = 4_096; -const SAMPLE_VALUE: f32 = 0.25; - -struct TrackedAudioStream(Arc); - -impl StreamTrait for TrackedAudioStream { - fn play(&self) -> Result<(), cpal::PlayStreamError> { - Ok(()) - } - - fn pause(&self) -> Result<(), cpal::PauseStreamError> { - Ok(()) - } -} - -impl Drop for TrackedAudioStream { - fn drop(&mut self) { - self.0.store(true, Ordering::SeqCst); - } -} - -fn decoder(sample_rate: u32) -> (AudioDecoder, Vec) { - ( - AudioDecoder::new(sample_rate, Stereo).unwrap(), - vec![0.0; sample_rate as usize * CHANNELS as usize], - ) -} - -fn active_handler(input_rate: u32) -> (AudioHandler, Arc) { - let dropped = Arc::new(AtomicBool::new(false)); - let handler = AudioHandler { - audio_decoder: Some(decoder(input_rate)), - audio_resampler: create_audio_resampler(input_rate, OUTPUT_RATE, CHANNELS).unwrap(), - sample_rate: (input_rate, OUTPUT_RATE), - audio_stream: Some(Box::new(TrackedAudioStream(dropped.clone()))), - channels: CHANNELS, - device_channel: CHANNELS, - ..Default::default() - }; - handler.playback_status.ready.store(true, Ordering::Release); - (handler, dropped) -} - -fn audio_frame() -> AudioFrame { - let samples = OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize; - let mut encoder = Encoder::new(OUTPUT_RATE, Stereo, LowDelay).unwrap(); - AudioFrame { - data: encoder - .encode_vec_float(&vec![SAMPLE_VALUE; samples], MAX_PACKET_BYTES) - .unwrap() - .into(), - ..Default::default() - } -} - -#[test] -fn failed_format_change_discards_old_playback_state() { - let (mut handler, dropped) = active_handler(INPUT_RATE); - handler - .audio_buffer - .0 - .lock() - .unwrap() - .push_slice(&[SAMPLE_VALUE; CHANNELS as usize]); - handler.audio_decoder = Some(decoder(OUTPUT_RATE)); - handler.sample_rate = (OUTPUT_RATE, OUTPUT_RATE); - - handler.handle_audio_start_result( - Err(anyhow!("Injected output stream startup failure")), - false, - ); - - assert!(dropped.load(Ordering::SeqCst)); - assert!(handler.audio_stream.is_none()); - assert!(handler.audio_resampler.is_none()); - assert!(handler.audio_decoder.is_none()); - assert!(!handler.playback_status.ready.load(Ordering::Acquire)); - handler.handle_frame(audio_frame()); - assert_eq!(handler.audio_buffer.0.lock().unwrap().occupied_len(), 0); -} - -#[test] -fn successful_start_or_compatible_failure_preserves_audio_packet_duration() { - for result in [ - Ok(()), - Err(anyhow!("Injected compatible stream replacement failure")), - ] { - let (mut handler, dropped) = active_handler(OUTPUT_RATE); - - handler.handle_audio_start_result(result, true); - handler.handle_frame(audio_frame()); - - assert!(!dropped.load(Ordering::SeqCst)); - assert_eq!( - handler.audio_buffer.0.lock().unwrap().occupied_len(), - OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize - ); - } -} diff --git a/src/client/tests/audio_playback_recovery_tests.rs b/src/client/tests/audio_playback_recovery_tests.rs new file mode 100644 index 000000000..663e88d48 --- /dev/null +++ b/src/client/tests/audio_playback_recovery_tests.rs @@ -0,0 +1,226 @@ +use super::*; +use crate::client::{ + audio_playback::AudioPlaybackStatus, audio_playback_recovery::STARTUP_CONFIRMATION_TIMEOUT, +}; +use cpal::StreamError; +use crossbeam_queue::SegQueue; +use hbb_common::tokio::time::Instant; +use std::time::Duration; + +const AFTER_COOLDOWN: Duration = Duration::from_secs(2); + +type PendingOutput = ( + Arc, + Arc, + Arc>, +); + +fn install_output(handler: &mut AudioHandler, dropped: Arc) { + handler.sample_rate = (INPUT_RATE, OUTPUT_RATE); + handler.device_channel = CHANNELS; + handler.audio_stream = Some(Box::new(TrackedAudioStream(dropped))); + handler.playback_status.ready.store(true, Ordering::Release); +} + +fn recovery_handler() -> (AudioHandler, Arc) { + let dropped = Arc::new(AtomicBool::new(false)); + let mut handler = AudioHandler::default(); + handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| { + install_output(candidate, dropped.clone()); + Ok(()) + }); + (handler, dropped) +} + +fn begin_pending(handler: &mut AudioHandler) -> PendingOutput { + let dropped = Arc::new(AtomicBool::new(false)); + let mut state = None; + handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| { + install_output(candidate, dropped.clone()); + candidate + .playback_status + .ready + .store(false, Ordering::Release); + state = Some(( + candidate.playback_status.clone(), + candidate.playback_recovery.errors.clone(), + )); + Ok(()) + }); + let (status, errors) = state.unwrap(); + (dropped, status, errors) +} + +#[test] +fn unconfirmed_start_retries_without_callback_or_error() { + let dropped = Arc::new(AtomicBool::new(false)); + let mut handler = AudioHandler::default(); + handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| { + install_output(candidate, dropped.clone()); + candidate + .playback_status + .ready + .store(false, Ordering::Release); + Ok(()) + }); + handler.recover_playback_with(Instant::now(), |_, _| { + panic!("Startup confirmation deadline has not elapsed") + }); + assert!(!dropped.load(Ordering::SeqCst)); + let expired = Instant::now() + STARTUP_CONFIRMATION_TIMEOUT; + let mut attempts = 0; + handler.recover_playback_with(expired, |candidate, requested| { + attempts += 1; + assert!(dropped.load(Ordering::SeqCst)); + assert_eq!(requested, format(INPUT_RATE, CHANNELS)); + install_output(candidate, Arc::new(AtomicBool::new(false))); + Ok(()) + }); + assert_eq!(attempts, 1); + handler.recover_playback_with(expired + STARTUP_CONFIRMATION_TIMEOUT, |_, _| { + panic!("Confirmed output must not be reopened") + }); +} + +#[test] +fn already_terminal_candidate_cannot_replace_compatible_output() { + let (mut handler, dropped) = recovery_handler(); + let candidate_dropped = Arc::new(AtomicBool::new(false)); + let old_buffer = handler.audio_buffer.0.clone(); + handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| { + install_output(candidate, candidate_dropped.clone()); + candidate + .playback_recovery + .errors + .push(StreamError::DeviceNotAvailable); + Ok(()) + }); + assert!(!dropped.load(Ordering::SeqCst)); + assert!(candidate_dropped.load(Ordering::SeqCst)); + assert!(Arc::ptr_eq(&old_buffer, &handler.audio_buffer.0)); + handler.recover_playback_with(Instant::now() + AFTER_COOLDOWN, |_, _| { + panic!("Compatible active output must not be reopened") + }); +} + +#[test] +fn pending_output_keeps_playing_and_transfers_decoder_history_on_commit() { + let (mut handler, old_dropped) = recovery_handler(); + let old_buffer = handler.audio_buffer.0.clone(); + let (_, status, _) = begin_pending(&mut handler); + let frame = audio_frame(); + let (mut reference, mut expected) = decoder(INPUT_RATE); + reference + .decode_float(&frame.data, &mut expected, false) + .unwrap(); + handler.handle_frame(frame.clone()); + assert!(Arc::ptr_eq(&old_buffer, &handler.audio_buffer.0)); + assert!(!drain_audio(&handler).is_empty()); + status.ready.store(true, Ordering::Release); + handler.recover_playback_with(Instant::now(), |_, _| panic!("Candidate already exists")); + assert!(old_dropped.load(Ordering::SeqCst)); + assert!(Arc::ptr_eq(&status, &handler.playback_status)); + let samples = reference + .decode_float(&frame.data, &mut expected, false) + .unwrap() + * CHANNELS as usize; + handler.handle_frame(frame); + assert_eq!( + &handler.audio_decoder.as_ref().unwrap().1[..samples], + &expected[..samples] + ); +} + +#[test] +fn rollback_keeps_the_restarted_decoders_accumulated_history() { + let (mut handler, old_dropped) = recovery_handler(); + let frame = audio_frame(); + handler.handle_frame(frame.clone()); + let (candidate_dropped, _, errors) = begin_pending(&mut handler); + let (mut reference, mut expected) = decoder(INPUT_RATE); + handler.handle_frame(frame.clone()); + reference + .decode_float(&frame.data, &mut expected, false) + .unwrap(); + errors.push(StreamError::DeviceNotAvailable); + handler.recover_playback_with(Instant::now(), |_, _| panic!("Old output still works")); + let samples = reference + .decode_float(&frame.data, &mut expected, false) + .unwrap() + * CHANNELS as usize; + handler.handle_frame(frame); + assert!(!old_dropped.load(Ordering::SeqCst)); + assert!(candidate_dropped.load(Ordering::SeqCst)); + assert_eq!( + &handler.audio_decoder.as_ref().unwrap().1[..samples], + &expected[..samples] + ); +} + +#[test] +fn later_compatible_format_retires_only_the_pending_attempt() { + let (mut handler, old_dropped) = recovery_handler(); + let (first_dropped, _, first_errors) = begin_pending(&mut handler); + let (second_dropped, second_status, _) = begin_pending(&mut handler); + assert!(first_dropped.load(Ordering::SeqCst)); + assert!(!old_dropped.load(Ordering::SeqCst)); + first_errors.push(StreamError::DeviceNotAvailable); + second_status.ready.store(true, Ordering::Release); + handler.recover_playback_with(Instant::now(), |_, _| { + panic!("Retired attempt affected current output") + }); + assert!(old_dropped.load(Ordering::SeqCst)); + assert!(!second_dropped.load(Ordering::SeqCst)); + assert!(Arc::ptr_eq(&second_status, &handler.playback_status)); +} + +#[test] +fn both_outputs_failing_retains_format_and_paces_recovery() { + let (mut handler, old_dropped) = recovery_handler(); + let old_errors = handler.playback_recovery.errors.clone(); + let (candidate_dropped, _, errors) = begin_pending(&mut handler); + old_errors.push(StreamError::DeviceNotAvailable); + errors.push(StreamError::DeviceNotAvailable); + handler.recover_playback_with(Instant::now(), |_, _| panic!("Retry must be paced")); + assert!(old_dropped.load(Ordering::SeqCst)); + assert!(candidate_dropped.load(Ordering::SeqCst)); + assert!(handler.audio_stream.is_none()); + let due = handler.playback_recovery.retry_at.unwrap(); + let mut attempts = 0; + handler.recover_playback_with(due, |_, requested| { + attempts += 1; + assert_eq!(requested, format(INPUT_RATE, CHANNELS)); + Ok(()) + }); + assert_eq!(attempts, 1); +} + +#[test] +fn superseding_format_keeps_ready_candidate_when_active_output_failed() { + let (mut handler, old_dropped) = recovery_handler(); + let old_errors = handler.playback_recovery.errors.clone(); + let (candidate_dropped, candidate_status, _) = begin_pending(&mut handler); + candidate_status.ready.store(true, Ordering::Release); + old_errors.push(StreamError::DeviceNotAvailable); + handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), failed_output); + assert!(!candidate_dropped.load(Ordering::SeqCst)); + assert!(old_dropped.load(Ordering::SeqCst)); + assert!(Arc::ptr_eq(&candidate_status, &handler.playback_status)); + assert!(handler.audio_decoder.is_some()); + assert!(handler.playback_recovery.retry_at.is_none()); +} + +#[test] +fn superseding_format_preserves_failure_when_both_outputs_failed() { + let (mut handler, old_dropped) = recovery_handler(); + let old_errors = handler.playback_recovery.errors.clone(); + let (candidate_dropped, status, errors) = begin_pending(&mut handler); + status.ready.store(true, Ordering::Release); + old_errors.push(StreamError::DeviceNotAvailable); + errors.push(StreamError::DeviceNotAvailable); + handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), failed_output); + assert!(candidate_dropped.load(Ordering::SeqCst)); + assert!(old_dropped.load(Ordering::SeqCst)); + assert!(handler.audio_stream.is_none()); + assert!(handler.playback_recovery.retry_at.is_some()); +} diff --git a/src/client/tests/audio_state_tests.rs b/src/client/tests/audio_state_tests.rs new file mode 100644 index 000000000..fa4c6efcc --- /dev/null +++ b/src/client/tests/audio_state_tests.rs @@ -0,0 +1,128 @@ +use super::{create_audio_resampler, AudioDecoder, AudioFormat, AudioFrame, AudioHandler, Stereo}; +use cpal::traits::StreamTrait; +use hbb_common::{anyhow::anyhow, ResultType}; +use magnum_opus::{Application::LowDelay, Encoder}; +use ringbuf::Rb; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; + +const INPUT_RATE: u32 = 24_000; +const OUTPUT_RATE: u32 = 48_000; +const CHANNELS: u16 = 2; +const PACKETS_PER_SECOND: usize = 100; +const MAX_PACKET_BYTES: usize = 4_096; +const SAMPLE_VALUE: f32 = 0.25; +const MONO_CHANNELS: u16 = 1; + +#[cfg(target_os = "windows")] +#[path = "audio_playback_recovery_tests.rs"] +mod recovery_tests; + +struct TrackedAudioStream(Arc); + +impl StreamTrait for TrackedAudioStream { + fn play(&self) -> Result<(), cpal::PlayStreamError> { + Ok(()) + } + + fn pause(&self) -> Result<(), cpal::PauseStreamError> { + Ok(()) + } +} + +impl Drop for TrackedAudioStream { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } +} + +fn decoder(sample_rate: u32) -> (AudioDecoder, Vec) { + ( + AudioDecoder::new(sample_rate, Stereo).unwrap(), + vec![0.0; sample_rate as usize * CHANNELS as usize], + ) +} + +fn active_handler(input_rate: u32) -> (AudioHandler, Arc) { + let dropped = Arc::new(AtomicBool::new(false)); + let handler = AudioHandler { + audio_decoder: Some(decoder(input_rate)), + audio_resampler: create_audio_resampler(input_rate, OUTPUT_RATE, CHANNELS).unwrap(), + sample_rate: (input_rate, OUTPUT_RATE), + audio_stream: Some(Box::new(TrackedAudioStream(dropped.clone()))), + channels: CHANNELS, + device_channel: CHANNELS, + ..Default::default() + }; + handler.playback_status.ready.store(true, Ordering::Release); + (handler, dropped) +} + +fn audio_frame() -> AudioFrame { + let samples = OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize; + let mut encoder = Encoder::new(OUTPUT_RATE, Stereo, LowDelay).unwrap(); + AudioFrame { + data: encoder + .encode_vec_float(&vec![SAMPLE_VALUE; samples], MAX_PACKET_BYTES) + .unwrap() + .into(), + ..Default::default() + } +} + +fn failed_output(candidate: &mut AudioHandler, _: AudioFormat) -> ResultType<()> { + candidate.sample_rate = (INPUT_RATE, INPUT_RATE); + candidate.device_channel = MONO_CHANNELS; + candidate + .audio_buffer + .resize(INPUT_RATE as _, MONO_CHANNELS as _); + candidate + .playback_status + .ready + .store(false, Ordering::Release); + Err(anyhow!("Injected playback failure")) +} + +fn format(sample_rate: u32, channels: u16) -> AudioFormat { + AudioFormat { + sample_rate, + channels: u32::from(channels), + ..Default::default() + } +} + +#[test] +fn identical_format_failure_preserves_playback_and_resampler_history() { + let (mut handler, dropped) = active_handler(INPUT_RATE); + let (mut reference, _) = active_handler(INPUT_RATE); + let (mut retained_decoder, _) = active_handler(INPUT_RATE); + retained_decoder.handle_frame(audio_frame()); + retained_decoder.handle_frame(audio_frame()); + handler.handle_frame(audio_frame()); + reference.handle_frame(audio_frame()); + let buffer = handler.audio_buffer.0.clone(); + let status = handler.playback_status.clone(); + + handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), failed_output); + reference.audio_decoder = Some(decoder(INPUT_RATE)); + handler.handle_frame(audio_frame()); + reference.handle_frame(audio_frame()); + + assert!(!dropped.load(Ordering::SeqCst)); + assert!(Arc::ptr_eq(&buffer, &handler.audio_buffer.0)); + assert!(Arc::ptr_eq(&status, &handler.playback_status)); + assert_eq!(handler.sample_rate, (INPUT_RATE, OUTPUT_RATE)); + assert_eq!(handler.device_channel, CHANNELS); + assert!(handler.playback_status.ready.load(Ordering::Acquire)); + let expected = drain_audio(&reference); + let actual = drain_audio(&handler); + assert!(!actual.is_empty()); + assert_ne!(drain_audio(&retained_decoder), expected); + assert_eq!(actual, expected); +} + +fn drain_audio(handler: &AudioHandler) -> Vec { + handler.audio_buffer.0.lock().unwrap().pop_iter().collect() +} diff --git a/src/server/audio_service.rs b/src/server/audio_service.rs index cfa98b047..f7ba94617 100644 --- a/src/server/audio_service.rs +++ b/src/server/audio_service.rs @@ -391,6 +391,15 @@ mod cpal_impl { if !audio_input.is_empty() { return get_audio_input(&audio_input); } + // The pinned CPAL uses event-driven WASAPI loopback here. Windows versions + // before Windows 10 1703 do not signal capture events, so system audio does + // not work on Win7. #16095 kept the same CPAL revision and loopback path; + // this limitation predates that PR. + // Ordinary microphone input is supported on Win7 and uses the branch above. + // #16095 added its callback-to-encoder wake dependency; see CapturePcmSender::wake + // for the new scheduling risk, whose audible impact on Win7 is unmeasured. + // https://learn.microsoft.com/en-us/windows/win32/coreaudio/loopback-recording + // https://learn.microsoft.com/en-us/windows/win32/coreaudio/capturesharedeventdriven let device = HOST .default_output_device() .with_context(|| "Failed to get default output device for loopback")?; diff --git a/src/server/audio_service/audio_capture_queue.rs b/src/server/audio_service/audio_capture_queue.rs index e46798e5b..aa64f9f00 100644 --- a/src/server/audio_service/audio_capture_queue.rs +++ b/src/server/audio_service/audio_capture_queue.rs @@ -97,6 +97,7 @@ impl Drop for CaptureEncoderWorker { fn drop(&mut self) { self.stop.store(true, Ordering::Release); if let Some(handle) = self.handle.take() { + // Owner-thread shutdown already waits via join(); see CapturePcmSender::wake for Win7. handle.thread().unpark(); if let Err(error) = handle.join() { log::error!("Failed to join audio encoder thread: {error:?}"); @@ -209,6 +210,16 @@ impl CapturePcmSender { fn wake(&self) { if let Some(thread) = self.handoff.wake_thread.get() { + // #16095 moved Opus encoding and message submission from the capture callback to a worker. + // Previously, the callback did that work directly, with allocations and blocking locks. + // On Win7 with Rust 1.75, if the worker is descheduled after publishing PARKED but before + // NtWaitForKeyedEvent, unpark() waits in NtReleaseKeyedEvent until the worker enters that wait. + // It does not wait for encoding; park_timeout() does not bound the callback's wait. + // Delays can cause gaps or stall teardown; a Win7 microphone regression has not been measured. + // System loopback already failed on Win7 before #16095 (see cpal_impl::get_device), + // so the affected path is microphone/input-device capture, including outgoing voice calls. + // Accept this risk to preserve Win7 input capture without a separate legacy notifier. + // https://github.com/rust-lang/rust/blob/1.75.0/library/std/src/sys/windows/thread_parking.rs thread.unpark(); } }