audio-opus-bsd 0.1.1

//! Worker-thread layer for Opus encode/decode — the RT-safety boundary.
//!
//! [`OpusEncoder`] and
//! [`OpusDecoder`] perform FFI and heap allocation, so
//! they are **not** real-time safe. This module moves them onto a dedicated
//! worker thread and exchanges data with the RT audio thread through
//! lock-free [`rtrb`] ring buffers.
//!
//! ## RT-safety contract
//!
//! Opus FFI is the single largest
//! source of unbounded allocation and syscall latency in the audio path. The
//! **only** correct place for it is a worker thread whose output the RT
//! thread consumes wait-free:
//!
//! ```text
//!   RT thread                      worker thread
//!   ─────────                      ─────────────
//!   submit_packet ───► [rtrb input ring] ──► OpusDecoder::decode_frame
//!                                                    │
//!   Consumer::pop ◄── [rtrb output ring] ◄───────────┘
//! ```
//!
//! The RT thread calls only [`rtrb::Consumer::pop`] (wait-free: no syscalls,
//! no allocation, no locks). All Opus work — including the planar↔interleaved
//! conversion and the libopus encode/decode call — happens on the worker.
//!
//! ## Shutdown
//!
//! [`OpusDecodeWorker::join`] (or [`OpusEncodeWorker::join`]) drops the input
//! [`rtrb::Producer`]. The worker's consumer observes
//! [`is_abandoned`](rtrb::Consumer::is_abandoned) and exits its loop; the
//! [`std::thread::JoinHandle`] then completes.

use crate::decoder::OpusDecoder;
use crate::encoder::OpusEncoder;
use crate::error;
use audio_core_bsd::AudioFrame;
use std::thread::JoinHandle;

/// Idle-poll interval for the worker loop when the input ring is transiently
/// empty (the producer is still alive but has no data ready). Short enough to
/// keep latency low, long enough to avoid a pure busy-spin.
const POLL_IDLE: std::time::Duration = std::time::Duration::from_micros(100);

// ===========================================================================
// OpusDecodeWorker
// ===========================================================================

/// Worker thread that decodes Opus packets into planar [`AudioFrame`]s and
/// hands them to the real-time audio thread through a lock-free `rtrb` ring.
///
/// Owns an [`OpusDecoder`] on a dedicated worker thread. Packet submission
/// (allocation- and FFI-safe) happens via the input ring; decoded frames are
/// consumed wait-free from the returned [`rtrb::Consumer`] on the RT thread.
///
/// See the [module docs](self) for the RT-safety contract.
pub struct OpusDecodeWorker {
    /// `JoinHandle` for the spawned worker thread (`None` after [`join`](Self::join)).
    handle: Option<JoinHandle<()>>,
    /// Input ring producer (packet submission). Dropped by [`join`](Self::join)
    /// to signal the worker to exit.
    input: rtrb::Producer<Vec<u8>>,
}

impl OpusDecodeWorker {
    /// Spawn the decode worker.
    ///
    /// Returns `(worker, rt_consumer)` so the RT thread takes ownership of
    /// the consumer (wait-free [`rtrb::Consumer::pop`] only). The worker
    /// thread is named `"opus-decoder"` for diagnostics.
    ///
    /// `capacity` sizes **both** the input (packet) and output (frame)
    /// rings. See [`OpusDecoder::new`](crate::OpusDecoder::new) for the
    /// valid `sample_rate` / `channels` ranges.
    ///
    /// # Errors
    ///
    /// Returns [`error::OpusError`] if the underlying [`OpusDecoder`] cannot
    /// be constructed (bad sample rate, unsupported channels, or a libopus
    /// internal error) or if the worker thread cannot be spawned. On error
    /// no thread is leaked.
    pub fn new(
        sample_rate: u32,
        channels: u16,
        capacity: usize,
    ) -> error::Result<(Self, rtrb::Consumer<AudioFrame>)> {
        // Pre-build the decoder before spawning so a construction failure is
        // reported to the caller without leaking a thread.
        let mut decoder = OpusDecoder::new(sample_rate, channels)?;
        let (input_prod, mut input_cons) = rtrb::RingBuffer::<Vec<u8>>::new(capacity);
        let (mut output_prod, output_cons) = rtrb::RingBuffer::<AudioFrame>::new(capacity);
        let handle = std::thread::Builder::new()
            .name("opus-decoder".to_string())
            .spawn(move || {
                loop {
                    match input_cons.pop() {
                        Ok(packet) => {
                            if let Ok(frame) = decoder.decode_frame(&packet) {
                                // Best-effort push: if the output ring is full
                                // (RT side not draining), the frame is dropped.
                                if output_prod.push(frame).is_err() {
                                    // Frame dropped — RT consumer will observe
                                    // an underrun.
                                }
                            }
                        }
                        Err(rtrb::PopError::Empty) => {
                            // Producer still alive but no data ready. If the
                            // producer has been dropped, the worker exits.
                            if input_cons.is_abandoned() {
                                break;
                            }
                            std::thread::sleep(POLL_IDLE);
                        }
                    }
                }
            })
            .map_err(|e| error::OpusError::WorkerThread(e.to_string()))?;
        Ok((
            Self {
                handle: Some(handle),
                input: input_prod,
            },
            output_cons,
        ))
    }

