diff --git a/Cargo.lock b/Cargo.lock index e58ffd397..f54e14e02 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#41c8a2a75903ffe9cd44cf176ce6b69e5d916d47" +source = "git+https://github.com/rustdesk-org/cpal?branch=osx-screencapturekit#69ad2578adc9200093fc81cdfbdad63dbc4274f9" dependencies = [ "alsa", "cidre", diff --git a/src/server/audio_capture_error.rs b/src/server/audio_capture_error.rs new file mode 100644 index 000000000..bb837d8bf --- /dev/null +++ b/src/server/audio_capture_error.rs @@ -0,0 +1,72 @@ +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; + +// Each stream owns its flag so a late callback cannot restart a replacement. +#[derive(Clone, Default)] +pub(super) struct CaptureErrorHandler { + interrupted: Arc, +} + +impl CaptureErrorHandler { + pub(super) fn handle(&self, error: cpal::StreamError) { + if matches!(error, cpal::StreamError::StreamInterrupted { .. }) { + // ScreenCaptureKit can stop capture while the remote session stays open. + // The observed -3821 error does not identify its underlying trigger. + // https://developer.apple.com/documentation/screencapturekit/scstreamdelegate/stream(_:didstopwitherror:) + hbb_common::log::error!("Audio capture stream interrupted: {error}"); + self.interrupted.store(true, Ordering::Relaxed); + } else { + // Keep frequent sample-buffer errors at the existing trace level. + hbb_common::log::trace!("an error occurred on stream: {error}"); + } + } + + pub(super) fn needs_restart(&self) -> bool { + self.interrupted.load(Ordering::Relaxed) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use cpal::{BackendSpecificError, StreamError}; + + fn system_interruption() -> StreamError { + StreamError::StreamInterrupted { + err: BackendSpecificError { + description: "ScreenCaptureKit system-stopped capture (-3821)".to_owned(), + }, + } + } + + #[test] + fn system_interruption_requests_recreation_of_the_active_stream() { + let errors = CaptureErrorHandler::default(); + let callback = errors.clone(); + assert!(!errors.needs_restart()); + callback.handle(system_interruption()); + assert!(errors.needs_restart()); + } + + #[test] + fn late_error_from_an_old_stream_does_not_restart_its_replacement() { + let old_callback = CaptureErrorHandler::default(); + let replacement = CaptureErrorHandler::default(); + old_callback.handle(system_interruption()); + assert!(old_callback.needs_restart()); + assert!(!replacement.needs_restart()); + } + + #[test] + fn other_backend_errors_do_not_request_recreation() { + let errors = CaptureErrorHandler::default(); + errors.handle(StreamError::BackendSpecific { + err: BackendSpecificError { + description: "A sample buffer could not be read".to_owned(), + }, + }); + assert!(!errors.needs_restart()); + } +} diff --git a/src/server/audio_service.rs b/src/server/audio_service.rs index b58f83bcf..52f2eff2e 100644 --- a/src/server/audio_service.rs +++ b/src/server/audio_service.rs @@ -171,9 +171,14 @@ pub fn is_screen_capture_kit_available() -> bool { .any(|host| *host == cpal::HostId::ScreenCaptureKit) } +#[cfg(not(any(target_os = "linux", target_os = "android")))] +#[path = "audio_capture_error.rs"] +mod audio_capture_error; + #[cfg(not(any(target_os = "linux", target_os = "android")))] mod cpal_impl { use self::service::{Reset, ServiceSwap}; + use super::audio_capture_error::CaptureErrorHandler; use super::*; use cpal::{ traits::{DeviceTrait, HostTrait, StreamTrait}, @@ -192,7 +197,7 @@ mod cpal_impl { #[derive(Default)] pub struct State { - stream: Option<(Box, Arc)>, + stream: Option<(Box, Arc, CaptureErrorHandler)>, } impl super::service::Reset for State { @@ -210,8 +215,10 @@ mod cpal_impl { } _ => {} } - if let Some((_, format)) = &state.stream { + if let Some((_, format, _)) = &state.stream { sp.send_shared(format.clone()); + #[cfg(target_os = "macos")] + log::info!("Audio capture stream recreated; replacement format sent"); } RESTARTING.store(false, Ordering::SeqCst); Ok(()) @@ -225,7 +232,7 @@ mod cpal_impl { } _ => {} } - if let Some((_, format)) = &state.stream { + if let Some((_, format, _)) = &state.stream { sps.send_shared(format.clone()); } Ok(()) @@ -234,6 +241,13 @@ 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"); + super::restart(); + } + } if !RESTARTING.load(Ordering::SeqCst) { run_serv_snapshot(sp, state) } else { @@ -353,7 +367,9 @@ mod cpal_impl { Ok((device, format)) } - fn play(sp: &GenericService) -> ResultType<(Box, Arc)> { + fn play( + sp: &GenericService, + ) -> ResultType<(Box, Arc, CaptureErrorHandler)> { use cpal::SampleFormat::*; let (device, config) = get_device()?; let sp = sp.clone(); @@ -371,7 +387,7 @@ mod cpal_impl { 48000 }; let ch = if config.channels() > 1 { Stereo } else { Mono }; - let stream = match config.sample_format() { + let (stream, errors) = match config.sample_format() { I8 => build_input_stream::(device, &config, sp, sample_rate, ch)?, I16 => build_input_stream::(device, &config, sp, sample_rate, ch)?, I32 => build_input_stream::(device, &config, sp, sample_rate, ch)?, @@ -385,9 +401,12 @@ mod cpal_impl { 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 _)), + errors, )) } @@ -397,14 +416,15 @@ mod cpal_impl { sp: GenericService, sample_rate: u32, encode_channel: magnum_opus::Channels, - ) -> ResultType + ) -> ResultType<(cpal::Stream, CaptureErrorHandler)> where T: cpal::SizedSample + dasp::sample::ToSample, { - let err_fn = move |err| { - // too many UnknownErrno, will improve later - log::trace!("an error occurred on stream: {}", err); - }; + let errors = CaptureErrorHandler::default(); + let callback_errors = errors.clone(); + let err_fn = move |err| callback_errors.handle(err); + #[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 { @@ -431,6 +451,25 @@ mod cpal_impl { &stream_config, move |data: &[T], _: &InputCallbackInfo| { let buffer: Vec = data.iter().map(|s| T::to_sample(*s)).collect(); + #[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 { @@ -449,7 +488,7 @@ mod cpal_impl { err_fn, timeout, )?; - Ok(stream) + Ok((stream, errors)) } }