fix(net): bound chunk delivery channel to apply backpressure
This commit is contained in:
parent
a20d98a107
commit
f465cfeeaf
|
|
@ -42,7 +42,7 @@ impl ChunkManager {
|
||||||
pub fn update(
|
pub fn update(
|
||||||
&mut self,
|
&mut self,
|
||||||
center: ChunkPos,
|
center: ChunkPos,
|
||||||
deliveries: &net::ChunkStream,
|
deliveries: &mut net::ChunkStream,
|
||||||
renderer: &mut renderer::Renderer,
|
renderer: &mut renderer::Renderer,
|
||||||
) {
|
) {
|
||||||
let unloaded = self.unload_outside(center, renderer);
|
let unloaded = self.unload_outside(center, renderer);
|
||||||
|
|
|
||||||
|
|
@ -262,10 +262,10 @@ impl ApplicationHandler for App {
|
||||||
// Apply queued server deliveries and reconcile the resident set against the camera.
|
// Apply queued server deliveries and reconcile the resident set against the camera.
|
||||||
if let (Some(chunks), Some(link), Some(renderer)) = (
|
if let (Some(chunks), Some(link), Some(renderer)) = (
|
||||||
self.chunks.as_mut(),
|
self.chunks.as_mut(),
|
||||||
self.link.as_ref(),
|
self.link.as_mut(),
|
||||||
self.renderer.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();
|
let view = self.camera.view_matrix();
|
||||||
|
|
|
||||||
|
|
@ -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.
|
//! 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 shared::protocol::chunk::{ChunkMessage, ChunkSubscribe};
|
||||||
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
|
use tokio::sync::mpsc::{Sender, UnboundedReceiver, UnboundedSender};
|
||||||
use tracing::{debug, warn};
|
use tracing::{debug, warn};
|
||||||
|
|
||||||
use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame};
|
use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame};
|
||||||
|
|
@ -125,7 +125,7 @@ impl ChunkSubscriber {
|
||||||
pub(crate) async fn client_chunk_task(
|
pub(crate) async fn client_chunk_task(
|
||||||
connection: quinn::Connection,
|
connection: quinn::Connection,
|
||||||
mut subscribe: UnboundedReceiver<ChunkSubscribe>,
|
mut subscribe: UnboundedReceiver<ChunkSubscribe>,
|
||||||
deliveries: crossbeam_channel::Sender<ChunkMessage>,
|
deliveries: Sender<ChunkMessage>,
|
||||||
) {
|
) {
|
||||||
// The client opens the chunk stream after the handshake; the server accepts it, mirroring the control-stream convention.
|
// 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 {
|
let (mut send, mut recv) = match connection.open_bi().await {
|
||||||
|
|
@ -155,8 +155,8 @@ pub(crate) async fn client_chunk_task(
|
||||||
frame = read_frame::<ChunkMessage>(&mut recv, MAX_CHUNK_FRAME_LEN) => {
|
frame = read_frame::<ChunkMessage>(&mut recv, MAX_CHUNK_FRAME_LEN) => {
|
||||||
match frame {
|
match frame {
|
||||||
Ok(message) => {
|
Ok(message) => {
|
||||||
// A closed delivery receiver means the UI is gone; nothing more to do.
|
// `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).is_err() {
|
if deliveries.send(message).await.is_err() {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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.
|
/// 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<Result<HandshakeAck, String>>;
|
pub type ConnectOutcome = crossbeam_channel::Receiver<Result<HandshakeAck, String>>;
|
||||||
|
|
||||||
/// Non-blocking receiver of chunks delivered by the server, drained by the UI thread with `try_recv`.
|
/// Bounded receiver of chunks delivered by the server, drained by the UI thread with `try_recv`.
|
||||||
pub type ChunkStream = crossbeam_channel::Receiver<ChunkMessage>;
|
pub type ChunkStream = tokio::sync::mpsc::Receiver<ChunkMessage>;
|
||||||
|
|
||||||
|
/// 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.
|
/// 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();
|
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).
|
// 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::<ChunkSubscribe>();
|
let (subscribe_tx, subscribe_rx) = tokio::sync::mpsc::unbounded_channel::<ChunkSubscribe>();
|
||||||
let (chunks_tx, chunks_rx) = crossbeam_channel::unbounded::<ChunkMessage>();
|
// 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::<ChunkMessage>(CHUNK_DELIVERY_CAPACITY);
|
||||||
|
|
||||||
let spawned = thread::Builder::new()
|
let spawned = thread::Builder::new()
|
||||||
.name("net-client".to_owned())
|
.name("net-client".to_owned())
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue