Crossbeam SPSC ring buffers for 60 FPS Rust audio meters

Why a lock-free SPSC ring buffer for audio meters

Audio callbacks cannot block, allocate, or take locks. The UI runs around 60 Hz and can jitter. A bounded single-producer/single-consumer ring fits that split:

• Constant-time, wait-free push/pop with only a couple of atomics
• Bounded memory with no allocation after init
• No syscalls (no parking/unparking), so latency stays flat
• A simple backpressure story (drop when full) that suits meters

Crossbeam has the pieces you want here (CachePadded, Backoff, AtomicCell) with solid atomics. The goal is a compact SPSC ring dedicated to RT audio → UI.

Design choices

• SPSC only: one audio thread produces; one UI thread consumes
• Bounded, power-of-two capacity with index masking for cheap modulo
• Avoid false sharing by padding head/tail with CachePadded
• Copy-only payloads for meter frames, so the RT path never runs Drop
• Drop-on-full on write, and pop_latest on the UI to merge bursts

If you truly need a general T with Drop, add MaybeUninit and do the drop work off the RT thread. For meters, a small Copy struct is the right trade.

Implementation

Cargo.toml (relevant parts):

[dependencies]
crossbeam-utils = "0.8"

SPSC ring buffer (Copy payload):

use core::mem::MaybeUninit;
use core::sync::atomic::{AtomicUsize, Ordering};
use crossbeam_utils::CachePadded;

pub struct SpscRing<T: Copy> {
    buf: Box<[MaybeUninit<T>]>,
    mask: usize,                   // capacity - 1 (power-of-two)
    head: CachePadded<AtomicUsize>,// write index (producer-only increments)
    tail: CachePadded<AtomicUsize>,// read index (consumer-only increments)
}

impl<T: Copy> SpscRing<T> {
    pub fn with_capacity_pow2(capacity: usize) -> Self {
        assert!(capacity >= 2 && capacity.is_power_of_two());
        let buf = {
            let mut v: Vec<MaybeUninit<T>> = Vec::with_capacity(capacity);
            // SAFETY: we’ll manage initialization manually; this is a fixed-size ring
            unsafe { v.set_len(capacity); }
            v.into_boxed_slice()
        };
        Self {
            mask: capacity - 1,
            buf,
            head: CachePadded::new(AtomicUsize::new(0)),
            tail: CachePadded::new(AtomicUsize::new(0)),
        }
    }

    #[inline]
    pub fn capacity(&self) -> usize { self.mask + 1 }

    #[inline]
    pub fn len(&self) -> usize {
        let head = self.head.load(Ordering::Acquire);
        let tail = self.tail.load(Ordering::Acquire);
        head - tail
    }

    // Producer: non-blocking; returns Err(value) if full
    #[inline]
    pub fn try_push(&self, value: T) -> Result<(), T> {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Acquire);
        if head - tail == self.capacity() {
            return Err(value); // drop-on-full policy
        }
        let idx = head & self.mask;
        // SAFETY: single-producer ensures exclusive write to this slot
        unsafe { self.buf.get_unchecked(idx).as_ptr().write(value); }
        // Publish the write
        self.head.store(head.wrapping_add(1), Ordering::Release);
        Ok(())
    }

    // Consumer: non-blocking; None if empty
    #[inline]
    pub fn try_pop(&self) -> Option<T> {
        let tail = self.tail.load(Ordering::Relaxed);
        let head = self.head.load(Ordering::Acquire);
        if tail == head { return None; }
        let idx = tail & self.mask;
        // SAFETY: single-consumer ensures exclusive read from this slot
        let v = unsafe { self.buf.get_unchecked(idx).as_ptr().read() };
        // Publish the read
        self.tail.store(tail.wrapping_add(1), Ordering::Release);
        Some(v)
    }

    // Consumer convenience: drain queue and return the most recent frame
    #[inline]
    pub fn pop_latest(&self) -> Option<T> {
        let mut last = self.try_pop()?;
        while let Some(v) = self.try_pop() { last = v; }
        Some(last)
    }
}

unsafe impl<T: Copy + Send> Send for SpscRing<T> {}
unsafe impl<T: Copy + Send> Sync for SpscRing<T> {}

Notes on memory ordering:

• Producer writes the slot, then Release-stores head. Consumer Acquire-loads head before reading so the write is visible.
• Consumer reads the slot, then Release-stores tail. Producer Acquire-loads tail before checking capacity so it observes consumption.
• CachePadded keeps head and tail on separate cache lines to avoid false sharing and reduce store/load ping-pong.

Meter frame type

Meters are tiny Copy structs:

#[derive(Copy, Clone, Default)]
pub struct MeterFrame {
    pub peak_l: f32,
    pub peak_r: f32,
    pub rms_l: f32,
    pub rms_r: f32,
}

Wiring it into an audio callback and a 60 FPS UI

Use cpal for the audio callback and a simple 60 Hz loop on the UI side. The producer must never block; the consumer coalesces with pop_latest().

use std::sync::Arc;
use std::time::{Duration, Instant};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};

