rill-io 0.6.0-M2

//! ALSA backend for Linux — callback-driven, event-driven via `snd_pcm_wait`.
//!
//! `run()` drives the graph on the audio thread: it waits on the device with
//! `snd_pcm_wait` (event-driven — no `thread::sleep`), and once a period is
//! ready fires the process callback to drive the signal graph.
//!
//! Zero-copy: playback writes directly into the period buffer through
//! `OutputWindow`; capture exposes the just-read period to `read_input` through
//! an input window (deinterleaved on demand). No ring buffers.

const MAX_BLOCK_SAMPLES: usize = 8192;

use std::cell::UnsafeCell;
use std::fmt;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;

use alsa::pcm::{Access, Format, HwParams};
use alsa::{Direction, ValueOr, PCM};

use crate::config::AudioConfig;
use crate::error::{IoError, IoResult};
use rill_core::io::{IoCapture, IoDriver, IoPlayback};
use rill_core::math::functions::{f32_to_i16_chunk, i16_to_f32_chunk};
use rill_core::time::ClockTick;

use crate::output_window::{OutputSlot, OutputWindow};

// ============================================================================
// Callback slot
// ============================================================================

/// Callback slot — stores a process callback via raw pointer for `Send`-safe
/// single-threaded access from the audio thread.
#[derive(Copy, Clone)]
struct CbSlot(usize);

impl CbSlot {
    fn new() -> Self {
        Self(Box::into_raw(Box::new(None::<Box<dyn FnMut(&ClockTick)>>)) as usize)
    }
    unsafe fn set(&self, cb: Box<dyn FnMut(&ClockTick)>) {
        (*(self.0 as *mut Option<Box<dyn FnMut(&ClockTick)>>)) = Some(cb);
    }
    /// Call the stored callback, if any (no-op when unset).
    unsafe fn call(&self, tick: &ClockTick) {
        if let Some(ref mut cb) = *(self.0 as *mut Option<Box<dyn FnMut(&ClockTick)>>) {
            cb(tick);
        }
    }
    unsafe fn take_box(&self) {
        let taken = (*(self.0 as *mut Option<Box<dyn FnMut(&ClockTick)>>)).take();
        drop(taken);
    }
}

// ============================================================================
// Input window — exposes the just-captured period to `read_input`
// ============================================================================

/// Interleaved capture window: `(ptr, channels, frames)`. Set by the audio
/// thread before firing the callbacks and read (deinterleaved) by `read_input`
/// on the same thread.
struct InputWindowSlot(UnsafeCell<Option<(*const f32, usize, usize)>>);

impl InputWindowSlot {
    fn new() -> Self {
        Self(UnsafeCell::new(None))
    }
    unsafe fn set(&self, ptr: *const f32, channels: usize, frames: usize) {
        *self.0.get() = Some((ptr, channels, frames));
    }
    unsafe fn clear(&self) {
        *self.0.get() = None;
    }
    unsafe fn get(&self) -> Option<(*const f32, usize, usize)> {
        *self.0.get()
    }
}

unsafe impl Send for InputWindowSlot {}
unsafe impl Sync for InputWindowSlot {}

// ============================================================================
// AlsaBackend
// ============================================================================

/// ALSA audio backend — callback-driven, event-driven (`snd_pcm_wait`).
///
/// Zero-copy DMA access: per-period f32 buffers live on the audio-thread stack;
/// graph nodes read/write through the input/output windows without ring buffers.
pub struct AlsaBackend {
    config: AudioConfig,
    process_cb: CbSlot,
    output_slot: OutputSlot,
    input_window: InputWindowSlot,
    xruns: Arc<AtomicU32>,
    running: Arc<AtomicBool>,
    sample_pos: Arc<AtomicU64>,
}

impl fmt::Debug for AlsaBackend {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("AlsaBackend")
            .field("config", &self.config)
            .field("running", &self.running.load(Ordering::Relaxed))
            .finish()
    }
}

impl AlsaBackend {
    /// Create a new ALSA backend.
    pub fn new(config: AudioConfig) -> IoResult<Self> {
        Ok(Self {
            config,
            process_cb: CbSlot::new(),
            output_slot: OutputSlot::new(),
            input_window: InputWindowSlot::new(),
            xruns: Arc::new(AtomicU32::new(0)),
            running: Arc::new(AtomicBool::new(false)),
            sample_pos: Arc::new(AtomicU64::new(0)),
        })
    }
}

// ============================================================================
// ALSA I/O loop — called from `run()` on the audio thread
// ============================================================================

