Merge pull request #81 from ferrumc-rs/fix/components_removal

Components removal^

src/lib/ecs/src/components/mod.rs

@@ -60,9 +60,20 @@ impl ComponentStorage {
             .map(|mut components| components.remove(entity));
     }
 
-    pub fn remove_all_components(&self, entity: Entity) {
+    pub fn remove_all_components(&self, entity: Entity) -> ECSResult<()> {
         self.components.iter_mut()
-            .for_each(|mut components| { components.remove(entity); });
+            .for_each(|mut components| {
+                // check if its locked or not
+                if let Some(component) = components.get_mut(entity) {
+                    let lock = component.write();
+                    // basically wait for component to be able to be written to (or have no readers & writers)
+                    drop(lock);
+                    // Remove else-wise
+                    components.remove(entity);
+                }
+            });
+
+        Ok(())
     }
 }
 impl ComponentStorage {

src/lib/ecs/src/components/sparse_set.rs

@@ -41,25 +41,38 @@ impl<T> SparseSet<T> {
     pub fn get(&self, entity: Entity) -> Option<&T> {
         self.indices
             .get(&entity)
-            .map(|index| &self.components[*index])
+            .and_then(|index| self.components.get(*index))
     }
 
     /// Get a mutable reference to a component for a specific Entity.
     /// Returns None if the component does not exist.
     pub fn get_mut(&mut self, entity: Entity) -> Option<&mut T> {
         self.indices
             .get(&entity)
-            .map(|index| &mut self.components[*index])
+            .and_then(|index| self.components.get_mut(*index))
     }
 
     /// Remove a component for a specific Entity.
     /// Returns the removed component if it existed.
     pub fn remove(&mut self, entity: Entity) -> Option<T> {
-        if let Some(index) = self.indices.remove(&entity) {
-            Some(self.components.remove(index))
-        } else {
-            None
+        let index = self.indices.remove(&entity)?;
+
+        let last_index = self.components.len() - 1;
+
+        // If we're not removing the last element, swap with the last.
+        if index != last_index {
+            self.components.swap(index, last_index);
+
+            let last_entity = self.indices.iter()
+                .find(|(_, &i)| i == last_index)
+                .map(|(e, _)| *e)
+                .expect("Entity not found. This should never happen.");
+
+            // Update the moved entity's index.
+            self.indices.insert(last_entity, index);
         }
+
+        self.components.pop()
     }
 
     pub fn entities(&self) -> Vec<Entity> {

src/lib/ecs/src/lib.rs

@@ -51,7 +51,7 @@ impl Universe {
         self.components.remove::<T>(entity)
     }
 
-    pub fn remove_all_components(&self, entity: Entity) {
+    pub fn remove_all_components(&self, entity: Entity) -> ECSResult<()> {
         self.components.remove_all_components(entity)
     }
 

src/lib/net/src/connection.rs

@@ -3,7 +3,6 @@ use crate::{handle_packet, NetResult, ServerState};
 use ferrumc_net_codec::encode::NetEncode;
 use ferrumc_net_codec::encode::NetEncodeOpts;
 use std::sync::Arc;
-use tokio::io::BufReader;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::net::TcpStream;
 use tracing::{debug, trace, warn};
@@ -29,11 +28,11 @@ impl ConnectionState {
 }
 
 pub struct StreamReader {
-    pub reader: BufReader<OwnedReadHalf>,
+    pub reader: OwnedReadHalf,
 }
 
 impl StreamReader {
-    pub fn new(reader: BufReader<OwnedReadHalf>) -> Self {
+    pub fn new(reader: OwnedReadHalf) -> Self {
         Self { reader }
     }
 }
@@ -81,7 +80,7 @@ pub async fn handle_connection(state: Arc<ServerState>, tcp_stream: TcpStream) -
     let entity = state
         .universe
         .builder()
-        .with(StreamReader::new(BufReader::new(reader)))
+        .with(StreamReader::new(reader))
         .with(StreamWriter::new(writer))
         .with(ConnectionState::Handshaking)
         .with(CompressionStatus::new())
@@ -108,19 +107,35 @@ pub async fn handle_connection(state: Arc<ServerState>, tcp_stream: TcpStream) -
             &mut packet_skele.data,
             Arc::clone(&state),
         )
-        .await
+            .await
         {
             warn!("Failed to handle packet: {:?}", e);
             // Kick the player (when implemented).
-            // Send a disconnect event
             break 'recv;
         };
     }
 
     debug!("Connection closed for entity: {:?}", entity);
 
     // Remove all components from the entity
-    state.universe.remove_all_components(entity);
+
+    drop(reader);
+
+    // Wait until anything that might be using the entity is done
+    if let Err(e) = remove_all_components_blocking(state.clone(), entity).await {
+        warn!("Failed to remove all components from entity: {:?}", e);
+    }
+
+    debug!("Dropped all components from entity: {:?}", entity);
 
     Ok(())
 }
+
+/// Since parking_lot is single-threaded, we use spawn_blocking to remove all components from the entity asynchronously (on another thread).
+async fn remove_all_components_blocking(state: Arc<ServerState>, entity: usize) -> NetResult<()> {
+    let res = tokio::task::spawn_blocking(move || {
+        state.universe.remove_all_components(entity)
+    }).await?;
+
+    Ok(res?)
+}

src/lib/net/src/errors.rs

@@ -17,6 +17,10 @@ pub enum NetError {
     #[error("IO Error: {0}")]
     IOError(#[from] std::io::Error),
 
+    #[error("Task Error: {0}")]
+    TaskError(#[from] tokio::task::JoinError),
+
+
     #[error("UTF8 Error: {0}")]
     UTF8Error(#[from] std::string::FromUtf8Error),
 

src/lib/utils/general_purpose/src/simd/arrays.rs

@@ -120,7 +120,7 @@ unsafe fn u8_slice_to_u32_be_simd(input: &[u8]) -> Vec<u32> {
 
 pub fn u8_slice_to_i32_be(input: &[u8]) -> Vec<i32> {
     let u32s = u8_slice_to_u32_be(input);
-    u32s.into_iter().map(|x| x as i32).collect()
+    unsafe { std::mem::transmute::<Vec<u32>, Vec<i32>>(u32s) }
 }
 
 pub fn u8_slice_to_u64_be(input: &[u8]) -> Vec<u64> {