Output callback is called less frequently than input callback

[rohansatapathy]

I'm working on writing a "feedback" program similar to this example where the samples in the input are directly fed to the output. The major difference between the sample code and my code is that there is no artificial delay, and this is because I'm working on eventually building a real-time MIDI-controlled harmonizer, for which I need to minimize latency as much as possible.

When using a fixed-size ringbuffer to send the samples between each thread, the input callback continuously reports that the output callback fell behind, which would only happen if the ring buffer was running out of space. This shouldn't be the case because I initialized the ring buffer with a size four times the buffer size in the config. When I switch to a MPSC channel (std::sync::mpsc), the issue is resolved, presumably because the channel can grow unbounded. As both the input and the output stream are using the same config (and thus the same buffer size and sample rate), I see no reason why the output stream should be falling behind in delivering samples. I would like to avoid potentially growing my memory usage as the program runs, so is there any way to fix this? Any help would be appreciated, thank you!

Relevant info:

• macOS Ventura 13.4 (CoreAudio)
• Rust 1.70.0

Here's the version of the code with the RingBuf:

use std::{thread, time};

use color_eyre::eyre::Result;
use cpal::{
    traits::{DeviceTrait, HostTrait, StreamTrait},
    StreamConfig,
};
use ringbuf::HeapRb;

fn main() -> Result<()> {
    // Set up error handling
    color_eyre::install()?;

    let host = cpal::default_host();

    let input_dev = host.default_input_device().expect("No input device.");
    let output_dev = host.default_output_device().expect("No output device.");

    let config = StreamConfig {
        channels: 1,
        sample_rate: cpal::SampleRate(44100),
        buffer_size: cpal::BufferSize::Fixed(64),
    };

    let ring = HeapRb::<f32>::new(256);
    let (mut producer, mut consumer) = ring.split();

    let input_callback = move |data: &[f32], _: &cpal::InputCallbackInfo| {
        let mut input_error = false;
        for &sample in data {
            if producer.push(sample).is_err() {
                input_error = true;
            }
        }
        if input_error {
            eprintln!("Output fell behind.");
        }
    };
    let output_callback = move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
        let mut output_error = false;
        for sample in data {
            *sample = match consumer.pop() {
                Some(s) => s,
                None => {
                    output_error = true;
                    0.0
                }
            }
        }
        if output_error {
            eprintln!("Input fell behind.")
        }
    };

    let err_callback = move |err: cpal::StreamError| {
        eprintln!("Error: {}", err);
    };

    let input_stream = input_dev.build_input_stream(
        &config, 
        input_callback, 
        err_callback, 
        None
    )?;
    let output_stream = output_dev.build_output_stream(
        &config,
        output_callback, 
        err_callback, 
        None
    )?;

    input_stream.play()?;
    output_stream.play()?;

    thread::sleep(time::Duration::from_secs(5));

    Ok(())
}

And the version using std::sync::mpsc:

use std::{thread, time};
use std::sync::mpsc;

use color_eyre::eyre::Result;
use cpal::{
    traits::{DeviceTrait, HostTrait, StreamTrait},
    StreamConfig,
};

fn main() -> Result<()> {
    // Set up error handling
    color_eyre::install()?;

    let host = cpal::default_host();

    let input_dev = host.default_input_device().expect("No input device.");
    let output_dev = host.default_output_device().expect("No output device.");

    let config = StreamConfig {
        channels: 1,
        sample_rate: cpal::SampleRate(44100),
        buffer_size: cpal::BufferSize::Fixed(64),
    };

    let (tx, rx) = mpsc::channel();

    let input_callback = move |data: &[f32], _: &cpal::InputCallbackInfo| {
        let mut input_error = false;
        for &sample in data {
            if tx.send(sample).is_err() {
                input_error = true;
            }
        }
        if input_error {
            eprintln!("Input error.");
        }
    };
    let output_callback = move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
        let mut output_error = false;
        for sample in data {
            *sample = match rx.try_recv() {
                Ok(s) => s,
                Err(_e) => {
                    output_error = true;
                    0.0
                }
            }
        }
        if output_error {
            eprintln!("Output error.")
        }
    };

    let err_callback = move |err: cpal::StreamError| {
        eprintln!("Error: {}", err);
    };

    let input_stream = input_dev.build_input_stream(
        &config, 
        input_callback, 
        err_callback, 
        None
    )?;
    let output_stream = output_dev.build_output_stream(
        &config,
        output_callback, 
        err_callback, 
        None
    )?;

    input_stream.play()?;
    output_stream.play()?;

    thread::sleep(time::Duration::from_secs(5));

    Ok(())
}

cargo.toml:

[package]
name = "harmonizer"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
color-eyre = "0.6.2"
cpal = "0.15.2"
ringbuf = "0.3.3"