Merge branches 'perf/path-shape-changes-once', 'perf/net-events-list' and 'perf/mobs-move-back' into bench/sp1-all

tick_creature_entities takes self.mobs out for the mob loop and appended it
back to the then empty list: Vec::append on an empty Vec copies every Mob
into a fresh buffer. With a server's 2000 mobs that is a multi-megabyte
allocation (mimalloc's huge-page path, madvise) and a memmove each tick,
1.5 % of the tick thread in a 60-player profile. Nothing adds to self.mobs
during the loop, so the list now moves back; the append stays for the case
where something did.

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

crates/horizon-client/src/game/creatures.rs

@@ -635,7 +635,7 @@ impl Game<'_> {
     pub(super) fn tick_creatures(&mut self) -> Result<()> {
         self.attack_ticks = self.attack_ticks.saturating_add(1);
         self.tick_leads();
-        self.spawn_natural();
+        horizon_client::spans::time("spawn_natural", || self.spawn_natural());
         // ServerChunkCache.tickChunks: the custom spawners tick only with spawn_mobs.
         if self.ui.rules.bool("spawn_mobs") {
             self.tick_wandering_trader();
@@ -644,7 +644,7 @@ impl Game<'_> {
             self.tick_village_siege();
         }
         self.tick_raids();
-        self.tick_creature_entities()
+        horizon_client::spans::time("tick_creature_entities", || self.tick_creature_entities())
     }
     /// `DistanceManager.inEntityTickingRange` in an entity-ticking chunk
     /// (block coordinates), from every player's tickets (`update_chunk_tickets`).
@@ -723,6 +723,7 @@ impl Game<'_> {
                 .map(super::boat_leads::boat_state)
                 .collect();
             let base = self.mob_world_base(&[]);
+            let path_changes = horizon_client::mobs::collision_changes(&base, &path_writes);
             // Each mob keeps its snapshot slot all tick; the first slot of an
             // id answers, as a scan would.
             let mut index = horizon_client::fast_hash::FxHashMap::<u32, u32>::default();
@@ -783,7 +784,7 @@ impl Game<'_> {
                     continue;
                 }
                 horizon_client::mobs::apply_mob_pushes(m, &mut pushes);
-                horizon_client::mobs::blocks_changed(m, &w, &path_writes);
+                horizon_client::mobs::blocks_changed(m, &path_changes);
                 let from = events.len();
                 let kept = m.tick(&w, &mut scratch, &mut events);
                 // A killed mob's body is removed: its effects that act on
@@ -853,7 +854,13 @@ impl Game<'_> {
             bindings.retain(|id, _| live.contains(id));
             host.mob_players = bindings;
         }
-        self.mobs.append(&mut mobs);
+        // The ticked list goes back whole: appended to an empty list, a
+        // server's thousands of mobs were copied into a new buffer each tick.
+        if self.mobs.is_empty() {
+            self.mobs = mobs;
+        } else {
+            self.mobs.append(&mut mobs);
+        }
         self.died = died;
         // ... and entity event 60 goes out as it is: the poof.
         for (id, at, width, height) in gone {

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();
     }
 }

crates/horizon-client/src/mobs.rs

@@ -1987,22 +1987,34 @@ impl PendingPushes {
         self.by_mob.entry(id).or_default().push(v);
     }
 }
-/// Adds the pending pushes aimed at `m` to its velocity, in push order.
 /// `ServerLevel.sendBlockUpdated`: for each write that changed a block's collision
 /// shape, a mob whose path runs near it re-paths (`PathNavigation.shouldRecomputePath`).
-pub fn blocks_changed(m: &mut Mob, w: &MobWorld, writes: &[((i32, i32, i32), u32, u32)]) {
+/// The writes whose shape changed are found once per tick ([`collision_changes`]).
+pub fn blocks_changed(m: &mut Mob, changed: &[[i32; 3]]) {
     if m.ai.nav.is_done() {
         return;
     }
-    for &((x, y, z), old, new) in writes {
-        let at = [x, y, z];
-        let shape = |state| w.collision_bounds_of(horizon_world::registry::BlockStateId(state), at);
-        if shape(old) != shape(new) && m.should_recompute_path(at) {
+    for &at in changed {
+        if m.should_recompute_path(at) {
             m.queue_path_recompute();
             return;
         }
     }
 }
+/// The writes of `writes` that changed the block's collision shape, in order:
+/// `sendBlockUpdated` compares the two shapes once per write, not per mob.
+pub fn collision_changes(w: &MobWorld, writes: &[((i32, i32, i32), u32, u32)]) -> Vec<[i32; 3]> {
+    writes
+        .iter()
+        .filter_map(|&((x, y, z), old, new)| {
+            let at = [x, y, z];
+            let shape =
+                |state| w.collision_bounds_of(horizon_world::registry::BlockStateId(state), at);
+            (shape(old) != shape(new)).then_some(at)
+        })
+        .collect()
+}
+/// Adds the pending pushes aimed at `m` to its velocity, in push order.
 pub fn apply_mob_pushes(m: &mut Mob, pending: &mut PendingPushes) {
     // Entity.push(x, y, z) sets needsSync.
     let mut add = |v: Vec3| {