merge: integrate Windows port from PR #725

cx
This commit is contained in:
vicrodh committed 2026-08-28 04:18:06 -06:00
commit 912638c387
93 files changed
+6794 -261

No files matched your search

+12
View File
@@ -64,3 +64,15 @@ jack = "0.13"
# CoreAudio for device probing, sample rate switching, and (future) hog mode / exclusive mode
coreaudio-rs = "0.14.1"
objc2-core-audio = "0.3.2"
[target.'cfg(windows)'.dependencies]
# WASAPI exclusive mode (phase B). HEnquist's crate, the one CamillaDSP
# uses. `windows` is pinned to the 0.62 line cpal already resolves, so
# this adds no second windows tree to the graph.
wasapi = "0.24"
windows = { version = "0.62", features = [
"Win32_Foundation",
"Win32_Media_Audio",
"Win32_System_Com",
"Win32_System_Threading",
] }
+221
View File
@@ -0,0 +1,221 @@
//! Throwaway proof for the WASAPI exclusive path (phase B, task 2 step 8).
//!
//! Opens a render endpoint in EXCLUSIVE mode and plays a 1 kHz sine, then
//! reports what the stream negotiated and how many underruns it took.
//!
//! Deliberately quiet: -20 dBFS. A stability proof does not need full scale,
//! and this runs on somebody's actual monitors.
//!
//! cargo run -p qbz-audio --example wasapi_play list endpoints
//! cargo run -p qbz-audio --example wasapi_play <id> [rate] [s] play
//!
//! What it proves that a unit test cannot: the DAC's own display shows the
//! rate we asked for. If the display says 48 kHz while this says 96, the
//! stream is being resampled somewhere and it is NOT bit-perfect.
fn main() {
#[cfg(not(windows))]
{
eprintln!("wasapi_play is Windows-only");
}
#[cfg(windows)]
{
use qbz_audio::wasapi_direct::{WasapiDirectStream, WasapiTiming};
let args: Vec<String> = std::env::args().skip(1).collect();
if args.is_empty() {
list_endpoints();
eprintln!();
eprintln!("pass an endpoint id to play, e.g.");
eprintln!(" cargo run -p qbz-audio --example wasapi_play \"{{0.0.0...}}\" 96000 10");
return;
}
let endpoint = args[0].clone();
let rate: u32 = args.get(1).and_then(|s| s.parse().ok()).unwrap_or(96_000);
let secs: u64 = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(10);
let stream = match WasapiDirectStream::new(&endpoint, rate, 2, WasapiTiming::Events) {
Ok(s) => s,
Err(e) => {
eprintln!("open failed: {e}");
return;
}
};
let info = stream.open_info();
println!("endpoint : {}", info.endpoint_name);
println!("rate : {} Hz", info.rate);
println!(
"format : {:?} ({}/{} bits)",
info.rung,
info.rung.valid_bits(),
info.rung.container_bits()
);
println!(
"period : {} hns buffer: {} frames",
info.period_hns, info.buffer_frames
);
println!("realigned: {} timing: {:?}", info.realigned, info.timing);
println!("bit-perfect mode: {:?}", stream.bit_perfect_mode());
println!();
println!("BEEP 1 of 3 - {secs}s of 1 kHz at -20 dBFS");
println!("CHECK THE DAC: it must read {rate}. Then expect TWO more one-second");
println!("beeps - one after the drain, one after the stop. Three beeps = PASS.");
// Feed in period-sized chunks so the queue's back-pressure paces us
// instead of a sleep. 1 kHz divides every rate in the ladder evenly
// enough that the phase never accumulates audible drift over 10 s.
let frames_per_chunk = info.buffer_frames.max(1) as usize;
let total_frames = rate as u64 * secs;
let amp = 0.1f32; // -20 dBFS
let step = 2.0 * std::f32::consts::PI * 1000.0 / rate as f32;
let mut phase = 0.0f32;
let mut written: u64 = 0;
while written < total_frames {
let n = frames_per_chunk.min((total_frames - written) as usize);
let mut buf = Vec::with_capacity(n * 2);
for _ in 0..n {
let v = phase.sin() * amp;
phase += step;
if phase > std::f32::consts::TAU {
phase -= std::f32::consts::TAU;
}
buf.push(v); // L
buf.push(v); // R
}
if let Err(e) = stream.write_f32(&buf) {
eprintln!("write failed after {written} frames: {e}");
break;
}
written += n as u64;
}
// The verdict must be judged on CONTINUOUS playback only. Everything
// after this point deliberately leaves the queue empty between beeps,
// and a genuine gap with the stream running IS an underrun -- counting
// those would make this instrument report a failure it created itself.
let underruns_playing = stream.underruns();
let timeouts_playing = stream.event_timeouts();
// Drain, then WRITE AGAIN. This is the discriminating test for the
// gapless contract: drain must block until the queue has played out
// AND must leave the stream usable for the next source. An early
// return, or a render thread that exits on drain, both show up here
// and nowhere else.
let t0 = std::time::Instant::now();
let drained = stream.drain();
let drain_ms = t0.elapsed().as_millis();
println!();
println!("drain : {drained:?} in {drain_ms} ms");
// A FULL SECOND, not one buffer. The point of these two checks is that
// a person can HEAR the second and third beeps; a 144-frame burst at
// 44.1 kHz is 3 ms, which is inaudible, so the test could pass or fail
// and sound identical either way.
let mut beep = |stream: &WasapiDirectStream, phase: &mut f32| -> Result<(), String> {
let mut left = rate as usize;
while left > 0 {
let n = frames_per_chunk.min(left);
let mut buf = Vec::with_capacity(n * 2);
for _ in 0..n {
let v = phase.sin() * amp;
*phase += step;
if *phase > std::f32::consts::TAU {
*phase -= std::f32::consts::TAU;
}
buf.push(v);
buf.push(v);
}
stream.write_f32(&buf)?;
left -= n;
}
Ok(())
};
println!("BEEP 2 of 3 - one second, after the drain");
match beep(&stream, &mut phase) {
Ok(()) => println!("write after drain: OK - the stream survived the drain"),
Err(e) => println!("write after drain: FAILED ({e}) - the stream died on drain"),
}
let _ = stream.drain();
// STOP, then WRITE AGAIN. This is the OTHER half of the gapless
// contract and the one that was missing: the player calls
// engine.stop() on EVERY track change and then reuses the same stream
// object for the next track, exactly as ALSA's stop() does
// ("prepare for next playback"). A render thread that exits on Stop
// passes every drain test and still cannot play a second track --
// which is what shipped, and what the app log caught as
// "Write failed: render thread is gone".
let t1 = std::time::Instant::now();
let stopped = stream.stop();
println!(
"stop : {stopped:?} in {} ms",
t1.elapsed().as_millis()
);
println!("BEEP 3 of 3 - one second, after the stop");
let survived_stop = match beep(&stream, &mut phase) {
Ok(()) => {
println!("write after stop : OK - the stream survived the stop");
true
}
Err(e) => {
println!("write after stop : FAILED ({e}) - the stream died on stop");
false
}
};
let _ = stream.drain();
println!();
println!("frames written : {written}");
println!(
"underruns : {} during playback ({} total, gaps between beeps included)",
underruns_playing,
stream.underruns()
);
println!(
"event timeouts : {} during playback ({} total)",
timeouts_playing,
stream.event_timeouts()
);
println!(
"VERDICT : {}",
if !survived_stop {
"FAIL - the stream did not survive stop(); a track change kills it"
} else if underruns_playing == 0 && timeouts_playing == 0 {
"PASS - survived drain and stop, no underruns, no timeouts"
} else {
"underruns occurred; try WasapiTiming::Polling"
}
);
}
}
#[cfg(windows)]
fn list_endpoints() {
use wasapi::{initialize_mta, DeviceEnumerator, Direction};
let _ = initialize_mta().ok();
let Ok(en) = DeviceEnumerator::new() else {
eprintln!("DeviceEnumerator failed");
return;
};
let Ok(coll) = en.get_device_collection(&Direction::Render) else {
eprintln!("device collection failed");
return;
};
let n = coll.get_nbr_devices().unwrap_or(0);
println!("render endpoints:");
for i in 0..n {
if let Ok(d) = coll.get_device_at_index(i) {
println!(
" [{i}] {}",
d.get_friendlyname().unwrap_or_else(|_| "<unnamed>".into())
);
println!(" {}", d.get_id().unwrap_or_else(|_| "<no id>".into()));
}
}
}
+33
View File
@@ -1227,3 +1227,36 @@ mod tests {
assert!(!uses_alsa_direct_route(&audio));
}
}
/// Forwards to the inherent methods, which is all this is: no behaviour is
/// added, moved or wrapped. See `backend::DirectSink`.
impl crate::backend::DirectSink for AlsaDirectStream {
fn write_f32(&self, samples: &[f32]) -> Result<(), String> {
AlsaDirectStream::write_f32(self, samples)
}
fn drain(&self) -> Result<(), String> {
AlsaDirectStream::drain(self)
}
fn stop(&self) -> Result<(), String> {
AlsaDirectStream::stop(self)
}
fn sample_rate(&self) -> u32 {
AlsaDirectStream::sample_rate(self)
}
fn channels(&self) -> u16 {
AlsaDirectStream::channels(self)
}
fn playback_delay_frames(&self) -> Result<u64, String> {
AlsaDirectStream::playback_delay_frames(self)
}
fn log_label(&self) -> &'static str {
"ALSA Direct Engine"
}
/// Linux only, because the inherent method is: the non-Linux stub of this
/// type has no mixer at all, and there the trait default refusal is the
/// honest answer.
#[cfg(target_os = "linux")]
fn set_hardware_volume(&self, volume: f32) -> Result<(), String> {
AlsaDirectStream::set_hardware_volume(self, volume)
}
}
+165 -5
View File
@@ -42,6 +42,16 @@ pub enum AudioBackendType {
/// bit-perfect ALSA-exclusive / DAC-passthrough paths.
Jack,
/// WASAPI EXCLUSIVE (Windows) - the bit-perfect path.
/// - Takes the endpoint exclusively; other apps go silent
/// - No resampling: the DAC is commanded to the track's own rate
/// - Selecting a device is required, by ENDPOINT ID (Windows gives several
/// endpoints the same friendly name)
/// The trait's `create_output_stream` for this type is the SHARED CPAL one:
/// the exclusive stream bypasses the trait entirely, exactly as ALSA Direct
/// and JACK do.
WasapiExclusive,
/// System default backend (non-Linux platforms)
/// - Uses CPAL default host (CoreAudio on macOS, WASAPI on Windows)
/// - Automatic device selection via OS audio system
@@ -308,6 +318,14 @@ impl BackendManager {
backends.push(AudioBackendType::SystemDefault);
}
#[cfg(windows)]
{
// Offered unconditionally: whether a given DEVICE accepts exclusive
// mode is answered per device when the stream opens, not here, and
// hiding the backend would hide the answer too.
backends.push(AudioBackendType::WasapiExclusive);
}
backends
}
@@ -333,6 +351,14 @@ impl BackendManager {
Ok(Box::new(CpalDefaultBackend::new()?))
}
}
AudioBackendType::WasapiExclusive => {
// The SHARED stream, deliberately. `create_stream_for_backend`
// tries the exclusive stream first and only reaches here when
// that is impossible - no device chosen, or the endpoint
// refusing every format - so this is the graceful fallback
// rather than the intended path.
Ok(Box::new(CpalDefaultBackend::new()?))
}
AudioBackendType::SystemDefault => {
// "System": play through the OS default output via CPAL's default
// host — CoreAudio/WASAPI off-Linux, the ALSA "default" PCM on Linux
@@ -456,6 +482,20 @@ impl AudioBackend for CpalDefaultBackend {
.map(|desc| desc.name().to_string())
.unwrap_or_else(|_| "Default Output".to_string());
// WINDOWS ONLY, additive. Endpoint FRIENDLY NAMES are not unique
// there: measured on a real box, three active render endpoints were
// all called "Altavoces" - the motherboard Realtek, a Bluetooth
// handsfree and a Cambridge Audio USB DAC. Matching the default by
// name would mark all three.
#[cfg(windows)]
let default_endpoint_id = default_device.id().ok().map(|i| i.1);
// Idempotent, and this is the natural place: the first time anything
// asks what devices exist is the first time it matters that the answer
// can go stale.
#[cfg(windows)]
crate::wasapi_backend::start_hotplug_watch();
let mut devices = Vec::new();
for device in self
.host
@@ -466,23 +506,92 @@ impl AudioBackend for CpalDefaultBackend {
.description()
.map(|desc| desc.name().to_string())
.unwrap_or_else(|_| "Unknown Device".to_string());
let is_default = name == default_name;
// On macOS, probe device capabilities via CoreAudio
#[cfg(target_os = "macos")]
let (supported_rates, max_rate, bus_type, is_hw) = { Self::probe_macos_device(&name) };
#[cfg(not(target_os = "macos"))]
// The placeholder for every platform that has no probe. On
// Windows `supported_rates`/`max_rate` are SHADOWED further down
// by the exclusive-mode sweep, which is why they carry an
// underscore prefix there and nowhere else.
#[cfg(all(not(target_os = "macos"), not(windows)))]
let (supported_rates, max_rate, bus_type, is_hw): (
Option<Vec<u32>>,
Option<u32>,
Option<String>,
bool,
) = (None, None, None, false);
#[cfg(windows)]
let (bus_type, is_hw): (Option<String>, bool) = (None, false);
// WINDOWS ONLY, additive: id, display and description all come
// from fields cpal already fills but this backend ignored.
//
// `name` is the endpoint ("Altavoces") and repeats across
// devices; `driver()` is the ADAPTER ("Cambridge Audio USB Audio
// 2.0") and is what tells them apart; `id()` is the stable WASAPI
// endpoint GUID. Using the name as the id meant device_filter
// folded every same-named endpoint into one entry, so a USB DAC
// sharing the name "Altavoces" with the onboard card could not be
// selected at all - and an unselectable DAC blocks bit-perfect.
#[cfg(windows)]
let (id, display, description, is_default) = {
let driver = device
.description()
.ok()
.and_then(|d| d.driver().map(str::to_string))
.filter(|d| !d.is_empty() && d.as_str() != name.as_str());
let endpoint_id = device.id().ok().map(|i| i.1);
let display = match &driver {
Some(d) => format!("{name} ({d})"),
None => name.clone(),
};
let is_default = match (&endpoint_id, &default_endpoint_id) {
(Some(a), Some(b)) => a == b,
_ => name == default_name,
};
(
endpoint_id.unwrap_or_else(|| name.clone()),
display,
driver,
is_default,
)
};
#[cfg(not(windows))]
let (id, display, description, is_default) =
(name.clone(), name.clone(), None, name == default_name);
// WINDOWS: the rates the endpoint accepts in EXCLUSIVE mode, which
// is the only list that describes what reaches the converter
// untouched. Shadows the `(None, None, ...)` placeholder above.
//
// It cannot be inferred, which is the whole reason it is measured:
// the owner's DAC accepts 44100/48000/88200/96000/192000 and
// REFUSES 176400, so the 44.1 family stops at 88.2 while the 48
// family reaches 192. Cached per endpoint, so this costs once.
//
// `supported_rates` answers `None` when the endpoint could not
// be asked at all -- busy, gone, or COM refused -- and that must
// not reach a caller as "supports nothing", which would narrow a
// quality cap on the strength of a failed question. An endpoint
// that answers and lists nothing is reported the same way for the
// same reason: there is nothing to offer either way, and neither
// is a fact worth caching upstream.
#[cfg(windows)]
let (supported_rates, max_rate) = {
match crate::wasapi_backend::supported_rates(&id) {
Some(rates) if !rates.is_empty() => {
let top = rates.iter().copied().max();
(Some(rates), top)
}
_ => (None, None),
}
};
devices.push(AudioDevice {
id: name.clone(),
name,
description: None,
id,
name: display,
description,
is_default,
max_sample_rate: max_rate,
supported_sample_rates: supported_rates,
@@ -563,6 +672,14 @@ impl AudioBackend for CpalDefaultBackend {
.output_devices()
.map_err(|e| format!("Failed to enumerate devices: {}", e))?
.find(|d| {
// WINDOWS ONLY, additive: enumerate_devices publishes the
// stable endpoint GUID as the id there, because friendly
// names repeat. The name comparison stays as the fallback
// so a setting saved before this change still resolves.
#[cfg(windows)]
if d.id().map(|i| i.1 == *device_id).unwrap_or(false) {
return true;
}
d.description()
.map(|desc| desc.name() == device_id.as_str())
.unwrap_or(false)
@@ -933,3 +1050,46 @@ impl CpalDefaultBackend {
}
}
}
/// The two-method surface a bit-perfect direct stream owes the player's writer
/// thread, plus the three it owes the engine around it.
///
/// It exists so ONE writer thread serves both `AlsaDirectStream` and
/// `WasapiDirectStream` instead of two copies drifting apart. The set is
/// exactly what `playback_engine.rs` already called on the ALSA stream - no
/// method was invented for the trait, and neither implementation gains or
/// loses behaviour by having it.
pub trait DirectSink: Send + Sync + 'static {
/// Interleaved f32, blocking when the device's queue is full. That block
/// IS the back-pressure that paces the writer thread.
fn write_f32(&self, samples: &[f32]) -> Result<(), String>;
/// Play out what is queued, then return. Bounded internally.
fn drain(&self) -> Result<(), String>;
fn stop(&self) -> Result<(), String>;
fn sample_rate(&self) -> u32;
fn channels(&self) -> u16;
/// Frames submitted to the sink but not yet heard, when the backend can
/// measure them. The visualizer uses this only as latency compensation;
/// an unsupported query keeps playback intact and falls back to the
/// bounded direct-writer pacing.
fn playback_delay_frames(&self) -> Result<u64, String> {
Err("this direct sink cannot measure queued playback frames".to_string())
}
/// Log prefix for this sink. On the trait rather than a parameter so a
/// call site cannot label an ALSA stream "WASAPI" or the reverse - which
/// is exactly what a hard-coded label at four call sites invites.
fn log_label(&self) -> &'static str;
/// Device-side volume, where the sink has one.
///
/// The default REFUSES rather than silently succeeding: a sink without a
/// hardware mixer must not pretend it moved the volume. Every call site is
/// already gated on the engine's `hardware_volume` flag, which is false for
/// any such sink - permanently so for WASAPI exclusive, where the phase B
/// contract rules out ISimpleAudioVolume.
fn set_hardware_volume(&self, _volume: f32) -> Result<(), String> {
Err("this sink has no hardware volume".to_string())
}
}
+14
View File
@@ -11,6 +11,8 @@
//! bit-perfect / sample-rate-passthrough path is untouched.
use serde::{Deserialize, Serialize};
// Every target but Windows still runs `pw-dump`.
#[cfg(not(windows))]
use std::process::Command;
/// The DAC's live, negotiated hardware state — what the card is REALLY clocked
@@ -94,6 +96,18 @@ pub fn parse_alsa_card_for_node(json: &str, node_name: &str) -> Option<u32> {
}
/// Resolve the ALSA card number for a sink node by running `pw-dump`.
///
/// W14, Windows only: `pw-dump` does not exist there, so the spawn already
/// failed into this same None. Scoped to Windows rather than `not(linux)` --
/// a macOS or BSD box can have a real `pw-dump`, and this function is not the
/// place to decide it may not speak.
#[cfg(windows)]
fn alsa_card_for_node(_node_name: &str) -> Option<u32> {
None
}
/// Resolve the ALSA card number for a sink node by running `pw-dump`.
#[cfg(not(windows))]
fn alsa_card_for_node(node_name: &str) -> Option<u32> {
let output = Command::new("pw-dump").output().ok()?;
if !output.status.success() {
+29 -1
View File
@@ -5,6 +5,8 @@
//! the check step can show the right `apt`/`dnf`/`pacman` remediation. None of
//! this opens a stream — purely diagnostic.
// Every target but Windows still probes; see `audio_stack_health`.
#[cfg(not(windows))]
use std::process::Command;
/// Result of the audio-stack probes. All best-effort; a failed probe reads as
@@ -32,7 +34,8 @@ impl AudioStackHealth {
}
}
/// True if `sh -c "<probe>"` exits 0.
/// True if `sh -c "<probe>"` exits 0. Windows never probes; see below.
#[cfg(not(windows))]
fn sh_ok(probe: &str) -> bool {
Command::new("sh")
.arg("-c")
@@ -44,6 +47,31 @@ fn sh_ok(probe: &str) -> bool {
/// Run the audio-stack probes. Linux-only meaningfully; elsewhere everything
/// reads false except where trivially true.
///
/// W14: on WINDOWS ONLY the answer is all-false, returned WITHOUT running the
/// probes. `systemctl`, `pactl` and `pw-dump` do not exist there (a spawn of a
/// missing image fails before a process is made, which is harmless), but `sh`
/// DOES exist wherever Git for Windows is installed, so the five `sh -c`
/// probes really run and flash five console windows past a GUI-subsystem
/// binary that owns no console.
///
/// Scoped to Windows, not to `not(linux)`: macOS and the BSDs can have a real
/// `pactl`/`pw-dump` from Homebrew or ports, and answering false for them
/// would be a lie this diff has no business telling.
#[cfg(windows)]
pub fn audio_stack_health() -> AudioStackHealth {
AudioStackHealth {
wireplumber_active: false,
has_pw_dump: false,
cpal_sees_pipewire: false,
has_pactl: false,
any_devices: false,
}
}
/// Run the audio-stack probes. Linux-only meaningfully; elsewhere everything
/// reads false except where trivially true.
#[cfg(not(windows))]
pub fn audio_stack_health() -> AudioStackHealth {
// systemd path first; fall back to a process check so non-systemd inits
// (Gentoo/OpenRC, Void/runit) don't read as "WirePlumber down".
+5
View File
@@ -29,6 +29,10 @@
#[cfg(target_os = "linux")]
pub mod alsa_backend;
pub mod alsa_direct;
pub mod wasapi_direct;
/// Endpoint capabilities: the exclusive-mode rate sweep and the hotplug watch.
#[cfg(windows)]
pub mod wasapi_backend;
#[cfg(target_os = "linux")]
pub mod alsa_error_handler;
pub mod analysis;
@@ -64,6 +68,7 @@ pub use alsa_backend::{
resolve_stable_to_current_hw,
};
pub use alsa_direct::AlsaDirectStream;
pub use wasapi_direct::{Rung, LADDER};
pub use analysis::SpectralAnalyzer;
pub use analyzer_tap::{AnalyzerMessage, AnalyzerTap, AnalyzerWaveformTrack};
pub use backend::{
+374
View File
@@ -0,0 +1,374 @@
//! WASAPI endpoint capabilities: what rates a device accepts in EXCLUSIVE
//! mode, and being told when the device list changes.
//!
//! The Windows counterpart of `alsa_backend::get_device_supported_rates`. It
//! exists because "what can this DAC actually do" cannot be inferred: the
//! spike measured a Cambridge Audio USB DAC accepting 44100, 48000, 88200,
//! 96000 and 192000 while REFUSING 176400 -- the 44.1 family stops at 88.2
//! while the 48 family reaches 192, which no rule about "higher rates imply
//! lower ones" predicts. The owner confirmed it from the hardware, and a
//! SECOND endpoint on the same machine does accept 176400, so the difference
//! is the device rather than the driver stack.
//!
//! ## Exclusive mode is the question, and it is a different question
//!
//! `IsFormatSupported(SHARED)` answers about the mixer, which resamples and
//! therefore says yes to almost everything. Only the EXCLUSIVE answer says
//! what reaches the converter untouched, which is the only answer bit-perfect
//! playback cares about.
//!
//! ## "I could not ask" is not "it supports nothing"
//!
//! A busy endpoint, a failed activation and a device that genuinely accepts no
//! rate all produce an empty list, and collapsing them was a real defect in the
//! first draft: the empty answer was cached as authoritative, and because a
//! stream CLOSING raises no device notification, it would have outlived the
//! condition that caused it. [`Caps`] keeps them apart, and only `Known` is
//! ever stored.
#![cfg(windows)]
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock};
use crate::wasapi_direct::LADDER;
/// The rates worth asking about: the two families QBZ streams and their
/// multiples, plus the DSD-over-PCM carriers. 352800 and 384000 are asked
/// because a DAC that takes them changes what a quality cap can offer.
const SWEEP: [u32; 8] = [
44_100, 48_000, 88_200, 96_000, 176_400, 192_000, 352_800, 384_000,
];
/// What a probe learned, which is not always a capability.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Caps {
/// The device answered. An empty vector HERE is a real "nothing".
Known(Vec<u32>),
/// The device could not be asked -- busy, gone, or COM refused. Never
/// cached: nothing about it is a property of the hardware.
Unavailable,
}
/// Bumped by every device notification. A cached answer is only good while
/// this has not moved.
///
/// A COUNTER rather than per-endpoint removal, and that is the point: the
/// notifications arrive on a thread the Windows audio system owns, and
/// Microsoft's rule there is DO NOT BLOCK. `fetch_add` cannot block, cannot
/// fail and cannot lose an invalidation, where taking the cache's mutex would
/// block a thread that was told not to, and a `try_lock` that skipped on
/// contention would silently keep a stale answer.
///
/// Invalidating EVERYTHING on any event is also the honest granularity: a
/// device arriving or leaving changes exclusive-mode contention for endpoints
/// other than itself, so "only that one changed" was never true.
static GENERATION: AtomicU64 = AtomicU64::new(0);
/// Endpoint id -> (the generation it was measured in, the rates it accepted).
fn cache() -> &'static Mutex<HashMap<String, (u64, Vec<u32>)>> {
static CACHE: OnceLock<Mutex<HashMap<String, (u64, Vec<u32>)>>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(HashMap::new()))
}
/// Serialises the probing itself, so two callers that miss together do the
/// work once instead of twice.
///
/// MEASURED, not theoretical: startup enumerates from two independent places
/// -- the device-capability pass and the DAC-capability pass -- and the log
/// showed every endpoint swept exactly twice, both racing past the empty
/// cache. Holding this across the sweep costs the second caller about a
/// millisecond per endpoint and saves it the same again.
///
/// SEPARATE from the cache mutex on purpose. The cache lock is taken for a
/// `get` and an `insert` and released immediately; this one is held across COM
/// calls, and merging them would make every reader wait behind a probe. The
/// hotplug callbacks touch NEITHER -- they only bump an atomic -- so nothing
/// the audio system owns can ever wait on either lock.
fn sweep_gate() -> &'static Mutex<()> {
static GATE: OnceLock<Mutex<()>> = OnceLock::new();
GATE.get_or_init(|| Mutex::new(()))
}
/// The cache key: the endpoint, and the QUESTION the stored answer answers.
///
/// The value is "some rung of the ladder accepts this rate", so naming one
/// rung would claim a precision it does not have. `LADDER.len()` is part of it
/// because growing the ladder changes the answer for the same device, and an
/// entry written under a shorter ladder must not survive that.
pub(crate) fn rate_sweep_key(endpoint_id: &str) -> String {
format!("{endpoint_id}|any-of-{}-rungs", LADDER.len())
}
/// COM is initialised at most ONCE per thread, and deliberately never undone.
///
/// `initialize_mta` should be balanced by a `CoUninitialize`, but the threads
/// that reach here are pool workers that get reused: uninitialising on the way
/// out would tear the apartment down under whatever COM work the same thread
/// runs next, and doing it per sweep would re-enter the apartment on every
/// call. One initialisation per worker for the life of the process is the
/// bounded cost, and it is the trade any COM-using code on a pool makes.
///
/// `RPC_E_CHANGED_MODE` -- this thread is already an STA -- is not an error to
/// report: `IsFormatSupported` is happy either way.
fn ensure_com() {
thread_local! {
static DONE: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
DONE.with(|d| {
if !d.get() {
let _ = wasapi::initialize_mta().ok();
d.set(true);
}
});
}
/// The rates `endpoint_id` accepts in EXCLUSIVE mode.
///
/// `None` = the endpoint could not be asked. Callers must NOT read that as
/// "supports nothing" -- see [`Caps`] -- and in particular must not use it to
/// justify sending a device a format it has not agreed to.
pub fn supported_rates(endpoint_id: &str) -> Option<Vec<u32>> {
let key = rate_sweep_key(endpoint_id);
// Read BEFORE probing, and store against THIS value rather than the
// current one. If a device event lands while the sweep runs, the result is
// already stale and the next caller re-measures instead of trusting it.
let generation = GENERATION.load(Ordering::SeqCst);
// The lock is never held across the sweep below: that is milliseconds of
// COM calls, and nothing else may be made to wait on it.
if let Ok(guard) = cache().lock() {
if let Some((gen, hit)) = guard.get(&key) {
if *gen == generation {
return Some(hit.clone());
}
}
}
// Take the gate, then look again: the caller that held it may have been
// measuring this very endpoint, in which case its answer is now in the
// cache and there is nothing left to do.
let _gate = sweep_gate().lock();
if let Ok(guard) = cache().lock() {
if let Some((gen, hit)) = guard.get(&key) {
if *gen == generation {
return Some(hit.clone());
}
}
}
match sweep(endpoint_id) {
Caps::Known(rates) => {
if let Ok(mut guard) = cache().lock() {
guard.insert(key, (generation, rates.clone()));
}
Some(rates)
}
// NOT cached. A busy or absent endpoint says nothing about the
// hardware, and a stream closing raises no notification to clear it.
Caps::Unavailable => None,
}
}
/// The highest rate the endpoint accepts, or `None` if it could not be asked
/// or accepted nothing.
pub fn max_supported_rate(endpoint_id: &str) -> Option<u32> {
supported_rates(endpoint_id)?.into_iter().max()
}
/// Drop everything we believe about every endpoint.
///
/// Lock-free by construction, because the hotplug callbacks call it from a
/// thread the audio system owns and are required not to block there.
pub fn invalidate_all() {
GENERATION.fetch_add(1, Ordering::SeqCst);
}
/// Ask the device about every rate in `SWEEP`, at every rung of the format
/// ladder, and keep a rate the moment ANY rung accepts it.
///
/// Walking the ladder rather than one rung matters: the spike's DAC accepts
/// ONLY 24-in-32 and refuses packed 24, 32-bit, 16-bit and float, so a sweep
/// pinned to a single rung would have reported that it supports nothing.
fn sweep(endpoint_id: &str) -> Caps {
use wasapi::{DeviceEnumerator, SampleType, ShareMode, WaveFormat};
ensure_com();
let Ok(enumerator) = DeviceEnumerator::new() else {
log::warn!("[wasapi-caps] no device enumerator; capabilities unknown");
return Caps::Unavailable;
};
let Ok(device) = enumerator.get_device(endpoint_id) else {
log::warn!("[wasapi-caps] endpoint {endpoint_id} unavailable; capabilities unknown");
return Caps::Unavailable;
};
// ONE client for every question. `IsFormatSupported` may be called at any
// time after activation and neither initialises nor reserves the endpoint,
// so a client per probe was forty activations buying nothing. Only an
// INITIALISED exclusive client reserves a device, and this one never is.
let Ok(client) = device.get_iaudioclient() else {
log::warn!("[wasapi-caps] cannot activate {endpoint_id}; capabilities unknown");
return Caps::Unavailable;
};
let mut rates = Vec::new();
for rate in SWEEP {
let accepted = LADDER.iter().any(|rung| {
// The same shape `wasapi_direct`'s open path builds, and it must
// stay the same shape: a rung probed one way and opened another is
// a device that says yes and then refuses.
let ty = if rung.is_float() {
SampleType::Float
} else {
SampleType::Int
};
let format = WaveFormat::new(
rung.container_bits() as usize,
rung.valid_bits() as usize,
&ty,
rate as usize,
2,
None,
);
// `is_supported`, NOT `is_supported_exclusive_with_quirks`. That
// helper retries as a plain WAVEFORMATEX and with alternative
// channel masks, and using it HERE ALONE would be a mistake: this
// sweep must predict what the OPEN path does, and that path asks
// the plain question. A rate accepted only through a quirk the
// opener never tries is a rate that appears in the UI and then
// fails to open -- worse than under-reporting.
//
// Teaching the opener those quirks is worth doing for DACs that
// need them, and it has to happen in BOTH places in one change, or
// the two disagree again in the other direction.
//
// In exclusive mode `ppClosestMatch` is null, so the answer is an
// exact yes or an error: there is no S_FALSE "closest match" to
// weigh, which is shared-mode behaviour.
client.is_supported(&format, &ShareMode::Exclusive).is_ok()
});
if accepted {
rates.push(rate);
}
}
log::info!("[wasapi-caps] {endpoint_id} accepts {rates:?} in exclusive mode");
Caps::Known(rates)
}
// ---------------------------------------------------------------------------
// Hotplug
// ---------------------------------------------------------------------------
/// Whether the watch thread is running. The REGISTRATION itself cannot live in
/// a static: it holds COM interface pointers and is therefore not `Send`.
static WATCH_STARTED: AtomicBool = AtomicBool::new(false);
/// Start listening for endpoints arriving, leaving or changing state.
///
/// Idempotent, and it RE-ARMS: if the audio service is unavailable when this
/// first runs, the flag is cleared so a later call can retry. A transient
/// failure must not disable the watch for the life of the process, which the
/// first draft did by resetting only when the thread failed to spawn.
///
/// SCOPE, stated plainly: this keeps the CAPABILITY CACHE honest. It does not
/// refresh an already-rendered device list -- a newly attached DAC appears
/// when something enumerates again, not the instant it is plugged in. Making
/// the settings model live needs a bridge signal and belongs with that model.
pub fn start_hotplug_watch() {
if WATCH_STARTED.swap(true, Ordering::SeqCst) {
return;
}
// A DEDICATED THREAD that owns the registration and then parks forever.
//
// Not a matter of style: `DeviceEventRegistration` holds COM interface
// pointers, so it is not `Send` and cannot be stored in a static. Keeping
// it on the stack of a thread that never returns is what keeps it alive,
// and dropping it is exactly what unregisters the callbacks -- so the
// thread outliving everything IS the lifetime management. It also gives
// the registration its own apartment rather than borrowing whichever
// thread happened to call first.
let spawned = std::thread::Builder::new()
.name("qbz-wasapi-hotplug".to_string())
.spawn(|| {
use wasapi::{DeviceEnumerator, DeviceEventCallbacks};
ensure_com();
let Ok(enumerator) = DeviceEnumerator::new() else {
log::warn!("[wasapi-caps] hotplug watch unavailable; will retry on next request");
WATCH_STARTED.store(false, Ordering::SeqCst);
return;
};
// EVERY callback below runs on a thread the Windows audio system
// owns, where the documented rule is that they return IMMEDIATELY.
// So each does one lock-free atomic increment and NOTHING else --
// no mutex, and no logging either: a log macro can allocate, take
// its own lock and perform synchronous output, and "it is only a
// log line" is exactly how that rule gets broken.
let mut callbacks = DeviceEventCallbacks::new();
callbacks.set_device_removed_callback(|_id| invalidate_all());
callbacks.set_device_added_callback(|_id| invalidate_all());
callbacks.set_device_state_callback(|_id, _state| invalidate_all());
// Reconfiguring a device in Windows' own Sound settings changes
// what it accepts without adding or removing anything. The first
// draft's comment promised this and installed nothing.
callbacks.set_property_value_callback(|_id, _key| invalidate_all());
let _registration = match enumerator.register_notification_callback(callbacks) {
Ok(reg) => reg,
Err(e) => {
log::warn!("[wasapi-caps] could not register the hotplug watch: {e}");
WATCH_STARTED.store(false, Ordering::SeqCst);
return;
}
};
log::info!("[wasapi-caps] hotplug watch registered");
// Park. `_registration` must not be dropped, and there is nothing
// for this thread to do: notifications arrive on WASAPI's own
// thread, not this one.
loop {
std::thread::park();
}
});
if let Err(e) = spawned {
WATCH_STARTED.store(false, Ordering::SeqCst);
log::warn!("[wasapi-caps] could not spawn the hotplug watch thread: {e}");
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_cache_key_says_what_the_answer_means() {
// The stored value is "some rung accepts this rate", so the key must
// not name one rung -- and it must change if the ladder grows, or an
// entry written under a shorter ladder would outlive its question.
let key = rate_sweep_key("{0.0.0.x}");
assert!(key.starts_with("{0.0.0.x}|"));
assert!(key.contains(&LADDER.len().to_string()));
assert_ne!(rate_sweep_key("{0.0.0.x}"), rate_sweep_key("{0.0.0.y}"));
}
#[test]
fn unavailable_is_not_an_empty_capability() {
// The distinction the first draft collapsed: a busy endpoint and one
// that genuinely accepts nothing are different answers, and only the
// second is a fact about the hardware.
assert_ne!(Caps::Unavailable, Caps::Known(Vec::new()));
assert_eq!(Caps::Known(vec![44_100]), Caps::Known(vec![44_100]));
}
#[test]
fn a_device_event_expires_every_cached_answer() {
let before = GENERATION.load(Ordering::SeqCst);
invalidate_all();
assert_ne!(GENERATION.load(Ordering::SeqCst), before);
}
}
File diff suppressed because it is too large. Load diff