diff --git a/crates/Cargo.lock b/crates/Cargo.lock index b00a064c6..90a046ade 100644 --- a/crates/Cargo.lock +++ b/crates/Cargo.lock @@ -5032,6 +5032,7 @@ dependencies = [ "log", "objc2-core-audio", "qbz-models", + "ringbuf", "rodio", "rusqlite", "rustfft", @@ -5825,6 +5826,17 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "ringbuf" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe47b720588c8702e34b5979cb3271a8b1842c7cb6f57408efa70c779363488c" +dependencies = [ + "crossbeam-utils", + "portable-atomic", + "portable-atomic-util", +] + [[package]] name = "rodio" version = "0.22.2" diff --git a/crates/qbz-audio/Cargo.toml b/crates/qbz-audio/Cargo.toml index 315db8610..c45465b9b 100644 --- a/crates/qbz-audio/Cargo.toml +++ b/crates/qbz-audio/Cargo.toml @@ -36,6 +36,10 @@ rusqlite = { version = "0.31", features = ["bundled"] } # System directories dirs = "6" +# Lock-free SPSC ring buffer (f32-native) for the JACK feeder -> RT callback +# handoff (#263 Tier 3). f32 elements avoid any byte/alignment handling. +ringbuf = "0.4" + [target.'cfg(target_os = "linux")'.dependencies] # ALSA for direct hardware access (bit-perfect) alsa = "0.11" diff --git a/crates/qbz-audio/src/backend.rs b/crates/qbz-audio/src/backend.rs index c4076192c..15b8f7ebc 100644 --- a/crates/qbz-audio/src/backend.rs +++ b/crates/qbz-audio/src/backend.rs @@ -296,11 +296,11 @@ impl BackendManager { backends.push(AudioBackendType::Pulse); } - // JACK (#263 Tier 3): foundation in place (jack_backend.rs + the enum + - // the factory) but NOT yet offered in the selector — the player wiring - // (StreamType::Jack + the dispatch + PlaybackEngine::Jack feeder/resampler) - // is still pending. Re-enable this push once that lands (Tier-3 handoff). - // backends.push(AudioBackendType::Jack); + // JACK (#263 Tier 3): offered now that the player wiring is in place + // (StreamType::Jack + dispatch + PlaybackEngine::Jack feeder/resampler). + // The binary links libjack, so reaching here means it is present; + // opening the client fails gracefully if no JACK server is reachable. + backends.push(AudioBackendType::Jack); } #[cfg(not(target_os = "linux"))] diff --git a/crates/qbz-audio/src/jack_backend.rs b/crates/qbz-audio/src/jack_backend.rs index 0f61bb05b..150b8bb22 100644 --- a/crates/qbz-audio/src/jack_backend.rs +++ b/crates/qbz-audio/src/jack_backend.rs @@ -2,35 +2,38 @@ //! //! QBZ appears as a first-class JACK client (`qbz`) with stable output ports //! `qbz:out_FL` / `qbz:out_FR`, patchable in qjackctl / qpwgraph / Reaper. The -//! routing survives track changes because the client + ports are created ONCE -//! and live for the whole session. +//! client + ports are created ONCE and live for the whole session, so routing +//! survives track changes. On activation the ports are auto-connected to the +//! system's physical playback (so it "just works" without a patchbay); a user +//! may re-patch them freely. //! -//! **NOT bit-perfect.** A JACK graph runs at ONE fixed sample rate, so audio is +//! **NOT bit-perfect.** A JACK graph runs at ONE fixed rate, so audio is //! resampled to the graph rate (by the player's feeder) before it reaches us. -//! This is the documented opt-in trade: routing freedom over per-track -//! bit-perfect. The ALSA-exclusive / DAC-passthrough paths are untouched. +//! Opt-in routing-freedom trade; the bit-perfect ALSA-exclusive / DAC-passthrough +//! paths are untouched. //! -//! Architecture: a lock-free SPSC ring buffer sits between the player's feeder -//! thread (push, via [`JackStream::write_f32`]) and the JACK `process` callback -//! (pop, runs in JACK's real-time thread). The callback never allocates or -//! locks beyond the lock-free ring read. +//! Architecture: a lock-free SPSC ring of **f32 samples** (`ringbuf`) sits +//! between the player's feeder thread (push, via [`JackStream::write_f32`]) and +//! the JACK `process` callback (pop, JACK's RT thread). f32 elements mean no +//! byte/alignment handling; writes/reads are kept to whole stereo frames. -use jack::{AudioOut, Client, ClientOptions, Control, Port, ProcessScope}; +use jack::{AudioOut, Client, ClientOptions, Control, Port, PortFlags, ProcessScope}; +use ringbuf::traits::{Consumer, Observer, Producer, Split}; +use ringbuf::{HeapCons, HeapProd, HeapRb}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; -/// Ring buffer headroom in stereo frames (~1.5 s at 44.1 kHz). Sized generously -/// so the feeder never blocks the audio decode under normal scheduling. +/// Ring capacity in stereo frames (~1.5 s at 44.1 kHz). Generous so the feeder +/// never blocks audio decode under normal scheduling. const RING_CAPACITY_FRAMES: usize = 1 << 16; // 65536 - -/// Max stereo frames a single `process` cycle will ever request; the reusable -/// de-interleave scratch is pre-sized to this so the RT callback never allocates. +/// Max stereo frames a single `process` cycle requests; the reusable scratch is +/// pre-sized to this so the RT callback never allocates. const MAX_NFRAMES: usize = 16384; -/// JACK `process` handler (runs in JACK's RT thread). Pops interleaved stereo -/// f32 from the ring buffer and de-interleaves into the two output ports. +/// JACK `process` handler (RT thread). Pops interleaved stereo f32 from the ring +/// and de-interleaves into the two output ports. Allocation- and lock-free. struct JackProcess { - reader: jack::RingBufferReader, + consumer: HeapCons, out_l: Port, out_r: Port, /// Reusable interleaved scratch (pre-sized; no RT allocation). @@ -41,29 +44,18 @@ struct JackProcess { impl jack::ProcessHandler for JackProcess { fn process(&mut self, _client: &Client, ps: &ProcessScope) -> Control { let nframes = (ps.n_frames() as usize).min(MAX_NFRAMES); - let need_samples = nframes * 2; // stereo interleaved - - // Read interleaved f32 (as bytes) from the lock-free ring buffer into the - // pre-allocated scratch. No allocation: scratch is sized to MAX_NFRAMES*2. - let scratch_bytes: &mut [u8] = unsafe { - std::slice::from_raw_parts_mut( - self.scratch.as_mut_ptr() as *mut u8, - need_samples * std::mem::size_of::(), - ) - }; - let got_bytes = self.reader.read_buffer(scratch_bytes); - let got_samples = got_bytes / std::mem::size_of::(); + let need = nframes * 2; // stereo interleaved + let got = self.consumer.pop_slice(&mut self.scratch[..need]); let l = self.out_l.as_mut_slice(ps); let r = self.out_r.as_mut_slice(ps); for i in 0..nframes { let li = i * 2; let ri = li + 1; - l[i] = if li < got_samples { self.scratch[li] } else { 0.0 }; - r[i] = if ri < got_samples { self.scratch[ri] } else { 0.0 }; + l[i] = if li < got { self.scratch[li] } else { 0.0 }; + r[i] = if ri < got { self.scratch[ri] } else { 0.0 }; } - - if got_samples < need_samples { + if got < need { self.underruns.fetch_add(1, Ordering::Relaxed); } Control::Continue @@ -72,25 +64,21 @@ impl jack::ProcessHandler for JackProcess { /// An active JACK client plus the producer side of the audio ring buffer. /// -/// Mirrors the role of `AlsaDirectStream`: the player feeds interleaved f32 via -/// [`write_f32`](Self::write_f32); a long-lived feeder thread paces the writes. -/// Dropping this deactivates + closes the JACK client (ports disappear). +/// Mirrors `AlsaDirectStream`: the player feeds interleaved stereo f32 via +/// [`write_f32`](Self::write_f32). Dropping this deactivates + closes the JACK +/// client (ports disappear). pub struct JackStream { - /// Activated async client; kept alive for the stream's lifetime. Its `Drop` - /// deactivates the client and unregisters the ports. + /// Activated async client; its `Drop` deactivates + unregisters the ports. _async_client: jack::AsyncClient<(), JackProcess>, - writer: Mutex, + producer: Mutex>, sample_rate: u32, channels: u16, underruns: Arc, } impl JackStream { - /// Open the JACK client, register stable stereo ports, and activate. - /// - /// `channels` is the player's channel count; JACK output is stereo here - /// (FL/FR). The player resamples + downmixes to stereo at the graph rate - /// before feeding, so the ring carries interleaved stereo f32. + /// Open the JACK client, register stable stereo ports, activate, and + /// auto-connect to the system's physical playback. pub fn new(channels: u16) -> Result { let (client, _status) = Client::new("qbz", ClientOptions::NO_START_SERVER) .map_err(|e| format!("JACK client open failed (is a JACK/pipewire-jack server running?): {e}"))?; @@ -104,14 +92,12 @@ impl JackStream { .register_port("out_FR", AudioOut::default()) .map_err(|e| format!("JACK register out_FR failed: {e}"))?; - // Lock-free SPSC ring buffer (bytes). Holds interleaved stereo f32. - let rb = jack::RingBuffer::new(RING_CAPACITY_FRAMES * 2 * std::mem::size_of::()) - .map_err(|e| format!("JACK ring buffer alloc failed: {e}"))?; - let (reader, writer) = rb.into_reader_writer(); + let rb = HeapRb::::new(RING_CAPACITY_FRAMES * 2); + let (producer, consumer) = rb.split(); let underruns = Arc::new(AtomicU64::new(0)); let process = JackProcess { - reader, + consumer, out_l, out_r, scratch: vec![0.0f32; MAX_NFRAMES * 2], @@ -122,6 +108,31 @@ impl JackStream { .activate_async((), process) .map_err(|e| format!("JACK activate failed: {e}"))?; + // Auto-connect qbz:out_FL/FR to the first two physical playback ports so + // audio reaches the hardware without a manual patchbay. A user may + // disconnect + re-patch freely (the ports are stable). + { + let c = async_client.as_client(); + let playback = c.ports(None, None, PortFlags::IS_INPUT | PortFlags::IS_PHYSICAL); + if playback.len() >= 2 { + if let Err(e) = c.connect_ports_by_name("qbz:out_FL", &playback[0]) { + log::warn!("[JACK] auto-connect out_FL -> {} failed: {e}", playback[0]); + } + if let Err(e) = c.connect_ports_by_name("qbz:out_FR", &playback[1]) { + log::warn!("[JACK] auto-connect out_FR -> {} failed: {e}", playback[1]); + } + log::info!( + "[JACK] auto-connected qbz:out_FL/FR -> {} / {}", + playback[0], playback[1] + ); + } else { + log::warn!( + "[JACK] no physical playback ports to auto-connect ({} found); patch qbz:out_FL/FR manually", + playback.len() + ); + } + } + log::info!( "[JACK] client 'qbz' active at {} Hz — ports qbz:out_FL / qbz:out_FR (NOT bit-perfect: resampled to the graph rate)", sample_rate @@ -129,7 +140,7 @@ impl JackStream { Ok(Self { _async_client: async_client, - writer: Mutex::new(writer), + producer: Mutex::new(producer), sample_rate, channels, underruns, @@ -137,8 +148,8 @@ impl JackStream { } /// The JACK graph sample rate. The player resamples each track to this rate - /// before feeding (the graph rate can change at runtime — the player should - /// re-read this on a `buffer_size`/`sample_rate` change; TODO Slice 3.2). + /// before feeding. (Runtime graph-rate changes are a slice-3.2 item; today + /// this is read once at open.) pub fn sample_rate(&self) -> u32 { self.sample_rate } @@ -147,22 +158,19 @@ impl JackStream { self.channels } - /// Push interleaved stereo f32 (already at the graph rate) into the ring - /// buffer. Called from the player's feeder thread. Returns the number of - /// *frames* accepted; fewer than requested means the ring is full and the - /// feeder should retry (it paces itself against real time). + /// Push interleaved stereo f32 (already at the graph rate) into the ring. + /// Called from the player's feeder thread. Returns the number of *frames* + /// accepted (whole stereo frames only); fewer than requested means the ring + /// is full and the feeder should retry (it paces against real time). pub fn write_f32(&self, samples: &[f32]) -> usize { - let bytes: &[u8] = unsafe { - std::slice::from_raw_parts( - samples.as_ptr() as *const u8, - std::mem::size_of_val(samples), - ) - }; - let written = { - let mut w = self.writer.lock().unwrap(); - w.write_buffer(bytes) - }; - (written / std::mem::size_of::()) / 2 + let mut p = self.producer.lock().unwrap(); + // Only push whole stereo frames so the L/R interleave never shifts. + let n = samples.len().min(p.vacant_len()) & !1; + if n == 0 { + return 0; + } + let pushed = p.push_slice(&samples[..n]); + pushed / 2 } /// Total underrun events since open (diagnostic). diff --git a/crates/qbz-player/src/player/mod.rs b/crates/qbz-player/src/player/mod.rs index f6b49ba39..c5bfa3231 100644 --- a/crates/qbz-player/src/player/mod.rs +++ b/crates/qbz-player/src/player/mod.rs @@ -495,6 +495,10 @@ enum StreamType { }, #[cfg(target_os = "linux")] AlsaDirect(Arc), + /// Native JACK output (#263 Tier 3). QBZ as a JACK client with stable ports; + /// NOT bit-perfect (resampled to the graph rate). + #[cfg(target_os = "linux")] + Jack(Arc), } impl StreamType { @@ -851,6 +855,19 @@ fn try_init_stream_with_backend( } } + // JACK (#263 Tier 3): create the JACK client/stream directly (not via the + // MixerDeviceSink trait). Opt-in routing-freedom mode, NOT bit-perfect. + #[cfg(target_os = "linux")] + if backend_type == AudioBackendType::Jack { + match qbz_audio::JackStream::new(config.channels) { + Ok(stream) => { + state.set_bit_perfect_mode(Some(qbz_audio::BitPerfectMode::Disabled)); + return Some(Ok(StreamType::Jack(Arc::new(stream)))); + } + Err(e) => return Some(Err(format!("JACK backend unavailable: {e}"))), + } + } + // Fallback to regular rodio stream (PipeWire, Pulse, ALSA via CPAL) match backend.create_output_stream_with_exclusive_guard(&config) { Ok((mixer_sink, _exclusive_guard)) => { @@ -1881,6 +1898,12 @@ impl Player { hardware_volume, ) } + #[cfg(target_os = "linux")] + StreamType::Jack(jack_stream) => { + *consecutive_sink_failures = 0; + thread_state.set_stream_error(false); + PlaybackEngine::new_jack(jack_stream.clone()) + } }; let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0; @@ -2209,6 +2232,10 @@ impl Player { hardware_volume, ) } + #[cfg(target_os = "linux")] + StreamType::Jack(jack_stream) => { + PlaybackEngine::new_jack(jack_stream.clone()) + } }; let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0; @@ -2507,6 +2534,10 @@ impl Player { hardware_volume, ) } + #[cfg(target_os = "linux")] + StreamType::Jack(jack_stream) => { + PlaybackEngine::new_jack(jack_stream.clone()) + } }; let volume = @@ -2709,6 +2740,10 @@ impl Player { hardware_volume, ) } + #[cfg(target_os = "linux")] + StreamType::Jack(jack_stream) => { + PlaybackEngine::new_jack(jack_stream.clone()) + } }; let volume = thread_state.volume.load(Ordering::SeqCst) as f32 / 100.0; diff --git a/crates/qbz-player/src/player/playback_engine.rs b/crates/qbz-player/src/player/playback_engine.rs index 59ede5d90..f44967d25 100644 --- a/crates/qbz-player/src/player/playback_engine.rs +++ b/crates/qbz-player/src/player/playback_engine.rs @@ -9,6 +9,8 @@ //! seamlessly without interrupting the PCM stream. use qbz_audio::AlsaDirectStream; +#[cfg(target_os = "linux")] +use qbz_audio::JackStream; use rodio::{mixer::Mixer, Player as RodioPlayer, Source}; use std::collections::VecDeque; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; @@ -81,6 +83,21 @@ pub enum PlaybackEngine { source_transition: Arc, hardware_volume: bool, }, + /// Native JACK output (#263 Tier 3). Mirrors AlsaDirect (gapless source queue + /// + a single long-lived feeder thread), but the feeder resamples each source + /// to the JACK graph rate and writes interleaved stereo f32 into the client's + /// lock-free ring buffer via `JackStream::write_f32`. NOT bit-perfect. + #[cfg(target_os = "linux")] + Jack { + is_playing: Arc, + should_stop: Arc, + position_frames: Arc, + duration_frames: Arc, + source_queue: Arc, + feeder_thread: Option>, + source_transition: Arc, + graph_rate: u32, + }, } impl PlaybackEngine { @@ -138,6 +155,46 @@ impl PlaybackEngine { } } + /// Create a JACK engine with a gapless source queue (#263 Tier 3). Spawns one + /// long-lived feeder thread that resamples each source to the JACK graph rate + /// and writes it to the client's ring buffer. + #[cfg(target_os = "linux")] + pub fn new_jack(stream: Arc) -> Self { + let is_playing = Arc::new(AtomicBool::new(false)); + let should_stop = Arc::new(AtomicBool::new(false)); + let position_frames = Arc::new(AtomicU64::new(0)); + let duration_frames = Arc::new(AtomicU64::new(0)); + let source_queue = Arc::new(SourceQueue::new()); + let source_transition = Arc::new(AtomicBool::new(false)); + let graph_rate = stream.sample_rate(); + + let handle = { + let stream_c = stream.clone(); + let playing_c = is_playing.clone(); + let stop_c = should_stop.clone(); + let pos_c = position_frames.clone(); + let dur_c = duration_frames.clone(); + let queue_c = source_queue.clone(); + let transition_c = source_transition.clone(); + thread::spawn(move || { + jack_feeder_thread( + stream_c, playing_c, stop_c, pos_c, dur_c, queue_c, transition_c, + ); + }) + }; + + Self::Jack { + is_playing, + should_stop, + position_frames, + duration_frames, + source_queue, + feeder_thread: Some(handle), + source_transition, + graph_rate, + } + } + /// Append audio source. /// For ALSA Direct: pushes to the source queue for gapless transition. /// For Rodio: delegates to Sink's built-in queue. @@ -177,6 +234,37 @@ impl PlaybackEngine { Ok(()) } + #[cfg(target_os = "linux")] + Self::Jack { + is_playing, + should_stop, + position_frames, + source_queue, + source_transition, + graph_rate, + .. + } => { + let is_first = source_queue.is_empty() && !is_playing.load(Ordering::SeqCst); + // Resample the track-native source to the JACK graph rate (stereo) so + // the feeder/ring always carry graph-rate interleaved stereo f32. + let resampled = rodio::source::UniformSourceIterator::new( + source, + std::num::NonZero::new(2u16).unwrap(), + std::num::NonZero::new(*graph_rate).unwrap(), + ); + let boxed: BoxedSampleIter = Box::new(resampled); + source_queue.push(boxed); + if is_first { + position_frames.store(0, Ordering::SeqCst); + should_stop.store(false, Ordering::SeqCst); + source_transition.store(false, Ordering::SeqCst); + is_playing.store(true, Ordering::SeqCst); + log::info!("[JACK Engine] First source queued, playback starting"); + } else { + log::info!("[JACK Engine] Source queued for gapless transition"); + } + Ok(()) + } } } @@ -188,6 +276,11 @@ impl PlaybackEngine { log::info!("[ALSA Direct Engine] Resume requested"); is_playing.store(true, Ordering::SeqCst); } + #[cfg(target_os = "linux")] + Self::Jack { is_playing, .. } => { + log::info!("[JACK Engine] Resume requested"); + is_playing.store(true, Ordering::SeqCst); + } } } @@ -199,6 +292,11 @@ impl PlaybackEngine { log::info!("[ALSA Direct Engine] Pause requested"); is_playing.store(false, Ordering::SeqCst); } + #[cfg(target_os = "linux")] + Self::Jack { is_playing, .. } => { + log::info!("[JACK Engine] Pause requested"); + is_playing.store(false, Ordering::SeqCst); + } } } @@ -237,6 +335,24 @@ impl PlaybackEngine { log::warn!("[ALSA Direct Engine] Stop failed: {}", e); } } + #[cfg(target_os = "linux")] + Self::Jack { + is_playing, + should_stop, + feeder_thread, + .. + } => { + if should_stop.load(Ordering::SeqCst) { + return; + } + log::info!("[JACK Engine] Stop requested"); + should_stop.store(true, Ordering::SeqCst); + is_playing.store(false, Ordering::SeqCst); + if let Some(handle) = feeder_thread.take() { + let _ = handle.join(); + } + // JackStream's Drop deactivates the client + unregisters the ports. + } } } @@ -262,6 +378,12 @@ impl PlaybackEngine { ); } } + #[cfg(target_os = "linux")] + Self::Jack { .. } => { + // JACK output volume is controlled in the JACK graph / DAW; the + // feeder writes unattenuated f32. (Software volume could later be + // applied by scaling in the feeder.) + } } } @@ -274,6 +396,12 @@ impl PlaybackEngine { source_queue, .. } => !is_playing.load(Ordering::SeqCst) && source_queue.is_empty(), + #[cfg(target_os = "linux")] + Self::Jack { + is_playing, + source_queue, + .. + } => !is_playing.load(Ordering::SeqCst) && source_queue.is_empty(), } } @@ -287,6 +415,12 @@ impl PlaybackEngine { } => source_transition .compare_exchange(true, false, Ordering::SeqCst, Ordering::SeqCst) .is_ok(), + #[cfg(target_os = "linux")] + Self::Jack { + source_transition, .. + } => source_transition + .compare_exchange(true, false, Ordering::SeqCst, Ordering::SeqCst) + .is_ok(), } } @@ -304,6 +438,15 @@ impl PlaybackEngine { let sample_rate = stream.sample_rate() as u64; Some(frames / sample_rate) } + #[cfg(target_os = "linux")] + Self::Jack { + position_frames, + graph_rate, + .. + } => { + let frames = position_frames.load(Ordering::SeqCst); + Some(frames / (*graph_rate as u64).max(1)) + } } } @@ -321,6 +464,15 @@ impl PlaybackEngine { let sample_rate = stream.sample_rate() as u64; Some(frames / sample_rate) } + #[cfg(target_os = "linux")] + Self::Jack { + duration_frames, + graph_rate, + .. + } => { + let frames = duration_frames.load(Ordering::SeqCst); + Some(frames / (*graph_rate as u64).max(1)) + } } } @@ -450,6 +602,103 @@ fn alsa_writer_thread( log::info!("[ALSA Direct Engine] Writer thread finished"); } +/// Single long-lived feeder thread for JACK (#263 Tier 3). +/// +/// Mirrors `alsa_writer_thread`, but writes graph-rate interleaved STEREO f32 +/// into the JACK client's lock-free ring buffer via `JackStream::write_f32` +/// (the RT process callback drains it), pacing itself when the ring is full. +/// Sources are resampled to the graph rate + stereo at `append` time. +#[cfg(target_os = "linux")] +fn jack_feeder_thread( + stream: Arc, + is_playing: Arc, + should_stop: Arc, + position_frames: Arc, + duration_frames: Arc, + source_queue: Arc, + source_transition: Arc, +) { + const CHUNK_FRAMES: usize = 4096; + const CHANNELS: usize = 2; + let chunk_samples = CHUNK_FRAMES * CHANNELS; + let mut buffer_f32: Vec = Vec::with_capacity(chunk_samples); + let mut current_source: Option = None; + let mut total_frames: u64 = 0; + + log::info!("[JACK Engine] Feeder thread started"); + + 'thread: loop { + if should_stop.load(Ordering::SeqCst) { + break 'thread; + } + if current_source.is_none() { + match source_queue.wait_for_source(Duration::from_millis(100)) { + Some(src) => { + current_source = Some(src); + total_frames = 0; + position_frames.store(0, Ordering::SeqCst); + } + None => continue 'thread, + } + } + while !is_playing.load(Ordering::SeqCst) { + if should_stop.load(Ordering::SeqCst) { + break 'thread; + } + std::thread::sleep(Duration::from_millis(50)); + } + + buffer_f32.clear(); + let source = current_source.as_mut().unwrap(); + let mut source_ended = false; + for _ in 0..chunk_samples { + match source.next() { + Some(s) => buffer_f32.push(s), + None => { + source_ended = true; + break; + } + } + } + + // Write to the ring, paced: write_f32 returns frames accepted; a full + // ring returns fewer (or 0) and we wait for the RT callback to drain. + let mut off_samples = 0usize; + while off_samples < buffer_f32.len() { + if should_stop.load(Ordering::SeqCst) { + break 'thread; + } + let frames = stream.write_f32(&buffer_f32[off_samples..]); + if frames == 0 { + std::thread::sleep(Duration::from_millis(2)); + continue; + } + off_samples += frames * CHANNELS; + total_frames += frames as u64; + position_frames.store(total_frames, Ordering::SeqCst); + duration_frames.store(total_frames, Ordering::SeqCst); + } + + if source_ended { + match source_queue.try_pop() { + Some(next_src) => { + current_source = Some(next_src); + total_frames = 0; + position_frames.store(0, Ordering::SeqCst); + source_transition.store(true, Ordering::SeqCst); + } + None => { + current_source = None; + is_playing.store(false, Ordering::SeqCst); + } + } + } + } + + is_playing.store(false, Ordering::SeqCst); + log::info!("[JACK Engine] Feeder thread finished"); +} + impl Drop for PlaybackEngine { fn drop(&mut self) { self.stop_inner();