diff --git a/src/core/messages.rs b/src/core/messages.rs index 9dd0547..e882e10 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -109,6 +109,80 @@ pub enum CoreCommand { SetGameProcessMap(std::collections::BTreeMap), } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DeliveryClass { + Reliable, + BestEffort, +} + +/// 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; +/// those are BestEffort. +pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass { + match cmd { + CoreCommand::SetPeerVolume(_, _) + | CoreCommand::SetPeerPan(_, _) + | CoreCommand::SetPeerGate(_, _) + | CoreCommand::SetPeerEq(_, _) + | CoreCommand::SetInputVolume(_) + | CoreCommand::SetOutputVolume(_) + | CoreCommand::SetNoiseGateThreshold(_) => DeliveryClass::BestEffort, + + 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::SetPixelpassPath(_) + | CoreCommand::ListAudioApps + | CoreCommand::StartScreenShare { audio_app: _ } + | CoreCommand::StopScreenShare + | CoreCommand::ViewShare(_) + | CoreCommand::RegenerateIdentity + | CoreCommand::AddFriend { + id: _, + name: _, + addr: _, + } + | CoreCommand::RemoveFriend(_) + | CoreCommand::RenameFriend(_, _) + | CoreCommand::SetPresenceMode(_) + | CoreCommand::SetGamePresenceEnabled(_) + | CoreCommand::SetGameOverride(_) + | CoreCommand::SetGameProcessMap(_) => DeliveryClass::Reliable, + } +} + #[derive(Debug, Clone)] pub enum UiEvent { RoomJoined { ticket: String, self_id: String }, @@ -199,3 +273,60 @@ pub enum UiEvent { ShutdownComplete, Error(String), } + +#[cfg(test)] +mod tests { + use super::{delivery_class, CoreCommand, DeliveryClass}; + use crate::audio::eq::EqSettings; + use crate::presence::PresenceMode; + use iroh::{EndpointId, SecretKey}; + + fn endpoint_id() -> EndpointId { + SecretKey::generate().public() + } + + #[test] + 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), + ]; + + for cmd in commands { + assert_eq!(delivery_class(&cmd), DeliveryClass::BestEffort); + } + } + + #[test] + fn discrete_user_actions_are_reliable() { + let peer = endpoint_id(); + let commands = [ + CoreCommand::ToggleMute, + CoreCommand::SetPttActive(false), + CoreCommand::Leave, + CoreCommand::RegenerateIdentity, + CoreCommand::Join { + name: "Peer".to_string(), + ticket: "create".to_string(), + room_name: "Room".to_string(), + input_device: None, + output_device: None, + echo_cancellation: true, + avatar: crate::avatar::Avatar::default(), + }, + CoreCommand::SetPeerMuted(peer, true), + CoreCommand::SetPresenceMode(PresenceMode::Normal), + CoreCommand::SendChat("hello".to_string()), + ]; + + for cmd in commands { + assert_eq!(delivery_class(&cmd), DeliveryClass::Reliable); + } + } +} diff --git a/src/core/mod.rs b/src/core/mod.rs index b665081..35d6491 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -11,7 +11,7 @@ use crate::network::{ iroh_impl::{IrohTransport, AudioRouter, FileRouter}, gossip::IrohGossipState, }; -use crate::core::messages::{CoreCommand, UiEvent}; +use crate::core::messages::{CoreCommand, DeliveryClass, UiEvent, delivery_class}; use crate::core::recovery::RecoveryCoordinator; use crate::config::{NetworkMode, RecordingMode}; @@ -26,38 +26,44 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; pub struct CoreController { - cmd_tx: mpsc::Sender, + reliable_tx: mpsc::UnboundedSender, + besteffort_tx: mpsc::Sender, } impl CoreController { pub fn new(ui_tx: mpsc::Sender) -> Self { - let (cmd_tx, cmd_rx) = mpsc::channel(100); + let (reliable_tx, reliable_rx) = mpsc::unbounded_channel(); + let (besteffort_tx, besteffort_rx) = mpsc::channel(100); 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(cmd_rx, ui_tx).await { + if let Err(e) = run_core_loop(reliable_rx, besteffort_rx, ui_tx).await { crate::log_msg(&format!("App core loop failed: {:?}", e)); } }); }); - Self { cmd_tx } + Self { reliable_tx, besteffort_tx } } - /// Queue a command for the core loop, best-effort. Returns `true` if it was - /// accepted, `false` if the channel is full or closed. (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.) + /// 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 { - self.cmd_tx.try_send(cmd).is_ok() + match delivery_class(&cmd) { + DeliveryClass::Reliable => self.reliable_tx.send(cmd).is_ok(), + DeliveryClass::BestEffort => self.besteffort_tx.try_send(cmd).is_ok(), + } } /// 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.cmd_tx.clone() + self.besteffort_tx.clone() } } @@ -952,7 +958,8 @@ async fn probe_friends_once( } async fn run_core_loop( - mut cmd_rx: mpsc::Receiver, + mut reliable_rx: mpsc::UnboundedReceiver, + mut besteffort_rx: mpsc::Receiver, ui_tx: mpsc::Sender, ) -> Result<(), anyhow::Error> { let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new(); @@ -1155,7 +1162,12 @@ async fn run_core_loop( ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { let cmd = tokio::select! { - maybe_cmd = cmd_rx.recv() => match maybe_cmd { + biased; + maybe_cmd = reliable_rx.recv() => match maybe_cmd { + Some(cmd) => cmd, + None => break, + }, + maybe_cmd = besteffort_rx.recv() => match maybe_cmd { Some(cmd) => cmd, None => break, },