feat: update to flecs with safety checks (#864)

andrewgazelka · web-flow · commit a62a9469388f · 2025-03-13T10:29:35.000Z
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/hyperion/Cargo.toml b/crates/hyperion/Cargo.toml
@@ -72,6 +72,7 @@ valence_registry = { workspace = true }
 valence_server = { workspace = true }
 valence_text = { workspace = true }
 ordered-float = { workspace = true }
+dashmap = "6.1.0"
 
 [dev-dependencies]
 approx = { workspace = true }
diff --git a/crates/hyperion/src/egress/sync_entity_state.rs b/crates/hyperion/src/egress/sync_entity_state.rs
@@ -73,59 +73,29 @@ impl Module for EntityStateSyncModule {
             .singleton()
             .multi_threaded()
             .kind::<flecs::pipeline::OnStore>()
-            .run(|mut table| {
+            .each_iter(|table, idx, (compose, net, prev_xp, current)| {
+                const {
+                    assert!(size_of::<Xp>() == size_of::<u16>());
+                    assert!(align_of::<Xp>() == align_of::<u16>());
+                }
                 let system = table.system();
-                while table.next() {
-                    let count = table.count();
-
-                    unsafe {
-                        const _: () = assert!(size_of::<Xp>() == size_of::<u16>());
-                        const _: () = assert!(align_of::<Xp>() == align_of::<u16>());
-
-                        /// Number of lanes in the SIMD vector
-                        const LANES: usize = 32; // up to AVX512
-
-                        let compose = table.field_unchecked::<Compose>(0);
-                        let compose = compose.first().unwrap();
-
-                        let net = table.field_unchecked::<ConnectionId>(1);
-                        let net = net.get(..).unwrap();
-
-                        let mut prev_xp = table.field_unchecked::<Xp>(2);
-                        let prev_xp = prev_xp.get_mut(..).unwrap();
-                        let prev_xp: &mut [u16] =
-                            core::slice::from_raw_parts_mut(prev_xp.as_mut_ptr().cast(), count);
-
-                        let mut xp = table.field_unchecked::<Xp>(3);
-                        let xp = xp.get_mut(..).unwrap();
-                        let xp: &mut [u16] =
-                            core::slice::from_raw_parts_mut(xp.as_mut_ptr().cast(), count);
 
-                        simd_utils::copy_and_get_diff::<_, LANES>(
-                            prev_xp,
-                            xp,
-                            |idx, prev, current| {
-                                debug_assert!(prev != current);
+                if prev_xp != current {
+                    let visual = current.get_visual();
 
-                                let net = net.get(idx).unwrap();
-
-                                let current = Xp::from(*current);
-                                let visual = current.get_visual();
-
-                                let packet = play::ExperienceBarUpdateS2c {
-                                    bar: visual.prop,
-                                    level: VarInt(i32::from(visual.level)),
-                                    total_xp: VarInt::default(),
-                                };
+                    let packet = play::ExperienceBarUpdateS2c {
+                        bar: visual.prop,
+                        level: VarInt(i32::from(visual.level)),
+                        total_xp: VarInt::default(),
+                    };
 
-                                let entity = table.entity(idx);
-                                entity.modified::<Xp>();
+                    let entity = table.entity(idx);
+                    entity.modified::<Xp>();
 
-                                compose.unicast(&packet, *net, system).unwrap();
-                            },
-                        );
-                    }
+                    compose.unicast(&packet, *net, system).unwrap();
                 }
+
+                *prev_xp = *current;
             });
 
         system!("entity_metadata_sync", world, &Compose($), &mut MetadataChanges)
diff --git a/crates/hyperion/src/ingress/mod.rs b/crates/hyperion/src/ingress/mod.rs
@@ -35,7 +35,7 @@ use crate::{
         skin::PlayerSkin,
     },
     storage::{Events, PlayerJoinServer, SkinHandler},
-    util::{SendableRef, TracingExt, mojang::MojangClient},
+    util::{TracingExt, mojang::MojangClient},
 };
 
 #[derive(Component, Debug)]
@@ -311,9 +311,9 @@ impl Module for IngressModule {
 
             let world = it.world();
 
-            let mut recv = receive.0.lock();
+            let recv = &receive.0;
 
-            for connect in recv.player_connect.drain(..) {
+            for connect in recv.player_connect.lock().drain(..) {
                 info!("player_connect");
                 let view = world
                     .entity()
@@ -328,7 +328,7 @@ impl Module for IngressModule {
                 lookup.insert(connect, view.id());
             }
 
-            for disconnect in recv.player_disconnect.drain(..) {
+            for disconnect in recv.player_disconnect.lock().drain(..) {
                 // will initiate the removal of entity
                 info!("queue pending remove");
                 let Some(id) = lookup.get(&disconnect).copied() else {
@@ -341,64 +341,28 @@ impl Module for IngressModule {
             }
         });
 
-        #[expect(
-            clippy::unwrap_used,
-            reason = "this is only called once on startup; it should be fine. we mostly care \
-                      about crashing during server execution"
-        )]
-        let num_threads = i32::try_from(rayon::current_num_threads()).unwrap();
-
-        let worlds = (0..num_threads)
-            // SAFETY: promoting world to static lifetime, system won't outlive world
-            .map(|i| unsafe { std::mem::transmute(world.stage(i)) })
-            .map(SendableRef)
-            .collect::<Vec<_>>();
-
-        system!(
-            "ingress_to_ecs",
-            world,
-            &StreamLookup($),
-            &ReceiveState($),
-        )
-        .immediate(true)
-        .kind::<flecs::pipeline::PostLoad>()
-        .each(move |(lookup, receive)| {
-            use rayon::prelude::*;
-
-            // 134µs with par_iter
-            // 150-208µs with regular drain
-            let span = info_span!("ingress_to_ecs");
-            let _enter = span.enter();
-
-            let mut recv = receive.0.lock();
-
-            recv.packets.par_drain().for_each(|(entity_id, bytes)| {
-                #[expect(
-                    clippy::indexing_slicing,
-                    reason = "it should be impossible to get a thread index that is out of bounds \
-                              unless the rayon thread pool changes size which does not occur"
-                )]
-                let world = &worlds[rayon::current_thread_index().unwrap_or_default()];
-                let world = &world.0;
-
-                let Some(entity_id) = lookup.get(&entity_id) else {
-                    // this is not necessarily a bug; race conditions occur
-                    warn!("player_packets: entity for {entity_id:?}");
-                    return;
-                };
-
-                if !world.is_alive(*entity_id) {
+        world
+            .system_named::<(&ReceiveState, &ConnectionId, &mut PacketDecoder)>("ingress_to_ecs")
+            .term_at(0u32)
+            .singleton() // StreamLookup
+            // .multi_threaded()
+            .immediate(true)
+            .kind::<flecs::pipeline::PostLoad>()
+            .each(move |(receive, connection_id, decoder)| {
+                // 134µs with par_iter
+                // 150-208µs with regular drain
+                let span = info_span!("ingress_to_ecs");
+                let _enter = span.enter();
+
+                let mut bytes = receive.0.packets.get_mut(&connection_id.inner()).unwrap();
+                if bytes.is_empty() {
                     return;
                 }
 
-                let entity = world.entity_from_id(*entity_id);
-
-                entity.get::<&mut PacketDecoder>(|decoder| {
-                    decoder.shift_excess();
-                    decoder.queue_slice(bytes.as_ref());
-                });
+                decoder.shift_excess();
+                decoder.queue_slice(bytes.as_ref());
+                bytes.clear();
             });
-        });
 
         system!(
             "remove_player_from_visibility",
@@ -591,9 +555,10 @@ impl Module for IngressModule {
                             // todo: better way?
                             if let Some((position, pose)) = position.as_mut().zip(pose.as_mut()) {
                                 let world = &world;
+                                let id = entity.id();
 
                                 let mut query = PacketSwitchQuery {
-                                    id: entity.id(),
+                                    id,
                                     view: entity,
                                     compose,
                                     io_ref,
diff --git a/crates/hyperion/src/net/proxy.rs b/crates/hyperion/src/net/proxy.rs
@@ -1,8 +1,9 @@
 //! Communication to a proxy which forwards packets to the players.
 
-use std::{collections::HashMap, io::Cursor, net::SocketAddr, process::Command, sync::Arc};
+use std::{io::Cursor, net::SocketAddr, process::Command, sync::Arc};
 
 use bytes::{Buf, BytesMut};
+use dashmap::DashMap;
 use flecs_ecs::macros::Component;
 use hyperion_proto::ArchivedProxyToServerMessage;
 use parking_lot::Mutex;
@@ -15,11 +16,11 @@ use crate::{runtime::AsyncRuntime, simulation::EgressComm};
 #[derive(Default)]
 pub struct ReceiveStateInner {
     /// All players who have recently connected to the server.
-    pub player_connect: Vec<u64>,
+    pub player_connect: Mutex<Vec<u64>>,
     /// All players who have recently disconnected from the server.
-    pub player_disconnect: Vec<u64>,
+    pub player_disconnect: Mutex<Vec<u64>>,
     /// A map of stream ids to the corresponding [`BytesMut`] buffers. This represents data from the client to the server.
-    pub packets: HashMap<u64, BytesMut>,
+    pub packets: DashMap<u64, BytesMut>,
 }
 
 fn get_pid_from_port(port: u16) -> Result<Option<u32>, std::io::Error> {
@@ -44,7 +45,7 @@ fn get_pid_from_port(port: u16) -> Result<Option<u32>, std::io::Error> {
 async fn inner(
     socket: SocketAddr,
     mut server_to_proxy: tokio::sync::mpsc::UnboundedReceiver<bytes::Bytes>,
-    shared: Arc<Mutex<ReceiveStateInner>>,
+    shared: Arc<ReceiveStateInner>,
 ) {
     let listener = match tokio::net::TcpListener::bind(socket).await {
         Ok(listener) => listener,
@@ -120,17 +121,16 @@ async fn inner(
                             ArchivedProxyToServerMessage::PlayerConnect(message) => {
                                 let Ok(stream) = rkyv::deserialize::<u64, !>(&message.stream);
 
-                                shared.lock().player_connect.push(stream);
+                                shared.player_connect.lock().push(stream);
                             }
                             ArchivedProxyToServerMessage::PlayerDisconnect(message) => {
                                 let Ok(stream) = rkyv::deserialize::<u64, !>(&message.stream);
-                                shared.lock().player_disconnect.push(stream);
+                                shared.player_disconnect.lock().push(stream);
                             }
                             ArchivedProxyToServerMessage::PlayerPackets(message) => {
                                 let Ok(stream) = rkyv::deserialize::<u64, !>(&message.stream);
 
                                 shared
-                                    .lock()
                                     .packets
                                     .entry(stream)
                                     .or_default()
@@ -152,13 +152,13 @@ async fn inner(
 
 /// A wrapper around [`ReceiveStateInner`]
 #[derive(Component)]
-pub struct ReceiveState(pub Arc<Mutex<ReceiveStateInner>>);
+pub struct ReceiveState(pub Arc<ReceiveStateInner>);
 
 /// Initializes proxy communications.
 #[must_use]
 pub fn init_proxy_comms(tasks: &AsyncRuntime, socket: SocketAddr) -> (ReceiveState, EgressComm) {
     let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
-    let shared = Arc::new(Mutex::new(ReceiveStateInner::default()));
+    let shared = Arc::new(ReceiveStateInner::default());
 
     tasks.block_on(async {
         inner(socket, rx, shared.clone()).await;
diff --git a/crates/hyperion/src/simulation/handlers.rs b/crates/hyperion/src/simulation/handlers.rs
@@ -88,25 +88,23 @@ fn change_position_or_correct_client(
             .entity_view(query.world)
             .set(PendingTeleportation::new(pose.position));
     }
-    query
-        .view
-        .get::<(&mut MovementTracking, &Yaw)>(|(tracking, yaw)| {
-            tracking.received_movement_packets += 1;
-            let y_delta = proposed.y - pose.y;
-
-            if y_delta > 0. && tracking.was_on_ground && !on_ground {
-                tracking.server_velocity.y = 0.419_999_986_886_978_15;
-
-                if tracking.sprinting {
-                    let smth = yaw.yaw * 0.017_453_292;
-                    tracking.server_velocity += DVec3::new(
-                        f64::from(-smth.sin()) * 0.2,
-                        0.0,
-                        f64::from(smth.cos()) * 0.2,
-                    );
-                }
+    query.view.get::<&mut MovementTracking>(|tracking| {
+        tracking.received_movement_packets += 1;
+        let y_delta = proposed.y - pose.y;
+
+        if y_delta > 0. && tracking.was_on_ground && !on_ground {
+            tracking.server_velocity.y = 0.419_999_986_886_978_15;
+
+            if tracking.sprinting {
+                let smth = query.yaw.yaw * 0.017_453_292;
+                tracking.server_velocity += DVec3::new(
+                    f64::from(-smth.sin()) * 0.2,
+                    0.0,
+                    f64::from(smth.cos()) * 0.2,
+                );
             }
-        });
+        }
+    });
 
     **pose = proposed;
 }
@@ -589,13 +587,13 @@ pub fn confirm_teleportation(
 ) -> anyhow::Result<()> {
     let entity = query.id.entity_view(query.world);
 
-    entity.get::<(Option<&PendingTeleportation>, &mut Position)>(|(pending_teleport, position)| {
+    entity.get::<Option<&PendingTeleportation>>(|pending_teleport| {
         if let Some(pending_teleport) = pending_teleport {
             if VarInt(pending_teleport.teleport_id) != pkt.teleport_id {
                 return;
             }
 
-            position.position = pending_teleport.destination;
+            **query.position = pending_teleport.destination;
             entity.remove::<PendingTeleportation>();
         }
     });
diff --git a/crates/hyperion/src/simulation/metadata/mod.rs b/crates/hyperion/src/simulation/metadata/mod.rs
@@ -47,34 +47,11 @@ where
         )>(system_name)
         .multi_threaded()
         .kind::<flecs::pipeline::OnUpdate>()
-        .run(move |mut table| {
-            while table.next() {
-                unsafe {
-                    let mut prev = table.field_unchecked::<T>(0);
-                    let prev = prev.get_mut(..).unwrap();
-
-                    let current = table.field_unchecked::<T>(1);
-                    let current = current.get(..).unwrap();
-
-                    let mut metadata_changes = table.field_unchecked::<MetadataChanges>(2);
-                    let metadata_changes = metadata_changes.get_mut(..).unwrap();
-
-                    // todo(perf): big optimization treating as raw bytes and SIMD
-                    // or code that can easily be compiled to SIMD
-                    // also can do copy_from_slice in one pass but want SIMD-optimized
-                    // first
-                    // todo(learn): reborrowing in-depth
-                    for (idx, (prev, current)) in itertools::zip_eq(&mut *prev, current).enumerate()
-                    {
-                        if prev != current {
-                            let metadata_changes = metadata_changes.get_unchecked_mut(idx);
-                            metadata_changes.encode(*current);
-                        }
-                    }
-
-                    prev.copy_from_slice(current);
-                }
+        .each(|(prev, current, metadata_changes)| {
+            if prev != current {
+                metadata_changes.encode(*current);
             }
+            *prev = *current;
         });
 
     let register = |view: &mut EntityView<'_>| {
diff --git a/justfile b/justfile
@@ -49,7 +49,7 @@ fmt:
     cargo fmt
 
 proxy:
-    ulimit -Sn {{fds}} && cargo run --profile release-full --bin hyperion-proxy
+    ulimit -Sn {{fds}} && cargo run --profile release-full --bin hyperion-proxy -- --server "127.0.0.1:35565" "0.0.0.0:25565"
 
 tag-full:
     cargo run --bin tag --profile release-full
@@ -83,7 +83,7 @@ watch:
 debug:
     #!/usr/bin/env -S parallel --shebang --ungroup --jobs 3
     RUST_BACKTRACE=full RUN_MODE=debug-{{arch}} cargo watch --postpone --no-vcs-ignores -w {{project_root}}/.trigger-debug -s './target/debug/tag'
-    RUST_BACKTRACE=full ulimit -Sn {{fds}} && cargo run --bin hyperion-proxy --release-full -- --server "127.0.0.1:35565" "0.0.0.0:25565"
+    RUST_BACKTRACE=full ulimit -Sn {{fds}} && cargo run --bin hyperion-proxy -- --server "127.0.0.1:35565" "0.0.0.0:25565"
     cargo watch -w '{{project_root}}/crates/hyperion' -w '{{project_root}}/events/tag' -s 'cargo check -p tag && cargo build -p tag' -s 'touch {{project_root}}/.trigger-debug'
 
 # run in release mode with tracy; auto-restarts on changes
