Joren/qbz
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(audio): keep the WASAPI render thread alive across a stop

Browse Source 
WASAPI Exclusive played nothing at all. The stream opened correctly every
time (Initialized EXCLUSIVE @ 44100 Hz, S24In32, DirectHardware); the render
thread was simply gone by the time audio arrived:

  Stop requested
  Writer thread started (gapless-capable)
  stream closed with 2 underrun(s)
  ERROR Write failed: render thread is gone

The player calls stop() on EVERY track change and then reuses the same stream
object for the next track. ALSA states that contract in its own comment --
"Stop PCM immediately (prepare for next playback)": DROP to halt and discard,
then prepare, and the object survives. Ctl::Stop here did break, so the stream
was single-use and the first track never reached the device.

Ctl::Stop now halts (stop_stream + reset_stream), discards the carry and the
queued PCM, answers any pending drain, and parks waiting for the next track's
first chunk, which it pre-rolls before restarting. The IAudioClient is never
re-Initialized, so the exclusive lock is held across the track change.

Three consequences, each found before shipping:

- Ctl::Shutdown is new, and only Drop sends it. Drop's comment asserted "the
  thread's own loop exits on Stop"; once that stopped being true, join()
  blocked forever at process exit. Measured as a 10-minute hang.
- stop() now blocks on an acknowledgement. Fire-and-forget raced the player,
  which queues the next track's PCM immediately after stop() returns, so the
  purge could eat audio belonging to the track that was starting.
- The halt and pre-roll results are no longer discarded, a Drain arriving
  while stopped completes at once instead of blocking for the deadline,
  drain() reports Err when a stop threw the audio away rather than claiming
  it played, and Resume no longer Starts an unprimed Reset stream.

The example only tested drain-then-write, which is why a stream that could not
play a second track passed it. It now plays three audible one-second beeps --
after playback, after the drain, after the stop -- and judges underruns on
continuous playback only, since the gaps between beeps are underruns the test
creates itself.

Measured against the DAC at 44100 and 192000: three beeps, stop acknowledged
in 109 ms, write after stop OK, 0 underruns, 0 event timeouts, exit 0.
Verified by smoke in the app: one exclusive open at 96000 Hz across two track
changes, zero "render thread is gone", and gapless confirmed audibly.

