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

feat(audio): native JACK backend, functional end to end (#263 Tier 3)

Browse Source 
Completes the JACK output on top of the foundation (fbc03f41):
- StreamType::Jack + the player dispatch (backend==Jack -> JackStream).
- PlaybackEngine::Jack: a feeder thread (mirrors AlsaDirect) that resamples each
  source to the JACK graph rate (UniformSourceIterator, stereo) and writes it to
  a lock-free f32 SPSC ring (ringbuf); wired at the 4 engine-creation sites.
- jack_backend.rs: f32-native ring (fixes a byte-alignment white-noise bug) plus
  auto-connect of qbz:out_FL/FR to the physical playback on activation, so audio
  reaches the hardware without a patchbay (a user may re-patch freely).
- Re-enabled the available_backends push (JACK is now offered in the selector).

Verified on hardware: select JACK -> QBZ registers as client 'qbz' with stable
ports qbz:out_FL/FR (client.api=jack) -> music plays through the DAC, resampled
to the graph rate (NOT bit-perfect; BitPerfectMode::Disabled, as designed). The
ALSA-exclusive / DAC-passthrough bit-perfect paths are untouched.

Slice 3.2 (pending, separate): JACK<->passthrough/exclusive mutual-exclusion UI,
the bit-perfect-OFF note, runtime graph-rate-change callback, mono/multichannel.

cc
This commit is contained in:
vicrodh committed 2026-05-31 14:09:45 -06:00
1 parent fbc03f41b7
commit 68b6cfbcfa
6 files changed
+381 -73
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/Cargo.lock
Generated
+12
View File
@@ -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"
crates/qbz-audio/Cargo.toml
+4
View File
@@ -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"
crates/qbz-audio/src/backend.rs
+5 -5
View File
@@ -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"))]
crates/qbz-audio/src/jack_backend.rs
+76 -68
View File
@@ -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<f32>,
out_l: Port<AudioOut>,
out_r: Port<AudioOut>,
/// 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::<f32>(),
)
};
let got_bytes = self.reader.read_buffer(scratch_bytes);
let got_samples = got_bytes / std::mem::size_of::<f32>();
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<jack::RingBufferWriter>,
producer: Mutex<HeapProd<f32>>,
sample_rate: u32,
channels: u16,
underruns: Arc<AtomicU64>,
}
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<Self, String> {
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::<f32>())
.map_err(|e| format!("JACK ring buffer alloc failed: {e}"))?;
let (reader, writer) = rb.into_reader_writer();
let rb = HeapRb::<f32>::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::<f32>()) / 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).
crates/qbz-player/src/player/mod.rs
+35
View File
@@ -495,6 +495,10 @@ enum StreamType {
},
#[cfg(target_os = "linux")]
AlsaDirect(Arc<qbz_audio::AlsaDirectStream>),
/// 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<qbz_audio::JackStream>),
}
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;
crates/qbz-player/src/player/playback_engine.rs
+249
View File
@@ -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<AtomicBool>,
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<AtomicBool>,
should_stop: Arc<AtomicBool>,
position_frames: Arc<AtomicU64>,
duration_frames: Arc<AtomicU64>,
source_queue: Arc<SourceQueue>,
feeder_thread: Option<thread::JoinHandle<()>>,
source_transition: Arc<AtomicBool>,
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<JackStream>) -> 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<JackStream>,
is_playing: Arc<AtomicBool>,
should_stop: Arc<AtomicBool>,
position_frames: Arc<AtomicU64>,
duration_frames: Arc<AtomicU64>,
source_queue: Arc<SourceQueue>,
source_transition: Arc<AtomicBool>,
) {
const CHUNK_FRAMES: usize = 4096;
const CHANNELS: usize = 2;
let chunk_samples = CHUNK_FRAMES * CHANNELS;
let mut buffer_f32: Vec<f32> = Vec::with_capacity(chunk_samples);
let mut current_source: Option<BoxedSampleIter> = 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();