    /// Submit an Opus packet for decoding.
    ///
    /// Returns `false` if the input ring is full (the worker is saturated or
    /// the RT side is not draining) — the caller should treat this as an
    /// xrun/backpressure signal. **Never blocks.**
    #[must_use]
    pub fn submit_packet(&mut self, packet: Vec<u8>) -> bool {
        self.input.push(packet).is_ok()
    }

    /// Signal the worker to stop (by dropping the input producer) and join.
    ///
    /// # Errors
    ///
    /// Returns the thread's panic payload if the worker thread panicked.
    pub fn join(mut self) -> std::thread::Result<()> {
        // Take the JoinHandle out so we can join it after signalling.
        let handle = self.handle.take();
        // Drop the input producer FIRST: the worker's consumer observes
        // is_abandoned() and exits its pop loop. Must happen before join to
        // avoid a deadlock (join blocks until the thread exits).
        drop(self.input);
        match handle {
            Some(h) => h.join(),
            None => Ok(()),
        }
    }
}

// ===========================================================================
// OpusEncodeWorker
// ===========================================================================

/// Worker thread that encodes planar [`AudioFrame`]s into Opus packets and
/// hands them to the caller (e.g. a network sender) through a lock-free
/// `rtrb` ring.
///
/// Owns an [`OpusEncoder`] on a dedicated worker thread. Frame submission
/// (allocation-safe) happens via the input ring; encoded packets are consumed
/// wait-free from the returned [`rtrb::Consumer`].
///
/// See the [module docs](self) for the RT-safety contract.
pub struct OpusEncodeWorker {
    /// `JoinHandle` for the spawned worker thread (`None` after [`join`](Self::join)).
    handle: Option<JoinHandle<()>>,
    /// Input ring producer (frame submission).
    input: rtrb::Producer<AudioFrame>,
}

impl OpusEncodeWorker {
    /// Spawn the encode worker.
    ///
    /// `application` selects the Opus application mode
    /// ([`opus::Application::Audio`] / [`opus::Application::Voip`] /
    /// [`opus::Application::LowDelay`]). The encoder is pre-built with a
    /// sane default bitrate of **64 kbps** before spawning (the bitrate
    /// request is best-effort; errors are swallowed). The worker thread is
    /// named `"opus-encoder"` for diagnostics.
    ///
    /// `capacity` sizes both the input (frame) and output (packet) rings.
    ///
    /// # Errors
    ///
    /// Returns [`error::OpusError`] if the underlying [`OpusEncoder`] cannot
    /// be constructed or if the worker thread cannot be spawned.
    pub fn new(
        sample_rate: u32,
        channels: u16,
        application: opus::Application,
        capacity: usize,
    ) -> error::Result<(Self, rtrb::Consumer<Vec<u8>>)> {
        let mut encoder = OpusEncoder::new(sample_rate, channels, application)?;
        // Sane default bitrate; error is intentionally swallowed (the encoder
        // still functions at libopus's default if this call fails).
        let _ = encoder.set_bitrate_checked(64_000);
        let (input_prod, mut input_cons) = rtrb::RingBuffer::<AudioFrame>::new(capacity);
        let (mut output_prod, output_cons) = rtrb::RingBuffer::<Vec<u8>>::new(capacity);
        let handle = std::thread::Builder::new()
            .name("opus-encoder".to_string())
            .spawn(move || {
                loop {
                    match input_cons.pop() {
                        Ok(frame) => {
                            if let Ok(packet) = encoder.encode_frame(&frame) {
                                // Best-effort push: if the output ring is
                                // full (consumer not draining), the packet is
                                // dropped.
                                if output_prod.push(packet).is_err() {
                                    // Packet dropped — caller-side underrun.
                                }
                            }
                        }
                        Err(rtrb::PopError::Empty) => {
                            if input_cons.is_abandoned() {
                                break;
                            }
                            std::thread::sleep(POLL_IDLE);
                        }
                    }
                }
            })
            .map_err(|e| error::OpusError::WorkerThread(e.to_string()))?;
        Ok((
            Self {
                handle: Some(handle),
                input: input_prod,
            },
            output_cons,
        ))
    }

    /// Submit an [`AudioFrame`] for encoding.
    ///
    /// Returns `false` if the input ring is full — treat as backpressure.
    /// **Never blocks.**
    #[must_use]
    pub fn submit_frame(&mut self, frame: AudioFrame) -> bool {
        self.input.push(frame).is_ok()
    }

    /// Signal the worker to stop and join.
    ///
    /// # Errors
    ///
    /// Returns the thread's panic payload if the worker thread panicked.
    pub fn join(mut self) -> std::thread::Result<()> {
        let handle = self.handle.take();
        drop(self.input);
        match handle {
            Some(h) => h.join(),
            None => Ok(()),
        }
    }
}

#[cfg(test)]
#[allow(clippy::cast_precision_loss)] // test-only: loop indices cast to f32
mod tests {
    use super::*;
    use crate::encoder::{AudioEncoder, OpusEncoder};

