audio-opus-bsd 0.1.2

//! RT-safety verification — measures **heap allocations == 0** on the
//! `rtrb::Consumer::pop` path (the sole RT entry point of the decode worker).
//!
//! This binary has its own `#[global_allocator]` ([`CountingAllocator`]) so it
//! is isolated from proptest/criterion/logging noise and measures only the RT
//! path's allocations. Warmup (worker construction + packet submission +
//! initial frame drain) is performed outside the measurement window; the
//! measurement itself covers only the steady-state `Consumer::pop` loop.

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};

use audio_opus_bsd::{AudioEncoder, OpusDecodeWorker, OpusEncoder};

/// Global allocator that counts allocations made while the RT path is running.
struct CountingAllocator;

/// Allocation count accumulated during the measurement window.
static ALLOC_COUNT: AtomicUsize = AtomicUsize::new(0);
/// When `1`, `alloc` calls are counted (measurement gate).
static COUNTING_ENABLED: AtomicUsize = AtomicUsize::new(0);

// SAFETY: the delegate `System` implements `GlobalAlloc`, and the counters use
// wait-free atomics.
unsafe impl GlobalAlloc for CountingAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        if COUNTING_ENABLED.load(Ordering::Relaxed) == 1 {
            ALLOC_COUNT.fetch_add(1, Ordering::Relaxed);
        }
        System.alloc(layout)
    }
    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        // dealloc is not counted — only "new allocations" on the RT path matter.
        System.dealloc(ptr, layout);
    }
}

#[global_allocator]
static A: CountingAllocator = CountingAllocator;

/// Returns the number of **allocations** made while `f` runs.
fn count_allocs<F: FnOnce()>(f: F) -> usize {
    ALLOC_COUNT.store(0, Ordering::Relaxed);
    COUNTING_ENABLED.store(1, Ordering::Relaxed);
    f();
    COUNTING_ENABLED.store(0, Ordering::Relaxed);
    ALLOC_COUNT.load(Ordering::Relaxed)
}

/// 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.
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.
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);
    }
    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 rt_consumer_pop_is_alloc_free() {
    // Setup (allocations allowed): build encoder, spawn decode worker,
    // encode packets, submit them, and drain initial frames so the output
    // ring has steady-state data ready for the measurement window.
    let mut enc =
        OpusEncoder::new(SAMPLE_RATE, 1, opus::Application::Audio).expect("encoder constructs");
    enc.set_bitrate_checked(64_000).expect("bitrate");
    let pcm = sine_440_mono();

    let (mut worker, mut rt_cons) =
        OpusDecodeWorker::new(SAMPLE_RATE, 1, 64).expect("decode worker spawns");

    // Submit 40 packets (40 × 20 ms = 800 ms of audio).
    for _ in 0..40 {
        let packet = enc.encode(&pcm).expect("encode");
        assert!(
            worker.submit_packet(packet),
            "submit must succeed while the ring has capacity"
        );
    }

    // Give the worker time to decode all submitted packets.
    std::thread::sleep(std::time::Duration::from_millis(50));

    // Warmup: drain initial frames to bring the worker into steady state.
    // (allocations allowed here)
    let _ = drain_n(&mut rt_cons, 40);

    // ★ Measurement: allocations on the RT path (Consumer::pop) must be 0.
    let allocs = count_allocs(|| {
        for _ in 0..32 {
            // Pop returns Err(Empty) when drained — that is fine; the
            // measurement covers the pop path itself, not the data flow.
            let _ = rt_cons.pop();
        }
    });
    assert_eq!(
        allocs, 0,
        "{allocs} allocations on the RT path — RT-safety violation \
         (alloc must be 0 on rtrb::Consumer::pop)"
    );
}

#[test]
fn construction_allocates_but_isolated_from_rt_path() {
    // Construction allocates (it is not RT-safe). This test confirms that
    // construction itself reports alloc > 0, which guarantees that the RT
    // test's warmup separation is meaningful (i.e. the RT test's 0 is not
    // coincidental — the measurement gate works).
    let allocs = count_allocs(|| {
        let (_worker, _rt_cons) =
            OpusDecodeWorker::new(SAMPLE_RATE, 1, 16).expect("decode worker spawns");
    });
    assert!(
        allocs > 0,
        "construction must allocate (it is not RT-safe) — measurement gate check"
    );
}