From f465cfeeaf8f8ec2976174694216b9b7eed254e2 Mon Sep 17 00:00:00 2001 From: Serkyo Date: Wed, 22 Jul 2026 05:19:22 +0200 Subject: [PATCH] fix(net): bound chunk delivery channel to apply backpressure --- crates/client/src/chunks.rs | 2 +- crates/client/src/main.rs | 4 ++-- crates/net/src/chunk.rs | 8 ++++---- crates/net/src/runtime.rs | 12 +++++++++--- 4 files changed, 16 insertions(+), 10 deletions(-) diff --git a/crates/client/src/chunks.rs b/crates/client/src/chunks.rs index 962ffeb..c20cffd 100644 --- a/crates/client/src/chunks.rs +++ b/crates/client/src/chunks.rs @@ -42,7 +42,7 @@ impl ChunkManager { pub fn update( &mut self, center: ChunkPos, - deliveries: &net::ChunkStream, + deliveries: &mut net::ChunkStream, renderer: &mut renderer::Renderer, ) { let unloaded = self.unload_outside(center, renderer); diff --git a/crates/client/src/main.rs b/crates/client/src/main.rs index 6f51997..44cb910 100644 --- a/crates/client/src/main.rs +++ b/crates/client/src/main.rs @@ -262,10 +262,10 @@ impl ApplicationHandler for App { // Apply queued server deliveries and reconcile the resident set against the camera. if let (Some(chunks), Some(link), Some(renderer)) = ( self.chunks.as_mut(), - self.link.as_ref(), + self.link.as_mut(), self.renderer.as_mut(), ) { - chunks.update(center, &link.chunks, renderer); + chunks.update(center, &mut link.chunks, renderer); } let view = self.camera.view_matrix(); diff --git a/crates/net/src/chunk.rs b/crates/net/src/chunk.rs index a2598e0..dea63b9 100644 --- a/crates/net/src/chunk.rs +++ b/crates/net/src/chunk.rs @@ -7,7 +7,7 @@ //! The two channels crossing the async/sync boundary run in opposite directions and therefore use different primitives. Inbound (`ChunkSubscribe` arriving async, consumed by the sync loop) reuses the crossbeam [`ServerEvent`] channel, whose sender is non-blocking. Outbound (a `ChunkMessage` produced by the sync loop, consumed async) uses a `tokio` unbounded MPSC: its `send` is synchronous, so the non-async simulation thread can push without a runtime, while the receiver's `recv().await` composes into the pump's `select!`. A blocking `crossbeam` receiver would instead freeze the current-thread runtime and cannot appear in a `select!` arm. use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe}; -use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender}; +use tokio::sync::mpsc::{Sender, UnboundedReceiver, UnboundedSender}; use tracing::{debug, warn}; use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame}; @@ -125,7 +125,7 @@ impl ChunkSubscriber { pub(crate) async fn client_chunk_task( connection: quinn::Connection, mut subscribe: UnboundedReceiver, - deliveries: crossbeam_channel::Sender, + deliveries: Sender, ) { // The client opens the chunk stream after the handshake; the server accepts it, mirroring the control-stream convention. let (mut send, mut recv) = match connection.open_bi().await { @@ -155,8 +155,8 @@ pub(crate) async fn client_chunk_task( frame = read_frame::(&mut recv, MAX_CHUNK_FRAME_LEN) => { match frame { Ok(message) => { - // A closed delivery receiver means the UI is gone; nothing more to do. - if deliveries.send(message).is_err() { + // `send` awaits when the delivery channel is full: the task suspends (yielding the runtime thread so the connection keeps ACKing) until the UI drains a slot, and until then reads no further frames, which backpressures the server via QUIC stream flow control. An error means the UI dropped its receiver, so the session ends. + if deliveries.send(message).await.is_err() { break; } } diff --git a/crates/net/src/runtime.rs b/crates/net/src/runtime.rs index 27b21c2..a81166a 100644 --- a/crates/net/src/runtime.rs +++ b/crates/net/src/runtime.rs @@ -17,8 +17,12 @@ use crate::handshake::{ServerConnection, accept_connection, connect}; /// Channel receiver delivering the outcome of a background client connect: the negotiated [`HandshakeAck`] on success, or a human-readable error string on failure. pub type ConnectOutcome = crossbeam_channel::Receiver>; -/// Non-blocking receiver of chunks delivered by the server, drained by the UI thread with `try_recv`. -pub type ChunkStream = crossbeam_channel::Receiver; +/// Bounded receiver of chunks delivered by the server, drained by the UI thread with `try_recv`. +pub type ChunkStream = tokio::sync::mpsc::Receiver; + +/// Capacity of the client's chunk-delivery channel, in [`ChunkMessage`]s. +// TODO: revisit once meshing moves to a worker pool; the right depth follows the UI's consume rate, so this is a candidate to derive from the meshing budget / view distance in a config layer rather than a hand-set constant. +const CHUNK_DELIVERY_CAPACITY: usize = 32; /// Handles a background client connection exposes to the synchronous UI thread. /// @@ -255,7 +259,9 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli let spawn_err_tx = outcome_tx.clone(); // Subscription updates flow UI -> network (sync send, async recv); chunk deliveries flow network -> UI (async send, sync try_recv). let (subscribe_tx, subscribe_rx) = tokio::sync::mpsc::unbounded_channel::(); - let (chunks_tx, chunks_rx) = crossbeam_channel::unbounded::(); + // The delivery channel is bounded so a slow (e.g. debug-build) UI thread applies backpressure to the network task instead of letting undelivered chunks accumulate without limit. + let (chunks_tx, chunks_rx) = + tokio::sync::mpsc::channel::(CHUNK_DELIVERY_CAPACITY); let spawned = thread::Builder::new() .name("net-client".to_owned())