fn main() -> anyhow::Result<()> {
    let ring = Arc::new(SpscRing::<MeterFrame>::with_capacity_pow2(64));

    // Audio setup (input or output callback depending on your app)
    let host = cpal::default_host();
    let device = host.default_input_device().expect("no input device");
    let config = device.default_input_config()?; // e.g., f32 mono/stereo

    let ring_p = ring.clone();
    let stream = match config.sample_format() {
        cpal::SampleFormat::F32 => build_stream::<f32>(&device, &config.into(), ring_p)?,
        cpal::SampleFormat::I16 => build_stream::<i16>(&device, &config.into(), ring_p)?,
        cpal::SampleFormat::U16 => build_stream::<u16>(&device, &config.into(), ring_p)?,
        _ => unimplemented!(),
    };
    stream.play()?;

    // UI loop @ ~60 FPS: pull latest frame, draw meters
    let mut last = MeterFrame::default();
    let frame = Duration::from_millis(16);
    let mut next = Instant::now();
    loop {
        if let Some(latest) = ring.pop_latest() { last = latest; }
        draw_meters(last); // your UI framework of choice
        next += frame;
        let now = Instant::now();
        if next > now { std::thread::sleep(next - now); } else { next = now; }
    }
}

fn build_stream<T>(device: &cpal::Device, config: &cpal::StreamConfig, ring: Arc<SpscRing<MeterFrame>>)
    -> Result<cpal::Stream, cpal::BuildStreamError>
where
    T: cpal::Sample + cpal::FromSample<f32>,
{
    let channels = config.channels as usize;
    device.build_input_stream(
        config,
        move |data: &[T], _| {
            // RT callback: compute per-block meters
            let mut peak_l = 0.0f32; let mut peak_r = 0.0f32;
            let mut sumsq_l = 0.0f32; let mut sumsq_r = 0.0f32; let mut n = 0u32;
            if channels == 1 {
                for &s in data { let x: f32 = s.to_sample();
                    let a = x.abs(); if a > peak_l { peak_l = a; }
                    sumsq_l += x * x; n += 1;
                }
                // mirror mono to R
                peak_r = peak_l; sumsq_r = sumsq_l;
            } else {
                for frame in data.chunks_exact(channels) {
                    let l: f32 = frame[0].to_sample();
                    let r: f32 = frame[1].to_sample();
                    let al = l.abs(); let ar = r.abs();
                    if al > peak_l { peak_l = al; }
                    if ar > peak_r { peak_r = ar; }
                    sumsq_l += l * l; sumsq_r += r * r; n += 1;
                }
            }
            if n > 0 {
                let invn = 1.0 / (n as f32);
                let frame = MeterFrame {
                    peak_l,
                    peak_r,
                    rms_l: (sumsq_l * invn).sqrt(),
                    rms_r: (sumsq_r * invn).sqrt(),
                };
                let _ = ring.try_push(frame); // drop on full: UI will coalesce
            }
        },
        move |err| eprintln!("stream error: {err}"),
        None,
    )
}

fn draw_meters(m: MeterFrame) {
    // Replace with egui, wgpu, iced, etc. Here we just log occasionally.
    static mut CNT: u32 = 0;
    unsafe {
        CNT += 1;
        if CNT % 60 == 0 { println!("peak L/R: {:.3} {:.3} | rms L/R: {:.3} {:.3}", m.peak_l, m.peak_r, m.rms_l, m.rms_r); }
    }
}

This keeps the audio thread allocation-free and lock-free. The UI always paints the freshest meter each frame, even with a 200–1000 Hz callback.

Why not crossbeam channels or ArrayQueue?

• crossbeam-channel (bounded) is great MPMC infrastructure, but it may park/unpark under contention and does extra bookkeeping you don’t need for SPSC. Any parking in a real-time callback is a non-starter.
• crossbeam_queue::ArrayQueue is fast and lock-free for MPMC, but SPSC lets you drop the extra indices/tickets and shave atomics.
• A small SPSC ring gives predictable cache behavior and fewer atomic ops.

If you need something working immediately and can afford a few extra atomics, ArrayQueue is acceptable; still pair it with drop-on-full and pop_latest to keep meters smooth.

Pitfalls and tuning

• Capacity: 32–256 frames is usually enough for meters. Too small drops more often; too large adds latency during UI hiccups.
• Use a power-of-two capacity and mask instead of %. It simplifies the hot path.
• Keep the payload Copy and small (<= 32 bytes) to stay cache-friendly and avoid drops on the RT thread.
• Don’t update tail from the producer: in SPSC only the consumer moves tail. Forcibly advancing tail to overwrite when full is possible, but you must synchronize carefully to avoid racing the consumer.
• Head/tail counter overflow is theoretical on 64-bit (2^64 wraps after centuries at kHz). Use wrapping_add and distance comparisons as shown.

Micro-benchmark sketch

On a modern x86_64 in release, expect roughly 5–20 ns per push+pop for a small Copy payload. To check, use criterion:

// Cargo.toml: criterion = "0.5"
use criterion::{criterion_group, criterion_main, Criterion, black_box};

fn bench_spsc(c: &mut Criterion) {
    let ring = SpscRing::<u64>::with_capacity_pow2(1024);
    c.bench_function("spsc push+pop", |b| {
        b.iter(|| {
            let v = black_box(42u64);
            let _ = ring.try_push(v);
            let _ = ring.try_pop();
        })
    });
}
criterion_group!(benches, bench_spsc);
criterion_main!(benches);

Takeaways

• For RT audio → UI, a tiny SPSC ring is the right tool: bounded, wait-free, no syscalls.
• crossbeam’s CachePadded and atomics make a robust implementation straightforward.
• Drop-on-full with pop_latest keeps meters responsive without growing latency when the UI stalls.
• Avoid general-purpose channels on the RT path; use them for orchestration away from the hot loop.