perf(core): coalesce continuous-control commands, last-value-wins (A15b)

A fast slider drag could burst past the bounded(100) best-effort command queue
and try_send would drop commands -- possibly the FINAL value of the drag, leaving
a gain/pan/volume stuck mid-drag until the next interaction. Replace the
best-effort queue with a coalescing latest-value map keyed by control
(CoalesceKey) plus a bounded(1) wake channel: send() overwrites the latest value
per control (never drops, never blocks) and wakes the loop, which pops one
coalesced command at a time and self-re-arms while entries remain. The existing
single-command match handler is reused unchanged.

command_sender() now returns a typed CoreCommandSender that routes by
delivery_class, so the window-close Shutdown (Reliable) goes through the
unbounded reliable channel (drained biased-first) instead of the best-effort
path -- a small correctness improvement. Mute/PTT remain Reliable, untouched.

Pure seams coalesce_key/coalesce_insert/coalesce_pop with unit tests
(overwrite-same-key, distinct-peers, global control, empty pop, drain-each-once)
and a coalesce_key<->BestEffort invariant assertion. No new deps, no wire change.
Codex-implemented (gpt-5.5 xhigh), senior-reviewed; tests-green (487 lib).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-29 01:42:41 -04:00
co-authored by Claude Opus 4.8
parent 10bd15aeaa
commit 8c33b5c70f
2 changed files with 351 additions and 36 deletions
+128 -9
View File
@@ -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<CoalesceKey> {
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
);
}
}
}
+223 -27
View File
@@ -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<StdMutex<HashMap<CoalesceKey, CoreCommand>>>;
pub struct CoreController {
reliable_tx: mpsc::UnboundedSender<CoreCommand>,
besteffort_tx: mpsc::Sender<CoreCommand>,
coalesce: CoalesceStore,
besteffort_wake_tx: mpsc::Sender<()>,
}
#[derive(Clone)]
pub struct CoreCommandSender {
reliable_tx: mpsc::UnboundedSender<CoreCommand>,
coalesce: CoalesceStore,
besteffort_wake_tx: mpsc::Sender<()>,
}
fn coalesce_insert(map: &mut HashMap<CoalesceKey, CoreCommand>, cmd: CoreCommand) -> bool {
let Some(key) = coalesce_key(&cmd) else {
return false;
};
map.insert(key, cmd);
true
}
fn coalesce_pop(map: &mut HashMap<CoalesceKey, CoreCommand>) -> Option<CoreCommand> {
let key = map.keys().next().cloned()?;
map.remove(&key)
}
impl CoreCommandSender {
pub fn send(&self, cmd: CoreCommand) -> std::future::Ready<Result<(), CoreCommand>> {
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<UiEvent>) -> 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<CoreCommand> {
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<CoreCommand>,
mut besteffort_rx: mpsc::Receiver<CoreCommand>,
coalesce: CoalesceStore,
mut besteffort_wake_rx: mpsc::Receiver<()>,
besteffort_wake_tx: mpsc::Sender<()>,
ui_tx: mpsc::Sender<UiEvent>,
) -> 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.