diff --git a/README.md b/README.md index 4e6726e..9250cd0 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ Look at [open pull requests](https://github.com/itz4blitz/tinyclient/pulls) befo | --- | --- | | **Cached network sidebars** | [#10](https://github.com/itz4blitz/tinyclient/pull/10) keeps each network's home in memory for the GUI session, so switching away and back does not show "Getting your chats…" while the local daemon is reread. It conflicts with `main` after the Discord loading fix and needs a rebase that keeps both. | | **Discord voice channels in the server list** | Voice works in a DM and in a channel you are already in. The server list does not yet show public or private voice channels, or let you join one from there. That is being built on top of `main`. | -| **Call audio** | Discord calls are Opus at 48 kHz. A click on both ends comes from the shared playback buffer: an underrun steps to silence, and an over-full buffer drops samples from the front. The fix underway is a short fade across both, and moving audio-device reopen off the 20 ms tick. Slack, Teams, Telegram, and Signal still hand the call to the official app. | +| **Call audio** | Discord calls are Opus at 48 kHz. An underrun fades to silence and an over-full playback buffer fades across the samples it drops, so neither end hears a click from that buffer. Reopening the audio device runs off the 20 ms tick. Slack, Teams, Telegram, and Signal still hand the call to the official app. | | **Camera and screen share** | Calls are audio only. The protocol already names video streams, and a voice occupant can show a monitor glyph when Discord says they are streaming or on camera. mediad does not decode or send video yet. Receive-only screen share is the first slice; sending a camera, then guild "Go Live", comes after. | Recently merged, and already in the sections above: cached message search from `tinyclient --text --search`, installable packages, in-app updates, Android Messages, animated images, an optional systemd user service for `chatd`, Signal's Note to Self, recovery when the Linux keyring unlocks late, native shortcut labels, and sidebar previews that stay inside their row. diff --git a/crates/mediad/src/audio.rs b/crates/mediad/src/audio.rs index c0e4275..8aecbf7 100644 --- a/crates/mediad/src/audio.rs +++ b/crates/mediad/src/audio.rs @@ -42,19 +42,52 @@ impl Playout { self.playing = true; self.fade = fade; } + let frames_left = self.ring.len() / CHANNELS; let (Some(left), Some(right)) = (self.ring.pop_front(), self.ring.pop_front()) else { // Ran dry: we fell behind the device. Fill again before playing on. self.playing = false; self.underruns += 1; return (0., 0.); }; - if self.fade == 0 { - return (left, right); + // Fade in at the start, and fade the last `fade` frames down to silence. A step from a + // loud sample to zero is the click both ends hear when the cushion runs out. + let mut gain = 1.; + if self.fade > 0 && fade > 0 { + gain *= 1. - self.fade as f32 / fade as f32; + self.fade -= 1; + } + if fade > 0 && frames_left <= fade { + gain *= frames_left.saturating_sub(1) as f32 / fade as f32; } - let gain = 1. - self.fade as f32 / fade as f32; - self.fade -= 1; (left * gain, right * gain) } + + /// Drops samples off the front until `keep` remain. The samples that were about to play are + /// blended into the new head, so the clock-drift skip is a ramp and not a splice. + fn trim_front(&mut self, keep: usize, fade_frames: usize) { + let len = self.ring.len(); + if len <= keep { + return; + } + let mut drop = len - keep; + drop -= drop % CHANNELS; + if drop == 0 { + return; + } + let fade = fade_frames + .min(drop / CHANNELS) + .min((len - drop) / CHANNELS); + for i in 0..fade { + let toward_new = (i + 1) as f32 / (fade as f32 + 1.); + for channel in 0..CHANNELS { + let at = i * CHANNELS + channel; + let kept = self.ring[at]; + let ahead = self.ring[drop + at]; + self.ring[drop + at] = kept * (1. - toward_new) + ahead * toward_new; + } + } + self.ring.drain(..drop); + } } /// Seconds of audio either ring may hold before the oldest is dropped: the device clock and ours @@ -63,7 +96,8 @@ const BACKLOG: f32 = 0.2; /// Seconds held before the speaker is given anything. The media loop hands over 20 ms at a time /// and not exactly on time; without a cushion the device runs dry between frames and clicks. const PREBUFFER: f32 = 0.06; -/// Seconds over which playing fades in when it starts. +/// Seconds over which playing fades in when it starts, and out before the cushion runs dry or +/// the ring skips ahead. const FADE: f32 = 0.005; #[derive(Debug)] @@ -100,15 +134,27 @@ pub fn listing() -> Listing { } let host = cpal::default_host(); let names = |devices: Option>| -> Vec { - let mut names: Vec = devices.unwrap_or_default().iter().filter_map(name).collect(); + let mut names: Vec = devices + .unwrap_or_default() + .iter() + .filter_map(name) + .collect(); names.dedup(); names }; Listing { inputs: names(host.input_devices().ok().map(Iterator::collect)), outputs: names(host.output_devices().ok().map(Iterator::collect)), - default_input: host.default_input_device().as_ref().and_then(name).unwrap_or_default(), - default_output: host.default_output_device().as_ref().and_then(name).unwrap_or_default(), + default_input: host + .default_input_device() + .as_ref() + .and_then(name) + .unwrap_or_default(), + default_output: host + .default_output_device() + .as_ref() + .and_then(name) + .unwrap_or_default(), } } @@ -142,7 +188,11 @@ fn table(cutoff: f64) -> Vec<[f32; TAPS]> { let mut row = [0f64; TAPS]; for (tap, weight) in row.iter_mut().enumerate() { let x = tap as f64 - CENTRE as f64 - offset; - let sinc = if x.abs() < 1e-9 { 1. } else { (std::f64::consts::PI * cutoff * x).sin() / (std::f64::consts::PI * cutoff * x) }; + let sinc = if x.abs() < 1e-9 { + 1. + } else { + (std::f64::consts::PI * cutoff * x).sin() / (std::f64::consts::PI * cutoff * x) + }; let across = (x / half).clamp(-1., 1.); let window = bessel_i0(BETA * (1. - across * across).sqrt()) / bessel_i0(BETA); *weight = cutoff * sinc * window; @@ -172,11 +222,21 @@ impl Resampler { pub(crate) fn new(from: u32, to: u32) -> Resampler { let step = f64::from(from) / f64::from(to); if from == to { - return Resampler { step, position: 0., held: Vec::new(), table: Vec::new() }; + return Resampler { + step, + position: 0., + held: Vec::new(), + table: Vec::new(), + }; } // Going down, everything above the new Nyquist frequency has to go first. let cutoff = 0.95 * (f64::from(to) / f64::from(from)).min(1.); - Resampler { step, position: 0., held: vec![0.; CENTRE * CHANNELS], table: table(cutoff) } + Resampler { + step, + position: 0., + held: vec![0.; CENTRE * CHANNELS], + table: table(cutoff), + } } pub(crate) fn push(&mut self, input: &[f32], output: &mut Vec) { @@ -184,7 +244,8 @@ impl Resampler { output.extend_from_slice(input); return; } - self.held.extend_from_slice(&input[..input.len() - input.len() % CHANNELS]); + self.held + .extend_from_slice(&input[..input.len() - input.len() % CHANNELS]); let frames = self.held.len() / CHANNELS; while self.position as usize + TAPS <= frames { let first = self.position as usize; @@ -193,7 +254,9 @@ impl Resampler { let blend = (between - phase as f64) as f32; let (near, far) = (&self.table[phase], &self.table[phase + 1]); let mut sums = [0f32; CHANNELS]; - let taps = self.held[first * CHANNELS..(first + TAPS) * CHANNELS].as_chunks::().0; + let taps = self.held[first * CHANNELS..(first + TAPS) * CHANNELS] + .as_chunks::() + .0; for (frame, (near, far)) in taps.iter().zip(near.iter().zip(far)) { let weight = near + (far - near) * blend; for (sum, sample) in sums.iter_mut().zip(frame) { @@ -234,13 +297,19 @@ struct Output { /// Samples the ring may hold, and how many it is cut back to when it is over. room: usize, fill: usize, + /// Frames of the ramp used when the ring skips ahead. + fade: usize, } enum Source { Silence, - Tone { phase: f64 }, + Tone { + phase: f64, + }, /// A word a second: half a second of nothing, 40 ms of a faint tone, then the tone. - Onset { at: u32 }, + Onset { + at: u32, + }, Device(Option), } @@ -267,6 +336,80 @@ pub struct Audio { sides: (bool, bool), } +/// Devices opened away from the call thread, ready to take over. +pub struct Fresh { + source: Source, + sink: Sink, + broken: Arc, + input: String, + output: String, +} + +/// Streams a call has stopped using. Drop this away from the 20 ms thread: stopping a device +/// is the slow half of a reopen. The fields exist so that drop runs here, not on the tick. +#[allow(dead_code)] +pub struct Retired { + source: Source, + sink: Sink, +} + +/// True when the system default we follow is no longer the device we have open. +pub(crate) fn route_moved( + want_input: &Option, + input: &str, + want_output: &Option, + output: &str, + sides: (bool, bool), +) -> bool { + let host = cpal::default_host(); + let moved = |wanted: &Option, current: &str, device: Option| { + wanted.is_none() + && device + .as_ref() + .and_then(name) + .is_some_and(|name| name != current) + }; + (sides.0 && moved(want_input, input, host.default_input_device())) + || (sides.1 && moved(want_output, output, host.default_output_device())) +} + +/// Opens the wanted devices without touching a call that is already playing. +pub(crate) fn open_fresh( + want_input: Option<&str>, + want_output: Option<&str>, + sides: (bool, bool), + mic_running: bool, +) -> Result { + let host = cpal::default_host(); + let broken = Arc::new(AtomicBool::new(false)); + let mut source = Source::Device(None); + let mut sink = Sink::Device(None); + let mut input_name = String::new(); + let mut output_name = String::new(); + if sides.0 { + let microphone = find(&host, want_input, true).ok_or(AudioError::NoMicrophone)?; + let mut input = open_input(µphone, &broken)?; + if !mic_running { + let _ = input.stream.pause(); + input.running = false; + } + input_name = name(µphone).unwrap_or_default(); + source = Source::Device(Some(input)); + } + if sides.1 { + let speaker = find(&host, want_output, false).ok_or(AudioError::NoSpeaker)?; + sink = Sink::Device(Some(open_output(&speaker, &broken)?)); + output_name = name(&speaker).unwrap_or_default(); + } + Ok(Fresh { + source, + sink, + broken, + input: input_name, + output: output_name, + }) +} + fn find(host: &cpal::Host, wanted: Option<&str>, input: bool) -> Option { if let Some(wanted) = wanted { let devices: Option> = if input { @@ -274,7 +417,11 @@ fn find(host: &cpal::Host, wanted: Option<&str>, input: bool) -> Option, input: bool) -> Option) -> Result { - let supported = device.default_input_config().map_err(|e| AudioError::Microphone(e.to_string()))?; + let supported = device + .default_input_config() + .map_err(|e| AudioError::Microphone(e.to_string()))?; let config = supported.config(); let (channels, rate) = (usize::from(config.channels), config.sample_rate); let ring: Ring = Arc::default(); @@ -313,12 +462,21 @@ fn open_input(device: &cpal::Device, broken: &Arc) -> Result) -> Result { - let supported = device.default_output_config().map_err(|e| AudioError::Speaker(e.to_string()))?; + let supported = device + .default_output_config() + .map_err(|e| AudioError::Speaker(e.to_string()))?; let config = supported.config(); let (channels, rate) = (usize::from(config.channels), config.sample_rate); let playout: Arc> = Arc::default(); @@ -348,9 +506,18 @@ fn open_output(device: &cpal::Device, broken: &Arc) -> Result, want_output: Option) -> Result { + pub fn open( + want_input: Option, + want_output: Option, + ) -> Result { Audio::open_sides(want_input, want_output, (true, true)) } /// Opens only the microphone (`sides.0`) or only the speaker (`sides.1`), for a test. - pub fn open_sides(want_input: Option, want_output: Option, sides: (bool, bool)) -> Result { + pub fn open_sides( + want_input: Option, + want_output: Option, + sides: (bool, bool), + ) -> Result { let mut audio = Audio { source: Source::Device(None), sink: Sink::Device(None), @@ -385,7 +559,9 @@ impl Audio { _ => Source::Silence, }; audio.sink = match std::env::var_os("TINYCLIENT_MEDIAD_SINK") { - Some(path) => Sink::File(File::create(path).map_err(|e| AudioError::Speaker(e.to_string()))?), + Some(path) => { + Sink::File(File::create(path).map_err(|e| AudioError::Speaker(e.to_string()))?) + } None => Sink::Null, }; audio.input = kind; @@ -396,61 +572,80 @@ impl Audio { Ok(audio) } - /// Opens the wanted devices again: at start, after a new choice, and when one went away. - pub fn reopen(&mut self) -> Result<(), AudioError> { - if !matches!(self.source, Source::Device(_)) { - return Ok(()); - } - // The old streams stop before the new ones start. - let was_running = match &self.source { + /// Real devices, not the stand-ins a test asks for. + pub(crate) fn on_devices(&self) -> bool { + matches!(self.source, Source::Device(_)) + } + + pub(crate) fn mic_running(&self) -> bool { + match &self.source { Source::Device(Some(input)) => input.running, _ => true, - }; - self.source = Source::Device(None); - self.sink = Sink::Device(None); - self.broken.store(false, Ordering::Relaxed); - let host = cpal::default_host(); - - if self.sides.0 { - let microphone = find(&host, self.want_input.as_deref(), true).ok_or(AudioError::NoMicrophone)?; - let mut input = open_input(µphone, &self.broken)?; - if !was_running { - let _ = input.stream.pause(); - input.running = false; - } - self.input = name(µphone).unwrap_or_default(); - self.source = Source::Device(Some(input)); } + } - if self.sides.1 { - let speaker = find(&host, self.want_output.as_deref(), false).ok_or(AudioError::NoSpeaker)?; - self.sink = Sink::Device(Some(open_output(&speaker, &self.broken)?)); - self.output = name(&speaker).unwrap_or_default(); + pub(crate) fn device_broken(&self) -> bool { + self.broken.load(Ordering::Relaxed) + } + + pub(crate) fn sides(&self) -> (bool, bool) { + self.sides + } + + /// Opens the wanted devices again: at start, before the 20 ms loop is running. + pub fn reopen(&mut self) -> Result<(), AudioError> { + if !self.on_devices() { + return Ok(()); } + let fresh = open_fresh( + self.want_input.as_deref(), + self.want_output.as_deref(), + self.sides, + self.mic_running(), + )?; + drop(self.install(fresh)); Ok(()) } + /// Takes devices that were opened off the call thread. The returned streams are the ones + /// just replaced; drop them off the 20 ms thread. + pub(crate) fn install(&mut self, fresh: Fresh) -> Retired { + let retired = Retired { + source: std::mem::replace(&mut self.source, fresh.source), + sink: std::mem::replace(&mut self.sink, fresh.sink), + }; + self.broken = fresh.broken; + self.input = fresh.input; + self.output = fresh.output; + self.captured.clear(); + self.scratch.clear(); + retired + } + /// True when a device stopped, or the system default we follow moved to another device. pub fn moved(&self) -> bool { - if !matches!(self.source, Source::Device(_)) { + if !self.on_devices() { return false; } - if self.broken.load(Ordering::Relaxed) { - return true; - } - let host = cpal::default_host(); - let default_moved = |wanted: &Option, current: &str, device: Option| { - wanted.is_none() && device.as_ref().and_then(name).is_some_and(|name| name != current) - }; - (self.sides.0 && default_moved(&self.want_input, &self.input, host.default_input_device())) - || (self.sides.1 && default_moved(&self.want_output, &self.output, host.default_output_device())) + self.device_broken() + || route_moved( + &self.want_input, + &self.input, + &self.want_output, + &self.output, + self.sides, + ) } /// Stops or starts the microphone itself, so a muted call shows no recording indicator. pub fn listen(&mut self, on: bool) { if let Source::Device(Some(input)) = &mut self.source { if input.running != on { - let changed = if on { input.stream.play() } else { input.stream.pause() }; + let changed = if on { + input.stream.play() + } else { + input.stream.pause() + }; if let Err(err) = changed { eprintln!("mediad: microphone: {err}"); } @@ -471,7 +666,8 @@ impl Audio { Source::Tone { phase } => { for pair in frame.as_chunks_mut::().0 { pair.fill((phase.sin() * 8000.) as i16); - *phase = (*phase + 440. * std::f64::consts::TAU / f64::from(RATE)) % std::f64::consts::TAU; + *phase = (*phase + 440. * std::f64::consts::TAU / f64::from(RATE)) + % std::f64::consts::TAU; } true } @@ -520,16 +716,16 @@ impl Audio { } Sink::Device(Some(output)) => { self.scratch.clear(); - self.scratch.extend(frame.iter().map(|&s| f32::from(s) / 32768.)); + self.scratch + .extend(frame.iter().map(|&s| f32::from(s) / 32768.)); let mut resampled = Vec::with_capacity(FRAME_SAMPLES * 2); output.resampler.push(&self.scratch, &mut resampled); let mut playout = output.playout.lock().unwrap(); playout.ring.extend(resampled); if playout.ring.len() > output.room { // Our clock runs ahead of the device's. Skip to the cushion: one short - // jump now rather than a delay that grows. - let over = playout.ring.len() - output.fill; - playout.ring.drain(..over - over % CHANNELS); + // ramp now rather than a delay that grows. + playout.trim_front(output.fill, output.fade); } } } @@ -564,7 +760,8 @@ impl Speaker { }; if Audio::simulated() { if let Some(path) = std::env::var_os("TINYCLIENT_MEDIAD_SINK") { - speaker.sink = Sink::File(File::create(path).map_err(|e| AudioError::Speaker(e.to_string()))?); + speaker.sink = + Sink::File(File::create(path).map_err(|e| AudioError::Speaker(e.to_string()))?); } return Ok(speaker); } @@ -601,8 +798,10 @@ impl Speaker { match &mut self.sink { Sink::Null | Sink::Device(None) => {} Sink::File(file) => { - let bytes: Vec = - pcm.iter().flat_map(|s| ((s.clamp(-1., 1.) * 32767.) as i16).to_le_bytes()).collect(); + let bytes: Vec = pcm + .iter() + .flat_map(|s| ((s.clamp(-1., 1.) * 32767.) as i16).to_le_bytes()) + .collect(); let _ = file.write_all(&bytes); } Sink::Device(Some(output)) => { @@ -643,7 +842,9 @@ mod tests { use super::*; fn tone(rate: u32, hertz: f32, frames: usize) -> Vec { - (0..frames).flat_map(|n| [(n as f32 * hertz * std::f32::consts::TAU / rate as f32).sin(); 2]).collect() + (0..frames) + .flat_map(|n| [(n as f32 * hertz * std::f32::consts::TAU / rate as f32).sin(); 2]) + .collect() } /// Resamples in uneven pieces, as a device delivers them. @@ -660,7 +861,10 @@ mod tests { fn worst(output: &[f32], rate: u32, hertz: f32) -> f32 { let frames = output.len() / 2; (200..frames) - .map(|n| (output[2 * n] - (n as f32 * hertz * std::f32::consts::TAU / rate as f32).sin()).abs()) + .map(|n| { + (output[2 * n] - (n as f32 * hertz * std::f32::consts::TAU / rate as f32).sin()) + .abs() + }) .fold(0f32, f32::max) } @@ -678,7 +882,12 @@ mod tests { #[test] fn resampling_keeps_high_frequencies_that_a_straight_line_would_bend() { // 9 kHz is where consonants live; linear interpolation is off by a quarter there. - for (from, to) in [(44_100, 48_000), (48_000, 44_100), (16_000, 48_000), (96_000, 48_000)] { + for (from, to) in [ + (44_100, 48_000), + (48_000, 44_100), + (16_000, 48_000), + (96_000, 48_000), + ] { let hertz = if from == 16_000 { 5_000. } else { 9_000. }; let output = resampled(from, to, &tone(from, hertz, from as usize / 10)); let off = worst(&output, to, hertz); @@ -697,7 +906,9 @@ mod tests { #[test] fn a_steady_level_stays_steady_at_every_phase() { let output = resampled(44_100, 48_000, &vec![0.5; 2 * 4410]); - let (low, high) = output[200..].iter().fold((1f32, 0f32), |(low, high), &s| (low.min(s), high.max(s))); + let (low, high) = output[200..] + .iter() + .fold((1f32, 0f32), |(low, high), &s| (low.min(s), high.max(s))); assert!(low > 0.4999 && high < 0.5001, "{low} to {high}"); } @@ -706,14 +917,19 @@ mod tests { let (fill, fade) = (8, 2); let mut playout = Playout::default(); playout.ring.extend([0.5; 6]); - assert_eq!(playout.next(fill, fade), (0., 0.), "three frames held, four wanted"); + assert_eq!( + playout.next(fill, fade), + (0., 0.), + "three frames held, four wanted" + ); assert_eq!(playout.ring.len(), 6, "and nothing was taken"); playout.ring.extend([0.5; 2]); // Playing starts faded in, so even a loud first sample does not click. assert_eq!(playout.next(fill, fade), (0., 0.)); assert_eq!(playout.next(fill, fade), (0.25, 0.25)); - assert_eq!(playout.next(fill, fade), (0.5, 0.5)); - assert_eq!(playout.next(fill, fade), (0.5, 0.5)); + // The last two frames ramp down, so the underrun that follows is already silence. + assert_eq!(playout.next(fill, fade), (0.25, 0.25)); + assert_eq!(playout.next(fill, fade), (0., 0.)); // Dry: silence, and the cushion is built again before anything plays. assert_eq!(playout.next(fill, fade), (0., 0.)); assert_eq!(playout.underruns, 1); @@ -722,6 +938,48 @@ mod tests { assert_eq!(playout.ring.len(), 4); } + /// The biggest jump between one played sample and the next. + fn steepest(played: &[(f32, f32)]) -> f32 { + played.windows(2).fold(0f32, |steepest, pair| { + let step = (pair[1].0 - pair[0].0) + .abs() + .max((pair[1].1 - pair[0].1).abs()); + steepest.max(step) + }) + } + + #[test] + fn running_dry_ramps_to_silence_instead_of_stepping() { + let (fill, fade) = (4, 8); + let mut playout = Playout::default(); + playout.ring.extend([1.; 40]); + let mut played = Vec::new(); + for _ in 0..24 { + played.push(playout.next(fill, fade)); + } + assert_eq!(playout.underruns, 1); + assert!(played.last() == Some(&(0., 0.))); + let step = steepest(&played); + assert!(step <= 1. / fade as f32 + 1e-4, "a step of {step}"); + } + + #[test] + fn skipping_ahead_crossfades_the_splice() { + let mut playout = Playout::default(); + playout.ring.extend([0.; 20]); + playout.ring.extend([1.; 8]); + playout.trim_front(8, 4); + assert_eq!(playout.ring.len(), 8); + let samples: Vec = playout.ring.iter().copied().collect(); + let played: Vec<(f32, f32)> = samples + .chunks(2) + .map(|frame| (frame[0], frame[1])) + .collect(); + let step = steepest(&played); + assert!(step < 0.25, "a splice of {step}: {played:?}"); + assert!(played[0].0 < 0.3 && played[3].0 > 0.7, "{played:?}"); + } + #[test] fn level_is_zero_for_silence_and_rises_with_loudness() { assert_eq!(level(&[0; 1920]), 0); diff --git a/crates/mediad/src/media.rs b/crates/mediad/src/media.rs index aa3a671..a71288a 100644 --- a/crates/mediad/src/media.rs +++ b/crates/mediad/src/media.rs @@ -1,8 +1,9 @@ -//! The 20 ms loop of a joined call. One thread owns the UDP socket, the codecs and the devices: -//! between ticks it takes packets off the socket into each sender's jitter buffer; on the tick -//! it sends one frame of the microphone and plays one frame of everyone mixed. +//! The 20 ms loop of a joined call. One thread owns the UDP socket and the codecs: between ticks +//! it takes packets off the socket into each sender's jitter buffer; on the tick it sends one +//! frame of the microphone and plays one frame of everyone mixed. Opening or replacing an audio +//! device happens on another thread. The tick only swaps in a device that is already open. -use crate::audio::{level, Audio, AudioError}; +use crate::audio::{level, open_fresh, route_moved, Audio, AudioError, Fresh, Retired}; use crate::cue::Playing; use crate::dave::{Dave, Outgoing}; use crate::jitter::{Jitter, Pop}; @@ -16,8 +17,9 @@ use std::collections::{HashMap, VecDeque}; use std::io::ErrorKind; use std::net::UdpSocket; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::mpsc; +use std::sync::mpsc::{self, TryRecvError}; use std::sync::{Arc, Mutex}; +use std::thread; use std::time::{Duration, Instant}; const TICK: Duration = Duration::from_millis(20); @@ -80,8 +82,10 @@ struct Stream { rebuilt: u64, } -pub const NO_MICROPHONE: &str = "Plug in a microphone, or choose another in Microphone and speaker."; -pub const NO_SPEAKER: &str = "Plug in headphones or a speaker, or choose another in Microphone and speaker."; +pub const NO_MICROPHONE: &str = + "Plug in a microphone, or choose another in Microphone and speaker."; +pub const NO_SPEAKER: &str = + "Plug in headphones or a speaker, or choose another in Microphone and speaker."; pub const MICROPHONE_FAULT: &str = "The microphone didn't start. Reconnect it or choose another in Microphone and speaker, then try again."; pub const SPEAKER_FAULT: &str = @@ -117,6 +121,10 @@ struct Loop { levels: (u8, u8), /// The cue being played under the voices. cue: Option, + /// Opens devices away from this thread. + devices: DeviceDesk, + /// Housekeeping asked for a reopen. The tick posts it; it does not open anything itself. + reopen_wanted: bool, } impl Loop { @@ -133,7 +141,9 @@ impl Loop { let stream = match self.streams.entry(packet.ssrc) { std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(), std::collections::hash_map::Entry::Vacant(entry) => { - let Some(decoder) = Decoder::new() else { return }; + let Some(decoder) = Decoder::new() else { + return; + }; entry.insert(Stream { jitter: Jitter::default(), decoder, @@ -153,9 +163,20 @@ impl Loop { stream.heard |= packet.payload != SILENCE; return; } - let user = self.setup.users.lock().unwrap().get(&packet.ssrc).and_then(|id| id.parse().ok()); + let user = self + .setup + .users + .lock() + .unwrap() + .get(&packet.ssrc) + .and_then(|id| id.parse().ok()); let mut again = Vec::new(); - let frame = self.setup.dave.lock().unwrap().decrypt(user, &packet.payload, &mut again); + let frame = self + .setup + .dave + .lock() + .unwrap() + .decrypt(user, &packet.payload, &mut again); if !again.is_empty() { let _ = self.setup.report.send(FromMedia::Dave(again)); } @@ -177,7 +198,10 @@ impl Loop { fn say(&self, speaking: bool) { let _ = self.setup.report.send(FromMedia::Speaking(speaking)); - out::send(&MediaResponse::Speaking { user_id: self.setup.user_id.clone(), speaking }); + out::send(&MediaResponse::Speaking { + user_id: self.setup.user_id.clone(), + speaking, + }); } fn encode(&mut self, frame: &[i16; FRAME_SAMPLES]) { @@ -264,7 +288,10 @@ impl Loop { Pop::Packet(packet) => stream.decoder.decode(Some(&packet), 0, &mut pcm), Pop::Lost => { // The packet after the hole may carry a copy of what was lost. - let rebuilt = stream.jitter.next().and_then(|next| stream.decoder.recover(next, &mut pcm)); + let rebuilt = stream + .jitter + .next() + .and_then(|next| stream.decoder.recover(next, &mut pcm)); stream.rebuilt += u64::from(rebuilt.is_some()); rebuilt.or_else(|| stream.decoder.decode(None, FRAME, &mut pcm)) } @@ -336,26 +363,83 @@ impl Loop { } fn devices(&self) { - let (input, output) = (self.setup.audio.input.clone(), self.setup.audio.output.clone()); + let (input, output) = ( + self.setup.audio.input.clone(), + self.setup.audio.output.clone(), + ); out::send(&MediaResponse::Devices { input, output }); } - /// Opens the devices again. A named device that is gone gives way to the system default, and - /// only having no device at all ends the call. - fn reopen(&mut self) -> bool { - let audio = &mut self.setup.audio; - let opened = audio.reopen().or_else(|err| { - eprintln!("mediad: {err:?}; falling back to the system default devices"); - audio.want_input = None; - audio.want_output = None; - audio.reopen() - }); - match opened { - Ok(()) => { + /// Takes a finished reopen, if one is ready, and posts the next one. Never opens a device + /// itself: that work is already running, or is posted here and finishes on a later tick. + fn service_devices(&mut self) -> bool { + loop { + let note = match self.devices.notes.try_recv() { + Ok(note) => note, + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => return false, + }; + self.devices.busy = false; + if !self.apply_devices(note) { + return false; + } + if let Some(ask) = self.devices.queued.take() { + let _ = self.devices.orders.send(Order::Open(ask)); + self.devices.busy = true; + } + } + if self.devices.busy { + return true; + } + let chosen = self.setup.controls.devices.lock().unwrap().take(); + if let Some((input, output)) = chosen { + self.setup.audio.want_input = input; + self.setup.audio.want_output = output; + self.reopen_wanted = true; + } + if self.setup.audio.device_broken() { + self.reopen_wanted = true; + } + let poll_route = self.ticks.is_multiple_of(SECOND) && self.setup.audio.on_devices(); + if self.reopen_wanted || poll_route { + let audio = &self.setup.audio; + let ask = OpenAsk { + want_input: audio.want_input.clone(), + want_output: audio.want_output.clone(), + current_input: audio.input.clone(), + current_output: audio.output.clone(), + sides: audio.sides(), + broken: audio.device_broken(), + force: self.reopen_wanted, + mic_running: audio.mic_running(), + simulated: !audio.on_devices(), + pause: Duration::ZERO, + }; + self.reopen_wanted = false; + let _ = self.devices.orders.send(Order::Open(ask)); + self.devices.busy = true; + } + true + } + + fn apply_devices(&mut self, note: Note) -> bool { + match note { + Note::Fresh { fresh, fell_back } => { + if fell_back { + self.setup.audio.want_input = None; + self.setup.audio.want_output = None; + } + let retired = self.setup.audio.install(fresh); + let _ = self.devices.orders.send(Order::Retire(retired)); self.devices(); true } - Err(err) => { + Note::Same => { + self.devices(); + true + } + Note::Quiet => true, + Note::Failed(err) => { let (reason, detail) = failure(err); let _ = self.setup.report.send(FromMedia::Fatal(reason, detail)); false @@ -363,18 +447,8 @@ impl Loop { } } - /// Once a second: what does not need 20 ms timing. + /// Once a second: what does not need 20 ms timing. Device reopen is not in here. fn housekeeping(&mut self) -> bool { - let chosen = self.setup.controls.devices.lock().unwrap().take(); - if let Some((input, output)) = chosen { - self.setup.audio.want_input = input; - self.setup.audio.want_output = output; - if !self.reopen() { - return false; - } - } else if self.setup.audio.moved() && !self.reopen() { - return false; - } if self.setup.microphone_unasked { match microphone() { Some(Microphone::Denied) => { @@ -388,9 +462,7 @@ impl Loop { // on hearing silence, so the devices are opened afresh. self.setup.microphone_unasked = false; eprintln!("mediad: microphone access was granted"); - if !self.reopen() { - return false; - } + self.reopen_wanted = true; } } } @@ -402,7 +474,9 @@ impl Loop { let _ = self.setup.socket.send(&probe); } let users = self.setup.users.lock().unwrap(); - self.streams.retain(|ssrc, stream| users.contains_key(ssrc) || stream.last_packet.elapsed() < Duration::from_secs(60)); + self.streams.retain(|ssrc, stream| { + users.contains_key(ssrc) || stream.last_packet.elapsed() < Duration::from_secs(60) + }); true } @@ -413,10 +487,14 @@ impl Loop { while !self.setup.stop.load(Ordering::Relaxed) { let now = Instant::now(); if now < next { - let _ = self.setup.socket.set_read_timeout(Some((next - now).max(Duration::from_micros(200)))); + let _ = self + .setup + .socket + .set_read_timeout(Some((next - now).max(Duration::from_micros(200)))); match self.setup.socket.recv(&mut datagram) { Ok(len) => self.receive(&datagram[..len]), - Err(err) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {} + Err(err) + if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {} // An ICMP error surfaces here once per packet we sent; it is not ours to act on. Err(_) => std::thread::sleep(Duration::from_millis(1)), } @@ -430,13 +508,19 @@ impl Loop { self.ticks += 1; let input = self.capture(); let output = self.play(); - if self.ticks.is_multiple_of(5) && self.setup.controls.metering() && (input, output) != self.levels { + if self.ticks.is_multiple_of(5) + && self.setup.controls.metering() + && (input, output) != self.levels + { self.levels = (input, output); out::send(&MediaResponse::Levels { input, output }); } if self.ticks.is_multiple_of(SECOND) && !self.housekeeping() { break; } + if !self.service_devices() { + break; + } } if self.talking { for _ in 0..TRAIL { @@ -446,7 +530,10 @@ impl Loop { } for (ssrc, stream) in &self.streams { let (jitter, rebuilt) = (&stream.jitter, stream.rebuilt); - eprintln!("mediad: ssrc {ssrc}: {} lost ({rebuilt} rebuilt), {} late", jitter.lost, jitter.late); + eprintln!( + "mediad: ssrc {ssrc}: {} lost ({rebuilt} rebuilt), {} late", + jitter.lost, jitter.late + ); } } } @@ -456,7 +543,10 @@ impl Loop { pub fn run(setup: Setup) -> Audio { let Some(encoder) = Encoder::new() else { eprintln!("mediad: the Opus encoder could not be created"); - let fatal = FromMedia::Fatal(CallFailure::Engine, "The call engine could not start. Try again.".to_owned()); + let fatal = FromMedia::Fatal( + CallFailure::Engine, + "The call engine could not start. Try again.".to_owned(), + ); let _ = setup.report.send(fatal); return setup.audio; }; @@ -472,7 +562,170 @@ pub fn run(setup: Setup) -> Audio { keepalive: 0, levels: (0, 0), cue: None, + devices: DeviceDesk::start(), + reopen_wanted: false, }; state.run(); + state.devices.stop(); state.setup.audio } + +/// What the call thread asks the device thread to do. +enum Order { + Open(OpenAsk), + Retire(Retired), + Stop, +} + +struct OpenAsk { + want_input: Option, + want_output: Option, + current_input: String, + current_output: String, + sides: (bool, bool), + broken: bool, + /// The user picked a device, the callback reported a failure, or permission was just granted. + force: bool, + mic_running: bool, + simulated: bool, + /// Test-only wait before the open, so the caller can show the tick is not inside it. + pause: Duration, +} + +enum Note { + Fresh { + fresh: Fresh, + fell_back: bool, + }, + /// Stand-in devices: nothing was opened, and the call should say which names it is using. + Same, + /// The route was checked and it has not moved. + Quiet, + Failed(AudioError), +} + +fn satisfy(ask: OpenAsk) -> Note { + if !ask.pause.is_zero() { + thread::sleep(ask.pause); + } + if ask.simulated { + return if ask.force { Note::Same } else { Note::Quiet }; + } + let moved = ask.broken + || route_moved( + &ask.want_input, + &ask.current_input, + &ask.want_output, + &ask.current_output, + ask.sides, + ); + if !ask.force && !moved { + return Note::Quiet; + } + match open_fresh( + ask.want_input.as_deref(), + ask.want_output.as_deref(), + ask.sides, + ask.mic_running, + ) { + Ok(fresh) => Note::Fresh { + fresh, + fell_back: false, + }, + Err(err) => { + eprintln!("mediad: {err:?}; falling back to the system default devices"); + match open_fresh(None, None, ask.sides, ask.mic_running) { + Ok(fresh) => Note::Fresh { + fresh, + fell_back: true, + }, + Err(err) => Note::Failed(err), + } + } + } +} + +struct DeviceDesk { + orders: mpsc::Sender, + notes: mpsc::Receiver, + busy: bool, + queued: Option, + join: Option>, +} + +impl DeviceDesk { + fn start() -> DeviceDesk { + let (orders_tx, orders_rx) = mpsc::channel(); + let (notes_tx, notes_rx) = mpsc::channel(); + let join = thread::Builder::new() + .name("mediad-devices".to_owned()) + .spawn(move || { + while let Ok(order) = orders_rx.recv() { + match order { + Order::Open(ask) => { + let _ = notes_tx.send(satisfy(ask)); + } + Order::Retire(retired) => drop(retired), + Order::Stop => break, + } + } + }) + .expect("the device thread starts"); + DeviceDesk { + orders: orders_tx, + notes: notes_rx, + busy: false, + queued: None, + join: Some(join), + } + } + + fn stop(&mut self) { + let _ = self.orders.send(Order::Stop); + if let Some(join) = self.join.take() { + let _ = join.join(); + } + } +} + +impl Drop for DeviceDesk { + fn drop(&mut self) { + self.stop(); + } +} + +#[cfg(test)] +mod device_tests { + use super::*; + + #[test] + fn reopening_a_device_does_not_block_the_caller() { + let desk = DeviceDesk::start(); + let started = Instant::now(); + desk.orders + .send(Order::Open(OpenAsk { + want_input: None, + want_output: None, + current_input: String::new(), + current_output: String::new(), + sides: (true, true), + broken: false, + force: true, + mic_running: true, + simulated: true, + pause: Duration::from_millis(200), + })) + .unwrap(); + assert!(desk.notes.try_recv().is_err(), "the open has not finished"); + assert!( + started.elapsed() < Duration::from_millis(50), + "posting the open stalled the caller" + ); + let note = desk + .notes + .recv_timeout(Duration::from_millis(500)) + .expect("the open finished"); + assert!(matches!(note, Note::Same)); + drop(desk); + } +}