cc
This commit is contained in:
vicrodh committed 2026-08-27 12:44:20 -06:00
1 parent 09fcdeaa06
commit b52c17ee24
2 files changed
+254 -29
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/qbz-audio/examples/wasapi_play.rs
+75 -13
View File
@@ -52,7 +52,9 @@ fn main() {
println!("realigned: {} timing: {:?}", info.realigned, info.timing);
println!("bit-perfect mode: {:?}", stream.bit_perfect_mode());
println!();
println!("playing {secs}s of 1 kHz at -20 dBFS - CHECK THE DAC DISPLAY, it must read {rate}");
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
@@ -83,6 +85,13 @@ fn main() {
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
@@ -94,27 +103,80 @@ fn main() {
println!();
println!("drain : {drained:?} in {drain_ms} ms");
let mut second = Vec::with_capacity(frames_per_chunk * 2);
for _ in 0..frames_per_chunk {
let v = phase.sin() * amp;
phase += step;
second.push(v);
second.push(v);
}
match stream.write_f32(&second) {
// 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 : {}", stream.underruns());
println!("event timeouts : {}", stream.event_timeouts());
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 stream.underruns() == 0 && stream.event_timeouts() == 0 {
"PASS - no underruns, no timeouts"
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"
}
crates/qbz-audio/src/wasapi_direct.rs
+179 -16
View File
@@ -323,7 +323,9 @@ mod imp {
use super::{aligned_period_hns, Rung, WasapiOpenInfo, WasapiTiming, LADDER};
use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{channel, sync_channel, Receiver, Sender, SyncSender, TryRecvError};
use std::sync::mpsc::{
channel, sync_channel, Receiver, RecvTimeoutError, Sender, SyncSender, TryRecvError,
};
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
@@ -341,8 +343,25 @@ mod imp {
/// return until the queue really has played out, because the player's
/// writer thread treats its return as "the source is finished" - and
/// the ALSA drain it shares that thread with does block.
Drain(Sender<()>),
Stop,
/// `true` = the queue really played out; `false` = a Stop threw it
/// away. The caller must be able to tell those apart.
Drain(Sender<bool>),
/// Halt and discard, then WAIT for the next track -- ALSA's
/// "prepare for next playback". The render thread survives it,
/// because the player reuses one stream across track changes.
/// Carries the acknowledgement channel. `stop()` MUST NOT return
/// before the halt-and-purge has run: the player sends Stop and then
/// immediately queues the next track's PCM on the other channel, so a
/// fire-and-forget Stop lets the purge below eat audio belonging to
/// the track that is starting. Same failure as the original bug, one
/// track later and intermittent.
Stop(Sender<()>),
/// Real teardown: leave the loop and let the thread end. Only `Drop`
/// sends this. Keeping it separate from `Stop` is the whole point --
/// when `Stop` also meant "exit", `Drop` joined a thread that had
/// already gone, and the day `Stop` stopped meaning that, `Drop`
/// joined a thread that never would.
Shutdown,
}
pub(super) struct Inner {
@@ -363,6 +382,10 @@ mod imp {
/// A drain cannot wait forever on a device that stopped consuming.
const DRAIN_DEADLINE: Duration = Duration::from_secs(10);
/// A Stop is acknowledged from the top of the render loop, so the wait is
/// at most one event timeout plus the halt itself. Generous by 10x, and
/// bounded so a dead device cannot wedge a track change.
const STOP_DEADLINE: Duration = Duration::from_secs(2);
fn wave_format(rung: Rung, rate: u32, channels: u16) -> WaveFormat {
let ty = if rung.is_float() {
@@ -606,7 +629,13 @@ mod imp {
let first_deadline = Instant::now() + Duration::from_secs(5);
while carry.is_empty() && Instant::now() < first_deadline {
match ctl.try_recv() {
Ok(Ctl::Stop) | Err(TryRecvError::Disconnected) => return,
Ok(Ctl::Shutdown) | Err(TryRecvError::Disconnected) => return,
// Nothing has started yet, so a Stop has nothing to halt;
// discard whatever arrived and keep waiting for real audio.
Ok(Ctl::Stop(ack)) => {
carry.clear();
let _ = ack.send(());
}
_ => {}
}
if let Ok(chunk) = pcm.recv_timeout(Duration::from_millis(20)) {
@@ -626,7 +655,10 @@ mod imp {
return;
}
let mut paused = false;
let mut draining: Option<(Instant, Sender<()>)> = None;
// Halted by Ctl::Stop and waiting for the next track's first chunk.
// NOT a teardown: see the Ctl::Stop arm below.
let mut stopped = false;
let mut draining: Option<(Instant, Sender<bool>)> = None;
// Starvation is counted as an EVENT, not per period: one real underrun
// otherwise reports as hundreds while the queue stays empty.
let mut starved = false;
@@ -642,15 +674,117 @@ mod imp {
}
Ok(Ctl::Resume) => {
if paused {
let _ = client.start_stream();
paused = false;
// While `stopped` the client is Reset and unprimed,
// and the re-arm block below owns the restart: it
// pre-rolls BEFORE Start. Starting here as well would
// play a buffer of silence and then, on the second
// Start, fail with AUDCLNT_E_NOT_STOPPED.
if !stopped {
let _ = client.start_stream();
}
}
}
Ok(Ctl::Drain(done)) => draining = Some((Instant::now() + DRAIN_DEADLINE, done)),
Ok(Ctl::Stop) | Err(TryRecvError::Disconnected) => break,
Ok(Ctl::Drain(done)) => {
if stopped {
// A halted, Reset stream holds nothing and plays
// nothing, so it has drained by definition. Parking
// this until the NEXT track arrives would block the
// caller for the whole drain deadline.
let _ = done.send(true);
} else {
draining = Some((Instant::now() + DRAIN_DEADLINE, done));
}
}
Ok(Ctl::Stop(ack)) => {
// ALSA's contract, which this mirrors deliberately:
// "Stop PCM immediately (prepare for next playback)" --
// DROP to halt and discard, then prepare, and the stream
// OBJECT survives because the player reuses it for the next
// track. Breaking here instead made the stream single-use:
// the player calls engine.stop() on every track change, so
// the render thread died before the first track's audio had
// even finished buffering and every later write answered
// "render thread is gone". Same shape as the drain defect,
// second instance -- a control that means "get ready" was
// read as "we are done".
// The results are NOT ignored. If the halt half fails the
// stream is not in a state a later Start can trust, and
// reporting it as stopped-and-ready would be the same kind
// of lie this whole fix exists to remove.
let halted = client
.stop_stream()
.map_err(|e| format!("stop_stream: {e}"))
.and_then(|()| {
client
.reset_stream()
.map_err(|e| format!("reset_stream: {e}"))
});
carry.clear();
// Discard what the previous track had already handed over,
// the way DROP discards queued frames. Safe here and ONLY
// here because `stop()` blocks on the ack below: nothing
// written after stop() returns can be in flight yet.
while pcm.try_recv().is_ok() {}
// A drain waiting on this stream will never see its audio
// play out -- it was just thrown away. Say so.
if let Some((_, done)) = draining.take() {
let _ = done.send(false);
}
starved = false;
paused = false;
stopped = true;
// Acknowledge before any bail-out, so a caller blocked in
// stop() is never stranded by a failure.
let _ = ack.send(());
if let Err(e) = halted {
log::error!("[WASAPI Direct] halt failed ({e}); stream is dead");
dead.store(true, Ordering::SeqCst);
break;
}
}
Ok(Ctl::Shutdown) | Err(TryRecvError::Disconnected) => break,
Err(TryRecvError::Empty) => {}
}
// Re-arm: hold the device open, idle, until the next track's first
// chunk arrives, then pre-roll and start again exactly as the
// initial start did. Keeping the client open is the point -- the
// exclusive lock stays ours across the track change.
if stopped && !paused {
match pcm.recv_timeout(Duration::from_millis(20)) {
Ok(chunk) => carry.extend(chunk),
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
}
// Pre-roll BEFORE Start, and only claim the stream is running
// if the pre-roll actually happened: a Start over an unprimed
// exclusive buffer plays a period of silence and then starves.
let primed = client
.get_available_space_in_frames()
.map_err(|e| format!("get_available_space_in_frames: {e}"))
.and_then(|n| {
let need = n as usize * block_align;
let take = need.min(carry.len());
let mut buf: Vec<u8> = carry.drain(..take).collect();
buf.resize(need, 0);
render
.write_to_device(n as usize, &buf, None)
.map_err(|e| format!("write_to_device: {e}"))
});
if let Err(e) = primed {
log::error!("[WASAPI Direct] Re-arm prefill failed ({e}); stream is dead");
dead.store(true, Ordering::SeqCst);
break;
}
if let Err(e) = client.start_stream() {
log::error!("[WASAPI Direct] Restart after stop failed: {e}");
dead.store(true, Ordering::SeqCst);
break;
}
stopped = false;
}
if paused {
std::thread::sleep(Duration::from_millis(period_ms as u64));
continue;
@@ -746,7 +880,7 @@ mod imp {
|| draining.is_some()
{
if let Some((_, done)) = draining.take() {
let _ = done.send(());
let _ = done.send(true);
}
}
continue;
@@ -871,10 +1005,18 @@ mod imp {
pub(super) fn drain_blocking(&self) -> Result<(), String> {
self.check_alive()?;
let (tx, rx) = channel::<()>();
let (tx, rx) = channel::<bool>();
self.send_ctl(Ctl::Drain(tx))?;
match rx.recv_timeout(DRAIN_DEADLINE + Duration::from_secs(2)) {
Ok(()) => Ok(()),
Ok(true) => Ok(()),
Ok(false) => {
// A Stop threw the queue away while this drain waited. The
// audio did NOT play out, and the caller's writer thread
// treats a successful drain as "the source finished", so
// saying Ok here would report a track as played that was
// discarded mid-flight.
Err("drain cancelled: the stream was stopped".to_string())
}
Err(_) => {
// Not an error to the caller: a stuck drain must not wedge
// playback, and the stream is still usable.
@@ -884,6 +1026,23 @@ mod imp {
}
}
/// Halt, discard, and leave the stream ready for the next track --
/// BLOCKING until the render thread confirms it has done so.
///
/// The blocking is the contract, not politeness. The player sends this
/// on a track change and then immediately queues the next track's PCM;
/// if this returned early, the render thread's purge would run after
/// that audio had been handed over and would eat it.
pub(super) fn stop_blocking(&self) -> Result<(), String> {
self.check_alive()?;
let (tx, rx) = channel::<()>();
self.send_ctl(Ctl::Stop(tx))?;
match rx.recv_timeout(STOP_DEADLINE) {
Ok(()) => Ok(()),
Err(_) => Err("stop was not acknowledged by the render thread".to_string()),
}
}
pub(super) fn send_ctl(&self, c: Ctl) -> Result<(), String> {
self.ctl
.send(c)
@@ -893,11 +1052,15 @@ mod imp {
impl Drop for Inner {
fn drop(&mut self) {
let _ = self.ctl.send(Ctl::Stop);
// MUST be Shutdown, not Stop. `Stop` means "prepare for the next
// track" and the thread deliberately survives it -- sending it here
// parked the thread waiting for audio that would never come and
// hung `join()` forever, at process exit, every time.
let _ = self.ctl.send(Ctl::Shutdown);
if let Some(join) = self.join.take() {
// The thread's own loop exits on Stop; joining without a
// deadline is safe because every wait inside it is bounded by
// the period timeout.
// Joining without a deadline is safe because every wait inside
// the loop is bounded: the event wait by `event_timeout_ms`,
// the re-arm wait by 20 ms, both looping back to this check.
let _ = join.join();
}
}
@@ -972,7 +1135,7 @@ impl WasapiDirectStream {
}
pub fn stop(&self) -> Result<(), String> {
self.inner.send_ctl(imp::Ctl::Stop)
self.inner.stop_blocking()
}
pub fn sample_rate(&self) -> u32 {