feat: true gapless playback for ALSA Direct mode

Refactor ALSA Direct engine from thread-per-source to a single
long-lived writer thread with a source queue. When one source ends,
the writer thread immediately picks up the next from the queue with
zero gap. Remove the ALSA Direct gapless rejection gate since the
engine now natively supports source queueing.
This commit is contained in:
vicrodh committed 2026-03-11 19:43:47 -06:00
1 parent 6cac85910a
commit c7f5dcf2f4
2 files changed
+277 -144

No files matched your search

+23 -7
View File
@@ -1903,13 +1903,6 @@ impl Player {
}
};
// Only Rodio supports sink queueing
if engine.is_alsa_direct() {
log::info!("Gapless: ALSA Direct backend, cannot queue — ignoring PlayNext for track {}", track_id);
thread_state.set_gapless_ready(false);
return;
}
// Verify format compatibility (same sample rate and channels)
if let (Some(cur_sr), Some(cur_ch)) = (*current_sample_rate, *current_channels) {
if sample_rate != cur_sr || channels != cur_ch {
@@ -2070,6 +2063,29 @@ impl Player {
}
}
// ALSA Direct gapless: the writer thread signals transitions
// via an atomic flag instead of position-based detection
if let Some(ref engine) = current_engine {
if engine.take_source_transition() {
if let Some(ref pending) = gapless_pending {
log::info!(
"ALSA Direct gapless transition: track {} -> {}",
thread_state.current_track_id.load(Ordering::SeqCst),
pending.track_id
);
thread_state.current_track_id.store(pending.track_id, Ordering::SeqCst);
thread_state.duration.store(pending.duration_secs, Ordering::SeqCst);
thread_state.start_playback_timer(0);
current_audio_data = Some(pending.data.clone());
current_normalization_gain = pending.normalization_gain;
thread_state.set_normalization_gain(pending.normalization_gain);
thread_state.set_gapless_next_track_id(0);
gapless_pending = None;
gapless_request_armed = false;
}
}
}
// Gapless readiness: signal frontend when ~5s remain
// Only when: playing, not streaming, no gapless already pending, gapless enabled
let gapless_enabled = thread_settings
+254 -137
View File
@@ -4,30 +4,84 @@
//! - Rodio (PipeWire, Pulse, ALSA via CPAL) - uses rodio::Sink
//! - ALSA Direct (hw: devices) - bypasses rodio, writes directly to ALSA PCM
//!
//! This abstraction allows the player to work with both approaches transparently.
//! ALSA Direct uses a single long-lived writer thread with a source queue
//! to enable gapless playback. When one source ends, the next is picked up
//! seamlessly without interrupting the PCM stream.
use qbz_audio::AlsaDirectStream;
use rodio::{mixer::Mixer, Player as RodioPlayer, Source};
use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;
/// A boxed sample iterator that can be sent across threads
type BoxedSampleIter = Box<dyn Iterator<Item = f32> + Send>;
/// Thread-safe source queue for gapless playback.
/// The writer thread consumes sources; append() pushes new ones.
pub(crate) struct SourceQueue {
queue: Mutex<VecDeque<BoxedSampleIter>>,
/// Notifies the writer thread that a new source is available
notify: Condvar,
}
impl SourceQueue {
fn new() -> Self {
Self {
queue: Mutex::new(VecDeque::new()),
notify: Condvar::new(),
}
}
/// Push a new source to the back of the queue
fn push(&self, source: BoxedSampleIter) {
let mut q = self.queue.lock().unwrap();
q.push_back(source);
self.notify.notify_one();
}
/// Try to pop the next source (non-blocking)
fn try_pop(&self) -> Option<BoxedSampleIter> {
let mut q = self.queue.lock().unwrap();
q.pop_front()
}
/// Wait for a source to become available (with timeout)
/// Returns None on timeout (used to check stop/pause flags)
fn wait_for_source(&self, timeout: Duration) -> Option<BoxedSampleIter> {
let mut q = self.queue.lock().unwrap();
if q.is_empty() {
let (guard, _) = self.notify.wait_timeout(q, timeout).unwrap();
q = guard;
}
q.pop_front()
}
fn is_empty(&self) -> bool {
self.queue.lock().unwrap().is_empty()
}
}
/// Unified playback engine
pub enum PlaybackEngine {
/// Rodio-based (PipeWire, Pulse, ALSA via CPAL)
Rodio {
sink: RodioPlayer,
},
/// Direct ALSA (hw: devices, bit-perfect)
/// Direct ALSA (hw: devices, bit-perfect) with gapless source queue
AlsaDirect {
stream: Arc<AlsaDirectStream>,
is_playing: Arc<AtomicBool>,
should_stop: Arc<AtomicBool>, // Separate flag for full stop vs pause
should_stop: Arc<AtomicBool>,
position_frames: Arc<AtomicU64>,
duration_frames: Arc<AtomicU64>,
source_queue: Arc<SourceQueue>,
playback_thread: Option<thread::JoinHandle<()>>,
hardware_volume: bool, // Use ALSA mixer for volume control
/// Signals that the writer thread has consumed a source and moved to next
source_transition: Arc<AtomicBool>,
hardware_volume: bool,
},
}
@@ -35,24 +89,54 @@ impl PlaybackEngine {
/// Create Rodio engine
pub fn new_rodio(mixer: &Mixer) -> Result<Self, String> {
let sink = RodioPlayer::connect_new(mixer);
Ok(Self::Rodio { sink })
}
/// Create ALSA Direct engine
/// Create ALSA Direct engine with gapless source queue.
/// Spawns a single writer thread that lives for the engine's lifetime.
pub fn new_alsa_direct(stream: Arc<AlsaDirectStream>, hardware_volume: bool) -> 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));
// Spawn the single long-lived writer thread
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();
let channels = stream.channels();
thread::spawn(move || {
alsa_writer_thread(
stream_c, playing_c, stop_c, pos_c, dur_c,
queue_c, transition_c, channels,
);
})
};
Self::AlsaDirect {
stream,
is_playing: Arc::new(AtomicBool::new(false)),
should_stop: Arc::new(AtomicBool::new(false)),
position_frames: Arc::new(AtomicU64::new(0)),
duration_frames: Arc::new(AtomicU64::new(0)),
playback_thread: None,
is_playing,
should_stop,
position_frames,
duration_frames,
source_queue,
playback_thread: Some(handle),
source_transition,
hardware_volume,
}
}
/// Append audio source
/// Append audio source.
/// For ALSA Direct: pushes to the source queue for gapless transition.
/// For Rodio: delegates to Sink's built-in queue.
pub fn append<S>(&mut self, source: S) -> Result<(), String>
where
S: Source<Item = f32> + Send + 'static,
@@ -63,109 +147,31 @@ impl PlaybackEngine {
Ok(())
}
Self::AlsaDirect {
stream,
is_playing,
should_stop,
position_frames,
duration_frames,
playback_thread,
hardware_volume: _,
source_queue,
source_transition,
..
} => {
// For ALSA Direct, we need to spawn a thread that:
// 1. Streams f32 samples from source (no buffering entire file)
// 2. Writes to ALSA PCM (converts f32 to hardware format)
// 3. Tracks position
// 4. Supports pause/resume without terminating
let is_first = source_queue.is_empty()
&& !is_playing.load(Ordering::SeqCst);
let stream_clone = stream.clone();
let is_playing_clone = is_playing.clone();
let should_stop_clone = should_stop.clone();
let position_clone = position_frames.clone();
let duration_clone = duration_frames.clone();
// Box the source iterator and push to queue
let boxed: BoxedSampleIter = Box::new(source.into_iter());
source_queue.push(boxed);
let channels = stream.channels();
if is_first {
// First source: reset position, clear stop, start playing
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!("[ALSA Direct Engine] First source queued, playback starting");
} else {
log::info!("[ALSA Direct Engine] Source queued for gapless transition");
}
is_playing.store(true, Ordering::SeqCst);
should_stop.store(false, Ordering::SeqCst);
position_clone.store(0, Ordering::SeqCst);
log::info!("[ALSA Direct Engine] Starting streaming playback thread");
let handle = thread::spawn(move || {
// Stream samples in chunks (no pre-buffering entire file)
const CHUNK_SIZE: usize = 8192; // frames per chunk
let chunk_samples = CHUNK_SIZE * channels as usize;
let mut buffer_f32 = Vec::with_capacity(chunk_samples);
let mut total_frames: u64 = 0;
let mut source_iter = source.into_iter();
let mut natural_end = false;
'playback: loop {
// Check if we should stop completely (not just pause)
if should_stop_clone.load(Ordering::SeqCst) {
log::info!("[ALSA Direct Engine] Stop requested, terminating thread");
break 'playback;
}
// Check if paused - wait instead of terminating
while !is_playing_clone.load(Ordering::SeqCst) {
// Still check for stop while paused
if should_stop_clone.load(Ordering::SeqCst) {
log::info!("[ALSA Direct Engine] Stop requested while paused");
break 'playback;
}
// Sleep briefly to avoid busy-waiting
std::thread::sleep(Duration::from_millis(50));
}
// Fill buffer from source
buffer_f32.clear();
for _ in 0..chunk_samples {
match source_iter.next() {
Some(sample) => buffer_f32.push(sample),
None => break, // End of stream
}
}
if buffer_f32.is_empty() {
// End of stream
log::info!("[ALSA Direct Engine] Stream ended (total frames: {})", total_frames);
natural_end = true;
break 'playback;
}
// Write to ALSA (auto-converts based on detected format)
// This is bit-perfect: no resampling, no mixing, direct to hardware
if let Err(e) = stream_clone.write_f32(&buffer_f32) {
log::error!("[ALSA Direct Engine] Write failed: {}", e);
break 'playback;
}
// Update position
let frames_written = buffer_f32.len() / channels as usize;
total_frames += frames_written as u64;
position_clone.store(total_frames, Ordering::SeqCst);
duration_clone.store(total_frames, Ordering::SeqCst);
}
// Only drain if song ended naturally (not skipped/stopped)
// This prevents 2-5s delay when rapidly changing tracks
if natural_end {
log::info!("[ALSA Direct Engine] Song ended naturally, draining buffer");
if let Err(e) = stream_clone.drain() {
log::warn!("[ALSA Direct Engine] Drain failed: {}", e);
}
} else {
log::info!("[ALSA Direct Engine] Playback interrupted, skipping drain for faster response");
}
is_playing_clone.store(false, Ordering::SeqCst);
log::info!("[ALSA Direct Engine] Playback thread finished");
});
*playback_thread = Some(handle);
Ok(())
}
}
@@ -207,16 +213,13 @@ impl PlaybackEngine {
..
} => {
log::info!("[ALSA Direct Engine] Stop requested");
// Signal thread to stop completely
should_stop.store(true, Ordering::SeqCst);
is_playing.store(false, Ordering::SeqCst);
// Wait for playback thread to finish
if let Some(handle) = playback_thread {
let _ = handle.join();
}
// Stop PCM
if let Err(e) = stream.stop() {
log::warn!("[ALSA Direct Engine] Stop failed: {}", e);
}
@@ -230,39 +233,42 @@ impl PlaybackEngine {
Self::Rodio { sink } => sink.set_volume(volume),
Self::AlsaDirect { stream, hardware_volume, .. } => {
if *hardware_volume {
// Try hardware mixer control
#[cfg(target_os = "linux")]
{
if let Err(e) = stream.set_hardware_volume(volume) {
log::warn!("[ALSA Direct Engine] Hardware volume failed: {}. Volume slider may not work.", e);
log::warn!("[ALSA Direct Engine] Hardware volume failed: {}", e);
}
}
} else {
// Hardware volume disabled - volume control is handled by DAC/amplifier
log::debug!("[ALSA Direct Engine] Hardware volume control disabled (use DAC/amplifier)");
}
}
}
}
/// Check if playback queue is empty
/// Check if playback queue is empty (all sources consumed, not playing)
pub fn empty(&self) -> bool {
match self {
Self::Rodio { sink } => sink.empty(),
Self::AlsaDirect {
is_playing,
position_frames,
duration_frames,
source_queue,
..
} => {
if !is_playing.load(Ordering::SeqCst) {
let pos = position_frames.load(Ordering::SeqCst);
let dur = duration_frames.load(Ordering::SeqCst);
// Consider empty if stopped and reached the end
pos >= dur && dur > 0
} else {
false
}
!is_playing.load(Ordering::SeqCst) && source_queue.is_empty()
}
}
}
/// Check if a gapless source transition just happened.
/// Returns true once, then resets the flag.
pub fn take_source_transition(&self) -> bool {
match self {
Self::Rodio { .. } => false,
Self::AlsaDirect { source_transition, .. } => {
source_transition.compare_exchange(
true, false, Ordering::SeqCst, Ordering::SeqCst
).is_ok()
}
}
}
@@ -271,12 +277,8 @@ impl PlaybackEngine {
#[allow(dead_code)]
pub fn position_secs(&self) -> Option<u64> {
match self {
Self::Rodio { .. } => None, // Rodio doesn't expose position directly
Self::AlsaDirect {
position_frames,
stream,
..
} => {
Self::Rodio { .. } => None,
Self::AlsaDirect { position_frames, stream, .. } => {
let frames = position_frames.load(Ordering::SeqCst);
let sample_rate = stream.sample_rate() as u64;
Some(frames / sample_rate)
@@ -289,11 +291,7 @@ impl PlaybackEngine {
pub fn duration_secs(&self) -> Option<u64> {
match self {
Self::Rodio { .. } => None,
Self::AlsaDirect {
duration_frames,
stream,
..
} => {
Self::AlsaDirect { duration_frames, stream, .. } => {
let frames = duration_frames.load(Ordering::SeqCst);
let sample_rate = stream.sample_rate() as u64;
Some(frames / sample_rate)
@@ -307,3 +305,122 @@ impl PlaybackEngine {
matches!(self, Self::AlsaDirect { .. })
}
}
/// Single long-lived writer thread for ALSA Direct.
///
/// Continuously reads samples from the current source and writes to ALSA.
/// When a source ends, seamlessly picks up the next one from the queue
/// (gapless transition). If no next source is available, drains the ALSA
/// buffer and waits for the next source or a stop signal.
fn alsa_writer_thread(
stream: Arc<AlsaDirectStream>,
is_playing: Arc<AtomicBool>,
should_stop: Arc<AtomicBool>,
position_frames: Arc<AtomicU64>,
duration_frames: Arc<AtomicU64>,
source_queue: Arc<SourceQueue>,
source_transition: Arc<AtomicBool>,
channels: u16,
) {
const CHUNK_FRAMES: usize = 8192;
let chunk_samples = CHUNK_FRAMES * channels as usize;
let mut buffer_f32 = Vec::with_capacity(chunk_samples);
let mut current_source: Option<BoxedSampleIter> = None;
let mut total_frames: u64 = 0;
log::info!("[ALSA Direct Engine] Writer thread started (gapless-capable)");
'thread: loop {
// Check global stop
if should_stop.load(Ordering::SeqCst) {
log::info!("[ALSA Direct Engine] Stop signal, writer thread exiting");
break 'thread;
}
// If no current source, try to get one
if current_source.is_none() {
// Wait for a source (with 100ms timeout to recheck stop flag)
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);
log::info!("[ALSA Direct Engine] Acquired new source from queue");
}
None => {
// No source available, loop back to check stop
continue 'thread;
}
}
}
// Wait while paused
while !is_playing.load(Ordering::SeqCst) {
if should_stop.load(Ordering::SeqCst) {
break 'thread;
}
std::thread::sleep(Duration::from_millis(50));
}
// Fill buffer from current source
buffer_f32.clear();
let source = current_source.as_mut().unwrap();
let mut source_ended = false;
for _ in 0..chunk_samples {
match source.next() {
Some(sample) => buffer_f32.push(sample),
None => {
source_ended = true;
break;
}
}
}
// Write whatever we have to ALSA (even partial chunks on source end)
if !buffer_f32.is_empty() {
if let Err(e) = stream.write_f32(&buffer_f32) {
log::error!("[ALSA Direct Engine] Write failed: {}", e);
break 'thread;
}
let frames_written = buffer_f32.len() / channels as usize;
total_frames += frames_written as u64;
position_frames.store(total_frames, Ordering::SeqCst);
duration_frames.store(total_frames, Ordering::SeqCst);
}
if source_ended {
log::info!(
"[ALSA Direct Engine] Source ended (total frames: {})",
total_frames
);
// Try to get next source immediately (gapless transition)
match source_queue.try_pop() {
Some(next_src) => {
log::info!("[ALSA Direct Engine] Gapless transition to next source");
current_source = Some(next_src);
total_frames = 0;
position_frames.store(0, Ordering::SeqCst);
// Signal that a transition happened
source_transition.store(true, Ordering::SeqCst);
// Continue immediately — no drain, no gap
}
None => {
// No next source — this is a natural end of playback
log::info!("[ALSA Direct Engine] No next source, draining ALSA buffer");
if let Err(e) = stream.drain() {
log::warn!("[ALSA Direct Engine] Drain failed: {}", e);
}
current_source = None;
is_playing.store(false, Ordering::SeqCst);
// Don't break — stay alive waiting for next append()
}
}
}
}
is_playing.store(false, Ordering::SeqCst);
log::info!("[ALSA Direct Engine] Writer thread finished");
}