tor_proto/congestion/
rtt.rs

1//! Round Trip Time measurement (§ 2.1)
2
3use std::cmp::{max, min};
4use std::collections::VecDeque;
5use std::sync::atomic::{AtomicBool, Ordering};
6use web_time_compat::{Duration, Instant};
7
8use super::params::RoundTripEstimatorParams;
9use super::{CongestionWindow, State};
10
11use thiserror::Error;
12use tor_error::{ErrorKind, HasKind};
13
14/// An error originating from the tor-congestion crate.
15#[derive(Error, Debug, Clone)]
16#[non_exhaustive]
17pub(crate) enum Error {
18    /// A call to `RoundtripTimeEstimator::sendme_received` was made without calling
19    /// `RoundtripTimeEstimator::expect_sendme` first.
20    #[error("Informed of a SENDME we weren't expecting")]
21    MismatchedEstimationCall,
22}
23
24impl HasKind for Error {
25    fn kind(&self) -> ErrorKind {
26        use Error as E;
27        match self {
28            E::MismatchedEstimationCall => ErrorKind::TorProtocolViolation,
29        }
30    }
31}
32
33/// Provides an estimate of the round-trip time (RTT) of a Tor circuit.
34#[derive(Debug)]
35#[allow(dead_code)]
36pub(crate) struct RoundtripTimeEstimator {
37    /// A queue of times we sent a cell that we'd expect a SENDME for.
38    ///
39    /// When a data cell is sent and for which we expect a SENDME next, the timestamp at the send
40    /// is kept in this queue so we can use it to measure the RTT when the SENDME is received.
41    ///
42    /// A queue is used here because the protocol allows to send all pending SENDMEs at once as
43    /// long as it is within one congestion window.
44    sendme_expected_from: VecDeque<Instant>,
45    /// The last *measured* round-trip time.
46    ///
47    /// This is `None` iff we have not managed to get any estimate yet.
48    last_rtt: Option<Duration>,
49    /// The current smoothed *estimate* of what the round-trip time is.
50    ///
51    /// This is `None` iff we have not managed to get any estimate yet.
52    ewma_rtt: Option<Duration>,
53    /// The minimum observed value of `last_rtt`.
54    ///
55    /// This is `None` iff we have not managed to get any estimate yet.
56    min_rtt: Option<Duration>,
57    /// The maximum observed value of `last_rtt`.
58    ///
59    /// This is `None` iff we have not managed to get any estimate yet.
60    max_rtt: Option<Duration>,
61    /// The network parameters we're using.
62    params: RoundTripEstimatorParams,
63    /// A reference to a shared boolean for storing if the clock is stalled or not.
64    /// Spec: CLOCK_HEURISTICS from prop324. See is_clock_stalled() for the implementation.
65    clock_stalled: AtomicBool,
66}
67
68#[allow(dead_code)]
69impl RoundtripTimeEstimator {
70    /// Create a new `RoundtripTimeEstimator`, using a set of `NetParameters` and a shared boolean
71    /// to cache clock stalled state in.
72    pub(crate) fn new(params: &RoundTripEstimatorParams) -> Self {
73        Self {
74            sendme_expected_from: Default::default(),
75            last_rtt: None,
76            ewma_rtt: None,
77            min_rtt: None,
78            max_rtt: None,
79            params: params.clone(),
80            clock_stalled: AtomicBool::default(),
81        }
82    }
83
84    /// Return true iff the estimator is ready to be used or read.
85    pub(crate) fn is_ready(&self) -> bool {
86        !self.clock_stalled() && self.last_rtt.is_some()
87    }
88
89    /// Return the state of the clock stalled indicator.
90    pub(crate) fn clock_stalled(&self) -> bool {
91        self.clock_stalled.load(Ordering::SeqCst)
92    }
93
94    /// Return the EWMA RTT in usec or `None` if we don't have an estimate yet.
95    pub(crate) fn ewma_rtt_usec(&self) -> Option<u32> {
96        self.ewma_rtt
97            .map(|rtt| u32::try_from(rtt.as_micros()).ok().unwrap_or(u32::MAX))
98    }
99
100    /// Return the Minimum RTT in usec or `None` if we don't have an estimate yet.
101    pub(crate) fn min_rtt_usec(&self) -> Option<u32> {
102        self.min_rtt
103            .map(|rtt| u32::try_from(rtt.as_micros()).ok().unwrap_or(u32::MAX))
104    }
105
106    /// Return the maximum observed RTT in usec or `None` if we don't have an estimate yet.
107    pub(crate) fn max_rtt_usec(&self) -> Option<u32> {
108        self.max_rtt
109            .map(|rtt| u32::try_from(rtt.as_micros()).ok().unwrap_or(u32::MAX))
110    }
111
112    /// Inform the estimator that we did (at time `now`) something that we'll expect a SENDME to
113    /// be received for.
114    pub(crate) fn expect_sendme(&mut self, now: Instant) {
115        self.sendme_expected_from.push_back(now);
116    }
117
118    /// Return whether we can use heuristics to sanity-check RTT values against our EWMA value.
119    /// Spec: 2.1.1. Clock Jump Heuristics CLOCK_HEURISTICS
120    ///
121    /// Used in [`is_clock_stalled`](RoundtripTimeEstimator::is_clock_stalled), to check the sanity of
122    /// a newly measured RTT value.
123    fn can_crosscheck_with_current_estimate(&self, in_slow_start: bool) -> bool {
124        // If we're in slow start, we don't perform any sanity checks, as per spec. If we don't
125        // have a current estimate, we can't use it for sanity checking, because it doesn't
126        // exist.
127        !in_slow_start && self.ewma_rtt.is_some()
128    }
129
130    /// Given a raw RTT value we just observed, compute whether or not we think the clock has
131    /// stalled or jumped, and we should throw it out as a result.
132    fn is_clock_stalled(&self, raw_rtt: Duration, in_slow_start: bool) -> bool {
133        if raw_rtt.is_zero() {
134            // Clock is stalled.
135            self.clock_stalled.store(true, Ordering::SeqCst);
136            true
137        } else if self.can_crosscheck_with_current_estimate(in_slow_start) {
138            let ewma_rtt = self
139                .ewma_rtt
140                .expect("ewma_rtt was not checked by can_crosscheck_with_current_estimate?!");
141
142            /// Discrepancy ratio of a new RTT value that we allow against the current RTT in order
143            /// to declare if the clock has stalled or not. This value is taken from proposal 324
144            /// section 2.1.1 CLOCK_HEURISTICS and has the same name as in C-tor.
145            const DELTA_DISCREPANCY_RATIO_MAX: u32 = 5000;
146            // If we have enough data, check the sanity of our measurement against our EWMA value.
147            if raw_rtt > ewma_rtt * DELTA_DISCREPANCY_RATIO_MAX {
148                // The clock significantly jumped forward.
149                //
150                // Don't update the global cache, though, since this is triggerable over the
151                // network.
152                //
153                // FIXME(eta): We should probably log something here?
154                true
155            } else if ewma_rtt > raw_rtt * DELTA_DISCREPANCY_RATIO_MAX {
156                // The clock might have stalled. We can't really make a decision just off this
157                // one measurement, though, so we'll use the stored stall value.
158                self.clock_stalled.load(Ordering::SeqCst)
159            } else {
160                // If we got here, we're not stalled.
161                self.clock_stalled.store(false, Ordering::SeqCst);
162                false
163            }
164        } else {
165            // If we don't have enough measurements to sanity check, assume it's okay.
166            false
167        }
168    }
169
170    /// Update the estimator on time `now` and at the congestion window `cwnd`.
171    ///
172    /// # Errors
173    ///
174    /// Each call to this function removes an entry from `sendme_expected_from` (the entries are
175    /// added using [`sendme_expected_from`](Self::sendme_expected_from)).
176    ///
177    /// Returns an error if are not expecting any SENDMEs at this time (if `expect_sendme` was
178    /// never called, or if we have exhausted all `sendme_expected_from` added by previous
179    /// `expect_sendme` calls).
180    ///
181    /// Spec: prop324 section 2.1 C-tor: congestion_control_update_circuit_rtt() in
182    /// congestion_control_common.c
183    pub(crate) fn update(
184        &mut self,
185        now: Instant,
186        state: &State,
187        cwnd: &CongestionWindow,
188    ) -> Result<ClockStall, Error> {
189        let data_sent_at = self
190            .sendme_expected_from
191            .pop_front()
192            .ok_or(Error::MismatchedEstimationCall)?;
193        let raw_rtt = now.saturating_duration_since(data_sent_at);
194
195        if self.is_clock_stalled(raw_rtt, state.in_slow_start()) {
196            return Ok(ClockStall::Detected);
197        }
198
199        self.max_rtt = self.max_rtt.max(Some(raw_rtt));
200        self.last_rtt = Some(raw_rtt);
201
202        // This is the "N" for N-EWMA.
203        let ewma_n = u64::from(if state.in_slow_start() {
204            self.params.ewma_ss_max()
205        } else {
206            min(
207                (cwnd.update_rate(state) * (self.params.ewma_cwnd_pct().as_percent())) / 100,
208                self.params.ewma_max(),
209            )
210        });
211        let ewma_n = max(ewma_n, 2);
212
213        // Get the USEC values.
214        let raw_rtt_usec = raw_rtt.as_micros() as u64;
215        let prev_ewma_rtt_usec = self.ewma_rtt.map(|rtt| rtt.as_micros() as u64);
216
217        // This is the actual EWMA calculation.
218        // C-tor simplifies this as follows for rounding error reasons:
219        //
220        // EWMA = value*2/(N+1) + EMA_prev*(N-1)/(N+1)
221        //      = (value*2 + EWMA_prev*(N-1))/(N+1)
222        //
223        // Spec: prop324 section 2.1.2 (N_EWMA_SMOOTHING)
224        let new_ewma_rtt_usec = match prev_ewma_rtt_usec {
225            None => raw_rtt_usec,
226            Some(prev_ewma_rtt_usec) => {
227                ((raw_rtt_usec * 2) + ((ewma_n - 1) * prev_ewma_rtt_usec)) / (ewma_n + 1)
228            }
229        };
230        let ewma_rtt = Duration::from_micros(new_ewma_rtt_usec);
231        self.ewma_rtt = Some(ewma_rtt);
232
233        let Some(min_rtt) = self.min_rtt else {
234            self.min_rtt = self.ewma_rtt;
235            return Ok(ClockStall::NotDetected);
236        };
237
238        if cwnd.get() == cwnd.min() && !state.in_slow_start() {
239            // The cast is OK even if lossy, we only care about the usec level.
240            let max = max(ewma_rtt, min_rtt).as_micros() as u64;
241            let min = min(ewma_rtt, min_rtt).as_micros() as u64;
242            let rtt_reset_pct = u64::from(self.params.rtt_reset_pct().as_percent());
243            let min_rtt = Duration::from_micros(
244                (rtt_reset_pct * max / 100) + (100 - rtt_reset_pct) * min / 100,
245            );
246
247            self.min_rtt = Some(min_rtt);
248        } else if self.ewma_rtt < self.min_rtt {
249            self.min_rtt = self.ewma_rtt;
250        }
251
252        Ok(ClockStall::NotDetected)
253    }
254}
255
256/// Whether a clock stall or jump was detected.
257#[derive(Copy, Clone, Debug, PartialEq, Eq)]
258pub(crate) enum ClockStall {
259    /// Clock stall or jump was detected.
260    Detected,
261    /// No clock stall or jump detected.
262    NotDetected,
263}
264
265#[cfg(test)]
266mod test {
267    // @@ begin test lint list maintained by maint/add_warning @@
268    #![allow(clippy::bool_assert_comparison)]
269    #![allow(clippy::clone_on_copy)]
270    #![allow(clippy::dbg_macro)]
271    #![allow(clippy::mixed_attributes_style)]
272    #![allow(clippy::print_stderr)]
273    #![allow(clippy::print_stdout)]
274    #![allow(clippy::single_char_pattern)]
275    #![allow(clippy::unwrap_used)]
276    #![allow(clippy::unchecked_time_subtraction)]
277    #![allow(clippy::useless_vec)]
278    #![allow(clippy::needless_pass_by_value)]
279    #![allow(clippy::string_slice)] // See arti#2571
280    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
281
282    use web_time_compat::{Duration, Instant, InstantExt};
283
284    use crate::congestion::test_utils::{new_cwnd, new_rtt_estimator};
285
286    use super::*;
287
288    #[derive(Debug)]
289    struct RttTestSample {
290        sent_usec_in: u64,
291        sendme_received_usec_in: u64,
292        cwnd_in: u32,
293        ss_in: bool,
294        last_rtt_usec_out: u64,
295        ewma_rtt_usec_out: u64,
296        min_rtt_usec_out: u64,
297    }
298
299    impl From<[u64; 7]> for RttTestSample {
300        fn from(arr: [u64; 7]) -> Self {
301            Self {
302                sent_usec_in: arr[0],
303                sendme_received_usec_in: arr[1],
304                cwnd_in: arr[2] as u32,
305                ss_in: arr[3] == 1,
306                last_rtt_usec_out: arr[4],
307                ewma_rtt_usec_out: arr[5],
308                min_rtt_usec_out: arr[6],
309            }
310        }
311    }
312    impl RttTestSample {
313        fn test(&self, estimator: &mut RoundtripTimeEstimator, start: Instant) {
314            let state = if self.ss_in {
315                State::SlowStart
316            } else {
317                State::Steady
318            };
319            let mut cwnd = new_cwnd();
320            cwnd.set(self.cwnd_in);
321            let sent = start + Duration::from_micros(self.sent_usec_in);
322            let sendme_received = start + Duration::from_micros(self.sendme_received_usec_in);
323
324            estimator.expect_sendme(sent);
325            estimator
326                .update(sendme_received, &state, &cwnd)
327                .expect("Error on RTT update");
328            assert_eq!(
329                estimator.last_rtt,
330                Some(Duration::from_micros(self.last_rtt_usec_out))
331            );
332            assert_eq!(
333                estimator.ewma_rtt,
334                Some(Duration::from_micros(self.ewma_rtt_usec_out))
335            );
336            assert_eq!(
337                estimator.min_rtt,
338                Some(Duration::from_micros(self.min_rtt_usec_out))
339            );
340        }
341    }
342
343    #[test]
344    fn test_vectors() {
345        let mut rtt = new_rtt_estimator();
346        let now = Instant::get();
347        // from C-tor src/test/test_congestion_control.c
348        let vectors = [
349            [100000, 200000, 124, 1, 100000, 100000, 100000],
350            [200000, 300000, 124, 1, 100000, 100000, 100000],
351            [350000, 500000, 124, 1, 150000, 133333, 100000],
352            [500000, 550000, 124, 1, 50000, 77777, 77777],
353            [600000, 700000, 124, 1, 100000, 92592, 77777],
354            [700000, 750000, 124, 1, 50000, 64197, 64197],
355            [750000, 875000, 124, 0, 125000, 104732, 104732],
356            [875000, 900000, 124, 0, 25000, 51577, 104732],
357            [900000, 950000, 200, 0, 50000, 50525, 50525],
358        ];
359        for vect in vectors {
360            let vect = RttTestSample::from(vect);
361            eprintln!("Testing vector: {:?}", vect);
362            vect.test(&mut rtt, now);
363        }
364    }
365}
tor_proto/congestion/rtt.rs

tor_proto/congestion/
rtt.rs
