quinn_proto/connection/
paths.rs

1use std::{cmp, net::SocketAddr};
2
3use tracing::trace;
4
5use super::{
6    mtud::MtuDiscovery,
7    pacing::Pacer,
8    spaces::{PacketSpace, SentPacket},
9};
10use crate::{Duration, Instant, TIMER_GRANULARITY, TransportConfig, congestion, packet::SpaceId};
11
12#[cfg(feature = "qlog")]
13use qlog::events::quic::MetricsUpdated;
14
15/// Description of a particular network path
16pub(super) struct PathData {
17    pub(super) remote: SocketAddr,
18    pub(super) rtt: RttEstimator,
19    /// Whether we're enabling ECN on outgoing packets
20    pub(super) sending_ecn: bool,
21    /// Congestion controller state
22    pub(super) congestion: Box<dyn congestion::Controller>,
23    /// Pacing state
24    pub(super) pacing: Pacer,
25    pub(super) challenge: Option<u64>,
26    pub(super) challenge_pending: bool,
27    /// Whether we're certain the peer can both send and receive on this address
28    ///
29    /// Initially equal to `use_stateless_retry` for servers, and becomes false again on every
30    /// migration. Always true for clients.
31    pub(super) validated: bool,
32    /// Total size of all UDP datagrams sent on this path
33    pub(super) total_sent: u64,
34    /// Total size of all UDP datagrams received on this path
35    pub(super) total_recvd: u64,
36    /// The state of the MTU discovery process
37    pub(super) mtud: MtuDiscovery,
38    /// Packet number of the first packet sent after an RTT sample was collected on this path
39    ///
40    /// Used in persistent congestion determination.
41    pub(super) first_packet_after_rtt_sample: Option<(SpaceId, u64)>,
42    pub(super) in_flight: InFlight,
43    /// Number of the first packet sent on this path
44    ///
45    /// Used to determine whether a packet was sent on an earlier path. Insufficient to determine if
46    /// a packet was sent on a later path.
47    first_packet: Option<u64>,
48
49    /// Snapshot of the qlog recovery metrics
50    #[cfg(feature = "qlog")]
51    recovery_metrics: RecoveryMetrics,
52
53    /// Tag uniquely identifying a path in a connection
54    generation: u64,
55}
56
57impl PathData {
58    pub(super) fn new(
59        remote: SocketAddr,
60        allow_mtud: bool,
61        peer_max_udp_payload_size: Option<u16>,
62        generation: u64,
63        now: Instant,
64        config: &TransportConfig,
65    ) -> Self {
66        let congestion = config
67            .congestion_controller_factory
68            .clone()
69            .build(now, config.get_initial_mtu());
70        Self {
71            remote,
72            rtt: RttEstimator::new(config.initial_rtt),
73            sending_ecn: true,
74            pacing: Pacer::new(
75                config.initial_rtt,
76                congestion.initial_window(),
77                config.get_initial_mtu(),
78                now,
79            ),
80            congestion,
81            challenge: None,
82            challenge_pending: false,
83            validated: false,
84            total_sent: 0,
85            total_recvd: 0,
86            mtud: config
87                .mtu_discovery_config
88                .as_ref()
89                .filter(|_| allow_mtud)
90                .map_or(
91                    MtuDiscovery::disabled(config.get_initial_mtu(), config.min_mtu),
92                    |mtud_config| {
93                        MtuDiscovery::new(
94                            config.get_initial_mtu(),
95                            config.min_mtu,
96                            peer_max_udp_payload_size,
97                            mtud_config.clone(),
98                        )
99                    },
100                ),
101            first_packet_after_rtt_sample: None,
102            in_flight: InFlight::new(),
103            first_packet: None,
104            #[cfg(feature = "qlog")]
105            recovery_metrics: RecoveryMetrics::default(),
106            generation,
107        }
108    }
109
110    pub(super) fn from_previous(
111        remote: SocketAddr,
112        prev: &Self,
113        generation: u64,
114        now: Instant,
115    ) -> Self {
116        let congestion = prev.congestion.clone_box();
117        let smoothed_rtt = prev.rtt.get();
118        Self {
119            remote,
120            rtt: prev.rtt,
121            pacing: Pacer::new(smoothed_rtt, congestion.window(), prev.current_mtu(), now),
122            sending_ecn: true,
123            congestion,
124            challenge: None,
125            challenge_pending: false,
126            validated: false,
127            total_sent: 0,
128            total_recvd: 0,
129            mtud: prev.mtud.clone(),
130            first_packet_after_rtt_sample: prev.first_packet_after_rtt_sample,
131            in_flight: InFlight::new(),
132            first_packet: None,
133            #[cfg(feature = "qlog")]
134            recovery_metrics: prev.recovery_metrics.clone(),
135            generation,
136        }
137    }
138
139    /// Resets RTT, congestion control, pacing and MTU states.
140    ///
141    /// This is useful when it is known the underlying path has changed.
142    pub(super) fn reset(&mut self, now: Instant, config: &TransportConfig) {
143        self.rtt = RttEstimator::new(config.initial_rtt);
144        self.congestion = config
145            .congestion_controller_factory
146            .clone()
147            .build(now, config.get_initial_mtu());
148        self.mtud.reset(config.get_initial_mtu(), config.min_mtu);
149        self.pacing = Pacer::new(
150            self.rtt.get(),
151            self.congestion.initial_window(),
152            self.current_mtu(),
153            now,
154        );
155    }
156
157    /// Indicates whether we're a server that hasn't validated the peer's address and hasn't
158    /// received enough data from the peer to permit sending `bytes_to_send` additional bytes
159    pub(super) fn anti_amplification_blocked(&self, bytes_to_send: u64) -> bool {
160        !self.validated && self.total_recvd * 3 < self.total_sent + bytes_to_send
161    }
162
163    /// Returns the path's current MTU
164    pub(super) fn current_mtu(&self) -> u16 {
165        self.mtud.current_mtu()
166    }
167
168    /// Account for transmission of `packet` with number `pn` in `space`
169    pub(super) fn sent(&mut self, pn: u64, packet: SentPacket, space: &mut PacketSpace) {
170        self.in_flight.insert(&packet);
171        if self.first_packet.is_none() {
172            self.first_packet = Some(pn);
173        }
174        if let Some(forgotten) = space.sent(pn, packet) {
175            self.remove_in_flight(&forgotten);
176        }
177    }
178
179    /// Remove `packet` with number `pn` from this path's congestion control counters, or return
180    /// `false` if `pn` was sent before this path was established.
181    pub(super) fn remove_in_flight(&mut self, packet: &SentPacket) -> bool {
182        if packet.path_generation != self.generation {
183            return false;
184        }
185        self.in_flight.remove(packet);
186        true
187    }
188
189    #[cfg(feature = "qlog")]
190    pub(super) fn qlog_recovery_metrics(&mut self, pto_count: u32) -> Option<MetricsUpdated> {
191        let controller_metrics = self.congestion.metrics();
192
193        let metrics = RecoveryMetrics {
194            min_rtt: Some(self.rtt.min),
195            smoothed_rtt: Some(self.rtt.get()),
196            latest_rtt: Some(self.rtt.latest),
197            rtt_variance: Some(self.rtt.var),
198            pto_count: Some(pto_count),
199            bytes_in_flight: Some(self.in_flight.bytes),
200            packets_in_flight: Some(self.in_flight.ack_eliciting),
201
202            congestion_window: Some(controller_metrics.congestion_window),
203            ssthresh: controller_metrics.ssthresh,
204            pacing_rate: controller_metrics.pacing_rate,
205        };
206
207        let event = metrics.to_qlog_event(&self.recovery_metrics);
208        self.recovery_metrics = metrics;
209        event
210    }
211
212    pub(super) fn generation(&self) -> u64 {
213        self.generation
214    }
215}
216
217/// Congestion metrics as described in [`recovery_metrics_updated`].
218///
219/// [`recovery_metrics_updated`]: https://datatracker.ietf.org/doc/html/draft-ietf-quic-qlog-quic-events.html#name-recovery_metrics_updated
220#[cfg(feature = "qlog")]
221#[derive(Default, Clone, PartialEq)]
222#[non_exhaustive]
223struct RecoveryMetrics {
224    pub min_rtt: Option<Duration>,
225    pub smoothed_rtt: Option<Duration>,
226    pub latest_rtt: Option<Duration>,
227    pub rtt_variance: Option<Duration>,
228    pub pto_count: Option<u32>,
229    pub bytes_in_flight: Option<u64>,
230    pub packets_in_flight: Option<u64>,
231    pub congestion_window: Option<u64>,
232    pub ssthresh: Option<u64>,
233    pub pacing_rate: Option<u64>,
234}
235
236#[cfg(feature = "qlog")]
237impl RecoveryMetrics {
238    /// Retain only values that have been updated since the last snapshot.
239    fn retain_updated(&self, previous: &Self) -> Self {
240        macro_rules! keep_if_changed {
241            ($name:ident) => {
242                if previous.$name == self.$name {
243                    None
244                } else {
245                    self.$name
246                }
247            };
248        }
249
250        Self {
251            min_rtt: keep_if_changed!(min_rtt),
252            smoothed_rtt: keep_if_changed!(smoothed_rtt),
253            latest_rtt: keep_if_changed!(latest_rtt),
254            rtt_variance: keep_if_changed!(rtt_variance),
255            pto_count: keep_if_changed!(pto_count),
256            bytes_in_flight: keep_if_changed!(bytes_in_flight),
257            packets_in_flight: keep_if_changed!(packets_in_flight),
258            congestion_window: keep_if_changed!(congestion_window),
259            ssthresh: keep_if_changed!(ssthresh),
260            pacing_rate: keep_if_changed!(pacing_rate),
261        }
262    }
263
264    /// Emit a `MetricsUpdated` event containing only updated values
265    fn to_qlog_event(&self, previous: &Self) -> Option<MetricsUpdated> {
266        let updated = self.retain_updated(previous);
267
268        if updated == Self::default() {
269            return None;
270        }
271
272        Some(MetricsUpdated {
273            min_rtt: updated.min_rtt.map(|rtt| rtt.as_secs_f32()),
274            smoothed_rtt: updated.smoothed_rtt.map(|rtt| rtt.as_secs_f32()),
275            latest_rtt: updated.latest_rtt.map(|rtt| rtt.as_secs_f32()),
276            rtt_variance: updated.rtt_variance.map(|rtt| rtt.as_secs_f32()),
277            pto_count: updated
278                .pto_count
279                .map(|count| count.try_into().unwrap_or(u16::MAX)),
280            bytes_in_flight: updated.bytes_in_flight,
281            packets_in_flight: updated.packets_in_flight,
282            congestion_window: updated.congestion_window,
283            ssthresh: updated.ssthresh,
284            pacing_rate: updated.pacing_rate,
285        })
286    }
287}
288
289/// RTT estimation for a particular network path
290#[derive(Copy, Clone)]
291pub struct RttEstimator {
292    /// The most recent RTT measurement made when receiving an ack for a previously unacked packet
293    latest: Duration,
294    /// The smoothed RTT of the connection, computed as described in RFC6298
295    smoothed: Option<Duration>,
296    /// The RTT variance, computed as described in RFC6298
297    var: Duration,
298    /// The minimum RTT seen in the connection, ignoring ack delay.
299    min: Duration,
300}
301
302impl RttEstimator {
303    pub(crate) fn new(initial_rtt: Duration) -> Self {
304        Self {
305            latest: initial_rtt,
306            smoothed: None,
307            var: initial_rtt / 2,
308            min: initial_rtt,
309        }
310    }
311
312    /// The current best RTT estimation.
313    pub fn get(&self) -> Duration {
314        self.smoothed.unwrap_or(self.latest)
315    }
316
317    /// Conservative estimate of RTT
318    ///
319    /// Takes the maximum of smoothed and latest RTT, as recommended
320    /// in 6.1.2 of the recovery spec (draft 29).
321    pub fn conservative(&self) -> Duration {
322        self.get().max(self.latest)
323    }
324
325    /// Minimum RTT registered so far for this estimator.
326    pub fn min(&self) -> Duration {
327        self.min
328    }
329
330    // PTO computed as described in RFC9002#6.2.1
331    pub(crate) fn pto_base(&self) -> Duration {
332        self.get() + cmp::max(4 * self.var, TIMER_GRANULARITY)
333    }
334
335    pub(crate) fn update(&mut self, ack_delay: Duration, rtt: Duration) {
336        self.latest = rtt;
337        // min_rtt ignores ack delay.
338        self.min = cmp::min(self.min, self.latest);
339        // Based on RFC6298.
340        if let Some(smoothed) = self.smoothed {
341            let adjusted_rtt = if self.min + ack_delay <= self.latest {
342                self.latest - ack_delay
343            } else {
344                self.latest
345            };
346            let var_sample = smoothed.abs_diff(adjusted_rtt);
347            self.var = (3 * self.var + var_sample) / 4;
348            self.smoothed = Some((7 * smoothed + adjusted_rtt) / 8);
349        } else {
350            self.smoothed = Some(self.latest);
351            self.var = self.latest / 2;
352            self.min = self.latest;
353        }
354    }
355}
356
357#[derive(Default)]
358pub(crate) struct PathResponses {
359    pending: Vec<PathResponse>,
360}
361
362impl PathResponses {
363    pub(crate) fn push(&mut self, packet: u64, token: u64, remote: SocketAddr) {
364        /// Arbitrary permissive limit to prevent abuse
365        const MAX_PATH_RESPONSES: usize = 16;
366        let response = PathResponse {
367            packet,
368            token,
369            remote,
370        };
371        let existing = self.pending.iter_mut().find(|x| x.remote == remote);
372        if let Some(existing) = existing {
373            // Update a queued response
374            if existing.packet <= packet {
375                *existing = response;
376            }
377            return;
378        }
379        if self.pending.len() < MAX_PATH_RESPONSES {
380            self.pending.push(response);
381        } else {
382            // We don't expect to ever hit this with well-behaved peers, so we don't bother dropping
383            // older challenges.
384            trace!("ignoring excessive PATH_CHALLENGE");
385        }
386    }
387
388    pub(crate) fn pop_off_path(&mut self, remote: SocketAddr) -> Option<(u64, SocketAddr)> {
389        let response = *self.pending.last()?;
390        if response.remote == remote {
391            // We don't bother searching further because we expect that the on-path response will
392            // get drained in the immediate future by a call to `pop_on_path`
393            return None;
394        }
395        self.pending.pop();
396        Some((response.token, response.remote))
397    }
398
399    pub(crate) fn pop_on_path(&mut self, remote: SocketAddr) -> Option<u64> {
400        let response = *self.pending.last()?;
401        if response.remote != remote {
402            // We don't bother searching further because we expect that the off-path response will
403            // get drained in the immediate future by a call to `pop_off_path`
404            return None;
405        }
406        self.pending.pop();
407        Some(response.token)
408    }
409
410    pub(crate) fn is_empty(&self) -> bool {
411        self.pending.is_empty()
412    }
413}
414
415#[derive(Copy, Clone)]
416struct PathResponse {
417    /// The packet number the corresponding PATH_CHALLENGE was received in
418    packet: u64,
419    token: u64,
420    /// The address the corresponding PATH_CHALLENGE was received from
421    remote: SocketAddr,
422}
423
424/// Summary statistics of packets that have been sent on a particular path, but which have not yet
425/// been acked or deemed lost
426pub(super) struct InFlight {
427    /// Sum of the sizes of all sent packets considered "in flight" by congestion control
428    ///
429    /// The size does not include IP or UDP overhead. Packets only containing ACK frames do not
430    /// count towards this to ensure congestion control does not impede congestion feedback.
431    pub(super) bytes: u64,
432    /// Number of packets in flight containing frames other than ACK and PADDING
433    ///
434    /// This can be 0 even when bytes is not 0 because PADDING frames cause a packet to be
435    /// considered "in flight" by congestion control. However, if this is nonzero, bytes will always
436    /// also be nonzero.
437    pub(super) ack_eliciting: u64,
438}
439
440impl InFlight {
441    fn new() -> Self {
442        Self {
443            bytes: 0,
444            ack_eliciting: 0,
445        }
446    }
447
448    fn insert(&mut self, packet: &SentPacket) {
449        self.bytes += u64::from(packet.size);
450        self.ack_eliciting += u64::from(packet.ack_eliciting);
451    }
452
453    /// Update counters to account for a packet becoming acknowledged, lost, or abandoned
454    fn remove(&mut self, packet: &SentPacket) {
455        self.bytes -= u64::from(packet.size);
456        self.ack_eliciting -= u64::from(packet.ack_eliciting);
457    }
458}
459
460#[cfg(test)]
461mod tests {
462    use super::*;
463
464    #[test]
465    fn reset_refreshes_pacer_budget() {
466        let now = Instant::now();
467        let config = TransportConfig::default();
468        let remote = "203.0.113.1:4433".parse().unwrap();
469        let mut path = PathData::new(remote, true, None, 0, now, &config);
470        let mtu = path.current_mtu();
471        let window = path.congestion.window();
472
473        for _ in 0..1000 {
474            if path
475                .pacing
476                .delay(path.rtt.get(), mtu.into(), mtu, window, now)
477                .is_some()
478            {
479                break;
480            }
481            path.pacing.on_transmit(mtu);
482        }
483        assert!(
484            path.pacing
485                .delay(path.rtt.get(), mtu.into(), mtu, window, now)
486                .is_some()
487        );
488
489        path.reset(now, &config);
490
491        assert_eq!(
492            path.pacing.delay(
493                path.rtt.get(),
494                path.current_mtu().into(),
495                path.current_mtu(),
496                path.congestion.window(),
497                now
498            ),
499            None
500        );
501    }
502}
quinn_proto/connection/paths.rs

quinn_proto/connection/
paths.rs
