diff --git a/crates/net/src/chunk.rs b/crates/net/src/chunk.rs index 8ab581f..d906675 100644 --- a/crates/net/src/chunk.rs +++ b/crates/net/src/chunk.rs @@ -2,12 +2,15 @@ //! Chunk-stream transport: the per-connection task that pumps chunk subscriptions and deliveries. +use std::sync::Arc; + use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe}; use tokio::sync::mpsc::{Sender, UnboundedReceiver, UnboundedSender}; use tracing::{debug, warn}; use crate::codec::{MAX_CHUNK_FRAME_LEN, read_frame, write_frame}; use crate::runtime::ServerEvent; +use crate::stats::NetCounters; /// A synchronous handle the simulation loop uses to hand [`ChunkMessage`]s to a connection's chunk-stream task. /// @@ -123,6 +126,7 @@ pub(crate) async fn client_chunk_task( connection: quinn::Connection, mut subscribe: UnboundedReceiver, deliveries: Sender, + counters: &Arc, ) { // 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 { @@ -138,6 +142,11 @@ pub(crate) async fn client_chunk_task( loop { match read_frame::(&mut recv, MAX_CHUNK_FRAME_LEN).await { Ok(message) => { + // Counted on arrival rather than on delivery to the UI, so the figure reflects what the transport received even while the UI thread is backpressuring below. + match message { + ChunkMessage::Chunk { .. } => counters.record_chunk(), + ChunkMessage::Drop { .. } => counters.record_drop(), + } // `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; @@ -159,6 +168,7 @@ pub(crate) async fn client_chunk_task( warn!(%error, "failed to write chunk subscribe; ending chunk stream"); break; } + counters.record_subscribe(); } }; diff --git a/crates/net/src/lib.rs b/crates/net/src/lib.rs index ff81cb3..9d22646 100644 --- a/crates/net/src/lib.rs +++ b/crates/net/src/lib.rs @@ -12,11 +12,13 @@ pub mod endpoint; pub mod error; pub mod handshake; pub mod runtime; +pub mod stats; pub use chunk::{ChunkSink, ChunkSubscriber}; pub use runtime::{ ChunkStream, ClientLink, ConnectOutcome, NetworkServer, ServerEvent, connect_in_background, }; +pub use stats::NetStats; /// Default UDP port the server binds and the client connects to when none is configured. // TODO: make the bind address and port configurable through server/client configuration. diff --git a/crates/net/src/runtime.rs b/crates/net/src/runtime.rs index a81166a..7ffbc51 100644 --- a/crates/net/src/runtime.rs +++ b/crates/net/src/runtime.rs @@ -3,6 +3,7 @@ //! Threaded `tokio` runtime bridge between the async network and the synchronous simulation. use std::net::SocketAddr; +use std::sync::{Arc, OnceLock}; use std::thread; use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe}; @@ -13,6 +14,7 @@ use crate::chunk::{ChunkSink, ChunkSubscriber, chunk_stream_task, client_chunk_t use crate::endpoint::{client_endpoint, server_endpoint}; use crate::error::NetError; use crate::handshake::{ServerConnection, accept_connection, connect}; +use crate::stats::{self, NetCounters, NetStats}; /// 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>; @@ -34,6 +36,20 @@ pub struct ClientLink { pub subscribe: ChunkSubscriber, /// Receives chunk deliveries from the server, drained non-blocking each frame. pub chunks: ChunkStream, + /// The live QUIC connection, published by the network thread once the handshake completes. Held privately so the `quinn` types stay inside this crate; the UI thread reads through [`ClientLink::stats`]. + connection: Arc>, + /// Application-level message counters shared with the chunk task. + counters: Arc, +} + +impl ClientLink { + /// Snapshots the connection's transport statistics. + /// + /// Safe to call before the handshake completes; the result then reports the disconnected state rather than failing. + #[must_use] + pub fn stats(&self) -> NetStats { + stats::snapshot(self.connection.get(), &self.counters) + } } /// An event surfaced by the network thread to the synchronous server loop. @@ -257,6 +273,11 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli let (outcome_tx, outcome_rx) = crossbeam_channel::bounded(1); // Retained so a failure to spawn the thread can still be reported to the caller. let spawn_err_tx = outcome_tx.clone(); + // Published by the network thread once the handshake succeeds, so the UI thread can read `quinn`'s own connection statistics without owning the connection. + let connection = Arc::new(OnceLock::new()); + let task_connection = Arc::clone(&connection); + let counters = Arc::new(NetCounters::default()); + let task_counters = Arc::clone(&counters); // 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::(); // 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. @@ -292,8 +313,16 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli if outcome_tx.send(Ok(connected.ack.clone())).is_err() { return; } + // The cell is written exactly once, here; a failure would mean a second handshake on one link, which cannot occur. + let _ = task_connection.set(connected.connection.clone()); // Pump the chunk stream on this thread until the UI drops its handles or the server closes the connection. - client_chunk_task(connected.connection, subscribe_rx, chunks_tx).await; + client_chunk_task( + connected.connection, + subscribe_rx, + chunks_tx, + &task_counters, + ) + .await; warn!("server connection closed"); } Err(error) => { @@ -311,5 +340,7 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Cli handshake: outcome_rx, subscribe: ChunkSubscriber::new(subscribe_tx), chunks: chunks_rx, + connection, + counters, } } diff --git a/crates/net/src/stats.rs b/crates/net/src/stats.rs new file mode 100644 index 0000000..5e4cdab --- /dev/null +++ b/crates/net/src/stats.rs @@ -0,0 +1,103 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +//! Transport statistics for a client connection. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Application-level message counters, incremented by the client's chunk task and read by the UI thread. +/// +/// Held behind an [`Arc`] and mutated with relaxed atomics: each counter is independent, nothing else is ordered against them, and a reader that observes a slightly stale value is reporting a diagnostic figure, not making a decision. +#[derive(Debug, Default)] +pub(crate) struct NetCounters { + /// Chunk deliveries received from the server. + chunks_received: AtomicU64, + /// Chunk drop notifications received from the server. + drops_received: AtomicU64, + /// Subscription updates written to the server. + subscribes_sent: AtomicU64, +} + +impl NetCounters { + /// Records one received chunk delivery. + pub(crate) fn record_chunk(&self) { + self.chunks_received.fetch_add(1, Ordering::Relaxed); + } + + /// Records one received chunk drop notification. + pub(crate) fn record_drop(&self) { + self.drops_received.fetch_add(1, Ordering::Relaxed); + } + + /// Records one subscription update written to the server. + pub(crate) fn record_subscribe(&self) { + self.subscribes_sent.fetch_add(1, Ordering::Relaxed); + } +} + +/// A snapshot of one client connection's transport state. +/// +/// Every field is a value copied at the moment of the call. A snapshot taken before the handshake completes reports `connected == false` and zeroes throughout, which is a meaningful state rather than missing data. +#[derive(Copy, Clone, Debug, Default, PartialEq)] +pub struct NetStats { + /// Whether a QUIC connection is currently established. + pub connected: bool, + /// Bytes carried in UDP datagrams sent on this connection, including QUIC framing and retransmissions. + pub bytes_sent: u64, + /// Bytes carried in UDP datagrams received on this connection, including QUIC framing. + pub bytes_received: u64, + /// UDP datagrams sent on this connection. + pub datagrams_sent: u64, + /// UDP datagrams received on this connection. + pub datagrams_received: u64, + /// Chunk deliveries received since the connection was established. + pub chunks_received: u64, + /// Chunk drop notifications received since the connection was established. + pub drops_received: u64, + /// Subscription updates written to the server since the connection was established. + pub subscribes_sent: u64, + /// The QUIC stack's current round-trip-time estimate, in milliseconds. + pub rtt_ms: f32, + /// Packets the congestion controller has declared lost on the current path. + pub lost_packets: u64, + /// Current congestion window, in bytes: the ceiling on data in flight. + pub congestion_window: u64, + /// Largest UDP payload the current path is known to carry, in bytes, as discovered by path MTU probing. + pub path_mtu: u16, +} + +/// Reads a connection's transport statistics, or reports the disconnected state. +/// +/// `connection` is [`None`] until the handshake completes and after the connection closes. +pub(crate) fn snapshot( + connection: Option<&quinn::Connection>, + counters: &Arc, +) -> NetStats { + let mut stats = NetStats { + chunks_received: counters.chunks_received.load(Ordering::Relaxed), + drops_received: counters.drops_received.load(Ordering::Relaxed), + subscribes_sent: counters.subscribes_sent.load(Ordering::Relaxed), + ..NetStats::default() + }; + + let Some(connection) = connection else { + return stats; + }; + + let quic = connection.stats(); + // A connection handle outlives the connection itself; a close reason is how a torn-down connection distinguishes itself from a live one, and its final counters stay readable either way. + stats.connected = connection.close_reason().is_none(); + stats.bytes_sent = quic.udp_tx.bytes; + stats.bytes_received = quic.udp_rx.bytes; + stats.datagrams_sent = quic.udp_tx.datagrams; + stats.datagrams_received = quic.udp_rx.datagrams; + stats.rtt_ms = quic.path.rtt.as_secs_f32() * 1000.0; + stats.lost_packets = quic.path.lost_packets; + stats.congestion_window = quic.path.cwnd; + stats.path_mtu = quic.path.current_mtu; + stats +} + +#[cfg(test)] +#[path = "tests/stats.rs"] +mod tests; diff --git a/crates/net/src/tests/stats.rs b/crates/net/src/tests/stats.rs new file mode 100644 index 0000000..ed6be00 --- /dev/null +++ b/crates/net/src/tests/stats.rs @@ -0,0 +1,31 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +//! Unit tests for the transport statistics snapshot. + +use super::*; + +#[test] +fn snapshot_without_a_connection_reports_the_disconnected_state() { + let counters = Arc::new(NetCounters::default()); + let stats = snapshot(None, &counters); + + assert!(!stats.connected); + assert_eq!(stats, NetStats::default()); +} + +#[test] +fn counters_are_reported_before_a_connection_exists() { + let counters = Arc::new(NetCounters::default()); + counters.record_chunk(); + counters.record_chunk(); + counters.record_drop(); + counters.record_subscribe(); + + let stats = snapshot(None, &counters); + assert_eq!(stats.chunks_received, 2); + assert_eq!(stats.drops_received, 1); + assert_eq!(stats.subscribes_sent, 1); + // Transport figures stay zero: they come from the QUIC stack, which has nothing to report yet. + assert_eq!(stats.bytes_sent, 0); + assert_eq!(stats.bytes_received, 0); +}