fix(net): break player receive loop into separate thread

src/net/io.rs

···
  * <https://www.gnu.org/licenses/>.
  */
 
-use std::{io::ErrorKind, net::SocketAddr, time::Duration};
+use std::{io::ErrorKind, time::Duration};
 
 use bytes::BytesMut;
 use color_eyre::eyre::{bail, Context, Result};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
-    net::TcpStream,
+    net::{
+        tcp::{OwnedReadHalf, OwnedWriteHalf},
+        TcpStream,
+    },
+    sync::{Mutex, RwLock},
 };
 
 use crate::protocol::{self, ClientboundPacket, Frame, ServerboundPacket};
 
 #[derive(Debug)]
 pub struct NetIo {
-    stream: TcpStream,
-    frame: Frame,
-    decoder: protocol::Decoder,
-    encoder: protocol::Encoder,
+    pub peer_addr: String,
+    pub connected: RwLock<bool>,
+    read_half: Mutex<OwnedReadHalf>,
+    write_half: Mutex<OwnedWriteHalf>,
+    frame: Mutex<Frame>,
+    decoder: Mutex<protocol::Decoder>,
+    encoder: Mutex<protocol::Encoder>,
 }
 
 const BUF_SIZE: usize = 4096;
···
             );
         }
 
+        let peer_addr = stream
+            .peer_addr()
+            .map_or("Unknown".to_owned(), |a| a.to_string());
+        let (read_half, write_half) = stream.into_split();
+
         Self {
-            stream,
-            frame: Frame {
+            peer_addr,
+            connected: RwLock::new(true),
+            read_half: Mutex::new(read_half),
+            write_half: Mutex::new(write_half),
+            frame: Mutex::new(Frame {
                 id: -1,
                 body: BytesMut::new(),
-            },
-            decoder: protocol::Decoder::new(),
-            encoder: protocol::Encoder::new(),
+            }),
+            decoder: Mutex::new(protocol::Decoder::new()),
+            encoder: Mutex::new(protocol::Encoder::new()),
         }
     }
 
