177 lines
8.7 KiB
Rust
177 lines
8.7 KiB
Rust
// SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
//! Chunk-stream transport: the per-connection task that pumps chunk subscriptions and deliveries.
|
|
//!
|
|
//! After the handshake, each connection carries a dedicated bidirectional QUIC stream for chunk sync (the canonical `StreamLayout::chunk_lod0` id). The client writes [`ChunkSubscribe`] requests on it and the server writes [`ChunkMessage`] deliveries back on the same stream. This module owns the server-side pump: a single [`tokio::select`] loop that reads subscriptions off the stream and forwards them to the synchronous simulation loop, while draining outbound [`ChunkMessage`]s handed to it by that loop.
|
|
//!
|
|
//! 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 tracing::{debug, warn};
|
|
|
|
use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame};
|
|
use crate::runtime::ServerEvent;
|
|
|
|
/// A synchronous handle the simulation loop uses to hand [`ChunkMessage`]s to a connection's chunk-stream task.
|
|
///
|
|
/// The wrapped channel is a `tokio` unbounded MPSC. Its `send` is synchronous and callable from the non-async simulation thread with no runtime in scope, while the connection's task drains the receiver with `recv().await` so it composes into the task's `select!`. The `tokio` sender type is kept private so the `server` crate never names it.
|
|
#[derive(Debug, Clone)]
|
|
pub struct ChunkSink {
|
|
/// Outbound queue drained by the connection's chunk-stream task.
|
|
tx: UnboundedSender<ChunkMessage>,
|
|
}
|
|
|
|
impl ChunkSink {
|
|
/// Wraps `tx` as a chunk sink.
|
|
pub(crate) fn new(tx: UnboundedSender<ChunkMessage>) -> Self {
|
|
Self { tx }
|
|
}
|
|
|
|
/// Queues `msg` for delivery on the connection's chunk stream.
|
|
///
|
|
/// Non-blocking. A send failure means the receiving task has ended (the connection dropped); it is logged at debug and swallowed, since the simulation loop cannot act on a departed connection.
|
|
pub fn send(&self, msg: ChunkMessage) {
|
|
if self.tx.send(msg).is_err() {
|
|
debug!("chunk sink send failed; connection task has ended");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Runs the server-side chunk-stream pump for one connection until the stream or connection closes.
|
|
///
|
|
/// Accepts the connection's chunk stream, then loops: inbound [`ChunkSubscribe`] frames are forwarded to the simulation loop as [`ServerEvent::ChunkSubscribe`], and outbound [`ChunkMessage`]s taken from `outbound` are written onto the stream. The loop ends when the peer closes the stream, when the events receiver is gone (the server is shutting down), or when the outbound sink is dropped.
|
|
pub(crate) async fn chunk_stream_task(
|
|
connection: quinn::Connection,
|
|
id: u64,
|
|
events: crossbeam_channel::Sender<ServerEvent>,
|
|
mut outbound: UnboundedReceiver<ChunkMessage>,
|
|
) {
|
|
// The client opens the chunk stream after the handshake; the server accepts it here, mirroring the control-stream convention.
|
|
let (mut send, mut recv) = match connection.accept_bi().await {
|
|
Ok(stream) => stream,
|
|
Err(error) => {
|
|
warn!(%error, id, "failed to accept chunk stream");
|
|
return;
|
|
}
|
|
};
|
|
|
|
loop {
|
|
tokio::select! {
|
|
// A subscription frame arrived from the client.
|
|
frame = read_frame::<ChunkSubscribe>(&mut recv, MAX_CHUNK_FRAME_LEN) => {
|
|
match frame {
|
|
Ok(request) => {
|
|
// A closed events receiver means the simulation loop is gone; nothing more to do.
|
|
if events
|
|
.send(ServerEvent::ChunkSubscribe { id, request })
|
|
.is_err()
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
Err(error) => {
|
|
// A read error is the normal end of a client session (stream finished or reset).
|
|
debug!(%error, id, "chunk stream read ended");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
// The simulation loop handed back a chunk to deliver.
|
|
msg = outbound.recv() => {
|
|
match msg {
|
|
Some(message) => {
|
|
if let Err(error) = write_frame(&mut send, &message).await {
|
|
warn!(%error, id, "failed to write chunk frame; ending chunk stream");
|
|
break;
|
|
}
|
|
}
|
|
// The sink was dropped: the connection is being torn down.
|
|
None => break,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A synchronous handle the client's UI thread uses to push [`ChunkSubscribe`] requests to its network task.
|
|
///
|
|
/// The client-side mirror of [`ChunkSink`]: `send` is synchronous and callable from the winit loop with no runtime in scope, while the client's chunk task drains the receiver with `recv().await` so it composes into that task's `select!`.
|
|
#[derive(Debug, Clone)]
|
|
pub struct ChunkSubscriber {
|
|
/// Outbound queue of subscription updates drained by the client's chunk task.
|
|
tx: UnboundedSender<ChunkSubscribe>,
|
|
}
|
|
|
|
impl ChunkSubscriber {
|
|
/// Wraps `tx` as a chunk subscriber.
|
|
pub(crate) fn new(tx: UnboundedSender<ChunkSubscribe>) -> Self {
|
|
Self { tx }
|
|
}
|
|
|
|
/// Queues a subscription update for the server.
|
|
///
|
|
/// Non-blocking. A send failure means the network task has ended (the connection dropped); it is logged at debug and swallowed, since the UI thread cannot act on a departed connection.
|
|
pub fn send(&self, request: ChunkSubscribe) {
|
|
if self.tx.send(request).is_err() {
|
|
debug!("chunk subscriber send failed; network task has ended");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Runs the client-side chunk-stream pump for one connection until the stream or connection closes.
|
|
///
|
|
/// Opens the chunk stream, then loops: [`ChunkSubscribe`] requests taken from `subscribe` are written to the server, and inbound [`ChunkMessage`] frames are forwarded to the UI thread over `deliveries`. The loop ends when the UI drops its subscriber, when the delivery receiver is gone, or when the stream closes.
|
|
pub(crate) async fn client_chunk_task(
|
|
connection: quinn::Connection,
|
|
mut subscribe: UnboundedReceiver<ChunkSubscribe>,
|
|
deliveries: crossbeam_channel::Sender<ChunkMessage>,
|
|
) {
|
|
// 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 {
|
|
Ok(stream) => stream,
|
|
Err(error) => {
|
|
warn!(%error, "failed to open chunk stream");
|
|
return;
|
|
}
|
|
};
|
|
|
|
loop {
|
|
tokio::select! {
|
|
// The UI thread pushed a subscription update to forward to the server.
|
|
request = subscribe.recv() => {
|
|
match request {
|
|
Some(request) => {
|
|
if let Err(error) = write_frame(&mut send, &request).await {
|
|
warn!(%error, "failed to write chunk subscribe; ending chunk stream");
|
|
break;
|
|
}
|
|
}
|
|
// The subscriber was dropped: the UI is shutting down.
|
|
None => break,
|
|
}
|
|
}
|
|
// A chunk arrived from the server.
|
|
frame = read_frame::<ChunkMessage>(&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() {
|
|
break;
|
|
}
|
|
}
|
|
Err(error) => {
|
|
// A read error is the normal end of the session (stream finished or reset).
|
|
debug!(%error, "chunk stream read ended");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "tests/chunk.rs"]
|
|
mod tests;
|