diff --git a/examples/exclusive.rs b/examples/exclusive.rs new file mode 100644 index 000000000..6567a38e0 --- /dev/null +++ b/examples/exclusive.rs @@ -0,0 +1,134 @@ +//! Silent exclusive-output lifecycle smoke test; requires a physical/virtual audio endpoint. +use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + mpsc, +}; +use std::time::{Duration, Instant}; + +#[cfg(target_os = "windows")] +mod exclusive_matrix; + +type Result = std::result::Result>; + +fn check_errors(errors: &mpsc::Receiver) -> Result<()> { + if let Ok(error) = errors.try_recv() { + return Err(error.into()); + } + Ok(()) +} + +fn wait_check_errors(errors: &mpsc::Receiver, duration: Duration) -> Result<()> { + match errors.recv_timeout(duration) { + Ok(error) => Err(error.into()), + Err(mpsc::RecvTimeoutError::Timeout) => Ok(()), + Err(mpsc::RecvTimeoutError::Disconnected) => { + Err("Stream error channel closed while awaiting callbacks".into()) + } + } +} + +fn run_stage( + stream: &cpal::Stream, + callbacks: &AtomicUsize, + errors: &mpsc::Receiver, + label: &str, +) -> Result<()> { + check_errors(errors)?; + let before = callbacks.load(Ordering::Relaxed); + let requested = Instant::now(); + stream.start()?; + println!("{label}: async start requested; waiting up to 10s for callback readiness"); + let deadline = requested + Duration::from_secs(10); + while callbacks.load(Ordering::Relaxed) == before { + check_errors(errors)?; + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(format!("{label}: no callback readiness within ten seconds").into()); + } + wait_check_errors(errors, remaining.min(Duration::from_millis(10)))?; + } + check_errors(errors)?; + let ready = callbacks.load(Ordering::Relaxed); + println!( + "{label}: callback ready after {:.3}s; observing another 1s", + requested.elapsed().as_secs_f64() + ); + let observation_end = Instant::now() + Duration::from_secs(1); + loop { + let remaining = observation_end.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + wait_check_errors(errors, remaining.min(Duration::from_millis(10)))?; + } + check_errors(errors)?; + let observed = callbacks.load(Ordering::Relaxed).saturating_sub(ready); + println!("{label}: {observed} additional callbacks observed"); + if observed == 0 { + return Err( + format!("{label}: callback progress stopped during one-second observation").into(), + ); + } + Ok(()) +} + +fn main() -> Result<()> { + let args: Vec<_> = std::env::args().skip(1).collect(); + if !args.is_empty() { + #[cfg(target_os = "windows")] + if args.first().map(String::as_str) == Some("--matrix") { + let name = match args.as_slice() { + [_, flag, name] if flag == "--device" => name.as_str(), + [_] => return Err( + "Usage: exclusive --matrix --device EXACT_NAME (the device name is required)" + .into(), + ), + _ => return Err("Usage: exclusive --matrix --device EXACT_NAME".into()), + }; + return exclusive_matrix::run(name); + } + return Err("Unsupported arguments for this platform".into()); + } + println!("Silent exclusive-output lifecycle smoke test (not a bit-perfect test)"); + let device = cpal::default_host() + .default_output_device() + .ok_or("No output device; an audio endpoint is required")?; + if !device.supports_exclusive() { + return Err("This device/backend does not implement exclusive output".into()); + } + let device = device.exclusive(true); + let supported = device.default_output_config()?; + println!("{}: {:?}", device, supported); + let callbacks = Arc::new(AtomicUsize::new(0)); + let callback_count = Arc::clone(&callbacks); + let (error_tx, errors) = mpsc::channel(); + let stream = device.build_output_stream_raw( + supported.config(), + supported.sample_format(), + move |_, _| { + callback_count.fetch_add(1, Ordering::Relaxed); + }, + move |error| { + let _ = error_tx.send(error); + }, + None, + )?; + println!("Actual callback frames: {}", stream.buffer_size()?); + run_stage(&stream, &callbacks, &errors, "Initial start")?; + stream.pause()?; + run_stage(&stream, &callbacks, &errors, "Resume after pause request")?; + stream.stop(Some(Duration::from_secs(1)))?; + run_stage( + &stream, + &callbacks, + &errors, + "Restart after bounded stop request", + )?; + stream.stop(Some(Duration::ZERO))?; + drop(stream); + check_errors(&errors)?; + println!("PASS: callback progress in all three running stages; no callback errors"); + Ok(()) +} diff --git a/examples/exclusive_matrix/mod.rs b/examples/exclusive_matrix/mod.rs new file mode 100644 index 000000000..5714f2b24 --- /dev/null +++ b/examples/exclusive_matrix/mod.rs @@ -0,0 +1,275 @@ +//! Silent, opt-in physical endpoint validation; not a hardware-rate/bit-perfect measurement. +use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; +use std::sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + mpsc, +}; +use std::time::{Duration, Instant}; + +type Result = std::result::Result>; + +struct Probe { + stream: cpal::Stream, + count: Arc, + errors: mpsc::Receiver, + timestamps: mpsc::Receiver, + diagnostics: mpsc::Receiver<(u64, usize, cpal::StreamTimestamp)>, +} + +impl Probe { + fn build(device: &cpal::Device, rate: u32) -> Result { + let count = Arc::new(AtomicU64::new(0)); + let expected = Arc::new(AtomicU64::new(0)); + let (tx, errors) = mpsc::channel(); + let (timestamp_tx, timestamps) = mpsc::sync_channel(1); + let (diagnostic_tx, diagnostics) = mpsc::sync_channel(8); + let error_tx = tx.clone(); + let callback_count = Arc::clone(&count); + let callback_expected = Arc::clone(&expected); + let mut previous: Option = None; + let stream = device.build_output_stream::( + cpal::StreamConfig { + channels: 2, + sample_rate: rate, + buffer_size: cpal::BufferSize::Default, + }, + move |data, info| { + // Never write audio: WASAPI supplies format-correct digital equilibrium. + let timestamp = info.timestamp(); + let sequence = callback_count.load(Ordering::Relaxed) + 1; + if sequence <= 8 { + let _ = diagnostic_tx.try_send((sequence, data.len() / 2, timestamp)); + } + let zero = cpal::StreamInstant::new(0, 0); + let mut failure = None; + if data.len() % 2 != 0 + || data.len() as u64 / 2 != callback_expected.load(Ordering::Relaxed) + { + failure = Some("callback frame count differs from stream.buffer_size()"); + } else if data.iter().any(|&sample| sample != 0) { + failure = Some("backend did not initialize I32 samples to digital silence"); + } else if timestamp.callback == zero { + // Only the first pre-start prime may have a zero clock snapshot. + if previous.is_some() || timestamp.device != zero { + failure = Some("zero clock outside the initial pre-start prime"); + } + } else { + let _ = timestamp_tx.try_send(timestamp); + if timestamp.device < timestamp.callback + || timestamp.device.duration_since(timestamp.callback) + > Duration::from_secs(1) + { + failure = Some("implausible device-versus-callback timestamp"); + } + } + if let Some(last) = previous { + if timestamp.callback < last.callback || timestamp.device < last.device { + failure = Some("callback/device timestamp regressed"); + } + } + if let Some(message) = failure { + let _ = tx.send(format!("{message}; callback_count={sequence} frames={} callback_ns={} device_ns={} previous={previous:?}", data.len() / 2, timestamp.callback.as_nanos(), timestamp.device.as_nanos())); + } + previous = Some(timestamp); + callback_count.fetch_add(1, Ordering::Relaxed); + }, + move |error| { + let _ = error_tx.send(format!("backend callback error: {error}")); + }, + None, + )?; + let frames = stream.buffer_size()?; + if frames == 0 { + return Err("zero actual buffer frames".into()); + } + expected.store(frames as u64, Ordering::Relaxed); + println!(" requested stereo I32 {rate} Hz; reported callback frames={frames}"); + Ok(Self { + stream, + count, + errors, + timestamps, + diagnostics, + }) + } + + fn print_diagnostics(&self) { + for (sequence, frames, timestamp) in self.diagnostics.try_iter() { + let now = self.stream.now(); + eprintln!( + "TIMESTAMP sample#{sequence} frames={frames} callback_ns={} device_ns={} sampled_now_ns={} callback_to_now_ns={:?} device_minus_callback_ns={:?}", + timestamp.callback.as_nanos(), + timestamp.device.as_nanos(), + now.as_nanos(), + now.checked_duration_since(timestamp.callback) + .map(|d| d.as_nanos()), + timestamp + .device + .checked_duration_since(timestamp.callback) + .map(|d| d.as_nanos()) + ); + } + } + + fn check(&self) -> Result<()> { + self.print_diagnostics(); + if let Ok(error) = self.errors.try_recv() { + return Err(format!("{error}; sampled_now_ns={}", self.stream.now().as_nanos()).into()); + } + Ok(()) + } + + fn observe(&self) -> Result<()> { + self.check()?; + let before = self.count.load(Ordering::Relaxed); + while self.timestamps.try_recv().is_ok() {} + self.stream.start()?; + let started = Instant::now(); + while self.count.load(Ordering::Relaxed).saturating_sub(before) < 10 { + self.check()?; + if started.elapsed() >= Duration::from_secs(10) { + return Err("fewer than ten callbacks within 10s".into()); + } + match self.errors.recv_timeout(Duration::from_millis(10)) { + Ok(error) => { + self.print_diagnostics(); + return Err(format!( + "{error}; sampled_now_ns={}", + self.stream.now().as_nanos() + ) + .into()); + } + Err(mpsc::RecvTimeoutError::Disconnected) => { + return Err("worker error channel disconnected".into()); + } + Err(mpsc::RecvTimeoutError::Timeout) => {} + } + } + self.check()?; + let timestamp = self + .timestamps + .try_recv() + .map_err(|_| "No live timestamp observed")?; + let now = self.stream.now(); + for instant in [timestamp.callback, timestamp.device] { + if instant > now + Duration::from_secs(1) + || now.duration_since(instant) > started.elapsed() + Duration::from_secs(1) + { + return Err("Callback/device timestamp not sane relative to stream.now()".into()); + } + } + println!( + " {} callbacks in {:.3}s (zero pre-start timestamps permitted)", + self.count.load(Ordering::Relaxed) - before, + started.elapsed().as_secs_f64() + ); + Ok(()) + } + + fn release(self) -> Result<()> { + let Self { stream, errors, .. } = self; + drop(stream); // joins worker; release exclusive ownership without an explicit Reset + if let Ok(error) = errors.try_recv() { + return Err(error.into()); + } + Ok(()) + } +} + +pub fn run(exact_name: &str) -> Result<()> { + let host = cpal::host_from_id(cpal::HostId::Wasapi)?; + let mut matches = Vec::new(); + for device in host.output_devices()? { + let name = device.description()?.name().to_owned(); + println!("Available output: {name:?}"); + if name == exact_name { + matches.push(device); + } + } + if matches.len() != 1 { + return Err(format!( + "Expected exactly one output named {exact_name:?}; found {}. No fallback selected.", + matches.len() + ) + .into()); + } + let device = matches.remove(0); + if !device.supports_exclusive() { + return Err("Selected endpoint does not support exclusive output".into()); + } + println!("Selected endpoint ID: {}", device.id()?); + let device = device.exclusive(true); + let configs: Vec<_> = device.supported_output_configs()?.collect(); + println!( + "MATRIX endpoint={exact_name:?}; requested configs only, not measured DAC physical rate or bit-perfect proof" + ); + let (mut supported, mut skipped, mut passed, mut failed) = (0, 0, 0, 0); + for rate in [44100, 48000, 96000, 192000] { + if !configs.iter().any(|c| { + c.channels() == 2 + && c.sample_format() == cpal::SampleFormat::I32 + && c.contains_rate(rate) + }) { + skipped += 1; + println!( + "SKIP {rate}: stereo I32 not advertised by exclusive configured-device enumeration" + ); + continue; + } + supported += 1; + let result = (|| -> Result<()> { + println!("RATE {rate}: first open"); + let first = Probe::build(&device, rate)?; + first.observe()?; + first.release()?; + println!("RATE {rate}: reopen and contention"); + let held = Probe::build(&device, rate)?; + held.observe()?; + match Probe::build(&device, rate) { + Ok(second) => { + second.release()?; + return Err("second exclusive build unexpectedly succeeded".into()); + } + Err(error) => { + let Some(error) = error.downcast_ref::() else { + return Err(error); + }; + if error.kind() != cpal::ErrorKind::DeviceBusy { + return Err( + format!("contention failed for unrelated reason: {error:?}").into() + ); + } + println!(" contention rejected: DeviceBusy"); + } + } + held.observe()?; + held.release()?; + println!("RATE {rate}: reopen after contention/release"); + let final_open = Probe::build(&device, rate)?; + final_open.observe()?; + final_open.release()?; + Ok(()) + })(); + match result { + Ok(()) => { + passed += 1; + println!("PASS {rate}"); + } + Err(error) => { + failed += 1; + eprintln!("FAIL {rate}: {error}"); + } + } + } + println!( + "MATRIX SUMMARY supported/attempted={supported} passed={passed} skipped={skipped} failed={failed}" + ); + if failed != 0 || supported == 0 { + return Err("Matrix failed or no supported configurations were exercised".into()); + } + println!( + "PASS for advertised configurations only; skipped rates are not passes; no mandatory baseline rate" + ); + Ok(()) +} diff --git a/src/host/mod.rs b/src/host/mod.rs index 34a92488d..d89a434d3 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -76,6 +76,9 @@ pub(crate) mod pulseaudio; #[cfg(windows)] pub(crate) mod wasapi; +#[cfg(any(windows, test))] +pub(crate) mod wasapi_policy; + #[cfg(all( target_arch = "wasm32", any(target_os = "emscripten", target_os = "unknown"), diff --git a/src/host/wasapi/device.rs b/src/host/wasapi/device.rs index 0d73be057..b451e1308 100644 --- a/src/host/wasapi/device.rs +++ b/src/host/wasapi/device.rs @@ -24,7 +24,7 @@ use crate::{ use windows::{ Win32::{ Devices::Properties, - Foundation::{ERROR_TIMEOUT, PROPERTYKEY}, + Foundation::{self, ERROR_TIMEOUT, PROPERTYKEY}, Media::{Audio, Audio::IAudioRenderClient, KernelStreaming, Multimedia}, System::{ Com, @@ -75,6 +75,7 @@ enum DeviceHandle { #[derive(Clone)] pub struct Device { device: DeviceHandle, + exclusive: bool, /// We cache an uninitialized `IAudioClient` so that we can call functions from it without /// having to create/destroy audio clients all the time. future_audio_client: Arc>>, // TODO: add NonZero around the ptr @@ -85,6 +86,10 @@ impl DeviceTrait for Device { type SupportedOutputConfigs = SupportedOutputConfigs; type Stream = Stream; + fn supports_exclusive(&self) -> bool { + self.data_flow() == Audio::eRender + } + fn description(&self) -> Result { Device::description(self) } @@ -94,7 +99,7 @@ impl DeviceTrait for Device { } fn supports_input(&self) -> bool { - self.data_flow() == Audio::eCapture + !self.exclusive && self.data_flow() == Audio::eCapture } fn supports_output(&self) -> bool { @@ -168,6 +173,19 @@ struct Endpoint { // Use RAII to make sure CoTaskMemFree is called when we are responsible for freeing. struct WaveFormatExPtr(*mut Audio::WAVEFORMATEX); +#[cfg(test)] +#[path = "device_tests.rs"] +mod tests; + +struct OwnedEvent(Foundation::HANDLE); +impl Drop for OwnedEvent { + fn drop(&mut self) { + unsafe { + let _ = Foundation::CloseHandle(self.0); + } + } +} + impl Drop for WaveFormatExPtr { fn drop(&mut self) { unsafe { @@ -518,9 +536,62 @@ impl Device { } } + /// Selects exclusive output access before querying configurations. + #[must_use] + pub fn exclusive(mut self, exclusive: bool) -> Self { + self.exclusive = exclusive; + self + } + + pub(crate) fn is_exclusive(&self) -> bool { + self.exclusive + } + + fn exclusive_formats(&self) -> Result { + if self.data_flow() != Audio::eRender { + return Err(Error::new(ErrorKind::UnsupportedOperation)); + } + com::com_initialized(); + let client = self.build_audioclient(None)?; + let mut result = Vec::new(); + for channels in [1, 2] { + for &sample_rate in COMMON_SAMPLE_RATES { + for sample_format in WAVEFORMATEXTENSIBLE_SAMPLE_FORMATS { + let config = StreamConfig { + channels, + sample_rate, + buffer_size: BufferSize::Default, + }; + if exclusive_waveformat(&client, config, sample_format)?.is_some() { + result.push(SupportedStreamConfigRange { + channels, + min_sample_rate: sample_rate, + max_sample_rate: sample_rate, + buffer_size: SupportedBufferSize::Unknown, + sample_format, + }); + } + } + } + } + Ok(result.into_iter()) + } + + fn check_mode(&self) -> Result<(), Error> { + if self.exclusive { + Err(Error::with_message( + ErrorKind::UnsupportedOperation, + "Exclusive WASAPI input and loopback are not supported", + )) + } else { + Ok(()) + } + } + fn from_immdevice(device: Audio::IMMDevice) -> Self { Device { device: DeviceHandle::Specific(device), + exclusive: false, future_audio_client: Arc::new(Mutex::new(None)), } } @@ -528,6 +599,7 @@ impl Device { fn default_output() -> Self { Device { device: DeviceHandle::DefaultOutput, + exclusive: false, future_audio_client: Arc::new(Mutex::new(None)), } } @@ -535,6 +607,7 @@ impl Device { fn default_input() -> Self { Device { device: DeviceHandle::DefaultInput, + exclusive: false, future_audio_client: Arc::new(Mutex::new(None)), } } @@ -624,6 +697,9 @@ impl Device { // parameter error. Thus, we just assume that the default number of channels is the only // number supported. fn supported_formats(&self) -> Result { + if self.exclusive { + return self.exclusive_formats(); + } // initializing COM because we call `CoTaskMemFree` to release the format. com::com_initialized(); @@ -742,6 +818,7 @@ impl Device { } pub fn supported_input_configs(&self) -> Result { + self.check_mode()?; if self.data_flow() == Audio::eCapture { self.supported_formats() // If it's an output device, assume no input formats. @@ -764,6 +841,13 @@ impl Device { // // One format is guaranteed to be supported, the one returned by `GetMixFormat`. fn default_format(&self) -> Result { + if self.exclusive { + return self + .exclusive_formats()? + .max_by(SupportedStreamConfigRange::cmp_default_heuristics) + .map(|range| range.with_max_sample_rate()) + .ok_or_else(|| Error::new(ErrorKind::UnsupportedConfig)); + } // initializing COM because we call `CoTaskMemFree` com::com_initialized(); @@ -817,6 +901,7 @@ impl Device { } pub fn default_input_config(&self) -> Result { + self.check_mode()?; if self.data_flow() == Audio::eCapture { self.default_format() } else { @@ -845,6 +930,7 @@ impl Device { sample_format: SampleFormat, activation_timeout: Option, ) -> Result { + self.check_mode()?; crate::validate_stream_config(&config)?; unsafe { // Making sure that COM is initialized. @@ -944,6 +1030,11 @@ impl Device { stream_latency, skip_callback: Arc::new(AtomicBool::new(false)), fill_usec: Arc::new(AtomicU64::new(0)), + exclusive: false, + qpc_frequency: 0, + needs_priming: false, + frames_written: 0, + real_frames_end: 0, }) } } @@ -960,48 +1051,102 @@ impl Device { // It's not actually sure that this is required, but when in doubt do it. com::com_initialized(); - // Obtaining a `IAudioClient`. - let audio_client = self - .build_audioclient(activation_timeout) - .context("Failed to build audio client")?; - - // No further range validation: IAudioClient::Initialize accepts any positive duration - // in shared mode. The callback period is always GetDevicePeriod() regardless of what - // is requested here; the value only affects ring-buffer latency. + // Shared duration controls latency; exclusive duration also controls its period. let buffer_duration = buffer_size_to_duration(&config.buffer_size, config.sample_rate); - // Computing the format and initializing the device. - let waveformatex = { - let format_attempt = config_to_waveformatextensible(config, sample_format, None) - .ok_or_else(|| { - Error::with_message( - ErrorKind::UnsupportedConfig, - "Stream configuration could not be converted to a compatible format", - ) - })?; - let share_mode = Audio::AUDCLNT_SHAREMODE_SHARED; - - // Finally, initializing the audio client + let endpoint = if self.exclusive { + if self.data_flow() != Audio::eRender { + return Err(Error::new(ErrorKind::UnsupportedOperation)); + } + Some( + self.immdevice() + .ok_or_else(|| Error::new(ErrorKind::DeviceNotAvailable))?, + ) + } else { + None + }; + let mut audio_client = if let Some(endpoint) = &endpoint { + endpoint.Activate::(Com::CLSCTX_ALL, None)? + } else { + self.build_audioclient(activation_timeout) + .context("Failed to build audio client")? + }; + let waveformatex = if let Some(endpoint) = &endpoint { + let format = exclusive_waveformat(&audio_client, config, sample_format)? + .ok_or_else(|| Error::new(ErrorKind::UnsupportedConfig))?; + let (mut default_period, mut minimum_period) = (0, 0); + audio_client + .GetDevicePeriod(Some(&mut default_period), Some(&mut minimum_period))?; + if default_period <= 0 || minimum_period <= 0 { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + "Endpoint returned an invalid exclusive period", + )); + } + let mut period = if buffer_duration == 0 { + default_period + } else { + buffer_duration + }; + period = period.max(minimum_period); + // Retry once with the device-reported buffer alignment if initialization requires it. + for attempt in 0..2 { + match audio_client.Initialize( + Audio::AUDCLNT_SHAREMODE_EXCLUSIVE, + Audio::AUDCLNT_STREAMFLAGS_EVENTCALLBACK, + period, + period, + &format.Format, + None, + ) { + Ok(()) => break, + Err(error) + if crate::host::wasapi_policy::retry_alignment( + attempt, + error.code() == Audio::AUDCLNT_E_BUFFER_SIZE_NOT_ALIGNED, + ) => + { + let frames = audio_client.GetBufferSize()?; + if frames == 0 { + return Err(Error::new(ErrorKind::UnsupportedConfig)); + } + period = crate::host::wasapi_policy::aligned_period( + frames, + config.sample_rate, + ) + .ok_or_else(|| Error::new(ErrorKind::UnsupportedConfig))?; + drop(audio_client); + audio_client = endpoint.Activate(Com::CLSCTX_ALL, None)?; + } + Err(error) => return Err(error.into()), + } + } + format.Format + } else { + let format = config_to_waveformatextensible(config, sample_format, None) + .ok_or_else(|| Error::new(ErrorKind::UnsupportedConfig))?; audio_client .Initialize( - share_mode, + Audio::AUDCLNT_SHAREMODE_SHARED, DEFAULT_FLAGS | Audio::AUDCLNT_STREAMFLAGS_SRC_DEFAULT_QUALITY | Audio::AUDCLNT_STREAMFLAGS_AUTOCONVERTPCM, buffer_duration, 0, - &format_attempt.Format, + &format.Format, None, ) .context("Failed to initialize audio client")?; - - format_attempt.Format + format.Format }; - // Creating the event that will be signalled whenever we need to submit some samples. - let event = - Threading::CreateEventA(None, false, false, windows::core::PCSTR(ptr::null())) - .context("Failed to create event")?; + let event_owner = OwnedEvent(Threading::CreateEventA( + None, + false, + false, + windows::core::PCSTR(ptr::null()), + )?); + let event = event_owner.0; audio_client .SetEventHandle(event) @@ -1015,13 +1160,17 @@ impl Device { let period_frames = shared_mode_period_frames(&audio_client, config.sample_rate, max_frames_in_buffer); + let period_frames = if self.exclusive { + max_frames_in_buffer + } else { + period_frames + }; + // Building a `IAudioRenderClient` that will be used to fill the samples buffer. let render_client = audio_client .GetService::() .context("Failed to get render client")?; - // Once we built the `StreamInner`, we add a command that will be picked up by the - // `run()` method and added to the `RunContext`. let client_flow = AudioClientFlow::Render { render_client }; let audio_clock = audio_client @@ -1035,6 +1184,7 @@ impl Device { Duration::from_nanos(hns.max(0) as u64 * 100) }; + mem::forget(event_owner); // ownership passes to StreamInner below Ok(StreamInner { audio_client, audio_clock, @@ -1049,6 +1199,11 @@ impl Device { stream_latency, skip_callback: Arc::new(AtomicBool::new(false)), fill_usec: Arc::new(AtomicU64::new(0)), + exclusive: self.exclusive, + qpc_frequency: 0, + needs_priming: true, + frames_written: 0, + real_frames_end: 0, }) } } @@ -1359,6 +1514,55 @@ const WAVEFORMATEXTENSIBLE_SAMPLE_FORMATS: [SampleFormat; 5] = [ SampleFormat::F32, ]; +fn exclusive_waveformat( + client: &Audio::IAudioClient, + config: StreamConfig, + sample_format: SampleFormat, +) -> Result, Error> { + if !WAVEFORMATEXTENSIBLE_SAMPLE_FORMATS.contains(&sample_format) { + return Ok(None); + } + let mask = match config.channels { + 1 => 4, // front center + 2 => 3, // front left/right + _ => return Ok(None), + }; + let Some(mut format) = config_to_waveformatextensible(config, sample_format, Some(mask)) else { + return Ok(None); + }; + for extensible in [false, true] { + if !extensible { + match sample_format { + SampleFormat::F32 => { + format.Format.wFormatTag = Multimedia::WAVE_FORMAT_IEEE_FLOAT as u16; + format.Format.cbSize = 0; + } + SampleFormat::I32 => { + format.Format.wFormatTag = Audio::WAVE_FORMAT_PCM as u16; + format.Format.cbSize = 0; + } + SampleFormat::I24 => continue, + _ => {} + } + } + if extensible { + format.Format.wFormatTag = KernelStreaming::WAVE_FORMAT_EXTENSIBLE as u16; + format.Format.cbSize = (mem::size_of::() + - mem::size_of::()) as u16; + } + let hr = unsafe { + client.IsFormatSupported(Audio::AUDCLNT_SHAREMODE_EXCLUSIVE, &format.Format, None) + }; + if hr.0 == 0 { + return Ok(Some(format)); + } + if hr.is_err() && hr != Audio::AUDCLNT_E_UNSUPPORTED_FORMAT { + return Err(windows::core::Error::from_hresult(hr).into()); + } + } + Ok(None) +} + // Turns a `Format` into a `WAVEFORMATEXTENSIBLE`. // // Returns `None` if the WAVEFORMATEXTENSIBLE does not support the given format. diff --git a/src/host/wasapi/device_tests.rs b/src/host/wasapi/device_tests.rs new file mode 100644 index 000000000..97ffb8f15 --- /dev/null +++ b/src/host/wasapi/device_tests.rs @@ -0,0 +1,168 @@ +use super::*; + +fn assert_exclusive_rejected(result: Result) { + let error = match result { + Err(error) => error, + Ok(_) => panic!("exclusive WASAPI operation unexpectedly succeeded"), + }; + assert_eq!(error.kind(), crate::ErrorKind::UnsupportedOperation); + assert_eq!( + error.message(), + Some("Exclusive WASAPI input and loopback are not supported") + ); +} + +#[test] +fn exclusive_input_rejects_before_validation_or_activation() { + use crate::traits::DeviceTrait; + + let device = Device::default_input().exclusive(true); + assert!(!device.supports_input()); + + // Zero channels/rate is deliberately invalid: check_mode must win before validation or any + // endpoint activation. The default-input constructor itself only stores a virtual handle. + let invalid = StreamConfig { + channels: 0, + sample_rate: 0, + buffer_size: BufferSize::Default, + }; + assert_exclusive_rejected(device.default_input_config()); + assert_exclusive_rejected(device.supported_input_configs()); + assert_exclusive_rejected(device.build_input_stream_raw( + invalid, + SampleFormat::F32, + |_, _| {}, + |_| {}, + None, + )); + assert_exclusive_rejected(device.build_input_stream::( + invalid, + |_, _| {}, + |_| {}, + None, + )); +} + +#[test] +fn exclusive_render_rejects_input_loopback_before_activation() { + use crate::traits::DeviceTrait; + + let device = Device::default_output().exclusive(true); + assert!(!device.supports_input()); + let invalid = StreamConfig { + channels: 0, + sample_rate: 0, + buffer_size: BufferSize::Default, + }; + assert_exclusive_rejected(device.build_input_stream_raw( + invalid, + SampleFormat::F32, + |_, _| {}, + |_| {}, + None, + )); +} + +#[test] +fn facade_exclusive_input_rejects_before_validation_or_activation() { + use crate::platform::Device as PlatformDevice; + use crate::traits::DeviceTrait; + + let device: PlatformDevice = Device::default_input().into(); + let exclusive = device.exclusive(true); + assert!(!exclusive.supports_input()); + let invalid = StreamConfig { + channels: 0, + sample_rate: 0, + buffer_size: BufferSize::Default, + }; + assert_exclusive_rejected(exclusive.default_input_config()); + assert_exclusive_rejected(exclusive.supported_input_configs()); + assert_exclusive_rejected(exclusive.build_input_stream_raw( + invalid, + SampleFormat::F32, + |_, _| {}, + |_| {}, + None, + )); + assert_exclusive_rejected(exclusive.build_input_stream::( + invalid, + |_, _| {}, + |_| {}, + None, + )); +} + +#[test] +fn facade_exclusive_render_rejects_loopback_input() { + use crate::platform::Device as PlatformDevice; + use crate::traits::DeviceTrait; + + let device: PlatformDevice = Device::default_output().into(); + let exclusive = device.exclusive(true); + assert!(!exclusive.supports_input()); + let invalid = StreamConfig { + channels: 0, + sample_rate: 0, + buffer_size: BufferSize::Default, + }; + assert_exclusive_rejected(exclusive.build_input_stream_raw( + invalid, + SampleFormat::F32, + |_, _| {}, + |_| {}, + None, + )); +} + +#[test] +fn wrapping_native_mode_preserves_exclusive_request() { + use crate::platform::{Device as PlatformDevice, DeviceInner}; + let native = Device::default_output().exclusive(true); + let wrapped = PlatformDevice::from(native.clone()); + assert!(!wrapped.supports_input()); + let unwrap = |device: PlatformDevice| match device.into_inner() { + DeviceInner::Wasapi(device) => device, + #[allow(unreachable_patterns)] + _ => panic!("wrong backend"), + }; + assert!(unwrap(wrapped.clone()).is_exclusive()); + assert!(unwrap(PlatformDevice::from(wrapped.into_inner())).is_exclusive()); + assert!(unwrap(PlatformDevice::from(DeviceInner::Wasapi(native))).is_exclusive()); + let mut replaced = PlatformDevice::from(Device::default_output()).exclusive(true); + *replaced.as_inner_mut() = DeviceInner::Wasapi(Device::default_output()); + assert!(unwrap(replaced.clone()).is_exclusive()); + assert!(!unwrap(replaced.exclusive(false)).is_exclusive()); + let mut shared = PlatformDevice::from(Device::default_output()); + *shared.as_inner_mut() = DeviceInner::Wasapi(Device::default_output().exclusive(true)); + assert!(unwrap(shared.clone()).is_exclusive()); + assert!(!unwrap(shared.exclusive(false)).is_exclusive()); +} + +#[test] +fn frame_duration_round_trips() { + for rate in [44_100, 48_000, 96_000] { + for frames in [128, 256, 512, 1024] { + let duration = buffer_size_to_duration(&BufferSize::Fixed(frames), rate); + assert!(duration > 0); + assert_eq!(buffer_duration_to_frames(duration, rate), frames); + } + } +} + +#[test] +fn i24_uses_a_four_byte_container() { + let config = StreamConfig { + channels: 2, + sample_rate: 48_000, + buffer_size: BufferSize::Default, + }; + let wave = config_to_waveformatextensible(config, SampleFormat::I24, Some(3)).unwrap(); + let bits = wave.Format.wBitsPerSample; + let align = wave.Format.nBlockAlign; + let valid = unsafe { wave.Samples.wValidBitsPerSample }; + assert_eq!(bits, 32); + assert_eq!(valid, 24); + assert_eq!(align, 8); + assert!(!WAVEFORMATEXTENSIBLE_SAMPLE_FORMATS.contains(&SampleFormat::U16)); +} diff --git a/src/host/wasapi/mod.rs b/src/host/wasapi/mod.rs index 999d1ab86..1b28ee470 100644 --- a/src/host/wasapi/mod.rs +++ b/src/host/wasapi/mod.rs @@ -1,6 +1,14 @@ //! WASAPI backend implementation. //! //! Default backend on Windows. +//! +//! Select `device.exclusive(true)` before querying output configurations to bypass +//! shared-engine conversion. Exclusive discovery is limited to formats and sample rates +//! accepted by the endpoint; packed 24-bit samples and exclusive input/loopback are +//! unsupported. Requested buffer sizes may be increased to satisfy the endpoint's +//! period/alignment requirements; query the stream's actual buffer size after construction. +//! Exclusive access can fail because another application owns the endpoint or system +//! policy disables it. Exclusive mode does not by itself guarantee bit-perfect playback. use std::io::Error as IoError; diff --git a/src/host/wasapi/stream.rs b/src/host/wasapi/stream.rs index a7723e7f3..24f8db90d 100644 --- a/src/host/wasapi/stream.rs +++ b/src/host/wasapi/stream.rs @@ -296,6 +296,11 @@ pub enum PlaybackState { } pub struct StreamInner { + pub exclusive: bool, + pub qpc_frequency: u64, + pub needs_priming: bool, + pub frames_written: u64, + pub real_frames_end: u64, pub audio_client: Audio::IAudioClient, pub audio_clock: Audio::IAudioClock, pub client_flow: AudioClientFlow, @@ -446,6 +451,8 @@ impl Stream { let mut latch = Latch::new(); let waiter = latch.waiter(); + let mut run_context = run_context; + run_context.stream.qpc_frequency = qpc_frequency as u64; let thread = thread::Builder::new() .name("cpal_wasapi_out".to_owned()) .spawn(move || { @@ -552,20 +559,7 @@ impl StreamTrait for Stream { } fn now(&self) -> StreamInstant { - let mut counter: i64 = 0; - unsafe { - Performance::QueryPerformanceCounter(&mut counter) - .expect("QueryPerformanceCounter failed"); - } - // Convert to 100-nanosecond units first, matching the precision of WASAPI QPCPosition - // values delivered to callbacks. This keeps `now()` on the same 100 ns grid as - // callback/capture/playback instants, avoiding false sub-100 ns deltas. - let units_100ns = counter as u128 * 10_000_000 / self.qpc_frequency as u128; - let nanos = units_100ns * 100; - StreamInstant::new( - (nanos / 1_000_000_000) as u64, - (nanos % 1_000_000_000) as u32, - ) + current_qpc(self.qpc_frequency).expect("QueryPerformanceCounter failed") } fn buffer_size(&self) -> Result { @@ -573,6 +567,31 @@ impl StreamTrait for Stream { } } +fn current_qpc(frequency: u64) -> Result { + if frequency == 0 { + return Err(Error::with_message( + ErrorKind::BackendError, + "QPC frequency is zero", + )); + } + let mut counter = 0i64; + unsafe { + Performance::QueryPerformanceCounter(&mut counter)?; + } + if counter < 0 { + return Err(Error::with_message( + ErrorKind::BackendError, + "QPC counter is negative", + )); + } + // Match WASAPI's 100ns grid, including Stream::now(). + let nanos = (counter as u128 * 10_000_000 / frequency as u128) * 100; + Ok(StreamInstant::new( + (nanos / 1_000_000_000) as u64, + (nanos % 1_000_000_000) as u32, + )) +} + impl Drop for StreamInner { fn drop(&mut self) { unsafe { @@ -593,7 +612,11 @@ fn process_commands(run_context: &mut RunContext) -> Result { // and may already hold real, unplayed data. let cold_start = match run_context.stream.client_flow { AudioClientFlow::Render { .. } => { - get_available_frames(&run_context.stream)? > 0 + if run_context.stream.exclusive { + run_context.stream.needs_priming + } else { + get_available_frames(&run_context.stream)? > 0 + } } AudioClientFlow::Capture { .. } => false, }; @@ -629,6 +652,10 @@ fn process_commands(run_context: &mut RunContext) -> Result { .audio_client .Reset() .context("Failed to reset audio client")?; + run_context.stream.frames_written = 0; + run_context.stream.real_frames_end = 0; + run_context.stream.needs_priming = true; + run_context.stream.fill_usec.store(0, Ordering::Relaxed); } }, Command::Terminate => { @@ -676,7 +703,11 @@ fn get_available_frames(stream: &StreamInner) -> Result { .audio_client .GetCurrentPadding() .context("Failed to get current padding")?; - Ok(stream.max_frames_in_buffer - padding) + Ok(crate::host::wasapi_policy::packet_frames( + false, + stream.max_frames_in_buffer, + padding, + )) } } @@ -757,7 +788,6 @@ fn run_output( return; } }; - let mut frames_written: u64 = 0; loop { match process_commands_and_await_signal(&mut run_ctxt, error_callback) { @@ -769,18 +799,20 @@ fn run_output( AudioClientFlow::Render { ref render_client } => render_client.clone(), _ => unreachable!(), }; - if let Err(err) = process_output( - &run_ctxt.stream, + let outcome = match process_output( + &mut run_ctxt.stream, render_client, data_callback, clock_frequency, - &mut frames_written, ) { - emit_error(error_callback, err); - break; - } + Ok(outcome) => outcome, + Err(err) => { + emit_error(error_callback, err); + break; + } + }; if run_ctxt.stream.playback_state == PlaybackState::Priming { - if run_ctxt.stream.skip_callback.load(Ordering::Relaxed) { + if !crate::host::wasapi_policy::start_after_prime(outcome) { // process_output submitted nothing this cycle; stay stopped. run_ctxt.stream.playback_state = PlaybackState::Stopped; } else { @@ -869,7 +901,9 @@ fn process_commands_and_await_signal( ); return ControlFlow::Continue(false); } - ControlFlow::Continue(handle_idx != 0) + ControlFlow::Continue( + handle_idx != 0 && run_context.stream.playback_state == PlaybackState::Playing, + ) } // The loop for processing pending input data. @@ -951,36 +985,81 @@ fn process_input( // The loop for writing output data. fn process_output( - stream: &StreamInner, + stream: &mut StreamInner, render_client: Audio::IAudioRenderClient, data_callback: &mut dyn FnMut(&mut Data, &CallbackInfo), clock_frequency: u64, - frames_written: &mut u64, -) -> Result<(), Error> { +) -> Result { + use crate::host::wasapi_policy::RenderOutcome; // The number of frames available for writing. - let frames_available = match get_available_frames(stream)? { - 0 => return Ok(()), // TODO: Can this happen? - n => n, + let frames_available = if stream.exclusive { + crate::host::wasapi_policy::packet_frames(true, stream.max_frames_in_buffer, 0) + } else { + match get_available_frames(stream)? { + 0 => return Ok(RenderOutcome::Skipped), + n => n, + } }; - let padding = stream.max_frames_in_buffer - frames_available; + let played_frames = if stream.exclusive { + let (_, position) = clock_position(stream)?; + crate::host::wasapi_policy::played_frames( + position, + stream.config.sample_rate, + clock_frequency, + ) + } else { + 0 + }; + let padding = if stream.exclusive { + stream + .real_frames_end + .saturating_sub(played_frames) + .min(u32::MAX as u64) as u32 + } else { + stream.max_frames_in_buffer - frames_available + }; let fill_usec = (padding as u64) .saturating_mul(1_000_000) .saturating_div(stream.config.sample_rate as u64) - .saturating_add( + .saturating_add(if stream.exclusive { + // IAudioClock already tracks device playback. Do not add the endpoint's + // maximum stream latency to the measured outstanding frame duration. + 0 + } else { stream .stream_latency .as_micros() .try_into() - .unwrap_or(u64::MAX), - ); + .unwrap_or(u64::MAX) + }); stream.fill_usec.store(fill_usec, Ordering::Relaxed); if stream.skip_callback.load(Ordering::Relaxed) { - // Skip the period instead of queuing silence a future resume would replay. - return Ok(()); + if stream.exclusive && stream.playback_state == PlaybackState::Playing { + unsafe { + let _ = render_client.GetBuffer(frames_available)?; + render_client + .ReleaseBuffer(frames_available, Audio::AUDCLNT_BUFFERFLAGS_SILENT.0 as u32)?; + } + stream.frames_written = crate::host::wasapi_policy::submission_end( + stream.frames_written, + played_frames, + frames_available, + ); + } + return Ok(RenderOutcome::Skipped); + } + if stream.exclusive { + stream.frames_written = stream.frames_written.max(played_frames); } + let timestamp = output_timestamp( + stream, + stream.config.sample_rate, + clock_frequency, + stream.frames_written, + )?; unsafe { let buffer = render_client.GetBuffer(frames_available)?; @@ -993,8 +1072,7 @@ fn process_output( let data = buffer as *mut (); let len = byte_count / stream.sample_format.sample_size(); let mut data = Data::from_parts(data, len, stream.sample_format); - let sample_rate = stream.config.sample_rate; - let timestamp = output_timestamp(stream, sample_rate, clock_frequency, *frames_written)?; + // WASAPI exposes no render-side xrun signal. data_callback( &mut data, @@ -1018,10 +1096,28 @@ fn process_output( render_client.ReleaseBuffer(frames_available, 0)?; - *frames_written += frames_available as u64; + stream.frames_written = crate::host::wasapi_policy::submission_end( + stream.frames_written, + played_frames, + frames_available, + ); + stream.real_frames_end = stream.frames_written; + stream.needs_priming = false; + let fill = if stream.exclusive { + stream + .real_frames_end + .saturating_sub(played_frames) + .saturating_mul(1_000_000) + / stream.config.sample_rate as u64 + } else { + fill_usec.saturating_add( + frames_available as u64 * 1_000_000 / stream.config.sample_rate as u64, + ) + }; + stream.fill_usec.store(fill, Ordering::Relaxed); } - Ok(()) + Ok(RenderOutcome::Submitted) } /// Reads the stream's `IAudioClock` in a single `GetPosition` call, returning the callback @@ -1093,6 +1189,15 @@ fn output_timestamp( u64::try_from(written_nanos.saturating_sub(consumed_nanos)).unwrap_or(u64::MAX); let buffered = Duration::from_nanos(buffered_nanos); + if stream.exclusive { + let now = current_qpc(stream.qpc_frequency)?; + return Ok(crate::host::wasapi_policy::exclusive_timestamp( + callback, now, buffered, + )); + } + + // Preserve the shared-mode snapshot and latency behavior. Exclusive playback + // is already clock-position based and must not add maximum stream latency again. let playback = callback + (buffered + stream.stream_latency); Ok(StreamTimestamp { callback, diff --git a/src/host/wasapi_policy.rs b/src/host/wasapi_policy.rs new file mode 100644 index 000000000..ca9372eb3 --- /dev/null +++ b/src/host/wasapi_policy.rs @@ -0,0 +1,155 @@ +//! Pure arithmetic shared by WASAPI rendering and host-independent tests. + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum RenderOutcome { + Submitted, + Skipped, +} + +pub(crate) fn exclusive_timestamp( + snapshot: crate::StreamInstant, + now: crate::StreamInstant, + queued: std::time::Duration, +) -> crate::StreamTimestamp { + // Some exclusive endpoints provide a zero QPC snapshot until playback advances. + // Position/frequency still determine queued duration; only its absolute anchor + // is unavailable. Estimate that startup anchor using the invocation's QPC. + let anchor = if snapshot == crate::StreamInstant::new(0, 0) { + now + } else { + snapshot + }; + crate::StreamTimestamp { + callback: now, + // A stale snapshot cannot predict new output delivery before invocation. + device: (anchor + queued).max(now), + } +} + +pub(crate) fn start_after_prime(outcome: RenderOutcome) -> bool { + outcome == RenderOutcome::Submitted +} + +pub(crate) fn retry_alignment(attempt: usize, unaligned: bool) -> bool { + attempt == 0 && unaligned +} + +pub(crate) fn packet_frames(exclusive: bool, capacity: u32, padding: u32) -> u32 { + if exclusive { + capacity + } else { + capacity.saturating_sub(padding) + } +} + +pub(crate) fn played_frames(position: u64, rate: u32, frequency: u64) -> u64 { + if frequency == 0 { + return 0; + } + (position as u128 * rate as u128 / frequency as u128).min(u64::MAX as u128) as u64 +} + +pub(crate) fn aligned_period(frames: u32, rate: u32) -> Option { + if frames == 0 || rate == 0 { + return None; + } + Some(((frames as u64 * 10_000_000 + rate as u64 / 2) / rate as u64).max(1) as i64) +} + +pub(crate) fn submission_end(written: u64, played: u64, frames: u32) -> u64 { + written.max(played).saturating_add(frames as u64) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn zero_qpc_startup_snapshots_use_the_current_clock_epoch() { + use crate::StreamInstant; + use std::time::Duration; + let zero = StreamInstant::new(0, 0); + let now = StreamInstant::new(1830, 0); + // The reported priming and first post-start snapshots both had QPC zero. + let priming = exclusive_timestamp(zero, now, Duration::ZERO); + assert_eq!(priming.callback, now); + assert_eq!(priming.device, now); + let next_now = now + Duration::from_millis(1); + let first_running = exclusive_timestamp(zero, next_now, Duration::from_millis(10)); + assert_eq!(first_running.callback, next_now); + assert_eq!(first_running.device, next_now + Duration::from_millis(10)); + assert!(first_running.device > priming.device); + } + + #[test] + fn valid_qpc_snapshot_anchors_playback_but_not_callback() { + use crate::StreamInstant; + use std::time::Duration; + let snapshot = StreamInstant::new(1830, 0); + let now = snapshot + Duration::from_millis(2); + let timestamp = exclusive_timestamp(snapshot, now, Duration::from_millis(10)); + assert_eq!(timestamp.callback, now); + assert_eq!(timestamp.device, snapshot + Duration::from_millis(10)); + // A stale prediction or underrun cannot put newly supplied output in the past. + let stale = exclusive_timestamp(snapshot, now, Duration::from_millis(1)); + assert_eq!(stale.callback, now); + assert_eq!(stale.device, now); + } + + #[test] + fn priming_uses_submission_not_a_later_skip_flag() { + for later_skip_flag in [false, true] { + let _ = later_skip_flag; + assert!(!start_after_prime(RenderOutcome::Skipped)); + assert!(start_after_prime(RenderOutcome::Submitted)); + } + } + + #[test] + fn alignment_is_rounded_and_retry_is_bounded() { + assert_eq!(aligned_period(0, 48000), None); + assert_eq!(aligned_period(256, 0), None); + for rate in [44100, 48000, 96000] { + for frames in [128, 256, 512, 1024] { + let period = aligned_period(frames, rate).unwrap() as u64; + assert_eq!( + (period * rate as u64 + 5_000_000) / 10_000_000, + frames as u64 + ); + } + } + assert!(retry_alignment(0, true)); + assert!(!retry_alignment(1, true)); + assert!(!retry_alignment(0, false)); + } + + #[test] + fn silence_remains_queued_on_resume_but_does_not_extend_real_drain() { + let real = submission_end(0, 0, 256); + let silent = submission_end(real, 128, 256); + assert_eq!(real, 256); + assert_eq!(silent, 512); + let resumed = submission_end(silent, 256, 256); + assert_eq!(resumed, 768); + assert_eq!(submission_end(0, 0, 256), 256); // reset restarts the epoch + assert_eq!(submission_end(resumed, 1024, 256), 1280); // underrun rebases + assert_eq!(submission_end(u64::MAX, 0, 256), u64::MAX); + } + + #[test] + fn exclusive_packets_are_always_full() { + for padding in [0, 1, 128, 256, u32::MAX] { + assert_eq!(packet_frames(true, 256, padding), 256); + } + assert_eq!(packet_frames(false, 256, 128), 128); + assert_eq!(packet_frames(false, 256, 257), 0); + } + + #[test] + fn clock_units_are_not_assumed_to_be_frames() { + assert_eq!(played_frames(10_000_000, 48_000, 10_000_000), 48_000); + assert_eq!(played_frames(44_100, 44_100, 44_100), 44_100); + assert_eq!(played_frames(0, 48_000, 10_000_000), 0); + assert_eq!(played_frames(u64::MAX, u32::MAX, 1), u64::MAX); + } +} diff --git a/src/platform/exclusive_tests.rs b/src/platform/exclusive_tests.rs new file mode 100644 index 000000000..d6b0b6bb1 --- /dev/null +++ b/src/platform/exclusive_tests.rs @@ -0,0 +1,126 @@ +use super::*; + +#[cfg(feature = "custom")] +mod custom_backend { + use super::*; + use crate::traits::DeviceTrait; + use std::collections::hash_map::DefaultHasher; + use std::fmt; + use std::hash::{Hash, Hasher}; + use std::time::Duration; + + #[derive(Clone, Debug, PartialEq, Eq, Hash)] + struct FakeDevice; + + impl fmt::Display for FakeDevice { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("fake exclusive device") + } + } + + impl DeviceTrait for FakeDevice { + type SupportedInputConfigs = std::iter::Empty; + type SupportedOutputConfigs = std::iter::Empty; + type Stream = crate::host::custom::Stream; + + fn description(&self) -> Result { + Ok(crate::DeviceDescriptionBuilder::new("Fake exclusive device").build()) + } + + fn id(&self) -> Result { + Ok(crate::DeviceId::new( + crate::platform::HostId::Custom, + "fake", + )) + } + + fn supported_input_configs(&self) -> Result { + Ok(std::iter::empty()) + } + + fn supported_output_configs(&self) -> Result { + Ok(std::iter::empty()) + } + + fn default_input_config(&self) -> Result { + Err(crate::Error::new(crate::ErrorKind::UnsupportedOperation)) + } + + fn default_output_config(&self) -> Result { + Err(crate::Error::new(crate::ErrorKind::UnsupportedOperation)) + } + + fn build_input_stream_raw( + &self, + _: crate::StreamConfig, + _: crate::SampleFormat, + _: D, + _: E, + _: Option, + ) -> Result + where + D: FnMut(&crate::Data, &crate::CallbackInfo) + Send + 'static, + E: FnMut(crate::Error) + Send + 'static, + { + Err(crate::Error::new(crate::ErrorKind::UnsupportedOperation)) + } + + fn build_output_stream_raw( + &self, + _: crate::StreamConfig, + _: crate::SampleFormat, + _: D, + _: E, + _: Option, + ) -> Result + where + D: FnMut(&mut crate::Data, &crate::CallbackInfo) + Send + 'static, + E: FnMut(crate::Error) + Send + 'static, + { + Err(crate::Error::new(crate::ErrorKind::UnsupportedOperation)) + } + } + + fn device() -> Device { + Device::from(crate::platform::CustomDevice::from_device(FakeDevice)) + } + + fn unsupported(result: Result) { + let error = match result { + Err(error) => error, + Ok(_) => panic!("exclusive operation unexpectedly succeeded"), + }; + assert_eq!(error.kind(), crate::ErrorKind::UnsupportedOperation); + } + + #[test] + fn unsupported_backend_fails_closed_for_exclusive_queries() { + let exclusive = device().exclusive(true); + assert!(!exclusive.supports_exclusive()); + assert!(!exclusive.supports_input()); + assert!(!exclusive.supports_output()); + assert!(!exclusive.supports_duplex()); + unsupported(exclusive.supported_input_configs()); + unsupported(exclusive.supported_output_configs()); + unsupported(exclusive.default_input_config()); + unsupported(exclusive.default_output_config()); + } + + #[test] + fn exclusive_selection_preserves_identity_hash_and_reversal() { + let shared = device(); + let exclusive = shared.clone().exclusive(true); + assert_eq!(shared, exclusive); + assert_eq!(shared.id().unwrap(), exclusive.id().unwrap()); + + let hash = |device: &Device| { + let mut hasher = DefaultHasher::new(); + device.hash(&mut hasher); + hasher.finish() + }; + assert_eq!(hash(&shared), hash(&exclusive)); + unsupported(exclusive.clone().default_output_config()); + assert!(exclusive.clone().exclusive(false).supports_output() == shared.supports_output()); + assert!(exclusive.exclusive(false).default_output_config().is_err()); + } +} diff --git a/src/platform/mod.rs b/src/platform/mod.rs index 96da987a3..0b4fe3f9c 100644 --- a/src/platform/mod.rs +++ b/src/platform/mod.rs @@ -6,6 +6,9 @@ pub use self::platform_impl::*; +#[cfg(all(test, feature = "custom"))] +mod exclusive_tests; + #[cfg(all( feature = "jack", any( @@ -84,7 +87,7 @@ macro_rules! impl_platform_host { /// The `Device` implementation associated with the platform's dynamically dispatched /// [`Host`] type. #[derive(Clone)] - pub struct Device(DeviceInner); + pub struct Device(DeviceInner, bool); /// The `Devices` iterator associated with the platform's dynamically dispatched [`Host`] /// type. @@ -273,6 +276,54 @@ macro_rules! impl_platform_host { } impl Device { + /// Requests exclusive access for subsequent configuration queries and streams. + /// + /// Select the mode before querying configurations. Unsupported backends return + /// [`crate::ErrorKind::UnsupportedOperation`] rather than falling back to shared + /// access. Selection is infallible; device availability is checked when building + /// the stream. `false` restores the backend's default mode. + /// + /// Capabilities and stream support remain backend- and device-dependent; query + /// [`DeviceTrait::supports_exclusive`](crate::traits::DeviceTrait::supports_exclusive) and configuration methods before use. + /// Exclusive mode selects an access mode and does not imply bit-perfect playback. + #[must_use] + pub fn exclusive(mut self, exclusive: bool) -> Self { + self.1 = exclusive; + #[cfg(windows)] + match &mut self.0 { + DeviceInner::Wasapi(device) => { + *device = device.clone().exclusive(exclusive); + } + #[allow(unreachable_patterns)] + _ => {} + } + self + } + + #[cfg(windows)] + fn wasapi_device(&self) -> Option { + match &self.0 { + DeviceInner::Wasapi(device) => Some(device.clone().exclusive(self.1 || device.is_exclusive())), + #[allow(unreachable_patterns)] + _ => None, + } + } + + fn check_exclusive(&self) -> Result<(), crate::Error> { + #[cfg(windows)] + if self.wasapi_device().is_some() { + return Ok(()); + } + if self.1 { + Err(crate::Error::with_message( + crate::ErrorKind::UnsupportedOperation, + "Exclusive access is not implemented by this backend", + )) + } else { + Ok(()) + } + } + /// Returns a reference to the underlying platform-specific [`DeviceInner`]. pub fn as_inner(&self) -> &DeviceInner { &self.0 } @@ -280,7 +331,11 @@ macro_rules! impl_platform_host { pub fn as_inner_mut(&mut self) -> &mut DeviceInner { &mut self.0 } /// Consumes this `Device`, returning the underlying platform-specific [`DeviceInner`]. - pub fn into_inner(self) -> DeviceInner { self.0 } + pub fn into_inner(self) -> DeviceInner { + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { return DeviceInner::Wasapi(device); } + self.0 + } } impl Host { @@ -388,6 +443,14 @@ macro_rules! impl_platform_host { type SupportedOutputConfigs = SupportedOutputConfigs; type Stream = Stream; + fn supports_exclusive(&self) -> bool { + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { + return device.supports_exclusive(); + } + false + } + fn description(&self) -> Result { match self.0 { $( @@ -407,6 +470,11 @@ macro_rules! impl_platform_host { } fn supports_input(&self) -> bool { + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { return device.supports_input(); } + if self.1 || self.check_exclusive().is_err() { + return false; + } match self.0 { $( $(#[cfg($feat)])? @@ -416,6 +484,9 @@ macro_rules! impl_platform_host { } fn supports_output(&self) -> bool { + if self.check_exclusive().is_err() { + return false; + } match self.0 { $( $(#[cfg($feat)])? @@ -425,6 +496,11 @@ macro_rules! impl_platform_host { } fn supported_input_configs(&self) -> Result { + self.check_exclusive()?; + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { + return device.supported_input_configs().map(SupportedInputConfigsInner::Wasapi).map(SupportedInputConfigs); + } match self.0 { $( $(#[cfg($feat)])? @@ -438,6 +514,11 @@ macro_rules! impl_platform_host { } fn supported_output_configs(&self) -> Result { + self.check_exclusive()?; + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { + return device.supported_output_configs().map(SupportedOutputConfigsInner::Wasapi).map(SupportedOutputConfigs); + } match self.0 { $( $(#[cfg($feat)])? @@ -451,6 +532,9 @@ macro_rules! impl_platform_host { } fn default_input_config(&self) -> Result { + self.check_exclusive()?; + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { return device.default_input_config(); } match self.0 { $( $(#[cfg($feat)])? @@ -460,6 +544,9 @@ macro_rules! impl_platform_host { } fn default_output_config(&self) -> Result { + self.check_exclusive()?; + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { return device.default_output_config(); } match self.0 { $( $(#[cfg($feat)])? @@ -480,6 +567,11 @@ macro_rules! impl_platform_host { D: FnMut(&crate::Data, &crate::CallbackInfo) + Send + 'static, E: FnMut(crate::Error) + Send + 'static, { + self.check_exclusive()?; + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { + return device.build_input_stream_raw(config, sample_format, data_callback, error_callback, timeout).map(StreamInner::Wasapi).map(Stream::from); + } match self.0 { $( $(#[cfg($feat)])? @@ -509,6 +601,11 @@ macro_rules! impl_platform_host { D: FnMut(&mut crate::Data, &crate::CallbackInfo) + Send + 'static, E: FnMut(crate::Error) + Send + 'static, { + self.check_exclusive()?; + #[cfg(windows)] + if let Some(device) = self.wasapi_device() { + return device.build_output_stream_raw(config, sample_format, data_callback, error_callback, timeout).map(StreamInner::Wasapi).map(Stream::from); + } match self.0 { $( $(#[cfg($feat)])? @@ -527,6 +624,9 @@ macro_rules! impl_platform_host { } fn supports_duplex(&self) -> bool { + if self.1 || self.check_exclusive().is_err() { + return false; + } match self.0 { $( $(#[cfg($feat)])? @@ -550,6 +650,8 @@ macro_rules! impl_platform_host { + 'static, E: FnMut(crate::Error) + Send + 'static, { + self.check_exclusive()?; + if self.1 { return Err(crate::Error::new(crate::ErrorKind::UnsupportedOperation)); } match self.0 { $( $(#[cfg($feat)])? @@ -756,7 +858,15 @@ macro_rules! impl_platform_host { impl From for Device { fn from(d: DeviceInner) -> Self { - Device(d) + #[cfg(not(windows))] + let exclusive = false; + #[cfg(windows)] + let exclusive = match &d { + DeviceInner::Wasapi(device) => device.is_exclusive(), + #[allow(unreachable_patterns)] + _ => false, + }; + Device(d, exclusive) } } diff --git a/src/traits.rs b/src/traits.rs index beeff09d2..6c7572721 100644 --- a/src/traits.rs +++ b/src/traits.rs @@ -127,6 +127,13 @@ pub trait HostTrait { /// Please note that `Device`s may become invalid if they get disconnected. Therefore, all the /// methods that involve a device return a `Result` allowing the user to handle this case. pub trait DeviceTrait: PartialEq + Eq + Hash + Debug + Display + Send + Sync { + /// Whether this backend implements exclusive access for this device. + /// + /// This does not guarantee that the device is currently available exclusively. + fn supports_exclusive(&self) -> bool { + false + } + /// The iterator type yielding supported input stream formats. type SupportedInputConfigs: Iterator; /// The iterator type yielding supported output stream formats.