diff --git a/src/core/messages.rs b/src/core/messages.rs index e163a06..b3f6eec 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -130,6 +130,17 @@ pub enum DeliveryClass { BestEffort, } +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum CoalesceKey { + InputVolume, + OutputVolume, + NoiseGate, + PeerVolume(EndpointId), + PeerPan(EndpointId), + PeerGate(EndpointId), + PeerEq(EndpointId), +} + /// Route a command by how bad it is to drop it. Discrete, human-paced user /// actions are Reliable (must land). The only high-frequency commands are the /// continuous audio sliders, where dropping intermediate values is harmless; @@ -215,6 +226,89 @@ pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass { } } +/// Coalescing bucket for high-frequency continuous controls. A key exists +/// exactly for [`DeliveryClass::BestEffort`] commands. +pub fn coalesce_key(cmd: &CoreCommand) -> Option { + match cmd { + CoreCommand::SetPeerVolume(peer_id, _) => Some(CoalesceKey::PeerVolume(*peer_id)), + CoreCommand::SetPeerPan(peer_id, _) => Some(CoalesceKey::PeerPan(*peer_id)), + CoreCommand::SetPeerGate(peer_id, _) => Some(CoalesceKey::PeerGate(*peer_id)), + CoreCommand::SetPeerEq(peer_id, _) => Some(CoalesceKey::PeerEq(*peer_id)), + CoreCommand::SetInputVolume(_) => Some(CoalesceKey::InputVolume), + CoreCommand::SetOutputVolume(_) => Some(CoalesceKey::OutputVolume), + CoreCommand::SetNoiseGateThreshold(_) => Some(CoalesceKey::NoiseGate), + + CoreCommand::Join { + name: _, + ticket: _, + room_name: _, + input_device: _, + output_device: _, + echo_cancellation: _, + avatar: _, + } + | CoreCommand::Leave + | CoreCommand::Shutdown + | CoreCommand::ToggleMute + | CoreCommand::SetAvatar(_) + | CoreCommand::ToggleDeafen + | CoreCommand::SetPttMode(_) + | CoreCommand::SetPttActive(_) + | CoreCommand::SetPeerMuted(_, _) + | CoreCommand::SetMicMonitor { + enabled: _, + input_device: _, + } + | CoreCommand::SetNetworkMode(_) + | CoreCommand::SetRecording(_) + | CoreCommand::SetRecordingMode(_) + | CoreCommand::SendChat(_) + | CoreCommand::SendChatFile { + text: _, + attachment: _, + data: _, + } + | CoreCommand::FetchAttachment { + from: _, + attachment: _, + } + | CoreCommand::ServeMusicTrack { + id: _, + data: _, + } + | CoreCommand::ForgetMusicTrack(_) + | CoreCommand::SetMusicPresence(_) + | CoreCommand::FetchMusic { + from: _, + id: _, + size: _, + } + | CoreCommand::PrefetchMusic { + from: _, + id: _, + size: _, + } + | CoreCommand::SetPixelpassPath(_) + | CoreCommand::ListAudioApps + | CoreCommand::StartScreenShare { audio_app: _ } + | CoreCommand::StopScreenShare + | CoreCommand::ViewShare(_) + | CoreCommand::RegenerateIdentity + | CoreCommand::AddFriend { + id: _, + name: _, + addr: _, + } + | CoreCommand::RemoveFriend(_) + | CoreCommand::RenameFriend(_, _) + | CoreCommand::RefreshFriends + | CoreCommand::SetPresenceMode(_) + | CoreCommand::SetGamePresenceEnabled(_) + | CoreCommand::SetGameOverride(_) + | CoreCommand::SetGameProcessMap(_) => None, + } +} + #[derive(Debug, Clone)] pub enum UiEvent { RoomJoined { ticket: String, self_id: String }, @@ -319,7 +413,7 @@ pub enum UiEvent { #[cfg(test)] mod tests { - use super::{delivery_class, CoreCommand, DeliveryClass}; + use super::{coalesce_key, delivery_class, CoalesceKey, CoreCommand, DeliveryClass}; use crate::audio::eq::EqSettings; use crate::presence::PresenceMode; use iroh::{EndpointId, SecretKey}; @@ -332,17 +426,37 @@ mod tests { fn continuous_audio_controls_are_best_effort() { let peer = endpoint_id(); let commands = [ - CoreCommand::SetPeerVolume(peer, 0.7), - CoreCommand::SetPeerPan(peer, -0.2), - CoreCommand::SetPeerGate(peer, 0.1), - CoreCommand::SetPeerEq(peer, EqSettings::default()), - CoreCommand::SetInputVolume(0.8), - CoreCommand::SetOutputVolume(0.9), - CoreCommand::SetNoiseGateThreshold(0.02), + ( + CoreCommand::SetPeerVolume(peer, 0.7), + CoalesceKey::PeerVolume(peer), + ), + ( + CoreCommand::SetPeerPan(peer, -0.2), + CoalesceKey::PeerPan(peer), + ), + ( + CoreCommand::SetPeerGate(peer, 0.1), + CoalesceKey::PeerGate(peer), + ), + ( + CoreCommand::SetPeerEq(peer, EqSettings::default()), + CoalesceKey::PeerEq(peer), + ), + (CoreCommand::SetInputVolume(0.8), CoalesceKey::InputVolume), + (CoreCommand::SetOutputVolume(0.9), CoalesceKey::OutputVolume), + ( + CoreCommand::SetNoiseGateThreshold(0.02), + CoalesceKey::NoiseGate, + ), ]; - for cmd in commands { + for (cmd, key) in commands { assert_eq!(delivery_class(&cmd), DeliveryClass::BestEffort); + assert_eq!(coalesce_key(&cmd), Some(key)); + assert_eq!( + coalesce_key(&cmd).is_some(), + delivery_class(&cmd) == DeliveryClass::BestEffort + ); } } @@ -370,6 +484,11 @@ mod tests { for cmd in commands { assert_eq!(delivery_class(&cmd), DeliveryClass::Reliable); + assert_eq!(coalesce_key(&cmd), None); + assert_eq!( + coalesce_key(&cmd).is_some(), + delivery_class(&cmd) == DeliveryClass::BestEffort + ); } } } diff --git a/src/core/mod.rs b/src/core/mod.rs index cc56f5a..dba9471 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -11,7 +11,9 @@ use crate::network::{ iroh_impl::{IrohTransport, AudioRouter, FileRouter}, gossip::IrohGossipState, }; -use crate::core::messages::{CoreCommand, DeliveryClass, UiEvent, delivery_class}; +use crate::core::messages::{ + CoalesceKey, CoreCommand, DeliveryClass, UiEvent, coalesce_key, delivery_class, +}; use crate::core::recovery::RecoveryCoordinator; use crate::config::{NetworkMode, RecordingMode}; @@ -21,49 +23,115 @@ use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::p use iroh_gossip::net::Gossip; use tokio::sync::{mpsc, Mutex}; use std::collections::{HashMap, HashSet}; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; +type CoalesceStore = Arc>>; + pub struct CoreController { reliable_tx: mpsc::UnboundedSender, - besteffort_tx: mpsc::Sender, + coalesce: CoalesceStore, + besteffort_wake_tx: mpsc::Sender<()>, +} + +#[derive(Clone)] +pub struct CoreCommandSender { + reliable_tx: mpsc::UnboundedSender, + coalesce: CoalesceStore, + besteffort_wake_tx: mpsc::Sender<()>, +} + +fn coalesce_insert(map: &mut HashMap, cmd: CoreCommand) -> bool { + let Some(key) = coalesce_key(&cmd) else { + return false; + }; + map.insert(key, cmd); + true +} + +fn coalesce_pop(map: &mut HashMap) -> Option { + let key = map.keys().next().cloned()?; + map.remove(&key) +} + +impl CoreCommandSender { + pub fn send(&self, cmd: CoreCommand) -> std::future::Ready> { + std::future::ready(match delivery_class(&cmd) { + DeliveryClass::Reliable => self.reliable_tx.send(cmd).map_err(|e| e.0), + DeliveryClass::BestEffort => { + let inserted = { + let mut map = self.coalesce.lock().unwrap(); + coalesce_insert(&mut map, cmd) + }; + debug_assert!(inserted); + let _ = self.besteffort_wake_tx.try_send(()); + Ok(()) + } + }) + } } impl CoreController { pub fn new(ui_tx: mpsc::Sender) -> Self { let (reliable_tx, reliable_rx) = mpsc::unbounded_channel(); - let (besteffort_tx, besteffort_rx) = mpsc::channel(100); + let coalesce = Arc::new(StdMutex::new(HashMap::new())); + let (besteffort_wake_tx, besteffort_wake_rx) = mpsc::channel(1); + let loop_coalesce = coalesce.clone(); + let loop_besteffort_wake_tx = besteffort_wake_tx.clone(); std::thread::spawn(move || { let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime"); rt.block_on(async move { crate::log_msg("Starting core network loop in dedicated Tokio runtime"); - if let Err(e) = run_core_loop(reliable_rx, besteffort_rx, ui_tx).await { + if let Err(e) = run_core_loop( + reliable_rx, + loop_coalesce, + besteffort_wake_rx, + loop_besteffort_wake_tx, + ui_tx, + ) + .await + { crate::log_msg(&format!("App core loop failed: {:?}", e)); } }); }); - Self { reliable_tx, besteffort_tx } - } - - /// Queue a command for the core loop. Reliable commands only fail when the - /// core loop is dead; best-effort slider commands keep today's bounded - /// try-send behavior. (We return a plain bool rather than the channel's - /// `Result` so the bulky `CoreCommand` isn't carried back by value in every - /// caller's error type.) - pub fn send(&self, cmd: CoreCommand) -> bool { - match delivery_class(&cmd) { - DeliveryClass::Reliable => self.reliable_tx.send(cmd).is_ok(), - DeliveryClass::BestEffort => self.besteffort_tx.try_send(cmd).is_ok(), + Self { + reliable_tx, + coalesce, + besteffort_wake_tx, } } - /// Clone the command sender for asynchronous one-shot sends that should wait - /// for channel capacity instead of failing immediately on a full queue. - pub fn command_sender(&self) -> mpsc::Sender { - self.besteffort_tx.clone() + /// Queue a command for the core loop. Reliable commands only fail when the + /// core loop is dead; best-effort slider commands are coalesced by control and + /// never fail to record their latest value. (We return a plain bool rather + /// than the channel's `Result` so the bulky `CoreCommand` isn't carried back by + /// value in every caller's error type.) + pub fn send(&self, cmd: CoreCommand) -> bool { + match delivery_class(&cmd) { + DeliveryClass::Reliable => self.reliable_tx.send(cmd).is_ok(), + DeliveryClass::BestEffort => { + let inserted = { + let mut map = self.coalesce.lock().unwrap(); + coalesce_insert(&mut map, cmd) + }; + debug_assert!(inserted); + let _ = self.besteffort_wake_tx.try_send(()); + true + } + } + } + + /// Clone a command sender for asynchronous one-shot sends. + pub fn command_sender(&self) -> CoreCommandSender { + CoreCommandSender { + reliable_tx: self.reliable_tx.clone(), + coalesce: self.coalesce.clone(), + besteffort_wake_tx: self.besteffort_wake_tx.clone(), + } } } @@ -1017,7 +1085,9 @@ async fn probe_friends_once( async fn run_core_loop( mut reliable_rx: mpsc::UnboundedReceiver, - mut besteffort_rx: mpsc::Receiver, + coalesce: CoalesceStore, + mut besteffort_wake_rx: mpsc::Receiver<()>, + besteffort_wake_tx: mpsc::Sender<()>, ui_tx: mpsc::Sender, ) -> Result<(), anyhow::Error> { let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new(); @@ -1226,8 +1296,22 @@ async fn run_core_loop( Some(cmd) => cmd, None => break, }, - maybe_cmd = besteffort_rx.recv() => match maybe_cmd { - Some(cmd) => cmd, + maybe_wake = besteffort_wake_rx.recv() => match maybe_wake { + Some(()) => { + let (cmd, has_more) = { + let mut map = coalesce.lock().unwrap(); + let cmd = coalesce_pop(&mut map); + let has_more = !map.is_empty(); + (cmd, has_more) + }; + if has_more { + let _ = besteffort_wake_tx.try_send(()); + } + match cmd { + Some(cmd) => cmd, + None => continue, + } + } None => break, }, game_change = next_game_change(&mut game_rx) => { @@ -2938,10 +3022,16 @@ async fn run_core_loop( mod tests { use super::{ admit_retained, apply_peer_volume, apply_volume, audio_datagram_len_ok, frame_level, - mix_frames, mix_stereo_frames, next_game_change, should_auto_fetch, stereo_to_mono, - KnownPeers, MicLevelMeter, PeerSpeakTicket, MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, - MIC_LEVEL_REPORT_SAMPLES, + coalesce_insert, coalesce_pop, mix_frames, mix_stereo_frames, next_game_change, + should_auto_fetch, stereo_to_mono, KnownPeers, MicLevelMeter, PeerSpeakTicket, + MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, MIC_LEVEL_REPORT_SAMPLES, }; + use crate::core::messages::{coalesce_key, CoalesceKey, CoreCommand}; + use std::collections::{HashMap, HashSet}; + + fn endpoint_id() -> iroh::EndpointId { + iroh::SecretKey::generate().public() + } #[test] fn admit_retained_rejects_only_new_ids_at_the_cap() { @@ -2957,6 +3047,112 @@ mod tests { assert!(admit_retained(MAX_RETAINED_PEERS + 5, false, MAX_RETAINED_PEERS)); } + #[test] + fn coalesce_insert_overwrites_same_peer_control_with_latest_value() { + let peer = endpoint_id(); + let mut map = HashMap::new(); + + assert!(coalesce_insert( + &mut map, + CoreCommand::SetPeerVolume(peer, 0.2) + )); + assert!(coalesce_insert( + &mut map, + CoreCommand::SetPeerVolume(peer, 0.9) + )); + + assert_eq!(map.len(), 1); + match coalesce_pop(&mut map) { + Some(CoreCommand::SetPeerVolume(id, volume)) => { + assert_eq!(id, peer); + assert_eq!(volume, 0.9); + } + other => panic!("expected latest peer volume, got {other:?}"), + } + assert!(map.is_empty()); + } + + #[test] + fn coalesce_insert_keeps_distinct_peers_for_same_control() { + let peer_a = endpoint_id(); + let peer_b = endpoint_id(); + let mut map = HashMap::new(); + + assert!(coalesce_insert( + &mut map, + CoreCommand::SetPeerVolume(peer_a, 0.2) + )); + assert!(coalesce_insert( + &mut map, + CoreCommand::SetPeerVolume(peer_b, 0.9) + )); + + assert_eq!(map.len(), 2); + let mut seen = HashMap::new(); + while let Some(cmd) = coalesce_pop(&mut map) { + match cmd { + CoreCommand::SetPeerVolume(id, volume) => { + seen.insert(id, volume); + } + other => panic!("expected peer volume command, got {other:?}"), + } + } + assert_eq!(seen.get(&peer_a), Some(&0.2)); + assert_eq!(seen.get(&peer_b), Some(&0.9)); + } + + #[test] + fn coalesce_insert_overwrites_global_control_with_latest_value() { + let mut map = HashMap::new(); + + assert!(coalesce_insert(&mut map, CoreCommand::SetInputVolume(0.1))); + assert!(coalesce_insert(&mut map, CoreCommand::SetInputVolume(0.4))); + + assert_eq!(map.len(), 1); + match coalesce_pop(&mut map) { + Some(CoreCommand::SetInputVolume(volume)) => assert_eq!(volume, 0.4), + other => panic!("expected latest input volume, got {other:?}"), + } + assert!(map.is_empty()); + } + + #[test] + fn coalesce_pop_empty_map_returns_none() { + let mut map = HashMap::new(); + + assert!(coalesce_pop(&mut map).is_none()); + } + + #[test] + fn coalesce_pop_drains_each_distinct_key_once() { + let peer_a = endpoint_id(); + let peer_b = endpoint_id(); + let mut map = HashMap::new(); + + assert!(coalesce_insert( + &mut map, + CoreCommand::SetPeerPan(peer_a, -0.3) + )); + assert!(coalesce_insert( + &mut map, + CoreCommand::SetPeerGate(peer_b, 0.4) + )); + assert!(coalesce_insert(&mut map, CoreCommand::SetOutputVolume(0.8))); + + let expected = HashSet::from([ + CoalesceKey::PeerPan(peer_a), + CoalesceKey::PeerGate(peer_b), + CoalesceKey::OutputVolume, + ]); + let mut seen = HashSet::new(); + while let Some(cmd) = coalesce_pop(&mut map) { + seen.insert(coalesce_key(&cmd).expect("coalesced command should have a key")); + } + + assert_eq!(seen, expected); + assert!(map.is_empty()); + } + #[test] fn auto_fetch_only_for_roster_images_not_already_inflight() { // The happy path: a roster author's brand-new image attachment.