fix(nodes/audio): audio input outputs partial buffers

Author: maxjoehnk

crates/runtime/pipeline/nodes/audio/src/input.rs

@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
 
 use mizer_node::*;
 
-use crate::AudioContext;
+use crate::{AudioContext, CHANNEL_COUNT, INPUT_BUFFER_SIZE};
 
 const AUDIO_OUTPUT: &str = "Stereo";
 
@@ -41,9 +41,9 @@ impl ProcessingNode for AudioInputNode {
             *state = AudioInputNodeState::new(context).context("Creating audio input state")?;
         }
         if let Some(state) = state {
-            let buffer = state.read(context)?;
-
-            context.write_port(AUDIO_OUTPUT, buffer);
+            if let Some(buffer) = state.read(context)? {
+                context.write_port(AUDIO_OUTPUT, buffer);
+            }
         }
 
         Ok(())
@@ -64,11 +64,12 @@ impl AudioInputNodeState {
     fn new(audio_context: &impl AudioContext) -> anyhow::Result<Option<Self>> {
         tracing::trace!("Opening audio input device");
         if let Some(device) = cpal::default_host().default_input_device() {
-            let buffer = SpscRb::new(audio_context.transfer_size() * 2);
+            let buffer = SpscRb::new(audio_context.transfer_size_per_channel() * CHANNEL_COUNT * INPUT_BUFFER_SIZE);
 
             let config = device.supported_input_configs()?;
             let configs = config.collect::<Vec<_>>();
             tracing::trace!("Supported Input Configs: {configs:?}");
+            // TODO: find device with lowest buffer size
             if let Some(config) = configs.into_iter().find(|c| {
                 c.channels() == 2
                     && c.sample_format() == SampleFormat::F32
@@ -112,14 +113,17 @@ impl AudioInputNodeState {
         }
     }
 
-    fn read(&mut self, audio_context: &impl AudioContext) -> anyhow::Result<Vec<f64>> {
-        let mut buffer = vec![0.; audio_context.transfer_size()];
-        let count = self
-            .buffer
-            .consumer()
-            .read(&mut buffer)
-            .map_err(|err| anyhow::anyhow!("Unable to read from Ringbuffer: {err:?}"))?;
+    fn read(&mut self, audio_context: &impl AudioContext) -> anyhow::Result<Option<Vec<f64>>> {
+        let mut buffer = vec![0.; audio_context.transfer_size_per_channel() * CHANNEL_COUNT];
+        let consumer = self.buffer.consumer();
+        let count = consumer.get(&mut buffer).unwrap_or(0);
+        if count < buffer.len() {
+            return Ok(None);
+        }
+
+        consumer.skip(count)
+            .map_err(|err| anyhow::anyhow!("Unable to skip from Ringbuffer: {err:?}"))?;
 
-        Ok(buffer[0..count].iter().map(|f| *f as f64).collect())
+        Ok(Some(buffer.into_iter().map(|f| f as f64).collect()))
     }
 }

crates/runtime/pipeline/nodes/audio/src/lib.rs

@@ -2,7 +2,7 @@ use std::vec::IntoIter;
 
 use dasp::frame::Stereo;
 use dasp::signal::{from_interleaved_samples_iter, FromInterleavedSamplesIterator};
-use dasp::Signal;
+pub use dasp::Signal;
 
 pub use file::*;
 pub use input::*;
@@ -12,7 +12,13 @@ use mizer_node::{NodeContext, PortId};
 pub use output::*;
 pub use volume::*;
 
+pub(crate) const CHANNEL_COUNT: usize = 2;
 pub(crate) const SAMPLE_RATE: u32 = 44_100;
+const BUFFER_SIZE: usize = 4;
+
+// TODO: decrease buffer size if possible
+pub(crate) const INPUT_BUFFER_SIZE: usize = BUFFER_SIZE;
+pub(crate) const OUTPUT_BUFFER_SIZE: usize = BUFFER_SIZE;
 
 mod file;
 mod input;
@@ -21,10 +27,10 @@ mod mix;
 mod output;
 mod volume;
 
-trait AudioContext {
+pub trait AudioContext {
     type InputSignal: Signal<Frame = Stereo<f64>>;
 
-    fn transfer_size(&self) -> usize;
+    fn transfer_size_per_channel(&self) -> usize;
     fn sample_rate(&self) -> u32 {
         SAMPLE_RATE
     }
@@ -40,7 +46,7 @@ trait AudioContext {
 impl<T: NodeContext> AudioContext for T {
     type InputSignal = FromInterleavedSamplesIterator<IntoIter<f64>, Stereo<f64>>;
 
-    fn transfer_size(&self) -> usize {
+    fn transfer_size_per_channel(&self) -> usize {
         (self.sample_rate() / self.fps().round() as u32) as usize
     }
 
@@ -58,7 +64,7 @@ impl<T: NodeContext> AudioContext for T {
         let buffer: Vec<f64> = signal
             .into_interleaved_samples()
             .into_iter()
-            .take(self.transfer_size() * 2) // Stereo Signal so take twice the sample count
+            .take(self.transfer_size_per_channel() * CHANNEL_COUNT) // Stereo Signal so take twice the sample count
             .collect();
         self.write_port(name, buffer);
     }

crates/runtime/pipeline/nodes/audio/src/meter.rs

@@ -9,8 +9,8 @@ use crate::AudioContext;
 
 const AUDIO_INPUT: &str = "Stereo";
 const VOLUME_OUTPUT: &str = "Volume";
-const VOLUME_L_OUTPUT: &str = "Volume (L)";
-const VOLUME_R_OUTPUT: &str = "Volume (R)";
+const VOLUME_L_OUTPUT: &str = "Volume L";
+const VOLUME_R_OUTPUT: &str = "Volume R";
 
 #[derive(Debug, Default, Clone, Deserialize, Serialize, PartialEq)]
 pub struct AudioMeterNode;
@@ -45,7 +45,7 @@ impl ProcessingNode for AudioMeterNode {
 
     fn process(&self, context: &impl NodeContext, _state: &mut Self::State) -> anyhow::Result<()> {
         if let Some(signal) = context.input_signal(AUDIO_INPUT) {
-            let buffer = Fixed::from(vec![[0.0; 2]; context.transfer_size()]);
+            let buffer = Fixed::from(vec![[0.0; 2]; context.transfer_size_per_channel()]);
             let rms = signal.rms(buffer).until_exhausted().last().unwrap();
             let stereo_rms = rms.into_iter().sum::<f64>() / 2.0;
             context.write_port(VOLUME_OUTPUT, stereo_rms);

crates/runtime/pipeline/nodes/audio/src/output.rs

@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
 
 use mizer_node::*;
 
-use crate::AudioContext;
+use crate::{AudioContext, CHANNEL_COUNT, OUTPUT_BUFFER_SIZE};
 
 const AUDIO_INPUT: &str = "Stereo";
 
@@ -87,7 +87,7 @@ impl AudioOutputNodeState {
     fn new(audio_context: &impl AudioContext) -> anyhow::Result<Option<Self>> {
         tracing::trace!("Opening audio output device");
         if let Some(device) = cpal::default_host().default_output_device() {
-            let buffer = SpscRb::new(audio_context.transfer_size() * 8);
+            let buffer = SpscRb::new(audio_context.transfer_size_per_channel() * CHANNEL_COUNT * OUTPUT_BUFFER_SIZE);
 
             let config = device.supported_output_configs()?;
             let configs = config.collect::<Vec<_>>();
@@ -137,6 +137,7 @@ impl AudioOutputNodeState {
 
     fn write(&self, buffer: Vec<f64>) -> anyhow::Result<()> {
         tracing::trace!("Received {} frames", buffer.len());
+        // TODO: count dropped frames
         self.buffer
             .producer()
             .write(