#[allow(clippy::too_many_arguments)]
fn alsa_io_loop(
    process_cb: CbSlot,
    output_slot: OutputSlot,
    input_window: &InputWindowSlot,
    xruns: Arc<AtomicU32>,
    sample_pos: Arc<AtomicU64>,
    config: &AudioConfig,
    running: &AtomicBool,
) {
    let out_dev = config.output_device.as_deref().unwrap_or("default");

    let mut negotiated_rate = config.sample_rate as f32;
    let period_frames: usize;
    let pcm_playback: Option<PCM> = if config.output_channels > 0 {
        match PCM::new(out_dev, Direction::Playback, false) {
            Ok(pcm) => {
                match configure_pcm(&pcm, config.output_channels, config) {
                    Ok((rate, period)) => {
                        negotiated_rate = rate as f32;
                        period_frames = period as usize;
                    }
                    Err(e) => {
                        log::error!("ALSA configure playback: {}", e);
                        return;
                    }
                }
                if let Ok(sw) = pcm.sw_params_current() {
                    let _ = sw.set_start_threshold((period_frames * 2) as alsa::pcm::Frames);
                    let _ = pcm.sw_params(&sw);
                }
                Some(pcm)
            }
            Err(e) => {
                log::error!("ALSA open {}: {}", out_dev, e);
                return;
            }
        }
    } else {
        None
    };

    let cap_dev = config.input_device.as_deref().unwrap_or("default");

    let pcm_capture: Option<PCM> = if config.input_channels > 0 {
        match PCM::new(cap_dev, Direction::Capture, true) {
            Ok(pcm) => {
                if let Err(e) = configure_pcm(&pcm, config.input_channels, config) {
                    log::warn!("ALSA configure capture: {}", e);
                }
                let _ = pcm.start();
                Some(pcm)
            }
            Err(e) => {
                log::warn!("ALSA capture {}: disabled", e);
                None
            }
        }
    } else {
        None
    };

    let buf_frames = config.buffer_size as usize;
    let has_playback = pcm_playback.is_some();
    let has_capture = pcm_capture.is_some();
    let out_ch = config.output_channels as usize;
    let in_ch = config.input_channels as usize;
    let chunk_samples = buf_frames * out_ch.max(1);
    let in_sz = (buf_frames * in_ch).clamp(1, MAX_BLOCK_SAMPLES);

    let mut cap_f32 = [0.0f32; MAX_BLOCK_SAMPLES];
    let mut play_f32 = [0.0f32; MAX_BLOCK_SAMPLES];
    let mut i16_buf = [0i16; MAX_BLOCK_SAMPLES];
    let mut cb_i16 = [0i16; MAX_BLOCK_SAMPLES];

    while running.load(Ordering::Acquire) {
        // Wait for device readiness via snd_pcm_wait — one period = one block.
        if has_playback {
            match pcm_playback.as_ref().unwrap().wait(Some(10u32)) {
                Ok(true) => {}
                Ok(false) => continue,
                Err(e) => {
                    let _ = pcm_playback.as_ref().unwrap().try_recover(e, true);
                    xruns.fetch_add(1, Ordering::Relaxed);
                    break;
                }
            }
        } else if has_capture {
            match pcm_capture.as_ref().unwrap().wait(Some(10u32)) {
                Ok(true) => {}
                Ok(false) => continue,
                Err(e) => {
                    let _ = pcm_capture.as_ref().unwrap().try_recover(e, true);
                    xruns.fetch_add(1, Ordering::Relaxed);
                    break;
                }
            }
        }

        if !running.load(Ordering::Acquire) {
            break;
        }

        // ── Capture: read interleaved i16 → f32, publish the input window ──
        let mut cap_frames = 0usize;
        if has_capture {
            let pcm = pcm_capture.as_ref().unwrap();
            if let Ok(io) = pcm.io_i16() {
                if let Ok(n_read) = io.readi(&mut cb_i16[..in_sz]) {
                    let n = (n_read * in_ch).min(in_sz);
                    cap_f32[..n].fill(0.0);
                    i16_to_f32_chunk(&cb_i16[..n], &mut cap_f32[..n]);
                    cap_frames = n_read;
                } else {
                    xruns.fetch_add(1, Ordering::Relaxed);
                }
            }
            unsafe {
                input_window.set(cap_f32.as_ptr(), in_ch.max(1), cap_frames);
            }
        }

        // ── Build the tick and expose the output window ──
        if has_playback {
            play_f32[..chunk_samples].fill(0.0);
            unsafe {
                output_slot.set(OutputWindow::new(play_f32.as_mut_ptr(), chunk_samples));
            }
        }
        let pos = sample_pos.fetch_add(buf_frames as u64, Ordering::Relaxed);
        let mut tick = ClockTick::new(pos, buf_frames as u32, negotiated_rate, "alsa".into());
        let config_rate = config.sample_rate as f64;
        let actual_rate = negotiated_rate as f64;
        tick.speed_ratio = if (config_rate - actual_rate).abs() > 1.0 {
            config_rate / actual_rate
        } else {
            1.0
        };

        // ── Drive the graph: capture chain first, then playback chain ──
        // In the split (duplex) wiring `input_cb` drives the recording roots and
        // `process_cb` drives the playback roots; in the single-callback wiring
        // only `process_cb` is registered and runs the whole graph. Calling both
        // is correct either way — an unset callback is a no-op.
        unsafe {
            process_cb.call(&tick);
            output_slot.clear();
            input_window.clear();
        }

        // ── Playback: convert f32 → i16 → write to PCM ──
        if has_playback {
            let pcm = pcm_playback.as_ref().unwrap();
            let total_samps = chunk_samples;
            f32_to_i16_chunk(&play_f32[..total_samps], &mut i16_buf[..total_samps]);

            let mut retries = 3usize;
            while let Ok(io) = pcm.io_i16() {
                match io.writei(&i16_buf[..total_samps]) {
                    Ok(_) => break,
                    Err(e) => {
                        xruns.fetch_add(1, Ordering::Relaxed);
                        if retries == 0 {
                            break;
                        }
                        retries -= 1;
                        if pcm.try_recover(e, true).is_err() {
                            break;
                        }
                    }
                }
            }
        }
    }

    if let Some(ref pcm) = pcm_playback {
        let _ = pcm.drain();
    }
    if let Some(ref pcm) = pcm_capture {
        let _ = pcm.drain();
    }
}

