LAN: connection threads report to a locked list the game takes whole, not a channel

The game drained the connection threads' channel once a tick with try_iter.
std's list channel makes the receiver spin and yield while a sender it raced
has claimed a slot but not written it yet; on a busy server a preempted
connection thread stretches that to a time slice, and the tick waited in
thread::yield_now (3.5 % of the tick thread with 60 players on the Mac mini).
Now each connection thread pushes onto a Vec behind a Mutex and the game takes
the Vec with try_lock, once: it never waits for a connection thread, and what
it misses while one holds the lock comes with the next tick. Order is kept;
sends fail once the Lan is dropped, as they did when the receiver went.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

crates/horizon-client/src/lan/conn.rs

@@ -1104,7 +1104,7 @@ fn wait_alive<T>(
 
 fn play(c: Connection, shared: &Arc<Shared>, parts: Parts, address: std::net::SocketAddr) {
     let conn_id = shared.next_id();
-    let events = shared.events.clone();
+    let events = &shared.events;
     let _ = c.reader.set_read_timeout(Some(READ_TIMEOUT));
     let (mut reader, writer) = c.split();
     let (tx, rx) = channel();
@@ -1152,7 +1152,7 @@ fn play(c: Connection, shared: &Arc<Shared>, parts: Parts, address: std::net::So
         match read_loop(
             &mut reader,
             conn_id,
-            &events,
+            events,
             &alive,
             required_pack,
             shared.config.rate_limit,
@@ -1237,7 +1237,7 @@ enum Ended {
 fn read_loop(
     reader: &mut ConnReader,
     conn: u32,
-    events: &Sender<NetEvent>,
+    events: &super::Events,
     alive: &AtomicI64,
     required_pack: bool,
     rate_limit: i32,

crates/horizon-client/src/lan/mod.rs

@@ -42,7 +42,7 @@ use std::{
     sync::{
         Arc, Mutex,
         atomic::{AtomicBool, AtomicI32, AtomicU32, AtomicUsize, Ordering},
-        mpsc::{Receiver, Sender, channel},
+        mpsc::{Receiver, channel},
     },
     thread::JoinHandle,
     time::Duration,
@@ -317,11 +317,54 @@ impl SecureChat {
     }
 }
 
+/// What the connection threads report to the game, in the order they
+/// reported it. A list behind a lock rather than a channel: the game takes
+/// it whole once a tick and never waits for a connection thread (a
+/// channel's receiver spins while a sender it raced is between claiming a
+/// slot and filling it, which a busy machine stretches to a time slice).
+#[derive(Default)]
+pub struct Events {
+    list: Mutex<Vec<NetEvent>>,
+    /// The game stopped taking events: a send fails, as a channel's does
+    /// once its receiver is gone.
+    closed: AtomicBool,
+}
+impl Events {
+    /// Report an event; `Err` once the game no longer takes them.
+    pub fn send(&self, event: NetEvent) -> Result<(), NetEvent> {
+        if self.closed.load(Ordering::Acquire) {
+            return Err(event);
+        }
+        self.list
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .push(event);
+        Ok(())
+    }
+    /// The game takes no more: what waits is dropped, as a channel's
+    /// receiver drops its messages, and later sends fail.
+    fn close(&self) {
+        self.closed.store(true, Ordering::Release);
+        drop(std::mem::take(
+            &mut *self.list.lock().unwrap_or_else(|e| e.into_inner()),
+        ));
+    }
+    /// Everything reported so far, or nothing while a connection thread
+    /// holds the list (it comes with the next call).
+    fn take(&self) -> Vec<NetEvent> {
+        match self.list.try_lock() {
+            Ok(mut list) => std::mem::take(&mut *list),
+            Err(std::sync::TryLockError::Poisoned(e)) => std::mem::take(&mut *e.into_inner()),
+            Err(std::sync::TryLockError::WouldBlock) => Vec::new(),
+        }
+    }
+}
+
 /// State shared with the connection threads.
 pub struct Shared {
     /// What the next login is sent; `PlayerList.reloadResources` swaps it.
     sync: Mutex<Arc<data::SyncData>>,
-    pub events: Sender<NetEvent>,
+    pub events: Events,
     pub config: NetConfig,
     motd: Mutex<String>,
     max_players: AtomicUsize,
@@ -500,7 +543,6 @@ pub fn base64(data: &[u8]) -> String {
 /// A running LAN server.
 pub struct Lan {
     pub shared: Arc<Shared>,
-    events: Receiver<NetEvent>,
     pub port: u16,
     listener: Option<JoinHandle<()>>,
     /// Stops the accept loop alone: local connections stay.
@@ -520,11 +562,10 @@ impl Lan {
     }
     /// ServerConnectionListener.startTcpServerListener with a configuration.
     pub fn start_with(sync: Arc<data::SyncData>, config: NetConfig) -> io::Result<Self> {
-        let (tx, events) = channel();
         let (listen, port, announce) = (config.listen, config.port, config.announce);
         let shared = Arc::new(Shared {
             sync: Mutex::new(sync),
-            events: tx,
+            events: Events::default(),
             motd: Mutex::new(config.motd.clone()),
             max_players: AtomicUsize::new(config.max_players),
             online: Mutex::new(Vec::new()),
@@ -536,7 +577,6 @@ impl Lan {
         });
         let mut lan = Self {
             shared,
-            events,
             port: 0,
             listener: None,
             listener_stop: Arc::new(AtomicBool::new(false)),
@@ -624,7 +664,7 @@ impl Lan {
     }
     /// Everything the connection threads reported since the last call.
     pub fn poll(&self) -> Vec<NetEvent> {
-        self.events.try_iter().collect()
+        self.shared.events.take()
     }
     /// Reserve a uuid for the host so a guest cannot take the name.
     pub fn reserve(&self, uuid: u128) {
@@ -637,6 +677,7 @@ impl Lan {
 }
 impl Drop for Lan {
     fn drop(&mut self) {
+        self.shared.events.close();
         self.stop();
     }
 }