From 5118d824a688a8ae3e4fb0ee25406b65f57aef5b Mon Sep 17 00:00:00 2001 From: Ludovic Marie <136255825+ludovic111@users.noreply.github.com> Date: Sat, 17 Oct 2026 04:47:35 +0200 Subject: [PATCH] LAN hub: hand a tick's packets to each connection together Every packet sent to a guest woke its writer thread, and at 25 walking bots those wake-ups (mpsc notify, semaphore_signal) were about a quarter of the server thread's tick. While tick_lan runs, packets for each guest are held and handed over in one Out::Packets at the end; a send outside a tick, or a disconnect, first hands over what is held, so the bytes and their order are unchanged. Co-Authored-By: Claude Opus 5.5 --- crates/horizon-client/src/game/lan.rs | 6 ++ crates/horizon-client/src/game/scoreboard.rs | 5 +- crates/horizon-client/src/lan/conn.rs | 108 +++++++++++++++++-- crates/horizon-client/src/lan/hub.rs | 4 +- crates/horizon-client/src/lan/mod.rs | 4 +- 5 files changed, 111 insertions(+), 16 deletions(-) diff --git a/crates/horizon-client/src/game/lan.rs b/crates/horizon-client/src/game/lan.rs index 1d00134e4..f6fc1e860 100644 --- a/crates/horizon-client/src/game/lan.rs +++ b/crates/horizon-client/src/game/lan.rs @@ -684,6 +684,8 @@ impl Game<'_> { let Some(mut host) = self.lan.take() else { return; }; + // What the tick sends a guest goes to its connection at the end. + let holding = lan::Holding::start(); let state = self.lan_host_state(&host); // A dedicated server runs the hub without a host player. if state.is_some() || host.hub.config.dedicated { @@ -783,6 +785,10 @@ impl Game<'_> { self.lan_swings(&mut host); self.lan_update_streaming(&host); self.lan_update_models(&mut host); + drop(holding); + for g in &host.hub.guests { + g.sender.flush(); + } self.lan = Some(host); } diff --git a/crates/horizon-client/src/game/scoreboard.rs b/crates/horizon-client/src/game/scoreboard.rs index 16297e930..ff1f602a4 100644 --- a/crates/horizon-client/src/game/scoreboard.rs +++ b/crates/horizon-client/src/game/scoreboard.rs @@ -737,10 +737,7 @@ mod tab_list_tests { name: name.into(), info: ClientInformation::new(8), brand: "test".into(), - sender: PlayerSender { - tx, - pending: Arc::default(), - }, + sender: PlayerSender::new(tx, Arc::default()), latency: Arc::new(AtomicI32::new(42)), address: "127.0.0.1:25565".parse().unwrap(), }, diff --git a/crates/horizon-client/src/lan/conn.rs b/crates/horizon-client/src/lan/conn.rs index a1ffac4fa..fa31f2275 100644 --- a/crates/horizon-client/src/lan/conn.rs +++ b/crates/horizon-client/src/lan/conn.rs @@ -21,9 +21,10 @@ use horizon_protocol::{ text::translatable, }; use std::{ + cell::Cell, net::TcpStream, sync::{ - Arc, + Arc, Mutex, PoisonError, atomic::{AtomicI32, AtomicI64, AtomicUsize, Ordering}, mpsc::{Receiver, RecvTimeoutError, Sender, channel}, }, @@ -122,16 +123,64 @@ pub fn valid_name(name: &str) -> bool { (1..=16).contains(&name.len()) && name.chars().all(|c| c as u32 > 32 && (c as u32) < 127) } +thread_local! { + static HOLDING: Cell = const { Cell::new(false) }; +} + +/// While one is alive, the packets this thread sends to each player are held +/// and handed to the player's writer thread together by +/// [`PlayerSender::flush`]: one wake-up per player per tick rather than one +/// per packet. The bytes and their order don't change. +pub struct Holding(bool); +impl Holding { + pub fn start() -> Self { + Self(HOLDING.replace(true)) + } +} +impl Drop for Holding { + fn drop(&mut self) { + HOLDING.set(self.0); + } +} + /// The sending half given to the game thread. #[derive(Clone)] pub struct PlayerSender { pub tx: Sender, pub pending: Arc, + held: Arc>>>, } impl PlayerSender { + pub fn new(tx: Sender, pending: Arc) -> Self { + Self { + tx, + pending, + held: Arc::default(), + } + } + fn held(&self) -> std::sync::MutexGuard<'_, Vec>> { + self.held.lock().unwrap_or_else(PoisonError::into_inner) + } pub fn send(&self, body: Vec) { self.pending.fetch_add(body.len(), Ordering::Relaxed); - let _ = self.tx.send(Out::Packet(body)); + if HOLDING.get() { + self.held().push(body); + } else { + self.flush(); + let _ = self.tx.send(Out::Packet(body)); + } + } + /// Hand the held packets to the writer thread. + pub fn flush(&self) { + let held = std::mem::take(&mut *self.held()); + if !held.is_empty() { + let _ = self.tx.send(Out::Packets(held)); + } + } + /// Send a play disconnect with this reason after what was sent before. + pub fn disconnect(&self, reason: horizon_world::nbt::Tag) { + self.flush(); + let _ = self.tx.send(Out::Disconnect(reason)); } } @@ -466,10 +515,7 @@ fn play(c: Connection, shared: &Arc, parts: Parts, address: std::net::So let pending = Arc::new(AtomicUsize::new(0)); let latency = Arc::new(AtomicI32::new(0)); let alive = Arc::new(AtomicI64::new(-1)); - let sender = PlayerSender { - tx: tx.clone(), - pending: pending.clone(), - }; + let sender = PlayerSender::new(tx.clone(), pending.clone()); let writer_thread = { let (alive, latency, pending) = (alive.clone(), latency.clone(), pending.clone()); std::thread::Builder::new() @@ -544,13 +590,18 @@ fn write_loop( let mut sent_at: Option<(i64, Instant)> = None; let mut next = Instant::now() + KEEP_ALIVE_EVERY; loop { + let mut write = |body: Vec| { + pending.fetch_sub(body.len(), Ordering::Relaxed); + writer.send(&body).is_ok() && pending.load(Ordering::Relaxed) <= MAX_PENDING + }; match rx.recv_timeout(Duration::from_millis(200)) { Ok(Out::Packet(body)) => { - pending.fetch_sub(body.len(), Ordering::Relaxed); - if writer.send(&body).is_err() { + if !write(body) { break; } - if pending.load(Ordering::Relaxed) > MAX_PENDING { + } + Ok(Out::Packets(bodies)) => { + if !bodies.into_iter().all(&mut write) { break; } } @@ -594,3 +645,42 @@ fn write_loop( } writer.shutdown(); } + +#[cfg(test)] +mod tests { + use super::*; + + fn bodies(rx: &Receiver) -> Vec>> { + rx.try_iter() + .map(|out| match out { + Out::Packet(body) => vec![body], + Out::Packets(bodies) => bodies, + Out::Disconnect(_) => vec![b"disconnect".to_vec()], + Out::Close => vec![b"close".to_vec()], + }) + .collect() + } + + #[test] + fn held_packets_go_together_and_in_order() { + let (tx, rx) = channel(); + let sender = PlayerSender::new(tx, Arc::default()); + sender.send(vec![1]); + assert_eq!(bodies(&rx), [vec![vec![1]]]); + let holding = Holding::start(); + sender.send(vec![2]); + sender.send(vec![3]); + assert!(bodies(&rx).is_empty()); + drop(holding); + // A packet sent after the tick still follows the held ones. + sender.send(vec![4]); + assert_eq!(bodies(&rx), [vec![vec![2], vec![3]], vec![vec![4]]]); + let holding = Holding::start(); + sender.send(vec![5]); + sender.disconnect(horizon_world::nbt::Tag::Byte(0)); + drop(holding); + sender.flush(); + assert_eq!(bodies(&rx), [vec![vec![5]], vec![b"disconnect".to_vec()]]); + assert_eq!(sender.pending.load(Ordering::Relaxed), 5); + } +} diff --git a/crates/horizon-client/src/lan/hub.rs b/crates/horizon-client/src/lan/hub.rs index cb438508e..414979a87 100644 --- a/crates/horizon-client/src/lan/hub.rs +++ b/crates/horizon-client/src/lan/hub.rs @@ -389,7 +389,7 @@ impl Hub { /// Disconnect a guest with a reason (ServerGamePacketListenerImpl.disconnect). pub fn kick(&mut self, conn: u32, reason: Tag) { if let Some(g) = self.guest(conn) { - let _ = g.sender.tx.send(super::Out::Disconnect(reason)); + g.sender.disconnect(reason); } } @@ -1638,7 +1638,7 @@ impl Hub { pub fn disconnect_all(&mut self) { let reason = translatable("multiplayer.disconnect.server_shutdown", &[]); for g in &self.guests { - let _ = g.sender.tx.send(super::Out::Disconnect(reason.clone())); + g.sender.disconnect(reason.clone()); } } } diff --git a/crates/horizon-client/src/lan/mod.rs b/crates/horizon-client/src/lan/mod.rs index 7084f20a6..6af4a1fef 100644 --- a/crates/horizon-client/src/lan/mod.rs +++ b/crates/horizon-client/src/lan/mod.rs @@ -21,7 +21,7 @@ pub mod playerdata; mod recipes; pub mod tracking; -pub use conn::PlayerSender; +pub use conn::{Holding, PlayerSender}; use horizon_protocol::packets::play; use horizon_protocol::{ GAME_VERSION, PROTOCOL_VERSION, crypto::uuid_string, packets::configuration::ClientInformation, @@ -69,6 +69,8 @@ pub struct Joined { pub enum Out { /// A packet body: VarInt id and payload. Packet(Vec), + /// Packet bodies held during a tick, in order. + Packets(Vec>), /// Send a play disconnect with this component, then close. Disconnect(horizon_world::nbt::Tag), Close,