// ============================================================================
// PCM configuration
// ============================================================================

fn configure_pcm(pcm: &PCM, channels: u32, config: &AudioConfig) -> IoResult<(u32, u32)> {
    let hw = HwParams::any(pcm).map_err(|e| IoError::Config(e.to_string()))?;
    hw.set_access(Access::RWInterleaved)
        .map_err(|e| IoError::Config(e.to_string()))?;
    hw.set_format(Format::s16())
        .map_err(|e| IoError::Config(e.to_string()))?;
    hw.set_rate(config.sample_rate, ValueOr::Nearest)
        .map_err(|e| IoError::Config(e.to_string()))?;
    hw.set_channels(channels)
        .map_err(|e| IoError::Config(e.to_string()))?;
    hw.set_buffer_size(config.buffer_size as alsa::pcm::Frames * 4)
        .map_err(|e| IoError::Config(e.to_string()))?;
    hw.set_period_size(config.buffer_size as alsa::pcm::Frames, ValueOr::Nearest)
        .map_err(|e| IoError::Config(e.to_string()))?;
    let negotiated_rate = hw.get_rate().map_err(|e| IoError::Config(e.to_string()))?;
    let negotiated_period = hw
        .get_period_size()
        .map_err(|e| IoError::Config(e.to_string()))?;
    if negotiated_period != config.buffer_size as alsa::pcm::Frames {
        return Err(IoError::Config(format!(
            "ALSA period mismatch: requested {}, got {}. Use a different backend (portaudio, pipewire, jack).",
            config.buffer_size, negotiated_period
        )));
    }
    pcm.hw_params(&hw)
        .map_err(|e| IoError::Config(e.to_string()))?;
    Ok((negotiated_rate, negotiated_period as u32))
}

// ============================================================================
// IoDriver impl
// ============================================================================

impl IoDriver for AlsaBackend {
    fn set_callback(&self, cb: Box<dyn FnMut(&ClockTick)>) {
        unsafe {
            self.process_cb.set(cb);
        }
    }

    fn run(&self, running: Arc<AtomicBool>) -> Result<(), String> {
        self.running.store(true, Ordering::Release);
        alsa_io_loop(
            self.process_cb,
            self.output_slot.clone(),
            &self.input_window,
            self.xruns.clone(),
            self.sample_pos.clone(),
            &self.config,
            &running,
        );
        Ok(())
    }

    fn stop(&self) -> Result<(), String> {
        self.running.store(false, Ordering::Release);
        Ok(())
    }
}

impl IoPlayback for AlsaBackend {
    fn write_output(&self, channel: usize, src: &[f32]) -> usize {
        unsafe {
            if let Some(window) = self.output_slot.as_mut() {
                let buf = window.as_mut_slice();
                let nch = self.config.output_channels as usize;
                let n_frames = buf.len() / nch.max(1);
                let n = src.len().min(n_frames);
                for i in 0..n {
                    buf[i * nch + channel] = src[i];
                }
                n
            } else {
                0
            }
        }
    }

    fn num_output_channels(&self) -> usize {
        self.config.output_channels as usize
    }
}

impl IoCapture for AlsaBackend {
    fn read_input(&self, channel: usize, dst: &mut [f32]) -> usize {
        unsafe {
            if let Some((ptr, channels, frames)) = self.input_window.get() {
                let n = dst.len().min(frames);
                for (i, d) in dst.iter_mut().enumerate().take(n) {
                    *d = *ptr.add(i * channels + channel);
                }
                if n < dst.len() {
                    dst[n..].fill(0.0);
                }
                n
            } else {
                dst.fill(0.0);
                dst.len()
            }
        }
    }

    fn num_input_channels(&self) -> usize {
        self.config.input_channels as usize
    }
}

impl Drop for AlsaBackend {
    fn drop(&mut self) {
        self.running.store(false, Ordering::Release);
        unsafe {
            self.process_cb.take_box();
        }
    }
}