diff --git a/src-tauri/src/commands/audio.rs b/src-tauri/src/commands/audio.rs index e663822..1e3bc7e 100644 --- a/src-tauri/src/commands/audio.rs +++ b/src-tauri/src/commands/audio.rs @@ -1,12 +1,12 @@ use std::{ - collections::VecDeque, + collections::{BTreeMap, VecDeque}, net::UdpSocket, sync::{ atomic::{AtomicBool, Ordering}, Arc, Mutex, }, thread, - time::Duration, + time::{Duration, Instant}, }; use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; @@ -14,45 +14,56 @@ use tauri::State; use crate::commands::config::ConfigState; -// ============================================================ -// AUDIO CONSTANTS -// ============================================================ +// ============================================================================ +// AUDIO FORMAT +// ============================================================================ -/// Network audio format. -/// -/// Everything sent over UDP is: -/// 44100 Hz -/// mono -/// signed i16 -/// big endian -/// -/// 20 ms @ 44100 Hz = 882 samples. -/// 882 * 2 = 1764 bytes. const NETWORK_SAMPLE_RATE: u32 = 44_100; + +// 20ms packets. +// +// 44100 * 0.020 = 882 samples const PACKET_DURATION_MS: usize = 20; const NETWORK_PACKET_SAMPLES: usize = NETWORK_SAMPLE_RATE as usize * PACKET_DURATION_MS / 1000; -/// Amount of audio that must be queued before playback begins. -const PREBUFFER_MS: usize = 40; +// PCM is i16 => 2 bytes/sample. +const NETWORK_PACKET_BYTES: usize = NETWORK_PACKET_SAMPLES * 2; -/// If the playback queue gets below this amount, we allow it to -/// continue normally but the callback will output silence if it -/// actually runs dry. -const LOW_WATERMARK_MS: usize = 20; +// Sequence number is a u32. +const AUDIO_HEADER_BYTES: usize = 4; -/// Maximum amount of queued audio. -/// -/// If this is exceeded, OLD audio is discarded. This is intentional: -/// for voice chat, dropping old audio is much better than accumulating -/// hundreds of milliseconds/seconds of latency. -const MAX_BUFFER_MS: usize = 100; - -/// Maximum UDP packet size we expect. +// Maximum UDP packet we accept. const MAX_UDP_PACKET_SIZE: usize = 4096; -// ============================================================ -// STATE -// ============================================================ +// ============================================================================ +// JITTER BUFFER +// ============================================================================ +// +// We intentionally keep this relatively small. +// +// Increasing these values reduces underruns but increases latency. +// +// Current target: +// +// startup: 40ms +// normal target: 40ms +// maximum: 100ms +// +// This is appropriate for low-latency voice. +// + +const JITTER_TARGET_MS: usize = 40; +const JITTER_MAX_MS: usize = 100; + +// If a packet is missing, wait this long before considering it lost. +// +// Since packets are 20ms, 30ms gives us enough room for modest +// out-of-order delivery without making latency enormous. +const PACKET_LOSS_WAIT_MS: u64 = 30; + +// ============================================================================ +// VOICE STATE +// ============================================================================ pub struct VoiceState { pub session: Mutex>, @@ -61,8 +72,6 @@ pub struct VoiceState { pub struct VoiceSession { pub input_stream: cpal::Stream, - // Wrapped so the watcher thread can replace the stream after - // the output device changes configuration. pub output_stream: Arc>, pub socket: Arc, @@ -80,9 +89,9 @@ impl Default for VoiceState { } } -// ============================================================ -// DEVICE ENUMERATION -// ============================================================ +// ============================================================================ +// DEVICE LISTING +// ============================================================================ #[tauri::command] pub fn list_input_devices() -> Result, String> { @@ -110,27 +119,22 @@ pub fn list_output_devices() -> Result, String> { .collect()) } -// ============================================================ +// ============================================================================ // DISCONNECT -// ============================================================ +// ============================================================================ #[tauri::command] pub fn disconnect_from_vc(voice_state: State) -> Result<(), String> { if let Some(session) = voice_state.session.lock().unwrap().take() { session.shutdown.store(true, Ordering::SeqCst); - - // Streams are dropped here. - // - // The UDP receiver has a 100ms read timeout, so it will notice - // shutdown shortly instead of remaining blocked forever. } Ok(()) } -// ============================================================ +// ============================================================================ // CONNECT -// ============================================================ +// ============================================================================ #[tauri::command] pub fn connect_to_vc( @@ -143,9 +147,9 @@ pub fn connect_to_vc( let config = config_state.0.lock().unwrap().clone(); - // ======================================================== - // UDP SOCKET - // ======================================================== + // ======================================================================== + // UDP + // ======================================================================== let socket = UdpSocket::bind("0.0.0.0:0").map_err(|e| format!("Failed to bind UDP socket: {e}"))?; @@ -154,24 +158,24 @@ pub fn connect_to_vc( .connect(&hostname) .map_err(|e| format!("Failed to connect UDP socket: {e}"))?; - // IMPORTANT: - // - // Without a timeout, recv() can remain blocked forever and the - // receiver thread can survive after disconnect. + // Required so disconnect can terminate the receiver thread. socket .set_read_timeout(Some(Duration::from_millis(100))) - .map_err(|e| format!("Failed to configure UDP timeout: {e}"))?; + .map_err(|e| format!("Failed to set UDP read timeout: {e}"))?; let socket = Arc::new(socket); - // PIN is always the first packet. + // ======================================================================== + // AUTHENTICATION + // ======================================================================== + socket .send(&pin.to_be_bytes()) .map_err(|e| format!("Failed to send pincode: {e}"))?; - // ======================================================== - // AUDIO DEVICES - // ======================================================== + // ======================================================================== + // DEVICES + // ======================================================================== let host = cpal::default_host(); @@ -211,35 +215,42 @@ pub fn connect_to_vc( .ok_or_else(|| "No default output device".to_string())?, }; - // ======================================================== - // SHARED PLAYBACK STATE - // ======================================================== - - // - // IMPORTANT: - // - // This queue contains samples already converted to the - // CURRENT output device's channel layout. - // - // Therefore: - // - // stereo device: - // L R L R L R ... - // - // mono device: - // M M M M ... - // - let playback_buffer = Arc::new(Mutex::new(VecDeque::::new())); - - let needs_output_rebuild = Arc::new(AtomicBool::new(false)); + // ======================================================================== + // SHUTDOWN + // ======================================================================== let shutdown = Arc::new(AtomicBool::new(false)); + // ======================================================================== + // PLAYBACK BUFFER + // ======================================================================== + // + // This contains samples already converted to the output device's + // channel layout. + // + + let playback_buffer = Arc::new(Mutex::new(VecDeque::::new())); + let playback_started = Arc::new(AtomicBool::new(false)); - // ======================================================== - // INPUT: MICROPHONE -> UDP - // ======================================================== + // ======================================================================== + // OUTPUT CONFIG + // ======================================================================== + + let output_config = build_output_config(&output_device)?; + + eprintln!( + "[vc] output: {} Hz / {} channels", + output_config.sample_rate, output_config.channels + ); + + let output_config_cell = Arc::new(Mutex::new(output_config.clone())); + + let needs_output_rebuild = Arc::new(AtomicBool::new(false)); + + // ======================================================================== + // INPUT + // ======================================================================== let input_socket = socket.clone(); @@ -248,7 +259,7 @@ pub fn connect_to_vc( .map_err(|e| format!("Failed to get input config: {e}"))? .into(); - // We explicitly capture mono. + // Network audio is mono. input_config.channels = 1; let input_sample_rate = input_config.sample_rate; @@ -260,16 +271,22 @@ pub fn connect_to_vc( let mut input_resampler = resampler::ResamplerFft::new( 1, - input_sample_rate - .try_into() - .map_err(|e| format!("Invalid input sample rate: {e:?}"))?, + input_sample_rate.try_into().map_err(|e| { + format!( + "Invalid input sample rate: \ + {e:?}" + ) + })?, resampler::SampleRate::Hz44100, ); let mut input_buffer = Vec::::new(); - // Resampled samples waiting to form a network packet. - let mut packet_buffer = Vec::::with_capacity(NETWORK_PACKET_SAMPLES * 2); + // Resampled samples waiting to form 20ms packets. + let mut packet_samples = Vec::::with_capacity(NETWORK_PACKET_SAMPLES * 2); + + // Sequence number for every audio packet. + let mut sequence: u32 = 0; let input_stream = input_device .build_input_stream( @@ -287,35 +304,50 @@ pub fn connect_to_vc( let mut output = vec![0.0f32; output_size]; if let Err(e) = input_resampler.resample(&input, &mut output) { - eprintln!("[vc] failed to resample input: {e}"); + eprintln!( + "[vc] input resample \ + failed: {e}" + ); + continue; } - packet_buffer.extend(output.into_iter().map(|sample| sample.clamp(-1.0, 1.0))); + packet_samples.extend(output.into_iter().map(|sample| sample.clamp(-1.0, 1.0))); - // ------------------------------------------------ - // Form exact 20ms network packets. - // ------------------------------------------------ + // ======================================================== + // CREATE EXACT 20ms PACKETS + // ======================================================== - while packet_buffer.len() >= NETWORK_PACKET_SAMPLES { - let packet_samples: Vec = - packet_buffer.drain(..NETWORK_PACKET_SAMPLES).collect(); + while packet_samples.len() >= NETWORK_PACKET_SAMPLES { + let samples: Vec = + packet_samples.drain(..NETWORK_PACKET_SAMPLES).collect(); - let mut packet = Vec::with_capacity(NETWORK_PACKET_SAMPLES * 2); + // Header: + // + // [u32 sequence] + // + // Then: + // + // [i16 PCM...] + let mut packet = + Vec::with_capacity(AUDIO_HEADER_BYTES + NETWORK_PACKET_BYTES); - for sample in packet_samples { - let pcm = (sample * i16::MAX as f32) as i16; + packet.extend_from_slice(&sequence.to_be_bytes()); + + for sample in samples { + let pcm = (sample.clamp(-1.0, 1.0) * i16::MAX as f32) as i16; packet.extend_from_slice(&pcm.to_be_bytes()); } - match input_socket.send(&packet) { - Ok(_) => {} - - Err(e) => { - eprintln!("[vc] UDP send failed: {e}"); - } + if let Err(e) = input_socket.send(&packet) { + eprintln!( + "[vc] UDP send failed: \ + {e}" + ); } + + sequence = sequence.wrapping_add(1); } } }, @@ -326,22 +358,9 @@ pub fn connect_to_vc( ) .map_err(|e| e.to_string())?; - // ======================================================== - // OUTPUT CONFIG - // ======================================================== - - let output_config = build_output_config(&output_device)?; - - eprintln!( - "[vc] output: {} Hz / {} channels", - output_config.sample_rate, output_config.channels - ); - - let output_config_cell = Arc::new(Mutex::new(output_config.clone())); - - // ======================================================== + // ======================================================================== // OUTPUT STREAM - // ======================================================== + // ======================================================================== let initial_output_stream = build_output_stream( &output_device, @@ -357,9 +376,9 @@ pub fn connect_to_vc( let output_stream = Arc::new(Mutex::new(initial_output_stream)); - // ======================================================== + // ======================================================================== // OUTPUT DEVICE WATCHER - // ======================================================== + // ======================================================================== { let output_device = output_device.clone(); @@ -370,12 +389,12 @@ pub fn connect_to_vc( let output_config_cell = output_config_cell.clone(); - let shutdown = shutdown.clone(); - let playback_buffer = playback_buffer.clone(); let playback_started = playback_started.clone(); + let shutdown = shutdown.clone(); + thread::spawn(move || { while !shutdown.load(Ordering::SeqCst) { thread::sleep(Duration::from_millis(100)); @@ -405,8 +424,8 @@ pub fn connect_to_vc( Ok((new_config, new_stream)) => { if let Err(e) = new_stream.play() { eprintln!( - "[vc] failed to start rebuilt \ - output stream: {e}" + "[vc] failed to start \ + rebuilt output: {e}" ); needs_output_rebuild.store(true, Ordering::SeqCst); @@ -414,8 +433,10 @@ pub fn connect_to_vc( continue; } - // Clear stale audio because it was generated - // for the old output timing/channel layout. + // The old output format is no longer valid. + // + // Throw away queued samples rather than playing + // them using the new device timing/layout. playback_buffer.lock().unwrap().clear(); playback_started.store(false, Ordering::SeqCst); @@ -425,13 +446,17 @@ pub fn connect_to_vc( *output_stream.lock().unwrap() = new_stream; eprintln!( - "[vc] output rebuilt: {} Hz / {} channels", + "[vc] output rebuilt: \ + {} Hz / {} channels", new_config.sample_rate, new_config.channels ); } Err(e) => { - eprintln!("[vc] failed to rebuild output: {e}"); + eprintln!( + "[vc] failed to rebuild \ + output: {e}" + ); needs_output_rebuild.store(true, Ordering::SeqCst); } @@ -440,9 +465,9 @@ pub fn connect_to_vc( }); } - // ======================================================== - // UDP RECEIVE -> RESAMPLE -> PLAYBACK QUEUE - // ======================================================== + // ======================================================================== + // UDP RECEIVE + JITTER BUFFER + // ======================================================================== { let recv_socket = socket.clone(); @@ -456,7 +481,20 @@ pub fn connect_to_vc( let shutdown = shutdown.clone(); thread::spawn(move || { - let mut buf = [0u8; MAX_UDP_PACKET_SIZE]; + // ================================================================ + // PACKET REORDER BUFFER + // ================================================================ + + let mut packets: BTreeMap> = BTreeMap::new(); + + let mut expected_sequence: Option = None; + + // When we first notice a missing packet, remember when. + let mut missing_since: Option = None; + + // ================================================================ + // RESAMPLER + // ================================================================ let initial_config = output_config_cell.lock().unwrap().clone(); @@ -467,84 +505,154 @@ pub fn connect_to_vc( Err(e) => { eprintln!( - "[vc] failed to create output \ - resampler: {e}" + "[vc] failed to create \ + output resampler: {e}" ); return; } }; - // Audio waiting to be fed into the resampler. let mut resample_input = Vec::::new(); - loop { - if shutdown.load(Ordering::SeqCst) { - return; - } + // ================================================================ + // UDP BUFFER + // ================================================================ - // ---------------------------------------------------- - // Receive UDP packet. - // ---------------------------------------------------- + let mut udp_buffer = [0u8; MAX_UDP_PACKET_SIZE]; - let len = match recv_socket.recv(&mut buf) { - Ok(len) => len, + // ================================================================ + // MAIN LOOP + // ================================================================ + + while !shutdown.load(Ordering::SeqCst) { + // ============================================================ + // RECEIVE PACKET + // ============================================================ + + match recv_socket.recv(&mut udp_buffer) { + Ok(len) => { + if len <= AUDIO_HEADER_BYTES { + continue; + } + + // ---------------------------------------------------- + // Read sequence number. + // ---------------------------------------------------- + + let sequence = u32::from_be_bytes([ + udp_buffer[0], + udp_buffer[1], + udp_buffer[2], + udp_buffer[3], + ]); + + let pcm_bytes = &udp_buffer[AUDIO_HEADER_BYTES..len]; + + // Must contain complete i16 samples. + let pcm_len = pcm_bytes.len() & !1; + + if pcm_len == 0 { + continue; + } + + let mut samples = Vec::with_capacity(pcm_len / 2); + + for chunk in pcm_bytes[..pcm_len].chunks_exact(2) { + let pcm = i16::from_be_bytes([chunk[0], chunk[1]]); + + samples.push(pcm as f32 / i16::MAX as f32); + } + + // ---------------------------------------------------- + // Ignore packets that are already too old. + // ---------------------------------------------------- + + if let Some(expected) = expected_sequence { + if sequence_before(sequence, expected) { + continue; + } + } + + // ---------------------------------------------------- + // Insert into jitter buffer. + // + // BTreeMap automatically keeps sequence numbers + // ordered. + // ---------------------------------------------------- + + packets.entry(sequence).or_insert(samples); + + // ---------------------------------------------------- + // Don't allow the packet jitter buffer itself to + // become a source of latency. + // ---------------------------------------------------- + + while packets.len() > 8 { + if let Some((&oldest, _)) = packets.iter().next() { + if let Some(expected) = expected_sequence { + if sequence_before(oldest, expected) { + packets.remove(&oldest); + } else { + break; + } + } else { + break; + } + } + } + } Err(e) - if e.kind() == std::io::ErrorKind::WouldBlock - || e.kind() == std::io::ErrorKind::TimedOut => + if e.kind() == std::io::ErrorKind::TimedOut + || e.kind() == std::io::ErrorKind::WouldBlock => { - continue; + // Timeout is expected. It gives us a chance to + // process jitter-buffer timeouts and shutdown. } Err(e) => { if !shutdown.load(Ordering::SeqCst) { - eprintln!("[vc] UDP receive failed: {e}"); + eprintln!( + "[vc] UDP receive failed: \ + {e}" + ); } continue; } - }; - - // ---------------------------------------------------- - // Ignore malformed packets. - // - // Every audio packet must contain complete i16 - // samples. - // ---------------------------------------------------- - - if len < 2 { - continue; } - let usable_len = len - (len % 2); + // ============================================================ + // INITIALIZE EXPECTED SEQUENCE + // ============================================================ - // ---------------------------------------------------- - // BIG-ENDIAN i16 -> f32 - // ---------------------------------------------------- + if expected_sequence.is_none() { + if let Some((&first_sequence, _)) = packets.iter().next() { + expected_sequence = Some(first_sequence); - for chunk in buf[..usable_len].chunks_exact(2) { - let pcm = i16::from_be_bytes([chunk[0], chunk[1]]); - - resample_input.push(pcm as f32 / i16::MAX as f32); + eprintln!( + "[vc] jitter buffer \ + synchronized at packet {}", + first_sequence + ); + } } - // ---------------------------------------------------- - // Check current output device configuration. - // ---------------------------------------------------- + // ============================================================ + // CURRENT OUTPUT CONFIG + // ============================================================ let current_config = output_config_cell.lock().unwrap().clone(); - // ---------------------------------------------------- - // Output sample rate changed. - // - // Recreate the resampler and discard samples from the - // old timing domain. - // ---------------------------------------------------- + // ============================================================ + // OUTPUT DEVICE SAMPLE RATE CHANGE + // ============================================================ if current_config.sample_rate != last_sample_rate { eprintln!( - "[vc] output rate changed: {} -> {}", + "[vc] output rate changed: \ + {} -> {}", last_sample_rate, current_config.sample_rate ); @@ -554,20 +662,26 @@ pub fn connect_to_vc( last_sample_rate = current_config.sample_rate; + packets.clear(); + resample_input.clear(); playback_buffer.lock().unwrap().clear(); playback_started.store(false, Ordering::SeqCst); + + expected_sequence = None; + + missing_since = None; } Err(e) => { eprintln!( - "[vc] unsupported output \ - sample rate {}: {e}", - current_config.sample_rate + "[vc] failed to create \ + output resampler: {e}" ); + packets.clear(); resample_input.clear(); continue; @@ -575,97 +689,200 @@ pub fn connect_to_vc( } } - // ---------------------------------------------------- - // Resample incoming 44.1kHz mono audio into the - // output device's sample rate. - // ---------------------------------------------------- + // ============================================================ + // MOVE READY PACKETS INTO RESAMPLER + // ============================================================ - let frame_size = output_resampler.chunk_size_input(); + loop { + let Some(expected) = expected_sequence else { + break; + }; - while resample_input.len() >= frame_size { - let input: Vec = resample_input.drain(..frame_size).collect(); + // -------------------------------------------------------- + // Expected packet exists. + // -------------------------------------------------------- - let output_size = output_resampler.chunk_size_output(); + if let Some(samples) = packets.remove(&expected) { + resample_input.extend_from_slice(&samples); - let mut resampled = vec![0.0f32; output_size]; + expected_sequence = Some(expected.wrapping_add(1)); + + missing_since = None; + + continue; + } + + // -------------------------------------------------------- + // Expected packet doesn't exist. + // + // If we don't have anything newer, there is nothing + // to do yet. + // -------------------------------------------------------- + + let has_newer_packet = packets.keys().any(|&seq| sequence_after(seq, expected)); + + if !has_newer_packet { + break; + } + + // -------------------------------------------------------- + // We have a later packet. + // + // Therefore the expected packet is either delayed or + // lost. + // + // Wait a short period before declaring it lost. + // -------------------------------------------------------- + + let now = Instant::now(); + + let since = missing_since.get_or_insert(now); + + if since.elapsed() < Duration::from_millis(PACKET_LOSS_WAIT_MS) { + break; + } + + // -------------------------------------------------------- + // Packet is considered lost. + // + // We insert a zero packet here. + // + // Because the missing packet is exactly 20ms, this + // results in a controlled 20ms gap rather than the + // playback clock getting permanently stuck. + // -------------------------------------------------------- + + resample_input.extend(std::iter::repeat(0.0f32).take(NETWORK_PACKET_SAMPLES)); + + expected_sequence = Some(expected.wrapping_add(1)); + + missing_since = None; + + eprintln!("[vc] lost UDP packet {}", expected); + } + + // ============================================================ + // RESAMPLE + // ============================================================ + + let input_frame_size = output_resampler.chunk_size_input(); + + while resample_input.len() >= input_frame_size { + let input: Vec = resample_input.drain(..input_frame_size).collect(); + + let output_frame_size = output_resampler.chunk_size_output(); + + let mut resampled = vec![0.0f32; output_frame_size]; if let Err(e) = output_resampler.resample(&input, &mut resampled) { eprintln!( - "[vc] failed to resample \ - output: {e}" + "[vc] output resample \ + failed: {e}" ); continue; } - // ------------------------------------------------ - // Mono -> device channels. - // ------------------------------------------------ + // ======================================================== + // MONO -> DEVICE CHANNELS + // ======================================================== let output = mono_to_output_channels(&resampled, current_config.channels); - // ------------------------------------------------ - // Push into bounded playback queue. - // ------------------------------------------------ + // ======================================================== + // PLAYBACK QUEUE + // ======================================================== + + let channels = current_config.channels.max(1) as usize; let mut queue = playback_buffer.lock().unwrap(); queue.extend(output); - let channels = current_config.channels.max(1) as usize; + // -------------------------------------------------------- + // Maximum playback queue. + // + // If this gets exceeded, THROW AWAY OLD AUDIO. + // + // This prevents latency from continuously growing. + // -------------------------------------------------------- - let max_frames = current_config.sample_rate as usize * MAX_BUFFER_MS / 1000; + let max_frames = current_config.sample_rate as usize * JITTER_MAX_MS / 1000; let max_samples = max_frames * channels; - // ------------------------------------------------ - // If we have accumulated too much audio, discard - // OLD audio. - // - // This is critical for voice chat latency. - // ------------------------------------------------ - while queue.len() > max_samples { queue.pop_front(); } - // ------------------------------------------------ - // Start playback only after we have a small - // amount of audio buffered. - // ------------------------------------------------ + // -------------------------------------------------------- + // Start playback once enough audio is available. + // -------------------------------------------------------- - if !playback_started.load(Ordering::SeqCst) { - let prebuffer_frames = - current_config.sample_rate as usize * PREBUFFER_MS / 1000; + if !playback_started.load(Ordering::Acquire) { + let target_frames = + current_config.sample_rate as usize * JITTER_TARGET_MS / 1000; - let prebuffer_samples = prebuffer_frames * channels; + let target_samples = target_frames * channels; - if queue.len() >= prebuffer_samples { - playback_started.store(true, Ordering::SeqCst); + if queue.len() >= target_samples { + playback_started.store(true, Ordering::Release); eprintln!( "[vc] playback started \ with ~{}ms buffered", - PREBUFFER_MS + JITTER_TARGET_MS ); } } } + + // ============================================================ + // RECOVER FROM PLAYBACK UNDERRUN + // ============================================================ + // + // If CPAL consumes everything, it will output silence. + // + // Once enough audio has accumulated again, playback can + // resume. + // + // We intentionally don't constantly toggle this state. + // That was one of the sources of the previous flicker. + // ============================================================ + + if playback_started.load(Ordering::Acquire) { + let channels = current_config.channels.max(1) as usize; + + let queue_len = playback_buffer.lock().unwrap().len(); + + let low_frames = current_config.sample_rate as usize * 10 / 1000; + + let low_samples = low_frames * channels; + + // We don't stop playback at 0ms. + // + // Only stop if the queue has actually become empty. + if queue_len == 0 { + playback_started.store(false, Ordering::Release); + } + + let _ = low_samples; + } } }); } - // ======================================================== + // ======================================================================== // START INPUT - // ======================================================== + // ======================================================================== input_stream .play() .map_err(|e| format!("Failed to start input stream: {e}"))?; - // ======================================================== - // STORE SESSION - // ======================================================== + // ======================================================================== + // SAVE SESSION + // ======================================================================== *voice_state.session.lock().unwrap() = Some(VoiceSession { input_stream, @@ -678,9 +895,9 @@ pub fn connect_to_vc( Ok(()) } -// ============================================================ +// ============================================================================ // OUTPUT CONFIG -// ============================================================ +// ============================================================================ fn build_output_config(output_device: &cpal::Device) -> Result { output_device @@ -689,27 +906,30 @@ fn build_output_config(output_device: &cpal::Device) -> Result Result { - let output_rate = output_sample_rate - .try_into() - .map_err(|e| format!("Invalid output sample rate: {e:?}"))?; + let rate = output_sample_rate.try_into().map_err(|e| { + format!( + "Invalid output sample rate: \ + {e:?}" + ) + })?; Ok(resampler::ResamplerFft::new( 1, resampler::SampleRate::Hz44100, - output_rate, + rate, )) } -// ============================================================ -// BUILD OUTPUT STREAM -// ============================================================ +// ============================================================================ +// OUTPUT STREAM +// ============================================================================ fn build_output_stream( output_device: &cpal::Device, @@ -721,36 +941,41 @@ fn build_output_stream( output_device .build_output_stream( output_config.clone(), - // ==================================================== - // CPAL OUTPUT CALLBACK - // ==================================================== + // ================================================================= + // AUDIO CALLBACK + // ================================================================= move |data: &mut [f32], _| { - let started = playback_started.load(Ordering::Acquire); + // ------------------------------------------------------------- + // Don't consume the queue until the jitter buffer has enough + // audio. + // ------------------------------------------------------------- - if !started { - // Do NOT consume audio before the prebuffer is - // ready. Just output silence. + if !playback_started.load(Ordering::Acquire) { data.fill(0.0); return; } let mut queue = playback_buffer.lock().unwrap(); + // ------------------------------------------------------------- + // CRITICAL: + // // VecDeque::pop_front() is O(1). // - // This is massively better than: - // - // Vec::remove(0) - // - // which shifts the entire vector every sample. + // NEVER use Vec::remove(0) here. + // ------------------------------------------------------------- - for sample in data.iter_mut() { - *sample = queue.pop_front().unwrap_or(0.0); + for output in data.iter_mut() { + *output = queue.pop_front().unwrap_or(0.0); } + + // If the callback consumed the entire queue, the receiver + // thread will refill it. We don't modify playback_started + // here because the CPAL callback should stay extremely cheap. }, - // ==================================================== - // OUTPUT ERROR CALLBACK - // ==================================================== + // ================================================================= + // ERROR CALLBACK + // ================================================================= { let needs_rebuild = needs_rebuild.clone(); @@ -772,9 +997,9 @@ fn build_output_stream( .map_err(|e| e.to_string()) } -// ============================================================ -// MONO -> OUTPUT CHANNELS -// ============================================================ +// ============================================================================ +// CHANNEL CONVERSION +// ============================================================================ fn mono_to_output_channels(mono: &[f32], channels: u16) -> Vec { let channels = channels.max(1) as usize; @@ -794,9 +1019,30 @@ fn mono_to_output_channels(mono: &[f32], channels: u16) -> Vec { output } -// ============================================================ +// ============================================================================ +// SEQUENCE NUMBER HELPERS +// ============================================================================ +// +// UDP sequence numbers eventually wrap around u32::MAX. +// +// These helpers make comparisons work correctly across the wrap. +// + +fn sequence_after(a: u32, b: u32) -> bool { + let diff = a.wrapping_sub(b); + + diff != 0 && diff < 0x8000_0000 +} + +fn sequence_before(a: u32, b: u32) -> bool { + let diff = a.wrapping_sub(b); + + diff != 0 && diff >= 0x8000_0000 +} + +// ============================================================================ // OPTIONAL UTILITY -// ============================================================ +// ============================================================================ fn stereo_to_mono(stereo_data: &[f32]) -> Vec { let mut mono = Vec::with_capacity(stereo_data.len() / 2);