From fd8e5bafe04788bbc969302ad65ce36ed02501ef Mon Sep 17 00:00:00 2001 From: Floris Bruynooghe Date: Tue, 24 Mar 2026 17:28:00 +0100 Subject: [PATCH] refactor(proto): Move FrameStats into PathStats (#521) ## Description This moves the FrameStats into the PathStats, allowing us to check what frames where sent on which paths. It then makes the ConnectionStats always be computed to be the sum of all the PathStats. This has as a nice side effect that some stats no longer need to be recorded twice. Which was also error-prone. The PathStats are now boxed in the PathEvent because that variant was now way bigger than any of the other variants. It also fixes a few missed fields in the stats code and uses more defensive code so that won't happen again. ## Breaking Changes Probably none? I'm only adding some fields to PathStats on the public API, while the fields on ConnectionStats remain the same. ## Notes & open questions I need this to be able to write tests that assert frames on specific paths. But I also think that generally this makes more sense. The previous version of the stats was thrown together without much thought I think (by me). I've cleaned up some cargo.toml mistakes that accumulated by using tooling that doesn't follow the sorting/and locations of where things live. --- Cargo.toml | 16 +- noq-proto/Cargo.toml | 6 +- noq-proto/src/connection/mod.rs | 127 +++++++-------- noq-proto/src/connection/paths.rs | 2 +- noq-proto/src/connection/stats.rs | 260 +++++++++++++++++++++++------- noq/src/connection.rs | 6 +- 6 files changed, 278 insertions(+), 139 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 694a8e09f..9e235ddec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,22 +15,25 @@ categories = ["network-programming", "asynchronous"] aes-gcm = { version = "0.10.3", default-features = false, features = ["aes"] } anyhow = "1.0.22" arbitrary = { version = "1.0.1", features = ["derive"] } -async-io = "2" assert_matches = "1.1" +async-io = "2" aws-lc-rs = { version = "1.9", default-features = false } -criterion = { version = "0.7", default-features = false, features = ["async_tokio"] } bytes = "1" +cfg_aliases = "0.2.1" clap = { version = "4.5", features = ["derive"] } crc = "3" +criterion = { version = "0.7", default-features = false, features = ["async_tokio"] } +derive_more = { version = "2.1.0", features = ["debug", "deref", "deref_mut", "display", "from", "add", "add_assign"] } directories-next = "2" +enum-assoc = "1.3.0" fastbloom = { version = "0.17", default-features = false } futures-io = "0.3.19" getrandom = { version = "0.4", default-features = false } hdrhistogram = { version = "7.2", default-features = false } hex-literal = "0.4" identity-hash = "0.1.0" -lru-slab = "0.1.2" log = "0.4" +lru-slab = "0.1.2" pin-project-lite = "0.2" proptest = { version = "1.9.0", default-features = false, features = ["std"] } qlog = { package = "n0-qlog", version = "0.1.0" } @@ -40,8 +43,8 @@ ring = "0.17" rustc-hash = "2" rustls = { version = "0.23.33", default-features = false, features = ["std"] } rustls-pemfile = "2" -rustls-platform-verifier = "0.6" rustls-pki-types = "1.7" +rustls-platform-verifier = "0.6" serde = { version = "1.0", features = ["derive"] } serde_json = "1" slab = "0.4.9" @@ -59,12 +62,11 @@ url = "2" wasm-bindgen-test = { version = "0.3.45" } web-time = "1" windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_System_IO", "Win32_Networking_WinSock", "Win32_System_LibraryLoader"] } -cfg_aliases = "0.2.1" # Fix minimal dependencies for indirect deps -async-global-executor = "2.4.1" -async-fs = "2.1" async-executor = "1.13.0" +async-fs = "2.1" +async-global-executor = "2.4.1" aws-lc-fips-sys = "0.13.10" aws-lc-sys = "0.39.0" gcc = "0.3.55" diff --git a/noq-proto/Cargo.toml b/noq-proto/Cargo.toml index 82ad4f463..beac687eb 100644 --- a/noq-proto/Cargo.toml +++ b/noq-proto/Cargo.toml @@ -57,13 +57,15 @@ arbitrary = { workspace = true, optional = true } aws-lc-rs = { workspace = true, optional = true } bytes = { workspace = true } criterion = { workspace = true, optional = true } +derive_more = { workspace = true } +enum-assoc = { workspace = true } fastbloom = { workspace = true, optional = true } identity-hash = { workspace = true } lru-slab = { workspace = true } qlog = { workspace = true, optional = true } -rustc-hash = { workspace = true } rand = { workspace = true } ring = { workspace = true, optional = true } +rustc-hash = { workspace = true } rustls = { workspace = true, optional = true } rustls-platform-verifier = { workspace = true, optional = true } slab = { workspace = true } @@ -71,8 +73,6 @@ sorted-index-buffer = { workspace = true } thiserror = { workspace = true } tinyvec = { workspace = true, features = ["alloc"] } tracing = { workspace = true } -derive_more = { version = "2.1.0", features = ["debug", "deref", "deref_mut", "display", "from"] } -enum-assoc = "1.3.0" # Feature flags & dependencies for wasm # wasm-bindgen is assumed for a wasm*-*-unknown target diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs index 18f01708c..c9dd8654e 100644 --- a/noq-proto/src/connection/mod.rs +++ b/noq-proto/src/connection/mod.rs @@ -27,6 +27,7 @@ use crate::{ connection::{ qlog::{QlogRecvPacket, QlogSink}, spaces::LostPacket, + stats::PathStatsMap, timer::{ConnTimer, PathTimer}, }, crypto::{self, Keys}, @@ -257,10 +258,14 @@ pub struct Connection { local_cid_state: FxHashMap, /// State of the unreliable datagram extension datagrams: DatagramState, - /// Connection level statistics - stats: ConnectionStats, - /// Path level statistics - path_stats: FxHashMap, + /// Path level statistics. + path_stats: PathStatsMap, + /// Accumulated stats of all discarded paths. + /// + /// The connection-level stats returned by [`Self::stats`] are the sum of the stats of + /// all the paths. However once a path is discarded it gets added to this field instead + /// so we do not have to keep an ever growing number of paths stats in memory. + partial_stats: ConnectionStats, /// QUIC version used for the connection. version: u32, @@ -421,8 +426,8 @@ impl Connection { config, remote_cids: FxHashMap::from_iter([(PathId::ZERO, CidQueue::new(remote_cid))]), rng, - stats: ConnectionStats::default(), path_stats: Default::default(), + partial_stats: ConnectionStats::default(), version, // peer params are not yet known, so multipath is not enabled @@ -1185,12 +1190,8 @@ impl Connection { self.path_data_mut(path_id) .inc_total_sent(transmit.len() as u64); - self.stats - .udp_tx - .on_sent(transmit.num_datagrams() as u64, transmit.len()); self.path_stats - .entry(path_id) - .or_default() + .for_path(path_id) .udp_tx .on_sent(transmit.num_datagrams() as u64, transmit.len()); @@ -1583,7 +1584,7 @@ impl Connection { &mut self.spaces[space_id], is_multipath_negotiated, &mut builder, - &mut self.stats.frame_tx, + &mut self.path_stats.for_path(path_id).frame_tx, self.crypto_state.has_keys(space_id.encryption_level()), ); } @@ -1599,7 +1600,7 @@ impl Connection { builder.frame_space_remaining() > frame::ConnectionClose::SIZE_BOUND, "ACKs should leave space for ConnectionClose" ); - let stats = &mut self.stats.frame_tx; + let stats = &mut self.path_stats.for_path(path_id).frame_tx; if frame::ConnectionClose::SIZE_BOUND < builder.frame_space_remaining() { let max_frame_size = builder.frame_space_remaining(); let close: Close = match self.state.as_type() { @@ -1764,19 +1765,19 @@ impl Connection { // We implement MTU probes as ping packets padded up to the probe size trace!(?probe_size, "writing MTUD probe"); - builder.write_frame(frame::Ping, &mut self.stats.frame_tx); + builder.write_frame(frame::Ping, &mut self.path_stats.for_path(path_id).frame_tx); // If supported by the peer, we want no delays to the probe's ACK if self.peer_supports_ack_frequency() { - builder.write_frame(frame::ImmediateAck, &mut self.stats.frame_tx); + builder.write_frame( + frame::ImmediateAck, + &mut self.path_stats.for_path(path_id).frame_tx, + ); } builder.finish_and_track(now, self, path_id, PadDatagram::ToSize(probe_size)); - self.path_stats - .entry(path_id) - .or_default() - .sent_plpmtud_probes += 1; + self.path_stats.for_path(path_id).sent_plpmtud_probes += 1; Some(self.build_transmit(path_id, transmit)) } @@ -1944,7 +1945,7 @@ impl Connection { let mut builder = PacketBuilder::new(now, SpaceId::Data, path_id, *prev_cid, buf, false, self)?; let challenge = frame::PathChallenge(token); - let stats = &mut self.stats.frame_tx; + let stats = &mut self.path_stats.for_path(path_id).frame_tx; builder.write_frame_with_log_msg(challenge, stats, Some("validating previous path")); // An endpoint MUST expand datagrams that contain a PATH_CHALLENGE frame @@ -1954,10 +1955,8 @@ impl Connection { builder.pad_to(MIN_INITIAL_SIZE); builder.finish(self, now); - self.stats.udp_tx.on_sent(1, buf.len()); self.path_stats - .entry(path_id) - .or_default() + .for_path(path_id) .udp_tx .on_sent(1, buf.len()); @@ -1992,7 +1991,7 @@ impl Connection { buf.start_new_datagram(); let mut builder = PacketBuilder::new(now, SpaceId::Data, path_id, cid, buf, false, self)?; - let stats = &mut self.stats.frame_tx; + let stats = &mut self.path_stats.for_path(path_id).frame_tx; builder.write_frame_with_log_msg(frame, stats, Some("(off-path)")); // Off-path: not tracked in congestion control. The packet is sent to a // different destination than path_id's network path. @@ -2001,12 +2000,7 @@ impl Connection { let size = buf.len(); - self.stats.udp_tx.on_sent(1, size); - self.path_stats - .entry(path_id) - .or_default() - .udp_tx - .on_sent(1, size); + self.path_stats.for_path(path_id).udp_tx.on_sent(1, size); Some(Transmit { destination: network_path.remote, size, @@ -2053,7 +2047,7 @@ impl Connection { let mut builder = PacketBuilder::new(now, SpaceId::Data, path_id, cid, &mut buf, false, self)?; - let stats = &mut self.stats.frame_tx; + let stats = &mut self.path_stats.for_path(path_id).frame_tx; builder.write_frame_with_log_msg(frame, stats, Some("(nat-traversal)")); // Off-path: not tracked in congestion control. The packet is sent to a // different destination than path_id's network path. @@ -2070,12 +2064,7 @@ impl Connection { let size = buf.len(); - self.stats.udp_tx.on_sent(1, size); - self.path_stats - .entry(path_id) - .or_default() - .udp_tx - .on_sent(1, size); + self.path_stats.for_path(path_id).udp_tx.on_sent(1, size); Some(Transmit { destination: remote, @@ -2163,9 +2152,7 @@ impl Connection { // anti-amplification blocked for it previously. .unwrap_or(false); - self.stats.udp_rx.datagrams += 1; - self.stats.udp_rx.bytes += first_decode.len() as u64; - let rx = &mut self.path_stats.entry(path_id).or_default().udp_rx; + let rx = &mut self.path_stats.for_path(path_id).udp_rx; rx.datagrams += 1; rx.bytes += first_decode.len() as u64; let data_len = first_decode.len(); @@ -2180,8 +2167,7 @@ impl Connection { } if let Some(data) = remaining { - self.stats.udp_rx.bytes += data.len() as u64; - self.path_stats.entry(path_id).or_default().udp_rx.bytes += data.len() as u64; + self.path_stats.for_path(path_id).udp_rx.bytes += data.len() as u64; self.handle_coalesced(now, network_path, path_id, ecn, data); } @@ -2483,13 +2469,23 @@ impl Connection { /// Returns connection statistics pub fn stats(&mut self) -> ConnectionStats { - self.stats.clone() + let mut stats = self.partial_stats.clone(); + + for path_stats in self.path_stats.iter_stats() { + // Self::path_stats() computes the path rtt, cwnd and current_mtu on access + // because they are not simple counters. When computing the connection stats we + // can skip that effort since those fields are not used in the `impl + // Add for ConnectionStats`. + stats += *path_stats; + } + + stats } /// Returns path statistics pub fn path_stats(&mut self, path_id: PathId) -> Option { let path = self.paths.get(&path_id)?; - let stats = self.path_stats.entry(path_id).or_default(); + let stats = self.path_stats.for_path(path_id); stats.rtt = path.data.rtt.get(); stats.cwnd = path.data.congestion.window(); stats.current_mtu = path.data.mtud.current_mtu(); @@ -2935,7 +2931,7 @@ impl Connection { } Ok(false) => {} Ok(true) => { - self.path_stats.entry(path).or_default().congestion_events += 1; + self.path_stats.for_path(path).congestion_events += 1; self.path_data_mut(path).congestion.on_congestion_event( now, largest_sent_time, @@ -3200,15 +3196,15 @@ impl Connection { } // Before removing the path, we fetch the final path stats via `Self::path_stats`. // This updates some values for the last time. - let path_stats = self.path_stats(path_id).unwrap_or_default(); - self.path_stats.remove(&path_id); + let path_stats = self.path_stats.discard(&path_id); + self.partial_stats += path_stats; self.paths.remove(&path_id); self.spaces[SpaceId::Data].number_spaces.remove(&path_id); self.events.push_back( PathEvent::Discarded { id: path_id, - path_stats, + path_stats: Box::new(path_stats), } .into(), ); @@ -3245,7 +3241,7 @@ impl Connection { .get(largest_lost) .unwrap() .time_sent; - let path_stats = self.path_stats.entry(path_id).or_default(); + let path_stats = self.path_stats.for_path(path_id); path_stats.lost_packets += lost_packets.len() as u64; path_stats.lost_bytes += size_of_lost_packets; trace!( @@ -3292,10 +3288,7 @@ impl Connection { self.datagrams.send_blocked = false; self.events.push_back(Event::DatagramsUnblocked); } - self.path_stats - .entry(path_id) - .or_default() - .black_holes_detected += 1; + self.path_stats.for_path(path_id).black_holes_detected += 1; } // Don't apply congestion penalty for lost ack-only packets @@ -3303,10 +3296,7 @@ impl Connection { old_bytes_in_flight != self.path_data_mut(path_id).in_flight.bytes; if lost_ack_eliciting { - self.path_stats - .entry(path_id) - .or_default() - .congestion_events += 1; + self.path_stats.for_path(path_id).congestion_events += 1; self.path_data_mut(path_id).congestion.on_congestion_event( now, largest_lost_sent, @@ -3329,10 +3319,7 @@ impl Connection { .unwrap() .remove_in_flight(&info); self.path_data_mut(path_id).mtud.on_probe_lost(); - self.path_stats - .entry(path_id) - .or_default() - .lost_plpmtud_probes += 1; + self.path_stats.for_path(path_id).lost_plpmtud_probes += 1; } } @@ -3955,8 +3942,7 @@ impl Connection { stateless_reset: bool, mut qlog: QlogRecvPacket, ) { - self.stats.udp_rx.ios += 1; - self.path_stats.entry(path_id).or_default().udp_rx.ios += 1; + self.path_stats.for_path(path_id).udp_rx.ios += 1; if let Some(ref packet) = packet { trace!( @@ -4224,7 +4210,10 @@ impl Connection { continue; }; - self.stats.frame_rx.record(frame.ty()); + self.path_stats + .for_path(path_id) + .frame_rx + .record(frame.ty()); if let Frame::Close(_error) = frame { self.state.move_to_draining(None); @@ -4527,7 +4516,10 @@ impl Connection { _ => Some(trace_span!("frame", ty = %frame.ty(), path = tracing::field::Empty)), }; - self.stats.frame_rx.record(frame.ty()); + self.path_stats + .for_path(path_id) + .frame_rx + .record(frame.ty()); let _guard = span.as_ref().map(|x| x.enter()); ack_eliciting |= frame.is_ack_eliciting(); @@ -4604,7 +4596,10 @@ impl Connection { _ => trace_span!("frame", ty = %frame.ty(), path = tracing::field::Empty), }; - self.stats.frame_rx.record(frame.ty()); + self.path_stats + .for_path(path_id) + .frame_rx + .record(frame.ty()); // Crypto, Stream and Datagram frames are special cased in order no pollute // the log with payload data match &frame { @@ -5617,7 +5612,7 @@ impl Connection { let is_multipath_negotiated = self.is_multipath_negotiated(); let space_has_keys = self.crypto_state.has_keys(space_id.encryption_level()); let is_0rtt = space_id == SpaceId::Data && !space_has_keys; - let stats = &mut self.stats.frame_tx; + let stats = &mut self.path_stats.for_path(path_id).frame_tx; let space = &mut self.spaces[space_id]; let path = &mut self.paths.get_mut(&path_id).expect("known path").data; space diff --git a/noq-proto/src/connection/paths.rs b/noq-proto/src/connection/paths.rs index a5c89066b..828c43399 100644 --- a/noq-proto/src/connection/paths.rs +++ b/noq-proto/src/connection/paths.rs @@ -1000,7 +1000,7 @@ pub enum PathEvent { /// The final path stats, they are no longer available via [`Connection::stats`] /// /// [`Connection::stats`]: super::Connection::stats - path_stats: PathStats, + path_stats: Box, }, /// The remote changed the status of the path /// diff --git a/noq-proto/src/connection/stats.rs b/noq-proto/src/connection/stats.rs index 8ced5f38d..697df60ff 100644 --- a/noq-proto/src/connection/stats.rs +++ b/noq-proto/src/connection/stats.rs @@ -1,19 +1,24 @@ //! Connection statistics +use rustc_hash::FxHashMap; + use crate::Duration; use crate::FrameType; -/// Statistics about UDP datagrams transmitted or received on a connection +use super::PathId; + +/// Statistics about UDP datagrams transmitted or received on a connection. /// -/// All QUIC packets are carried by UDP datagrams. Hence, these statistics cover all traffic on a connection. -#[derive(Default, Debug, Copy, Clone, PartialEq, Eq)] +/// All QUIC packets are carried by UDP datagrams. Hence, these statistics cover all traffic +/// on a connection. +#[derive(Default, Debug, Copy, Clone, PartialEq, Eq, derive_more::Add, derive_more::AddAssign)] #[non_exhaustive] pub struct UdpStats { - /// The amount of UDP datagrams observed + /// The number of UDP datagrams observed. pub datagrams: u64, - /// The total amount of bytes which have been transferred inside UDP datagrams + /// The total amount of bytes which have been transferred inside UDP datagrams. pub bytes: u64, - /// The amount of I/O operations executed + /// The number of I/O operations executed. /// /// Can be less than `datagrams` when GSO, GRO, and/or batched system calls are in use. pub ios: u64, @@ -27,8 +32,8 @@ impl UdpStats { } } -/// Number of frames transmitted or received of each frame type -#[derive(Default, Copy, Clone)] +/// Number of frames transmitted or received of each frame type. +#[derive(Default, Copy, Clone, PartialEq, Eq, derive_more::Add, derive_more::AddAssign)] #[non_exhaustive] #[allow(missing_docs)] pub struct FrameStats { @@ -123,83 +128,220 @@ impl FrameStats { impl std::fmt::Debug for FrameStats { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let Self { + acks, + path_acks, + ack_frequency, + crypto, + connection_close, + data_blocked, + datagram, + handshake_done, + immediate_ack, + max_data, + max_stream_data, + max_streams_bidi, + max_streams_uni, + new_connection_id, + path_new_connection_id, + new_token, + path_challenge, + path_response, + ping, + reset_stream, + retire_connection_id, + path_retire_connection_id, + stream_data_blocked, + streams_blocked_bidi, + streams_blocked_uni, + stop_sending, + stream, + observed_addr, + path_abandon, + path_status_available, + path_status_backup, + max_path_id, + paths_blocked, + path_cids_blocked, + add_address, + reach_out, + remove_address, + } = self; f.debug_struct("FrameStats") - .field("ACK", &self.acks) - .field("ACK_FREQUENCY", &self.ack_frequency) - .field("CONNECTION_CLOSE", &self.connection_close) - .field("CRYPTO", &self.crypto) - .field("DATA_BLOCKED", &self.data_blocked) - .field("DATAGRAM", &self.datagram) - .field("HANDSHAKE_DONE", &self.handshake_done) - .field("IMMEDIATE_ACK", &self.immediate_ack) - .field("MAX_DATA", &self.max_data) - .field("MAX_PATH_ID", &self.max_path_id) - .field("MAX_STREAM_DATA", &self.max_stream_data) - .field("MAX_STREAMS_BIDI", &self.max_streams_bidi) - .field("MAX_STREAMS_UNI", &self.max_streams_uni) - .field("NEW_CONNECTION_ID", &self.new_connection_id) - .field("NEW_TOKEN", &self.new_token) - .field("PATHS_BLOCKED", &self.paths_blocked) - .field("PATH_ABANDON", &self.path_abandon) - .field("PATH_ACK", &self.path_acks) - .field("PATH_STATUS_AVAILABLE", &self.path_status_available) - .field("PATH_STATUS_BACKUP", &self.path_status_backup) - .field("PATH_CHALLENGE", &self.path_challenge) - .field("PATH_CIDS_BLOCKED", &self.path_cids_blocked) - .field("PATH_NEW_CONNECTION_ID", &self.path_new_connection_id) - .field("PATH_RESPONSE", &self.path_response) - .field("PATH_RETIRE_CONNECTION_ID", &self.path_retire_connection_id) - .field("PING", &self.ping) - .field("RESET_STREAM", &self.reset_stream) - .field("RETIRE_CONNECTION_ID", &self.retire_connection_id) - .field("STREAM_DATA_BLOCKED", &self.stream_data_blocked) - .field("STREAMS_BLOCKED_BIDI", &self.streams_blocked_bidi) - .field("STREAMS_BLOCKED_UNI", &self.streams_blocked_uni) - .field("STOP_SENDING", &self.stop_sending) - .field("STREAM", &self.stream) + .field("ACK", acks) + .field("ACK_FREQUENCY", ack_frequency) + .field("CONNECTION_CLOSE", connection_close) + .field("CRYPTO", crypto) + .field("DATA_BLOCKED", data_blocked) + .field("DATAGRAM", datagram) + .field("HANDSHAKE_DONE", handshake_done) + .field("IMMEDIATE_ACK", immediate_ack) + .field("MAX_DATA", max_data) + .field("MAX_PATH_ID", max_path_id) + .field("MAX_STREAM_DATA", max_stream_data) + .field("MAX_STREAMS_BIDI", max_streams_bidi) + .field("MAX_STREAMS_UNI", max_streams_uni) + .field("NEW_CONNECTION_ID", new_connection_id) + .field("NEW_TOKEN", new_token) + .field("PATHS_BLOCKED", paths_blocked) + .field("PATH_ABANDON", path_abandon) + .field("PATH_ACK", path_acks) + .field("PATH_STATUS_AVAILABLE", path_status_available) + .field("PATH_STATUS_BACKUP", path_status_backup) + .field("PATH_CHALLENGE", path_challenge) + .field("PATH_CIDS_BLOCKED", path_cids_blocked) + .field("PATH_NEW_CONNECTION_ID", path_new_connection_id) + .field("PATH_RESPONSE", path_response) + .field("PATH_RETIRE_CONNECTION_ID", path_retire_connection_id) + .field("PING", ping) + .field("RESET_STREAM", reset_stream) + .field("RETIRE_CONNECTION_ID", retire_connection_id) + .field("STREAM_DATA_BLOCKED", stream_data_blocked) + .field("STREAMS_BLOCKED_BIDI", streams_blocked_bidi) + .field("STREAMS_BLOCKED_UNI", streams_blocked_uni) + .field("STOP_SENDING", stop_sending) + .field("STREAM", stream) + .field("OBSERVED_ADDRESS", observed_addr) + .field("ADD_ADDRESS", add_address) + .field("REACH_OUT", reach_out) + .field("REMOVE_ADDRESS", remove_address) .finish() } } -/// Statistics related to a transmission path +/// Statistics related to a transmission path. #[derive(Debug, Default, Copy, Clone, PartialEq, Eq)] #[non_exhaustive] pub struct PathStats { - /// Current best estimate of this connection's latency (round-trip-time) + /// Current best estimate of this connection's latency (round-trip-time). pub rtt: Duration, - /// Statistics about datagrams and bytes sent on this path + /// Statistics about datagrams and bytes sent on this path. pub udp_tx: UdpStats, - /// Statistics about datagrams and bytes received on this path + /// Statistics about datagrams and bytes received on this path. pub udp_rx: UdpStats, - /// Current congestion window of the connection + /// Statistics about frames transmitted on this path. + pub frame_tx: FrameStats, + /// Statistics about frames received on this path. + pub frame_rx: FrameStats, + /// Current congestion window of the connection. pub cwnd: u64, - /// Congestion events on the connection + /// Congestion events on the connection. pub congestion_events: u64, - /// The amount of packets lost on this path + /// The amount of packets lost on this path. pub lost_packets: u64, - /// The amount of bytes lost on this path + /// The amount of bytes lost on this path. pub lost_bytes: u64, - /// The amount of PLPMTUD probe packets sent on this path (also counted by `udp_tx.datagrams`) + /// The number of PLPMTUD probe packets sent on this path. + /// + /// These are also counted by [`UdpStats::datagrams`]. pub sent_plpmtud_probes: u64, - /// The amount of PLPMTUD probe packets lost on this path (ignored by `lost_packets` and - /// `lost_bytes`) + /// The number of PLPMTUD probe packets lost on this path. + /// + /// These are not included in [`Self::lost_packets`] and [`Self::lost_bytes`]. pub lost_plpmtud_probes: u64, - /// The number of times a black hole was detected in the path + /// The number of times a black hole was detected in the path. pub black_holes_detected: u64, - /// Largest UDP payload size the path currently supports + /// Largest UDP payload size the path currently supports. pub current_mtu: u16, } -/// Connection statistics +/// Connection statistics. +/// +/// The fields here are a sum of the respective fields in the [`PathStats`] for all the +/// paths that exist as well as all paths that previously existed. #[derive(Debug, Default, Clone)] #[non_exhaustive] pub struct ConnectionStats { - /// Statistics about UDP datagrams transmitted on a connection + /// Statistics about UDP datagrams transmitted on a connection. pub udp_tx: UdpStats, - /// Statistics about UDP datagrams received on a connection + /// Statistics about UDP datagrams received on a connection. pub udp_rx: UdpStats, - /// Statistics about frames transmitted on a connection + /// Statistics about frames transmitted on a connection. pub frame_tx: FrameStats, - /// Statistics about frames received on a connection + /// Statistics about frames received on a connection. pub frame_rx: FrameStats, } + +impl std::ops::Add for ConnectionStats { + type Output = Self; + + fn add(self, rhs: PathStats) -> Self::Output { + // Be aware that Connection::stats() relies on the fact this function ignores the + // rtt, cwnd and current_mtu fields. + let PathStats { + rtt: _, + udp_tx, + udp_rx, + frame_tx, + frame_rx, + cwnd: _, + congestion_events: _, + lost_packets: _, + lost_bytes: _, + sent_plpmtud_probes: _, + lost_plpmtud_probes: _, + black_holes_detected: _, + current_mtu: _, + } = rhs; + Self { + udp_tx: self.udp_tx + udp_tx, + udp_rx: self.udp_rx + udp_rx, + frame_tx: self.frame_tx + frame_tx, + frame_rx: self.frame_rx + frame_rx, + } + } +} + +impl std::ops::AddAssign for ConnectionStats { + fn add_assign(&mut self, rhs: PathStats) { + // Be aware that Connection::stats() relies on the fact this function ignores the + // rtt, cwnd and current_mtu fields. + let PathStats { + rtt: _, + udp_tx, + udp_rx, + frame_tx, + frame_rx, + cwnd: _, + congestion_events: _, + lost_packets: _, + lost_bytes: _, + sent_plpmtud_probes: _, + lost_plpmtud_probes: _, + black_holes_detected: _, + current_mtu: _, + } = rhs; + self.udp_tx += udp_tx; + self.udp_rx += udp_rx; + self.frame_tx += frame_tx; + self.frame_rx += frame_rx; + } +} + +/// Helper to make [`PathStats`] infallibly available. +/// +/// This helper also helps with borrowing issues compared to having the [`Self::for_path`] +/// function as a helper directly on [`Connection`]. +/// +/// [`Connection`]: super::Connection +#[derive(Debug, Default)] +pub(super) struct PathStatsMap(FxHashMap); + +impl PathStatsMap { + /// Returns the [`PathStats`] for the path. + pub(super) fn for_path(&mut self, path_id: PathId) -> &mut PathStats { + self.0.entry(path_id).or_default() + } + + /// An iterator over all contained [`PathStats`]. + pub(super) fn iter_stats(&self) -> impl Iterator { + self.0.values() + } + + /// Removes the stats for a given path. + /// + /// Only do this once you are discarding the path. + pub(super) fn discard(&mut self, path_id: &PathId) -> PathStats { + self.0.remove(path_id).unwrap_or_default() + } +} diff --git a/noq/src/connection.rs b/noq/src/connection.rs index 7de2bb090..ad500f9ca 100644 --- a/noq/src/connection.rs +++ b/noq/src/connection.rs @@ -1580,11 +1580,11 @@ impl State { sender.send_modify(|value| *value = Ok(())); } } - Path(evt @ PathEvent::Discarded { id, path_stats }) => { + Path(ref evt @ PathEvent::Discarded { id, ref path_stats }) => { if self.path_refs.contains_key(&id) { - self.final_path_stats.insert(id, path_stats); + self.final_path_stats.insert(id, *path_stats.clone()); } - self.path_events.send(evt).ok(); + self.path_events.send(evt.clone()).ok(); } Path(ref evt @ PathEvent::Abandoned { id, .. }) => { if let Some(sender) = self.open_path.remove(&id) {