collective
Browse / github.com/storytold/soundcraft / patch
Playback: strips render on the audio thread itself; count underruns and late callbacks
Playback: strips render on the audio thread itself; count underruns and late callbacks Playback dropped out under CPU load: the realtime callback handed strips to rayon's global pool whenever a session had four or more, then waited for them. The pool's workers run at normal priority, so with the machine busy the callback, a realtime thread, waited on starved workers (priority inversion). Waiting there also breaks AGENTS.md's "the callback never blocks". The global pool is also shared with bounces and video decoding, so offline work could delay callbacks too. Two commits: 1. **Strips render on the audio thread itself.** `run_strips` keeps rayon for offline renders (bounces, `re
- Status
- open
- Score
- 0.00
- Adopters
- 0 other operators
- Author
- forwardaudio (imported from pull request #1074, open)
- Base
541096d2cb5f96a13d5b58935e45711530471480- Size
- +265 −34 in 6 files
- Review
- pending
- Shared
- 2026-10-28
- Id
1d0fd6c02b6e24046a641e1f
How it went
No other operator has reported using it yet.
Diff
raw · apps/soundcraft-cli/src/main.rs apps/soundcraft/src/logging.rs crates/mix/src/lib.rs crates/playback/src/lib.rs crates/ui-egui/src/control.rs docs/control-protocol.md
@@ -458,6 +458,7 @@ fn is_control_method(method: &str) -> bool { | "ui.key" | "ui.text" | "ui.screenshot"+ | "audio.health" | "app.quit" ) }@@ -608,7 +609,7 @@ mod tests { assert_eq!(m, "engine.execute"); let (m, _) = app_request("track.new", json!({})); assert_eq!(m, "engine.execute");- for method in ["session.inspect", "engine.commands", "engine.parity", "ui.screenshot", "app.quit"] {+ for method in ["session.inspect", "engine.commands", "engine.parity", "ui.screenshot", "audio.health", "app.quit"] { assert_eq!(app_request(method, json!({})).0, method); } }@@ -22,8 +22,8 @@ //! callback must never do. Code that runs on the audio thread logs anyway (the cpal error callback, //! the synth's full event queue, a hosted plugin's failed `process`, a CLAP plugin calling the //! host's `log` from its audio thread, the panic hook). So the cpal callbacks mark their thread-//! (`soundcraft_playback::mark_audio_thread`; the mix engine carries the mark into the worker-//! threads that render strips for it), and a record from a marked thread is not written+//! (`soundcraft_playback::mark_audio_thread`; the mix engine renders every strip on the marked+//! thread itself), and a record from a marked thread is not written //! there: the first one is kept in a fixed buffer (`try_lock`, no allocation, never waits) and the //! rest are only counted (an atomic). Another thread writes them out: the UI thread every frame //! ([`AppLogger::report_audio_thread`]) and any other thread before its own next record.@@ -664,7 +664,8 @@ impl MixEngine { let order = std::mem::take(&mut self.order); let live = |t: &Track| !(t.kind == TrackKind::Master || t.inactive || t.kind == TrackKind::Vca); let is_aux = |t: &Track| matches!(t.kind, TrackKind::Aux | TrackKind::Folder);- // Phase 1: independent strips (audio, instrument, MIDI) in parallel.+ // Phase 1: independent strips (audio, instrument, MIDI), in parallel except on the audio+ // thread. let mut work = std::mem::take(&mut self.work); work.clear(); for &i in &order {@@ -1293,42 +1294,29 @@ std::thread_local! { } /// Mark the calling thread as the realtime audio thread (`soundcraft_playback`'s cpal callbacks-/// do, every call: one thread-local store). Strips [`MixEngine::render`] hands to worker threads-/// count as the audio thread while they render for it. Code that might run there asks+/// do, every call: one thread-local store). [`MixEngine::render`] renders every strip on that+/// thread itself instead of handing strips to worker threads. Code that might run there asks /// [`on_audio_thread`] before doing anything that blocks; the desktop app's logger does, so a /// `log::` record from the audio thread never takes a lock or touches a file. pub fn mark_audio_thread() { AUDIO_THREAD.set(true); } -/// True on the thread [`mark_audio_thread`] marked, and on a worker while it renders strips for-/// it. Lock-free and allocation-free.+/// True on the thread [`mark_audio_thread`] marked. Lock-free and allocation-free. pub fn on_audio_thread() -> bool { AUDIO_THREAD.get() } -/// Run `f` on this thread as (or as not) the audio thread, then restore the thread's own mark-/// (also when `f` unwinds): a pool worker renders for realtime playback and offline bounces alike.-#[cfg(not(target_arch = "wasm32"))]-fn as_audio_thread(audio: bool, f: impl FnOnce()) {- struct Restore(bool);- impl Drop for Restore {- fn drop(&mut self) {- AUDIO_THREAD.set(self.0);- }- }- let _restore = Restore(AUDIO_THREAD.replace(audio));- f();-}--/// Process strips, in parallel on native targets when there are enough of them.+/// Process strips: in parallel on native targets when there are enough of them, except on the+/// realtime audio thread. There they render one after the other on the thread itself: the pool's+/// workers run at normal priority, so under load the callback would wait for starved workers+/// (priority inversion) and miss its deadline, and waiting on them blocks, which the callback+/// never may. Offline renders (bounces) keep the parallelism. #[cfg(not(target_arch = "wasm32"))] fn run_strips<T: Send>(work: &mut [T], f: impl Fn(&mut T) + Sync + Send) { use rayon::prelude::*;- if work.len() >= 4 {- // The workers render on behalf of this thread: on the audio thread, they are it too.- let audio = on_audio_thread();- work.par_iter_mut().for_each(|w| as_audio_thread(audio, || f(w)));+ if work.len() >= 4 && !on_audio_thread() {+ work.par_iter_mut().for_each(f); } else { work.iter_mut().for_each(f); }@@ -2449,10 +2437,39 @@ mod audio_thread_tests { .unwrap() } + /// The threads each strip job of `run_strips` ran on, seen from a thread that is (or is not)+ /// the audio thread, and that thread's own id.+ fn strip_threads(audio: bool) -> (Vec<std::thread::ThreadId>, std::thread::ThreadId) {+ std::thread::spawn(move || {+ if audio {+ mark_audio_thread();+ }+ let mut work = vec![None; 64];+ run_strips(&mut work, |w| {+ std::thread::sleep(std::time::Duration::from_millis(1));+ *w = Some(std::thread::current().id());+ });+ (work.into_iter().map(|w| w.unwrap()).collect(), std::thread::current().id())+ })+ .join()+ .unwrap()+ }++ #[test]+ fn strips_for_the_audio_thread_render_on_it_and_bounces_use_the_pool() {+ // The realtime callback must never wait on other threads: pool workers run at normal+ // priority, so under load the callback would wait on starved workers (priority inversion).+ let (threads, caller) = strip_threads(true);+ assert!(threads.iter().all(|t| *t == caller), "every strip renders on the audio thread itself");+ // Offline renders (bounces) keep the parallelism.+ let (threads, caller) = strip_threads(false);+ assert!(threads.iter().any(|t| *t != caller), "a bounce spreads strips over the pool");+ }+ #[test] fn strips_rendered_for_the_audio_thread_count_as_the_audio_thread_and_only_then() {- assert!(seen_by_strips(true).iter().all(|&rt| rt), "a worker rendering for the audio thread is on it");- assert!(seen_by_strips(false).iter().all(|&rt| !rt), "the workers don't stay marked afterwards");+ assert!(seen_by_strips(true).iter().all(|&rt| rt), "strips rendered for the audio thread are on it");+ assert!(seen_by_strips(false).iter().all(|&rt| !rt), "pool workers are never marked"); } } @@ -41,8 +41,8 @@ const BLOCK: usize = 512; /// Room for replaced snapshots on their way back; `update_session` empties it before each send. const OLD_SESSIONS: usize = 16; -/// The realtime audio thread marker lives in the mix engine, which carries it into the strips it-/// renders on worker threads; the cpal callbacks here set it.+/// The realtime audio thread marker lives in the mix engine, which renders every strip on the+/// marked thread itself; the cpal callbacks here set it. pub use soundcraft_mix::{mark_audio_thread, on_audio_thread}; /// Meter values for the UI.@@ -103,6 +103,29 @@ struct Fault { beats: std::sync::atomic::AtomicU64, /// The UI has been told. reported: AtomicBool,+ /// Underruns the device reported, callbacks that overran their buffer period, and the worst+ /// callback's share of its buffer time (‰), since the stream opened or the last+ /// [`Player::take_health`], and the device callbacks in that time.+ xruns: std::sync::atomic::AtomicU64,+ late_callbacks: std::sync::atomic::AtomicU64,+ worst_load: AtomicU32,+ callbacks: std::sync::atomic::AtomicU64,+}++/// How the realtime audio path keeps up, for the UI and for stress tests: since the output+/// opened, or since the last [`Player::take_health`].+#[derive(Debug, Clone, Copy, Default, PartialEq, serde::Serialize)]+pub struct AudioHealth {+ /// Device callbacks (none for the silent clock).+ pub callbacks: u64,+ /// Underruns (dropouts) the device reported.+ pub xruns: u64,+ /// Callbacks that overran their buffer period. Not every one is an audible gap, and the+ /// device may report the same overrun as an xrun too (CoreAudio does).+ pub late_callbacks: u64,+ /// The slowest callback's render time as a share of its buffer period, to 0.001 (1.0 = it+ /// overran).+ pub worst_load: f64, } impl Fault {@@ -119,6 +142,51 @@ impl Fault { }; (!self.reported.swap(true, Ordering::Relaxed)).then_some(trouble) }++ /// A device callback rendered `frames` at `rate` in `took`. Lock-free and allocation-free.+ #[cfg(not(target_arch = "wasm32"))]+ fn note_callback(&self, took: std::time::Duration, frames: usize, rate: f64) {+ self.callbacks.fetch_add(1, Ordering::Relaxed);+ if frames == 0 || rate <= 0.0 {+ return;+ }+ let load = took.as_secs_f64() * rate / frames as f64;+ if load > 1.0 {+ self.late_callbacks.fetch_add(1, Ordering::Relaxed);+ }+ self.worst_load.fetch_max((load * 1000.0).min(f64::from(u32::MAX)) as u32, Ordering::Relaxed);+ }++ /// The stream reported an error (on the audio thread or another).+ fn note_stream_error(&self, e: &cpal::Error) {+ match e.kind() {+ cpal::ErrorKind::DeviceNotAvailable => self.gone.store(true, Ordering::Relaxed),+ cpal::ErrorKind::Xrun => {+ self.xruns.fetch_add(1, Ordering::Relaxed);+ }+ _ => {}+ }+ }++ fn health(&self) -> AudioHealth {+ AudioHealth {+ callbacks: self.callbacks.load(Ordering::Relaxed),+ xruns: self.xruns.load(Ordering::Relaxed),+ late_callbacks: self.late_callbacks.load(Ordering::Relaxed),+ worst_load: f64::from(self.worst_load.load(Ordering::Relaxed)) / 1000.0,+ }+ }++ /// The health so far, counting afresh from here: each counter is swapped for zero, so an+ /// event the callback records meanwhile lands in this reading or the next, never in neither.+ fn take_health(&self) -> AudioHealth {+ AudioHealth {+ callbacks: self.callbacks.swap(0, Ordering::Relaxed),+ xruns: self.xruns.swap(0, Ordering::Relaxed),+ late_callbacks: self.late_callbacks.swap(0, Ordering::Relaxed),+ worst_load: f64::from(self.worst_load.swap(0, Ordering::Relaxed)) / 1000.0,+ }+ } } /// How long a playing output may go without a block before it counts as stalled.@@ -507,6 +575,19 @@ impl Player { self.fault.trouble(stalled) } + /// How the realtime path has kept up since the output opened or the last+ /// [`Player::take_health`]: callbacks, device-reported underruns, late callbacks and the+ /// worst callback load.+ pub fn health(&self) -> AudioHealth {+ self.fault.health()+ }++ /// [`Player::health`] so far, then count afresh (a stress run measures from a take). No+ /// event is lost between the reading and the restart.+ pub fn take_health(&self) -> AudioHealth {+ self.fault.take_health()+ }+ fn side(&self) -> std::sync::MutexGuard<'_, Side> { self.side.lock().unwrap_or_else(PoisonError::into_inner) }@@ -967,10 +1048,14 @@ where let mut scratch = vec![0.0f32; 8192 * frame]; let chunk = scratch.len(); let (failed, guard, gone) = (Arc::clone(fault), Arc::clone(shared), Arc::clone(fault));+ #[cfg(not(target_arch = "wasm32"))]+ let rate = f64::from(config.sample_rate); device.build_output_stream( *config, move |data: &mut [T], _: &cpal::OutputCallbackInfo| { mark_audio_thread();+ #[cfg(not(target_arch = "wasm32"))]+ let started = std::time::Instant::now(); for out in data.chunks_mut(chunk) { let Some(s) = scratch.get_mut(..out.len()) else { return }; guarded_block(&failed, &guard, s, |d| state.fill(d, channels));@@ -978,14 +1063,14 @@ where *o = T::from_sample(*x); } }+ #[cfg(not(target_arch = "wasm32"))]+ failed.note_callback(started.elapsed(), data.len() / frame, rate); }, // cpal may call this on the audio thread (on ALSA in a loop while the device stays // broken): marked, so the logger hands the record off instead of writing it. move |e: cpal::Error| { mark_audio_thread();- if e.kind() == cpal::ErrorKind::DeviceNotAvailable {- gone.gone.store(true, Ordering::Relaxed);- }+ gone.note_stream_error(&e); log::warn!("audio stream error: {e}"); }, None,@@ -1176,6 +1261,27 @@ mod tests { assert_eq!((fault.trouble(false), fault.trouble(true)), (Some(OutputTrouble::Gone), None)); } + #[test]+ fn underruns_and_late_callbacks_are_counted_until_taken() {+ use std::time::Duration;+ let f = Fault::default();+ f.note_stream_error(&cpal::ErrorKind::Xrun.into());+ f.note_stream_error(&cpal::ErrorKind::Xrun.into());+ // 512 frames at 48 kHz last 10.67 ms.+ f.note_callback(Duration::from_millis(4), 512, 48_000.0);+ f.note_callback(Duration::from_millis(16), 512, 48_000.0);+ f.note_callback(Duration::from_millis(2), 512, 48_000.0);+ let h = f.health();+ assert_eq!(h, AudioHealth { callbacks: 3, xruns: 2, late_callbacks: 1, worst_load: 1.5 });+ assert!(!f.gone.load(Ordering::Relaxed), "an underrun is not a lost device");+ assert_eq!(f.take_health(), h);+ assert_eq!(f.health(), AudioHealth::default());+ f.note_callback(Duration::from_millis(2), 512, 48_000.0);+ assert_eq!(f.health().callbacks, 1);+ f.note_stream_error(&cpal::ErrorKind::DeviceNotAvailable.into());+ assert!(f.gone.load(Ordering::Relaxed));+ }+ #[test] fn audio_state_plays_and_stops_at_end() { let s = Arc::new(soundcraft_model::Session::default());@@ -1194,6 +1300,99 @@ mod tests { assert!(!shared.playing.load(Ordering::Relaxed)); } + /// A session like a real mix: `tracks` stereo tracks, each looping 10 s of noise through an+ /// EQ and a compressor.+ fn busy_session(tracks: usize) -> Arc<soundcraft_model::Session> {+ use soundcraft_model::{ChannelFormat, Clip, Insert, SourceAudio, SourceId, TrackKind};+ let mut s = soundcraft_model::Session::default();+ let mut seed = 1u32;+ let mut noise = || {+ seed = seed.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);+ (seed >> 8) as f32 / (1u32 << 24) as f32 - 0.5+ };+ let len = 480_000;+ let x: Vec<f32> = (0..len).map(|_| noise() * 0.5).collect();+ s.pool+ .insert(SourceId(1), Arc::new(SourceAudio::new(soundcraft_audio_io::AudioBuffer { sample_rate: 48_000, channels: vec![x.clone(), x] })));+ for _ in 0..tracks {+ let t = s.add_track(TrackKind::Audio, ChannelFormat::Stereo, None);+ let id = s.new_clip_id();+ let track = s.track_mut(t).unwrap();+ track.playlist_mut().unwrap().clips.push(Clip::audio(id, "noise", SourceId(1), 0, 0, len as i64));+ track.mixer.inserts[0] = Some(Insert::new("eq_7band"));+ track.mixer.inserts[1] = Some(Insert::new("compressor"));+ }+ Arc::new(s)+ }++ /// The realtime path under load, without a device: callbacks come once per buffer period (so+ /// helper threads, if the engine used any, would park between them, as with a real device)+ /// while all cores but one are kept busy. One core stays free, so late blocks come from waiting+ /// on other threads rather than from the load itself (the test thread is not realtime, so other+ /// processes can still preempt it, and a slow machine may lack the throughput). Opt-in, as it measures+ /// time: `cargo test -p soundcraft-playback -- --ignored`.+ #[test]+ #[ignore = "timing stress test, run explicitly"]+ fn no_callback_misses_its_deadline_while_every_core_but_one_is_busy() {+ use std::time::{Duration, Instant};+ const FRAMES: usize = 512;+ const CALLBACKS: usize = 1500;+ let period = Duration::from_secs_f64(FRAMES as f64 / 48_000.0);+ let session = busy_session(16);+ let stop = Arc::new(AtomicBool::new(false));+ let cores = std::thread::available_parallelism().map_or(2, usize::from);+ let load: Vec<_> = (1..cores)+ .map(|_| {+ let stop = Arc::clone(&stop);+ std::thread::spawn(move || {+ let mut x = 0u64;+ while !stop.load(Ordering::Relaxed) {+ x = std::hint::black_box(x.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1));+ }+ })+ })+ .collect();+ let mut times = std::thread::spawn(move || {+ mark_audio_thread();+ let shared = Arc::new(Shared {+ position: AtomicI64::new(0),+ playing: AtomicBool::new(false),+ speed: AtomicU32::new(1000),+ meters: Mutex::new(MeterSnapshot::default()),+ });+ let (tx, rx) = channel();+ let mut st = AudioState::new(rx, shared, session, 48_000.0, channel().0);+ tx.send(Cmd::Play { from: 0, end: None, looped: Some(Range::new(0, 480_000)) }).unwrap();+ let mut out = vec![0.0f32; 2 * FRAMES];+ let mut times = Vec::with_capacity(CALLBACKS);+ let mut next = Instant::now();+ for i in 0..CALLBACKS + 50 {+ let t0 = Instant::now();+ st.fill(&mut out, 2);+ // The first blocks create the plugins and warm the caches.+ if i >= 50 {+ times.push(t0.elapsed());+ }+ next += period;+ std::thread::sleep(next.saturating_duration_since(Instant::now()));+ }+ times+ })+ .join()+ .unwrap();+ stop.store(true, Ordering::Relaxed);+ load.into_iter().for_each(|h| h.join().unwrap());+ times.sort();+ let at = |q: f64| times[((times.len() - 1) as f64 * q) as usize];+ let (p50, p99, max) = (at(0.5), at(0.99), at(1.0));+ let late = times.iter().filter(|t| **t > period).count();+ eprintln!("{cores} cores, period {period:?}: p50 {p50:?}, p99 {p99:?}, max {max:?}, {late} of {CALLBACKS} late");+ // A callback that takes longer than its period is a dropout on a real device. Two of+ // 1500 are allowed: the test thread is not realtime, so another process can preempt it+ // (the waits this test is about made 23 and 30 late).+ assert!(late <= 2, "{late} callbacks missed their deadline, the slowest took {max:?} of a {period:?} period");+ }+ #[test] fn short_device_callbacks_render_short_mix_blocks() { let s = Arc::new(soundcraft_model::Session::default());@@ -124,6 +124,19 @@ fn handle(app: &mut SoundApp, ctx: &egui::Context, req: &ControlRequest) -> Opti "session.inspect" | "document.inspect" => { ok(soundcraft_engine::inspect::session(&app.engine, p.get("detail").and_then(Value::as_str) == Some("full"))) }+ "audio.health" => match &app.player {+ Some(player) => {+ let h = if p.get("reset").and_then(Value::as_bool) == Some(true) { player.take_health() } else { player.health() };+ ok(json!({+ "device": player.device_name,+ "silent": player.silent,+ "sample_rate": player.device_rate,+ "buffer_frames": player.device_buffer,+ "health": h,+ }))+ }+ None => err("no audio output"),+ }, "ui.inspect" => { let mut v = app.inspect(ctx); v["edit_layout"] = json!(app.edit_layout);@@ -44,6 +44,7 @@ Errors: `{"id": 1, "ok": false, "error": "bad parameters for `mix.volume`: no tr | `engine.parity` | `{}` | Parity report against the incumbent's menu catalog. | | `session.inspect` | `{detail?: "summary"|"full"}` | The document: tracks, clips (with times), inserts (+params when full), sends, routing, automation, markers, groups, busses, sources (`loaded` true when decoded audio is in the pool), selection, zoom (`samples_per_px`, `scroll`), transport, undo label. | | `ui.inspect` | `{}` | UI state (`window`, panels, dialogs), window size, playing/position, audio device, and `edit_layout` (timeline rect and every track row's rect, for clicking). |+| `audio.health` | `{reset?: bool}` | How the realtime audio path keeps up since the output opened (or the last reset): `health` holds `callbacks`, `xruns` (underruns the device reported), `late_callbacks` (callbacks that overran their buffer period; not every one is audible, and the device may report the same overrun as an xrun) and `worst_load` (slowest callback as a share of its buffer period, to 0.001; 1.0 = it overran). Also `device`, `silent`, `sample_rate` and `buffer_frames` (`null`: the device default). The first blocks create plugins and may read late, so a stress run resets once playback is under way. `reset: true` returns the reading and counts afresh from it; no event falls between the two. | | `ui.set` | any `UiState` fields | e.g. `{"window": "Mix", "narrow_mix": true}`. | | `ui.menu.list` | `{}` | Every catalog menu path and the command that implements it (or null). | | `ui.menu.invoke` | `{path}` or `{id}` | Like clicking the menu item: may open its dialog (e.g. `"Track > New..."`). |