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) {