diff --git a/crates/collab-net/src/hub.rs b/crates/collab-net/src/hub.rs index b79eb1652..6fadf0946 100644 --- a/crates/collab-net/src/hub.rs +++ b/crates/collab-net/src/hub.rs @@ -737,6 +737,9 @@ impl Hub { continue; } let Some(&conn) = room.ids.get(&id) else { continue }; + if matches!(msg, ServerMsg::Presence { .. }) && self.conns.get(&conn).is_some_and(|c| !c.out.takes_presence()) { + continue; + } let Some(b) = bytes.get_or_insert_with(|| encode(&msg)).clone() else { continue }; if !deliver(&self.conns, conn, &b, &mut held, &mut stuck) { stuck.push(conn); diff --git a/crates/collab-net/src/hub/barred_tests.rs b/crates/collab-net/src/hub/barred_tests.rs index a88caf633..eeb293231 100644 --- a/crates/collab-net/src/hub/barred_tests.rs +++ b/crates/collab-net/src/hub/barred_tests.rs @@ -74,6 +74,8 @@ fn a_kicked_connection_with_a_full_queue_is_shut() { let out = Line { tx, queued: Arc::default(), + waiting: Arc::default(), + max_presence: 32, block: Arc::new(AtomicUsize::new(0)), all: Arc::new(AtomicUsize::new(0)), max: 1 << 20, diff --git a/crates/collab-net/src/hub/conn.rs b/crates/collab-net/src/hub/conn.rs index 1f365cbab..08b9c355e 100644 --- a/crates/collab-net/src/hub/conn.rs +++ b/crates/collab-net/src/hub/conn.rs @@ -235,6 +235,8 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender, limits: NetLimit let out = Line { tx, queued: Arc::default(), + waiting: Arc::default(), + max_presence: limits.queue.max(4) / 2, block: block.queued.clone(), all: shared.queued.clone(), max: limits.max_queued_bytes, diff --git a/crates/collab-net/src/hub/queue.rs b/crates/collab-net/src/hub/queue.rs index 1c6a5dfcf..3559b0f10 100644 --- a/crates/collab-net/src/hub/queue.rs +++ b/crates/collab-net/src/hub/queue.rs @@ -48,6 +48,7 @@ pub(crate) struct Charges { pub(super) struct Charge { n: usize, mine: Arc, + waiting: Arc, _all: Arc, _block: Option>, } @@ -55,6 +56,7 @@ pub(super) struct Charge { impl Drop for Charge { fn drop(&mut self) { self.mine.fetch_sub(self.n, Ordering::Relaxed); + self.waiting.fetch_sub(1, Ordering::Relaxed); } } @@ -76,6 +78,10 @@ pub(crate) enum Push { pub(crate) struct Line { pub(crate) tx: SyncSender, pub(crate) queued: Arc, + /// Messages waiting in the queue. + pub(crate) waiting: Arc, + /// With this many waiting, presence is left out (see [`Line::takes_presence`]). + pub(crate) max_presence: usize, pub(crate) block: Arc, pub(crate) all: Arc, pub(crate) max: usize, @@ -84,6 +90,15 @@ pub(crate) struct Line { } impl Line { + /// Whether presence goes in the queue now. A client still taking in a long message (the welcome + /// of a large file on a slow link) would otherwise have its queue filled by everyone's cursors + /// and be dropped, again on every try. Presence is latest-wins, so what's left out is made up + /// for by the next update, and this keeps room for what can't be: other people's edits. A client + /// that reads nothing at all still ends at the write timeout. + pub(crate) fn takes_presence(&self) -> bool { + self.waiting.load(Ordering::Relaxed) < self.max_presence + } + /// Queue `b`. One message always fits an empty queue (a welcome as large as the largest file /// gets out). `charges` is what the message counts against, shared by its recipients; /// `over_budget` lets it past the server's budget (the hub just dropped the biggest consumer @@ -118,7 +133,8 @@ impl Line { } }; self.queued.fetch_add(n, Ordering::Relaxed); - let charge = Charge { n, mine: self.queued.clone(), _all: all, _block: block }; + self.waiting.fetch_add(1, Ordering::Relaxed); + let charge = Charge { n, mine: self.queued.clone(), waiting: self.waiting.clone(), _all: all, _block: block }; self.tx.try_send(Out::Bytes(b.clone(), Some(charge))).map_err(|e| match e { TrySendError::Full(_) => Push::Full, TrySendError::Disconnected(_) => Push::Gone, diff --git a/crates/collab-net/src/hub/queue_tests.rs b/crates/collab-net/src/hub/queue_tests.rs index 13879bd1e..57447cc4a 100644 --- a/crates/collab-net/src/hub/queue_tests.rs +++ b/crates/collab-net/src/hub/queue_tests.rs @@ -1,7 +1,8 @@ //! The hub's send queues (no writers: nothing is ever sent, so what is queued stays queued): a //! broadcast counts once against the server's budget however many copies are queued, a block of //! addresses can't take more than its share, and when the server's budget is used up the -//! connection holding the most is shed, not the next client that needs a little. +//! connection holding the most is shed, not the next client that needs a little. Presence stays out +//! of a busy queue. use std::collections::HashMap; use std::net::{IpAddr, TcpListener, TcpStream}; @@ -16,7 +17,20 @@ use super::{BlockUse, Blocks, Charges, Conn, Kind, Line, Out, Push, deliver}; /// A line with its own queue counter, the given block counter and the server's. fn line(block: &Arc, all: &Arc, max_block: usize, max_all: usize) -> (Line, Receiver) { let (tx, rx) = sync_channel(64); - (Line { tx, queued: Arc::default(), block: block.clone(), all: all.clone(), max: 1 << 20, max_block, max_all }, rx) + ( + Line { + tx, + queued: Arc::default(), + waiting: Arc::default(), + max_presence: 32, + block: block.clone(), + all: all.clone(), + max: 1 << 20, + max_block, + max_all, + }, + rx, + ) } fn socket() -> TcpStream { @@ -120,3 +134,25 @@ fn blocks_nobody_uses_are_forgotten() { let _c = blocks.get("203.0.113.3".parse().unwrap()); assert_eq!(blocks.len(), 2, "203.0.113.1 went"); } + +#[test] +fn a_busy_queue_leaves_presence_out_and_keeps_room_for_edits() { + let (l, rx) = line(&Arc::default(), &Arc::default(), usize::MAX, usize::MAX); + let msg = Bytes::from_static(b"x"); + for _ in 0..32 { + assert!(l.takes_presence()); + l.push(&msg, &mut Charges::default(), false).unwrap(); + } + assert!(!l.takes_presence(), "half full"); + // Edits still fit in the half kept for them. + for _ in 0..32 { + l.push(&msg, &mut Charges::default(), false).unwrap(); + } + assert_eq!(l.push(&msg, &mut Charges::default(), false), Err(Push::Full), "a client that takes in nothing still ends"); + // It catches up: presence goes in again. + for _ in 0..33 { + drop(rx.recv().unwrap()); + } + assert!(l.takes_presence()); + assert_eq!(l.waiting.load(Ordering::Relaxed), 31); +}