-    pub fn peer_addr(&self) -> Result<SocketAddr> {
-        Ok(self.stream.peer_addr()?)
+    pub async fn connected(&self) -> bool {
+        let c = self.connected.read().await;
+        *c
     }
 
-    pub async fn rx<'a, 'b, P>(&'a mut self) -> Result<P>
+    pub async fn rx<'a, 'b, P>(&'a self) -> Result<Frame>
     where
         P: ServerboundPacket<'a>,
     {
         // TODO: maybe move this somewhere else? i don't know if a global timeout of 5 seconds per
         // packet is realistic but for testing it's chill i suppose
         tokio::time::timeout(Duration::from_secs(5), async move {
+            let mut decoder = self.decoder.lock().await;
             loop {
-                if let Some(frame) = self
-                    .decoder
-                    .try_read_next()
-                    .context("failed try_read_next")?
-                {
+                if let Some(frame) = decoder.try_read_next().context("failed try_read_next")? {
                     if frame.id != P::ID {
                         debug!(
                             "Got packet ID {} while awaiting {}, discarding",
···
                         continue;
                     }
 
-                    self.frame = frame;
-                    let r = self.frame.decode()?;
-                    debug!("Got packet {:?}", r);
-                    return Ok(r);
+                    return Ok(frame);
                 };
 
-                self.decoder.reserve_additional(BUF_SIZE);
-                let mut buf = self.decoder.take_all();
+                decoder.reserve_additional(BUF_SIZE);
+                let mut buf = decoder.take_all();
 
-                if self
-                    .stream
+                let mut read_half = self.read_half.lock().await;
+                if read_half
                     .read_buf(&mut buf)
                     .await
                     .context("failed read_buf")?
                     == 0
                 {
+                    let mut c = self.connected.write().await;
+                    *c = false;
                     return Err(std::io::Error::from(ErrorKind::UnexpectedEof).into());
                 }
 
-                self.decoder.add_bytes(buf);
+                decoder.add_bytes(buf);
             }
         })
         .await?
     }
 
-    pub async fn tx<P>(&mut self, packet: &P) -> Result<()>
+    pub async fn tx<P>(&self, packet: &P) -> Result<()>
     where
         P: ClientboundPacket,
     {
         trace!("Sending packet {:?}", packet);
-        self.encoder.append_packet(packet)?;
-        let bytes = self.encoder.take();
+        let mut encoder = self.encoder.lock().await;
+        encoder.append_packet(packet)?;
+        let bytes = encoder.take();
         trace!("raw packet is {} bytes", bytes.len());
-        Ok(self.stream.write_all(&bytes).await?)
+        let mut writer = self.write_half.lock().await;
+        Ok(writer.write_all(&bytes).await?)
     }
 
-    pub async fn tx_raw(&mut self, packet: &[u8]) -> Result<()> {
+    pub async fn tx_raw(&self, packet: &[u8]) -> Result<()> {
         trace!("Sending packet {:?}", packet);
-        Ok(self.stream.write_all(packet).await?)
-    }
-
-    pub async fn flush(&mut self) -> Result<()> {
-        self.stream.flush().await?;
-        Ok(())
+        let mut writer = self.write_half.lock().await;
+        Ok(writer.write_all(packet).await?)
     }
 
-    pub async fn rx_raw(&mut self) -> Result<Frame> {
-        if let Some(frame) = self
-            .decoder
-            .try_read_next()
-            .context("failed try_read_next")?
-        {
+    pub async fn rx_raw(&self) -> Result<Frame> {
+        let mut decoder = self.decoder.lock().await;
+        if let Some(frame) = decoder.try_read_next().context("failed try_read_next")? {
             return Ok(frame);
         };
 
-        self.decoder.reserve_additional(BUF_SIZE);
-        let mut buf = self.decoder.take_all();
+        decoder.reserve_additional(BUF_SIZE);
+        let mut buf = decoder.take_all();
 
-        if self
-            .stream
-            .read_buf(&mut buf)
-            .await
-            .context("failed read_buf")?
-            == 0
         {
-            return Err(std::io::Error::from(ErrorKind::UnexpectedEof).into());
+            let mut reader = self.read_half.lock().await;
+            if reader.read_buf(&mut buf).await.context("failed read_buf")? == 0 {
+                let mut c = self.connected.write().await;
+                *c = false;
+                return Err(std::io::Error::from(ErrorKind::UnexpectedEof).into());
+            }
         }
 
-        self.decoder.add_bytes(buf);
+        decoder.add_bytes(buf);
 
         bail!("No packet available")
     }

src/net/player.rs

···
     sync::{Mutex, OwnedSemaphorePermit, RwLock},
     time::{self, timeout, Instant},
 };
+use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
 use crate::{
···
 pub struct Player {
     pub id: u16,
     _permit: OwnedSemaphorePermit,
-    pub io: Mutex<NetIo>,
+    pub io: NetIo,
+    frame_queue: Mutex<Vec<Frame>>,
 
     crawlstate: CrawlState,
     packet_state: RwLock<PacketState>,
···
     ) -> Self {
         Self(Arc::new(Player {
             id,
-            io: Mutex::new(NetIo::new(connection)),
+            io: NetIo::new(connection),
+            frame_queue: Mutex::new(Vec::new()),
             _permit: permit,
 
             crawlstate,
···
 
     pub async fn connect(&self) {
         {
-            let io = self.0.io.lock().await;
-            let addy = io
-                .peer_addr()
-                .map_or("Unknown".to_string(), |a| a.to_string());
-            debug!("Handling new player (id {}) from {}", self.0.id, addy);
+            debug!(
+                "Handling new player (id {}) from {}",
+                self.0.id, self.0.io.peer_addr
+            );
         }
 
         // crawlspace intentionally doesn't support legacy pings :3
···
 
     async fn handshake(&self) -> Result<()> {
         let state = self.0.crawlstate.clone();
-        let mut io = self.0.io.lock().await;
 
-        let p = io.rx::<HandshakeS>().await?;
+        let p = self.0.io.rx::<HandshakeS>().await?;
+        let p = p.decode::<HandshakeS>()?;
 
         if p.protocol_version.0 != state.version_number {
             warn!(
···
         }
 
         let next_state = p.next_state;
-        drop(io);
 
         let mut s = self.0.packet_state.write().await;
         match next_state {
···
     }
 
     async fn handle_status(&self) -> Result<()> {
-        let mut io = self.0.io.lock().await;
-
-        io.rx::<StatusRequestS>().await?;
+        self.0.io.rx::<StatusRequestS>().await?;
         let state = self.0.crawlstate.clone();
 
         let res = json!({
···
             json_respose: &res.to_string(),
         };
 
-        io.tx(&res).await?;
-        let ping = io.rx::<Ping>().await?;
-        io.tx(&ping).await?;
+        self.0.io.tx(&res).await?;
+        let ping: Ping = self.0.io.rx::<Ping>().await?.decode()?;
+
+        self.0.io.tx(&ping).await?;
 
         Ok(())
     }
 
     async fn login(&self) -> Result<()> {
         let state = self.0.crawlstate.clone();
-        let mut io = self.0.io.lock().await;
 
-        let login = io.rx::<LoginStartS>().await?;
+        let login = self.0.io.rx::<LoginStartS>().await?;
+        let login: LoginStartS = login.decode()?;
 
         // need to manually clone this or else the reference to self.io lives too long
         // TODO: clean up lifetimes on encode/decode - possibly just clone strings?
···
             *own_uuid = Some(uuid);
         }
 
-        io.tx(&success).await?;
-        io.rx::<LoginAckS>().await?;
+        self.0.io.tx(&success).await?;
+        self.0.io.rx::<LoginAckS>().await?;
 
         let clientbound_known_packs = KnownPacksC::of_version(&state.version_name);
-        io.tx(&clientbound_known_packs).await?;
+        self.0.io.tx(&clientbound_known_packs).await?;
 
         // TODO: maybe(?) actually handle this
-        io.rx::<KnownPacksS>().await?;
+        self.0.io.rx::<KnownPacksS>().await?;
 
-        io.tx_raw(&state.registry_cache.encoded).await?;
+        self.0.io.tx_raw(&state.registry_cache.encoded).await?;
 
-        io.tx(&FinishConfigurationC).await?;
-        io.rx::<FinishConfigurationAckS>().await?;
+        self.0.io.tx(&FinishConfigurationC).await?;
+        self.0.io.rx::<FinishConfigurationAckS>().await?;
 
         Ok(())
     }
···
         *packet_state = PacketState::Play;
 
         let state = self.0.crawlstate.clone();
-        let mut io = self.0.io.lock().await;
 
         let max_players: i32 = state.max_players.try_into().unwrap_or(50);
 
···
             enforces_secure_chat: false,
         };
 
-        io.tx(&login).await?;
+        self.0.io.tx(&login).await?;
 
-        io.tx(&SetTickingStateC {
-            tick_rate: 20.0,
-            is_frozen: false,
-        })
-        .await?;
-
-        io.tx(&StepTicksC(10)).await?;
+        self.0
+            .io
+            .tx(&SetTickingStateC {
+                tick_rate: 20.0,
+                is_frozen: false,
+            })
+            .await?;
 
-        drop(io);
+        self.0.io.tx(&StepTicksC(10)).await?;
 
         let spawnpoint = state.spawnpoint;
         self.teleport_awaiting(spawnpoint.0, spawnpoint.1, spawnpoint.2, 0.0, 0.0)
             .await?;
 
-        let mut io = self.0.io.lock().await;
-
-        io.tx(&SetBorderCenterC {
-            x: spawnpoint.0,
-            z: spawnpoint.2,
-        })
-        .await?;
+        self.0
+            .io
+            .tx(&SetBorderCenterC {
+                x: spawnpoint.0,
+                z: spawnpoint.2,
+            })
+            .await?;
 
-        io.tx(&SetBorderSizeC(state.border_radius as f64 * 2.0))
+        self.0
+            .io
+            .tx(&SetBorderSizeC(state.border_radius as f64 * 2.0))
             .await?;
 
-        io.tx(&PlayerInfoUpdateC {
-            players: &[PlayerStatus::for_player(self.uuid().await)
-                .add_player("You're alone...", &[])
-                .update_listed(true)],
-        })
-        .await?;
+        self.0
+            .io
+            .tx(&PlayerInfoUpdateC {
+                players: &[PlayerStatus::for_player(self.uuid().await)
+                    .add_player("You're alone...", &[])
+                    .update_listed(true)],
+            })
+            .await?;
 
         let await_chunks = GameEventC::from(GameEvent::StartWaitingForLevelChunks);
-        io.tx(&await_chunks).await?;
+        self.0.io.tx(&await_chunks).await?;
 
         let set_center = SetCenterChunkC {
             x: VarInt(spawnpoint.0.floor() as i32 / 16),
             y: VarInt(spawnpoint.2.floor() as i32 / 16),
         };
-        io.tx(&set_center).await?;
-        drop(io);
+        self.0.io.tx(&set_center).await?;
 
         // FIXME: GROSS LOL?????? this should(?) change ownership of the player to the server
         // thread but realistically who knows burhhhh
         state.player_send.send(self.clone()).await?;
+        self.spawn_read_loop();
 
         Ok(())
     }
 
     pub async fn handle_all_packets(&self) -> Result<()> {
-        let Some(packets) = self.rx_all().await else {
-            bail!("Connection for player {} was closed", self.id())
+        let packets = {
+            let mut frame_queue = self.0.frame_queue.lock().await;
+            std::mem::take(&mut *frame_queue)
         };
 
-        debug!("player {} {:?}", self.id(), packets);
-        self.handle_frames(packets).await
+        for packet in packets {
+            self.handle_frame(packet).await?;
+        }
+
+        Ok(())
     }
 
     pub async fn keepalive(&self) -> Result<()> {
···
         }
     }
 
-    async fn rx_all(&self) -> Option<Vec<Frame>> {
-        let mut io = self.0.io.lock().await;
-
-        let mut frames = Vec::new();
-
-        loop {
-            match io.rx_raw().await {
-                Ok(frame) => frames.push(frame),
-                Err(why) => match why.downcast_ref::<tokio::io::Error>().map(|e| e.kind()) {
-                    Some(tokio::io::ErrorKind::UnexpectedEof) => return None,
-                    _ => break,
-                },
-            }
-        }
-
-        Some(frames)
-    }
-
     async fn ping(&self, id: i64) -> Result<()> {
-        let mut io = self.0.io.lock().await;
         let ka = KeepAliveC(id);
-        io.tx(&ka).await?;
+        self.0.io.tx(&ka).await?;
         // TODO: check return keepalive, kick
         Ok(())
     }
···
             };
         }
 
-        let mut io = self.0.io.lock().await;
-
         let tp = SynchronisePositionC::new(x, y, z, yaw, pitch);
         {
             let mut tp_state = self.0.tp_state.write().await;
             // player will be given 5 (FIVE) SECONDS TO ACK!!!!!
             *tp_state = TeleportState::Pending(tp.id, Instant::now());
         }
-        io.tx(&tp).await?;
+        self.0.io.tx(&tp).await?;
 
-        let tp_ack = io.rx::<ConfirmTeleportS>().await?;
+        let tp_ack = self.0.io.rx::<ConfirmTeleportS>().await?;
+        let tp_ack = tp_ack.decode::<ConfirmTeleportS>()?;
 
         match tokio::time::timeout(Duration::from_secs(5), self.confirm_teleport(tp_ack.id)).await {
             Ok(Ok(())) => {
···
         }
     }
 
-    async fn handle_frames(&self, frames: Vec<Frame>) -> Result<()> {
-        for frame in frames {
-            match frame.id {
-                SetPlayerPositionS::ID => {
-                    let packet: SetPlayerPositionS = frame.decode()?;
+    fn spawn_read_loop(&self) {
+        let player = self.clone();
 
-                    let tp_state = self.0.tp_state.read().await;
-                    match *tp_state {
-                        TeleportState::Clear => {
-                            let mut entity = self.0.entity.write().await;
-                            entity.reposition(packet.x, packet.feet_y, packet.z);
-                        }
+        tokio::spawn(async move {
+            loop {
+                match player.0.io.rx_raw().await {
+                    Ok(frame) => {
+                        let mut queue = player.0.frame_queue.lock().await;
+                        queue.push(frame);
+                    }
+                    Err(why) => match why.downcast_ref::<tokio::io::Error>().map(|e| e.kind()) {
+                        Some(tokio::io::ErrorKind::UnexpectedEof) => return,
                         _ => (),
+                    },
+                }
+            }
+        });
+    }
+
+    async fn handle_frame(&self, frame: Frame) -> Result<()> {
+        match frame.id {
+            SetPlayerPositionS::ID => {
+                let packet: SetPlayerPositionS = frame.decode()?;
+
+                let tp_state = self.0.tp_state.read().await;
+                match *tp_state {
+                    TeleportState::Clear => {
+                        let mut entity = self.0.entity.write().await;
+                        entity.reposition(packet.x, packet.feet_y, packet.z);
                     }
+                    _ => (),
                 }
+            }
 
-                SetPlayerPositionAndRotationS::ID => {
-                    let packet: SetPlayerPositionAndRotationS = frame.decode()?;
+            SetPlayerPositionAndRotationS::ID => {
+                let packet: SetPlayerPositionAndRotationS = frame.decode()?;
 
-                    let tp_state = self.0.tp_state.read().await;
-                    match *tp_state {
-                        TeleportState::Clear => {
-                            let mut entity = self.0.entity.write().await;
-                            entity.reposition(packet.x, packet.feet_y, packet.z);
-                            entity.rotate(packet.yaw, packet.pitch);
-                        }
-                        _ => (),
+                let tp_state = self.0.tp_state.read().await;
+                match *tp_state {
+                    TeleportState::Clear => {
+                        let mut entity = self.0.entity.write().await;
+                        entity.reposition(packet.x, packet.feet_y, packet.z);
+                        entity.rotate(packet.yaw, packet.pitch);
                     }
+                    _ => (),
                 }
+            }
 
-                ConfirmTeleportS::ID => {
-                    let packet: ConfirmTeleportS = frame.decode()?;
-                    self.check_teleports(Some(packet)).await?;
-                }
+            ConfirmTeleportS::ID => {
+                let packet: ConfirmTeleportS = frame.decode()?;
+                self.check_teleports(Some(packet)).await?;
+            }
 
-                UseItemOnS::ID => {
-                    let packet: UseItemOnS = frame.decode()?;
-                    self.handle_use_item(packet).await?;
-                }
+            UseItemOnS::ID => {
+                let packet: UseItemOnS = frame.decode()?;
+                self.handle_use_item(packet).await?;
+            }
 
-                id => {
-                    debug!(
-                        "Got packet with id {id} from player {}, ignoring",
-                        self.0.id
-                    );
-                }
+            id => {
+                debug!(
+                    "Got packet with id {id} from player {}, ignoring",
+                    self.0.id
+                );
             }
         }
 
···
             title: "Hi".into(),
         };
 
-        {
-            let mut io = self.0.io.lock().await;
-            io.tx(&OpenScreenC::from(&window)).await?;
-        }
+        self.0.io.tx(&OpenScreenC::from(&window)).await?;
 
         {
             let mut sw = self.0.window.write().await;

src/server/mod.rs

···
 pub mod ticker;
 pub mod window;
 
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 
 use color_eyre::eyre::Result;
 use tokio::time::Instant;
···
             tokio::spawn(Self::send_world_to(p.clone(), self.world_cache.clone()));
         }
 
-        let mut invalid_players: Vec<u16> = Vec::new();
+        let mut invalid_players: HashSet<u16> = HashSet::new();
 
         for (id, player) in &self.players {
             let _ = player.keepalive().await;
···
                 Ok(()) => (),
                 Err(why) => {
                     error!("error handling packets for player {}: {why}", player.id());
-                    invalid_players.push(*id);
+                    invalid_players.insert(*id);
                     continue;
                 }
             }
 
+            {
+                if player.0.io.connected().await == false {
+                    invalid_players.insert(*id);
+                }
+            }
+
             match player.check_teleports(None).await {
                 Err(TeleportError::TimedOut) | Err(TeleportError::WrongId(..)) => {
                     warn!("Player {} teleport failed, removing", player.0.id);
-                    invalid_players.push(*id);
+                    invalid_players.insert(*id);
                 }
                 _ => (),
             }
···
     }
 
     async fn send_world_to(player: SharedPlayer, world_cache: Arc<WorldCache>) -> Result<()> {
-        let mut io = player.0.io.lock().await;
-
         for packet in world_cache.encoded.iter() {
-            io.tx_raw(packet).await?;
+            player.0.io.tx_raw(packet).await?;
         }
 
         Ok(())