    /// Canonical Opus block: 960 samples/channel = 20 ms @ 48 kHz.
    const FRAME_SIZE: usize = 960;
    const SAMPLE_RATE: u32 = 48_000;

    /// Cap on the poll loop when collecting frames from the RT consumer.
    /// 2000 × 1 ms = 2 s — far longer than any worker should need for a
    /// handful of packets, short enough to fail fast rather than hang CI.
    const POLL_CAP: usize = 2000;
    const POLL_SLEEP: std::time::Duration = std::time::Duration::from_millis(1);

    /// Generate `FRAME_SIZE` mono samples of a 440 Hz sine at -6 dBFS.
    fn sine_440_mono() -> Vec<f32> {
        let step = 2.0 * std::f32::consts::PI * 440.0 / SAMPLE_RATE as f32;
        (0..FRAME_SIZE)
            .map(|i| 0.5 * (i as f32 * step).sin())
            .collect()
    }

    /// Collect exactly `want` items from a consumer, polling with a short
    /// sleep until they arrive or `POLL_CAP` iterations elapse. Asserts the
    /// exact count was collected — never busy-loops forever.
    fn drain_n<T>(cons: &mut rtrb::Consumer<T>, want: usize) -> Vec<T> {
        let mut out = Vec::with_capacity(want);
        for _ in 0..POLL_CAP {
            while let Ok(item) = cons.pop() {
                out.push(item);
                if out.len() == want {
                    return out;
                }
            }
            std::thread::sleep(POLL_SLEEP);
        }
        // One last drain in case the final batch arrived during the last sleep.
        while let Ok(item) = cons.pop() {
            out.push(item);
        }
        assert_eq!(
            out.len(),
            want,
            "expected {want} items from consumer, got {}",
            out.len()
        );
        out
    }

    #[test]
    fn decode_worker_round_trips_frames() {
        // Encode 8 mono 440 Hz blocks into Opus packets on the calling thread.
        let mut enc = OpusEncoder::new(SAMPLE_RATE, 1, opus::Application::Audio).unwrap();
        let pcm = sine_440_mono();
        let packets: Vec<Vec<u8>> = (0..8).map(|_| enc.encode(&pcm).unwrap()).collect();

        let (mut worker, mut rt_cons) =
            OpusDecodeWorker::new(SAMPLE_RATE, 1, 32).expect("decode worker spawns");
        for pkt in packets {
            assert!(
                worker.submit_packet(pkt),
                "submit must succeed while the ring has capacity"
            );
        }

        let frames = drain_n(&mut rt_cons, 8);
        for f in &frames {
            assert_eq!(f.channels, 1, "decoded mono frame must have 1 channel");
            assert_eq!(
                f.num_frames(),
                FRAME_SIZE,
                "decoded frame must be the canonical 960-sample block"
            );
        }
        worker.join().expect("worker must join cleanly");
    }

    #[test]
    fn decode_worker_submit_returns_false_when_full() {
        // Capacity-2 ring: flood it with a tight push loop. The worker's
        // per-packet FFI decode (even for an invalid packet that libopus
        // rejects instantly) takes ~microseconds; each push takes ~nanoseconds.
        // A 50-packet burst therefore guarantees overflow regardless of how
        // fast the worker drains.
        let (mut worker, _cons) =
            OpusDecodeWorker::new(SAMPLE_RATE, 1, 2).expect("decode worker spawns");
        let pkt = vec![0u8; 8];
        let mut rejected = false;
        for _ in 0..50 {
            if !worker.submit_packet(pkt.clone()) {
                rejected = true;
            }
        }
        assert!(
            rejected,
            "at least one submit on a capacity-2 ring must signal backpressure"
        );
    }

    #[test]
    fn decode_worker_join_terminates() {
        let (mut worker, _cons) =
            OpusDecodeWorker::new(SAMPLE_RATE, 1, 4).expect("decode worker spawns");
        // Submit a couple of valid packets so the worker does real work.
        let mut enc = OpusEncoder::new(SAMPLE_RATE, 1, opus::Application::Audio).unwrap();
        let pkt = enc.encode(&sine_440_mono()).unwrap();
        let _ = worker.submit_packet(pkt);
        let res = worker.join();
        assert!(
            res.is_ok(),
            "join must return Ok(()) for a clean shutdown, got {res:?}"
        );
    }

    #[test]
    fn encode_worker_round_trips_packets() {
        let (mut worker, mut rt_cons) =
            OpusEncodeWorker::new(SAMPLE_RATE, 2, opus::Application::Audio, 16)
                .expect("encode worker spawns");

        for _ in 0..4 {
            let frame = AudioFrame::silence(2, FRAME_SIZE, SAMPLE_RATE);
            assert!(
                worker.submit_frame(frame),
                "submit must succeed while the ring has capacity"
            );
        }

        let packets = drain_n(&mut rt_cons, 4);
        for p in &packets {
            assert!(!p.is_empty(), "encoded packet must not be empty");
        }
        worker.join().expect("encoder must join cleanly");
    }
}