From 1bbe820853c759d0f724ef7f9cb56e4c1414cdf3 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Wed, 26 Aug 2026 23:51:56 +0200 Subject: [PATCH] Using ring buffer --- src-tauri/Cargo.lock | 12 +++++ src-tauri/Cargo.toml | 1 + src-tauri/src/commands/audio.rs | 84 ++++++++++++++++++++++----------- 3 files changed, 70 insertions(+), 27 deletions(-) diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 29c7181..e5acfe6 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -963,6 +963,7 @@ version = "0.1.0" dependencies = [ "cpal", "resampler", + "ringbuf", "serde", "serde_json", "tauri", @@ -2998,6 +2999,17 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28fdbea87ff02ebbfd904079d1e52138c0c7fbaa3aaddca4a1b9c7d3f85749f2" +[[package]] +name = "ringbuf" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a158e09ede21a14b172ca6cdd6208386c6ae2cb6acef58d774368ef8c450dfa7" +dependencies = [ + "crossbeam-utils", + "portable-atomic", + "portable-atomic-util", +] + [[package]] name = "rustc-hash" version = "2.1.3" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index c24ab14..8b27ac7 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -24,4 +24,5 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" cpal = "0.18.2" resampler = "0.5.1" +ringbuf = "0.5.1" diff --git a/src-tauri/src/commands/audio.rs b/src-tauri/src/commands/audio.rs index 62367da..a07d009 100644 --- a/src-tauri/src/commands/audio.rs +++ b/src-tauri/src/commands/audio.rs @@ -10,6 +10,10 @@ use std::{ }; use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; +use ringbuf::{ + traits::{Consumer, Producer, Split}, + HeapRb, +}; use tauri::State; use crate::commands::config::ConfigState; @@ -159,10 +163,17 @@ pub fn connect_to_vc( let shutdown = Arc::new(AtomicBool::new(false)); // ------------------------------------------------------------------------ - // PLAYBACK BUFFER + // PLAYBACK BUFFER (RingBuffer) // ------------------------------------------------------------------------ - let playback_buffer = Arc::new(Mutex::new(std::collections::VecDeque::::new())); + // Buffer 2 seconds of audio max (prevents infinite latency growth) + let max_capacity = 48000; + let rb = HeapRb::::new(max_capacity); + let (producer, consumer) = rb.split(); + + // Split locks: UDP thread and Audio thread will never block each other + let producer_lock = Arc::new(Mutex::new(producer)); + let consumer_lock = Arc::new(Mutex::new(consumer)); // ------------------------------------------------------------------------ // OUTPUT @@ -176,7 +187,7 @@ pub fn connect_to_vc( let output_stream = build_output_stream( &output_device, &output_config.lock().unwrap(), - playback_buffer.clone(), + consumer_lock.clone(), needs_output_rebuild.clone(), )?; @@ -267,7 +278,7 @@ pub fn connect_to_vc( let output_device = output_device.clone(); let output_stream = output_stream.clone(); let output_config = output_config.clone(); - let playback_buffer = playback_buffer.clone(); + let consumer_lock = consumer_lock.clone(); let needs_rebuild = needs_output_rebuild.clone(); let shutdown = shutdown.clone(); @@ -287,7 +298,7 @@ pub fn connect_to_vc( let Ok(new_stream) = build_output_stream( &output_device, &new_config, - playback_buffer.clone(), + consumer_lock.clone(), needs_rebuild.clone(), ) else { needs_rebuild.store(true, Ordering::SeqCst); @@ -299,8 +310,6 @@ pub fn connect_to_vc( continue; } - playback_buffer.lock().unwrap().clear(); - *output_config.lock().unwrap() = new_config; *output_stream.lock().unwrap() = new_stream; } @@ -313,7 +322,6 @@ pub fn connect_to_vc( { let socket = socket.clone(); - let playback_buffer = playback_buffer.clone(); let output_config = output_config.clone(); let shutdown = shutdown.clone(); @@ -388,7 +396,6 @@ pub fn connect_to_vc( packets.clear(); resample_buffer.clear(); - playback_buffer.lock().unwrap().clear(); expected = None; } @@ -445,20 +452,16 @@ pub fn connect_to_vc( let channels = config.channels.max(1) as usize; - let mut queue = playback_buffer.lock().unwrap(); + let mut prod = producer_lock.lock().unwrap(); for sample in output { for _ in 0..channels { - queue.push_back(sample); + // try_push will return an error if the buffer is full (2 seconds). + // By ignoring the error, we simply drop the oldest overflow packets, + // which naturally prevents latency from growing forever. + let _ = prod.try_push(sample); } } - - // Don't let latency grow forever. - let max_samples = config.sample_rate as usize * channels / 5; - - while queue.len() > max_samples { - queue.pop_front(); - } } } }); @@ -520,30 +523,57 @@ fn create_output_resampler( // OUTPUT STREAM // ============================================================================ -fn build_output_stream( +fn build_output_stream( device: &cpal::Device, config: &cpal::StreamConfig, - playback_buffer: Arc>>, + consumer_lock: Arc>, needs_rebuild: Arc, -) -> Result { +) -> Result +where + C: Consumer + Send + 'static, +{ + // Jitter Buffer: wait for ~60ms of audio before starting playback + let channels = config.channels as usize; + let sample_rate = config.sample_rate as usize; + let jitter_cushion_samples = (sample_rate / 1000) * 60 * channels; + + let mut is_buffering = true; + device .build_output_stream( config.clone(), move |data: &mut [f32], _| { - let mut queue = playback_buffer.lock().unwrap(); + // The audio thread only locks the consumer side. + // Contention is zero unless the stream is actively crashing. + let mut cons = consumer_lock.lock().unwrap(); - for sample in data { - *sample = queue.pop_front().unwrap_or(0.0); + // 1. Jitter Buffer State Machine + if is_buffering { + if cons.occupied_len() >= jitter_cushion_samples { + is_buffering = false; // We have enough cushion, start! + } else { + data.fill(0.0); // Output silence while we wait + return; + } + } + + // 2. Lock-free pop + let read = cons.pop_slice(data); + + // 3. Underrun Detection + if read < data.len() { + // We ran out of data. Fill remainder with silence to avoid static + data[read..].fill(0.0); + + // Re-enter buffering mode to rebuild our cushion + is_buffering = true; } }, { let needs_rebuild = needs_rebuild.clone(); - move |err| { eprintln!("[vc] output error: {err}"); - let message = err.to_string(); - if message.contains("sample rate changed") || message.contains("DeviceNotAvailable") || message.contains("device not available")