feat: #1 optimize backpressure handling with improved buffer sizes and data types

Author: ronakgh97

README.md

@@ -12,7 +12,7 @@
 [![Dialga](https://img.pokemondb.net/sprites/black-white/anim/normal/dialga.gif)](https://pokemondb.net/pokedex/dialga)
 [![Palkia](https://img.pokemondb.net/sprites/black-white/anim/normal/palkia.gif)](https://pokemondb.net/pokedex/palkia)
 
-Networking in games is pure suffering.
+***Networking in games is pure suffering.***
 
 You start with “just a simple relay server.” _(alone let authoritative servers)_
 Just a few sockets. A couple messages. Clean.
@@ -81,4 +81,4 @@ TODO
 - Add tuned buffering and improve performance by lessen serialization and deserialization, where possible (Zero-copy,
   etc.)
 - Fixed the DAMN BACKPRESSURE, CHANNEL GETS FULL AND THEN EVERYTHING BLOWS UP
-- Somehow increase TCP write throughput, the only bottleneck right now is TCP write, which is fucking I/O bound!! 
+- Somehow increase TCP write throughput, the only bottleneck right now is TCP write, which is fucking I/O bound!!

examples/bot.rs

@@ -11,8 +11,8 @@ use wincode::{SchemaRead, SchemaWrite};
 
 const ADDR: &str = "127.0.0.1:7777";
 const ROOM: &str = "test-room";
-const BOT_COUNT: usize = 144;
-const TICK_RATE: u64 = 8;
+const BOT_COUNT: usize = 64;
+const TICK_RATE: u64 = 24;
 const MAX_SPEED: f32 = 400.0;
 const SCREEN: f32 = 600.0;
 

examples/daemon.rs

@@ -1,4 +1,5 @@
 use anyhow::Result;
+use dashmap::DashMap;
 use ghost_sync::{info, warn, Server, ServerHandle, ServerHandler, Uuid};
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -13,26 +14,31 @@ struct Daemon {
 }
 
 struct Metrics {
-    total_connections: AtomicUsize,
     start_time: Instant,
+    total_connections: AtomicUsize,
+    // Key - ClientID, Value - Number of frames dropped
+    backpressure_tracker: DashMap<Uuid, AtomicUsize>,
 }
 
 impl Daemon {
     fn new() -> Self {
         Self {
             metrics: Metrics {
-                total_connections: AtomicUsize::new(0),
                 start_time: Instant::now(),
+                total_connections: AtomicUsize::new(0),
+                backpressure_tracker: DashMap::with_capacity(1024),
             },
         }
     }
 }
 
-struct DaemonHandler(Arc<Daemon>);
+struct DaemonHandler {
+    daemon: Arc<Daemon>,
+}
 
 impl DaemonHandler {
-    fn new(demon: Arc<Daemon>) -> Self {
-        Self(demon)
+    fn new(daemon: Arc<Daemon>) -> Self {
+        Self { daemon }
     }
 }
 
@@ -46,7 +52,7 @@ struct RoomMeta {
 impl ServerHandler for DaemonHandler {
     fn on_connect(&self, addr: SocketAddr) -> bool {
         let count = self
-            .0
+            .daemon
             .metrics
             .total_connections
             .fetch_add(1, Ordering::Relaxed)
@@ -77,6 +83,12 @@ impl ServerHandler for DaemonHandler {
             "Backpressure: Client {} in room '{}' has a full write channel, frame dropped",
             client_id, room_id
         );
+        let tracker = &self.daemon.metrics.backpressure_tracker;
+        _ = tracker
+            .entry(client_id)
+            .or_insert_with(|| AtomicUsize::new(0))
+            .fetch_add(1, Ordering::Relaxed)
+            + 1;
     }
 }
 

examples/test_daemon.py

@@ -44,13 +44,13 @@ def main():
         room_count = resp.count("clients\n")
         print(f"Room count: {room_count}")
 
-        print("> CREATE_ROOM test-room")
+        print("> CREATE_ROOM foo-bar-room")
         print(send_cmd(sock, "CREATE_ROOM\ntest-room\n"))
 
         print("> ROOM_CLIENTS test-room")
         print(send_cmd(sock, "ROOM_CLIENTS\ntest-room\n"))
 
-        print("> DELETE_ROOM test-room")
+        print("> DELETE_ROOM foo-bar-room")
         print(send_cmd(sock, "DELETE_ROOM\ntest-room\n"))
 
         print("> METRICS")

src/client.rs

@@ -1,3 +1,4 @@
+use bytes::Bytes;
 use tokio::io::{BufReader, BufWriter};
 use tokio::net::TcpStream;
 use uuid::Uuid;
@@ -109,7 +110,10 @@ impl Client {
             ServerWire::PlayerJoined { client_id } => ServerEvent::PlayerJoined { client_id },
             ServerWire::PlayerLeft { client_id } => ServerEvent::PlayerLeft { client_id },
             ServerWire::Error(msg) => ServerEvent::Error(msg),
-            ServerWire::Broadcast { sender_id, data } => ServerEvent::Broadcast { sender_id, data },
+            ServerWire::Broadcast { sender_id, data } => ServerEvent::Broadcast {
+                sender_id,
+                data: Bytes::from(data),
+            },
             // Ping/Pong are handled internally before reaching here
             _ => unreachable!("ping/pong should be handled in recv()"),
         }
@@ -155,8 +159,8 @@ impl ClientBuilder {
         let (read_half, write_half) = stream.into_split();
 
         Ok(Client {
-            reader: BufReader::with_capacity(2 * 1024 * 1024, read_half),
-            writer: BufWriter::with_capacity(1024 * 1024, write_half),
+            reader: BufReader::with_capacity(4 * 1024 * 1024, read_half),
+            writer: BufWriter::with_capacity(6 * 1024 * 1024, write_half),
             max_payload: self.max_payload,
             client_id: None,
         })

src/protocol.rs

@@ -1,4 +1,5 @@
 use crate::types::Result;
+use bytes::Bytes;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use wincode::config::DefaultConfig;
 use wincode::SchemaWrite;
@@ -21,7 +22,7 @@ where
 
 /// Read a length-prefixed wincode-deserialized frame with a max payload size check.
 #[inline(always)]
-pub async fn read_frame_raw<R>(reader: &mut R, max_payload: usize) -> Result<Vec<u8>>
+pub async fn read_frame_raw<R>(reader: &mut R, max_payload: usize) -> Result<Bytes>
 where
     R: AsyncRead + Unpin,
 {
@@ -38,18 +39,18 @@ where
 
     let mut payload = vec![0u8; len];
     reader.read_exact(&mut payload).await?;
-    Ok(payload)
+    Ok(Bytes::from(payload))
 }
 
 /// Write a raw byte slice as a length-prefixed frame.
 #[inline(always)]
-pub async fn write_frame_raw<W>(writer: &mut W, payload: &[u8]) -> Result<()>
+pub async fn write_frame_raw<W>(writer: &mut W, payload: Bytes) -> Result<()>
 where
     W: AsyncWrite + Unpin,
 {
     let len = payload.len() as u32;
     writer.write_all(&len.to_be_bytes()).await?;
-    writer.write_all(payload).await?;
+    writer.write_all(&payload).await?;
     writer.flush().await?;
     Ok(())
 }

src/server.rs

@@ -366,8 +366,8 @@ impl Server {
         let client_id = Uuid::new_v4();
         stream.set_nodelay(true).ok();
         let (read_half, write_half) = stream.into_split();
-        let mut reader = tokio::io::BufReader::with_capacity(1024 * 1024, read_half);
-        let mut writer = tokio::io::BufWriter::with_capacity(4 * 1024 * 1024, write_half);
+        let mut reader = tokio::io::BufReader::with_capacity(6 * 1024 * 1024, read_half);
+        let mut writer = tokio::io::BufWriter::with_capacity(12 * 1024 * 1024, write_half);
 
         // This is the main bottleneck for backpressure handling. Each client has a dedicated writer task
         // that receives frames to send via this channel. If the channel is full, we know the client is falling behind and can drop frames or disconnect as needed.
@@ -376,7 +376,7 @@ impl Server {
         // Spawn writer task
         let writer_handle = tokio::spawn(async move {
             while let Some(frame) = write_rx.recv().await {
-                if let Err(e) = protocol::write_frame_raw(&mut writer, &frame).await {
+                if let Err(e) = protocol::write_frame_raw(&mut writer, frame).await {
                     #[allow(clippy::needless_ifs)]
                     if !e.is_connection_closed() {}
                     warn!("write error: {e}");
@@ -617,6 +617,7 @@ impl Server {
     #[inline(always)]
     async fn send_to_client(&self, client_id: Uuid, tx: &mpsc::Sender<Bytes>, msg: &ServerWire) {
         if let Ok(payload) = wincode::serialize(msg) {
+            // If the channel is full, we drop the frame and call the backpressure hook.
             if tx.try_send(Bytes::from(payload)).is_err() {
                 if let Some(state) = self.clients.get(&client_id) {
                     if let Some(ref room_id) = state.room_id {
@@ -711,7 +712,7 @@ impl ServerBuilder {
 
     /// Build the server. Call [`Server::run`] to start accepting connections.
     pub fn build(self) -> Server {
-        let (tx, _) = broadcast::channel(1);
+        let (tx, _) = broadcast::channel(4);
         Server {
             config: self.config.clone(),
             handler: self.handler,

src/types.rs

@@ -1,3 +1,4 @@
+use bytes::Bytes;
 use std::time::Duration;
 use uuid::Uuid;
 use wincode::{SchemaRead, SchemaWrite};
@@ -37,7 +38,7 @@ pub enum ServerEvent {
     /// Server error.
     Error(String),
     /// Relayed data from a peer.
-    Broadcast { sender_id: Uuid, data: Vec<u8> },
+    Broadcast { sender_id: Uuid, data: Bytes },
 }
 
 /// Server configuration.
@@ -74,7 +75,7 @@ impl Default for ServerConfig {
             // Non-divisible defaults to avoid idle_timeout and ping_interval
             idle_timeout: Duration::from_secs(31),
             ping_interval: Duration::from_secs(13),
-            channel_capacity: 64,
+            channel_capacity: 1024,
         }
     }
 }

tests/integration.rs

@@ -1,10 +1,10 @@
 #![allow(deprecated)]
-use std::sync::{Arc, Mutex};
-use std::time::Duration;
-
+use bytes::Bytes;
 use ghost_sync::{Client, Server, ServerEvent, SyncError};
 use rand::rngs::StdRng;
 use rand::{RngExt, SeedableRng};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
 use tokio::sync::{mpsc, watch};
 
 /// Helper: start a server on a random port with a "test" room.
@@ -97,7 +97,7 @@ async fn broadcast_relay() {
     // A should receive it
     match a.recv().await.unwrap() {
         Some(ServerEvent::Broadcast { data, .. }) => {
-            assert_eq!(&data, b"hello");
+            assert_eq!(data, Bytes::from_static(b"hello"));
         }
         other => panic!("expected Broadcast, got: {:?}", other.is_some()),
     }