quinn_proto/congestion/
cubic.rs

1use std::any::Any;
2use std::cmp;
3use std::sync::Arc;
4
5use super::{BASE_DATAGRAM_SIZE, Controller, ControllerFactory};
6use crate::connection::RttEstimator;
7use crate::{Duration, Instant};
8
9/// CUBIC Constants.
10///
11/// These are recommended value in RFC8312.
12const BETA_CUBIC: f64 = 0.7;
13
14const C: f64 = 0.4;
15
16/// CUBIC State Variables.
17///
18/// We need to keep those variables across the connection.
19/// k, w_max are described in the RFC.
20#[derive(Debug, Default, Clone)]
21pub(super) struct State {
22    k: f64,
23
24    w_max: f64,
25
26    // Store cwnd increment during congestion avoidance.
27    cwnd_inc: u64,
28}
29
30/// CUBIC Functions.
31///
32/// Note that these calculations are based on a count of cwnd as bytes,
33/// not packets.
34/// Unit of t (duration) and RTT are based on seconds (f64).
35impl State {
36    // K = cbrt(w_max * (1 - beta_cubic) / C) (Eq. 2)
37    fn cubic_k(&self, max_datagram_size: u64) -> f64 {
38        let w_max = self.w_max / max_datagram_size as f64;
39        (w_max * (1.0 - BETA_CUBIC) / C).cbrt()
40    }
41
42    // W_cubic(t) = C * (t - K)^3 - w_max (Eq. 1)
43    fn w_cubic(&self, t: Duration, max_datagram_size: u64) -> f64 {
44        let w_max = self.w_max / max_datagram_size as f64;
45
46        (C * (t.as_secs_f64() - self.k).powi(3) + w_max) * max_datagram_size as f64
47    }
48
49    // W_est(t) = w_max * beta_cubic + 3 * (1 - beta_cubic) / (1 + beta_cubic) *
50    // (t / RTT) (Eq. 4)
51    fn w_est(&self, t: Duration, rtt: Duration, max_datagram_size: u64) -> f64 {
52        let w_max = self.w_max / max_datagram_size as f64;
53        (w_max * BETA_CUBIC
54            + 3.0 * (1.0 - BETA_CUBIC) / (1.0 + BETA_CUBIC) * t.as_secs_f64() / rtt.as_secs_f64())
55            * max_datagram_size as f64
56    }
57}
58
59/// The RFC8312 congestion controller, as widely used for TCP
60#[derive(Debug, Clone)]
61pub struct Cubic {
62    config: Arc<CubicConfig>,
63    /// Maximum number of bytes in flight that may be sent.
64    window: u64,
65    /// Slow start threshold in bytes. When the congestion window is below ssthresh, the mode is
66    /// slow start and the window grows by the number of bytes acknowledged.
67    ssthresh: u64,
68    /// The time when QUIC first detects a loss, causing it to enter recovery. When a packet sent
69    /// after this time is acknowledged, QUIC exits recovery.
70    recovery_start_time: Option<Instant>,
71    cubic_state: State,
72    current_mtu: u64,
73}
74
75impl Cubic {
76    /// Construct a state using the given `config` and current time `now`
77    pub fn new(config: Arc<CubicConfig>, _now: Instant, current_mtu: u16) -> Self {
78        Self {
79            window: config.initial_window,
80            ssthresh: u64::MAX,
81            recovery_start_time: None,
82            config,
83            cubic_state: Default::default(),
84            current_mtu: current_mtu as u64,
85        }
86    }
87
88    fn minimum_window(&self) -> u64 {
89        2 * self.current_mtu
90    }
91}
92
93impl Controller for Cubic {
94    fn on_ack(
95        &mut self,
96        now: Instant,
97        sent: Instant,
98        bytes: u64,
99        app_limited: bool,
100        rtt: &RttEstimator,
101    ) {
102        if app_limited
103            || self
104                .recovery_start_time
105                .map(|recovery_start_time| sent <= recovery_start_time)
106                .unwrap_or(false)
107        {
108            return;
109        }
110
111        if self.window < self.ssthresh {
112            // Slow start
113            self.window += bytes;
114        } else {
115            // Congestion avoidance.
116            let ca_start_time;
117
118            match self.recovery_start_time {
119                Some(t) => ca_start_time = t,
120                None => {
121                    // When we come here without congestion_event() triggered,
122                    // initialize congestion_recovery_start_time, w_max and k.
123                    ca_start_time = now;
124                    self.recovery_start_time = Some(now);
125
126                    self.cubic_state.w_max = self.window as f64;
127                    self.cubic_state.k = 0.0;
128                }
129            }
130
131            let t = now - ca_start_time;
132
133            // w_cubic(t + rtt)
134            let w_cubic = self.cubic_state.w_cubic(t + rtt.get(), self.current_mtu);
135
136            // w_est(t)
137            let w_est = self.cubic_state.w_est(t, rtt.get(), self.current_mtu);
138
139            let mut cubic_cwnd = self.window;
140
141            if w_cubic < w_est {
142                // TCP friendly region.
143                cubic_cwnd = cmp::max(cubic_cwnd, w_est as u64);
144            } else if cubic_cwnd < w_cubic as u64 {
145                // Concave region or convex region use same increment.
146                let cubic_inc =
147                    (w_cubic - cubic_cwnd as f64) / cubic_cwnd as f64 * self.current_mtu as f64;
148
149                // w_cubic grows cubically with the time since the last congestion
150                // event and can exceed `u64::MAX` after a long lossless period.
151                cubic_cwnd = cubic_cwnd.saturating_add(cubic_inc as u64);
152            }
153
154            // Update the increment and increase cwnd by MSS.
155            self.cubic_state.cwnd_inc += cubic_cwnd - self.window;
156
157            // cwnd_inc can be more than 1 MSS in the late stage of max probing.
158            // however RFC9002 §7.3.3 (Congestion Avoidance) limits
159            // the increase of cwnd to 1 max_datagram_size per cwnd acknowledged.
160            if self.cubic_state.cwnd_inc >= self.current_mtu {
161                self.window += self.current_mtu;
162                self.cubic_state.cwnd_inc = 0;
163            }
164        }
165    }
166
167    fn on_congestion_event(
168        &mut self,
169        now: Instant,
170        sent: Instant,
171        is_persistent_congestion: bool,
172        _lost_bytes: u64,
173    ) {
174        if self
175            .recovery_start_time
176            .map(|recovery_start_time| sent <= recovery_start_time)
177            .unwrap_or(false)
178        {
179            return;
180        }
181
182        self.recovery_start_time = Some(now);
183        let window = self.window as f64;
184
185        // Fast convergence lowers W_max first; the 0.7 loss reduction still
186        // applies to the old window, not to that already-reduced W_max.
187        // https://www.rfc-editor.org/rfc/rfc9438.html#section-4.7
188        // https://www.rfc-editor.org/rfc/rfc9438.html#section-4.6
189        if window < self.cubic_state.w_max {
190            self.cubic_state.w_max = window * (1.0 + BETA_CUBIC) / 2.0;
191        } else {
192            self.cubic_state.w_max = window;
193        }
194
195        self.ssthresh = cmp::max((window * BETA_CUBIC) as u64, self.minimum_window());
196        self.window = self.ssthresh;
197        self.cubic_state.k = self.cubic_state.cubic_k(self.current_mtu);
198
199        self.cubic_state.cwnd_inc = (self.cubic_state.cwnd_inc as f64 * BETA_CUBIC) as u64;
200
201        if is_persistent_congestion {
202            self.recovery_start_time = None;
203            self.cubic_state.w_max = self.window as f64;
204
205            // 4.7 Timeout - reduce ssthresh based on BETA_CUBIC
206            self.ssthresh = cmp::max(
207                (self.window as f64 * BETA_CUBIC) as u64,
208                self.minimum_window(),
209            );
210
211            self.cubic_state.cwnd_inc = 0;
212
213            self.window = self.minimum_window();
214        }
215    }
216
217    fn on_mtu_update(&mut self, new_mtu: u16) {
218        self.current_mtu = new_mtu as u64;
219        self.window = self.window.max(self.minimum_window());
220    }
221
222    fn window(&self) -> u64 {
223        self.window
224    }
225
226    fn metrics(&self) -> super::ControllerMetrics {
227        super::ControllerMetrics {
228            congestion_window: self.window(),
229            ssthresh: Some(self.ssthresh),
230            pacing_rate: None,
231        }
232    }
233
234    fn clone_box(&self) -> Box<dyn Controller> {
235        Box::new(self.clone())
236    }
237
238    fn initial_window(&self) -> u64 {
239        self.config.initial_window
240    }
241
242    fn into_any(self: Box<Self>) -> Box<dyn Any> {
243        self
244    }
245}
246
247/// Configuration for the `Cubic` congestion controller
248#[derive(Debug, Clone)]
249pub struct CubicConfig {
250    initial_window: u64,
251}
252
253impl CubicConfig {
254    /// Default limit on the amount of outstanding data in bytes.
255    ///
256    /// Recommended value: `min(10 * max_datagram_size, max(2 * max_datagram_size, 14720))`
257    pub fn initial_window(&mut self, value: u64) -> &mut Self {
258        self.initial_window = value;
259        self
260    }
261}
262
263impl Default for CubicConfig {
264    fn default() -> Self {
265        Self {
266            initial_window: 14720.clamp(2 * BASE_DATAGRAM_SIZE, 10 * BASE_DATAGRAM_SIZE),
267        }
268    }
269}
270
271impl ControllerFactory for CubicConfig {
272    fn build(self: Arc<Self>, now: Instant, current_mtu: u16) -> Box<dyn Controller> {
273        Box::new(Cubic::new(self, now, current_mtu))
274    }
275}
276
277#[cfg(test)]
278mod tests {
279    use super::*;
280
281    #[test]
282    fn fast_convergence_reduces_w_max_without_double_reducing_window() {
283        let now = Instant::now();
284        let config = Arc::new(CubicConfig::default());
285        let mut cubic = Cubic::new(config, now, BASE_DATAGRAM_SIZE as u16);
286        let window = 8 * BASE_DATAGRAM_SIZE;
287
288        cubic.window = window;
289        cubic.ssthresh = window;
290        cubic.cubic_state.w_max = 12.0 * BASE_DATAGRAM_SIZE as f64;
291
292        cubic.on_congestion_event(now, now + Duration::from_millis(1), false, 0);
293
294        assert_eq!(
295            cubic.cubic_state.w_max,
296            window as f64 * (1.0 + BETA_CUBIC) / 2.0
297        );
298        assert_eq!(cubic.ssthresh, (window as f64 * BETA_CUBIC) as u64);
299        assert_eq!(cubic.window, cubic.ssthresh);
300    }
301
302    #[test]
303    fn congestion_avoidance_does_not_overflow_after_long_lossless_period() {
304        let now = Instant::now();
305        let rtt = RttEstimator::new(Duration::from_millis(100));
306        let config = Arc::new(CubicConfig::default());
307        let mut cubic = Cubic::new(config, now, BASE_DATAGRAM_SIZE as u16);
308
309        // Put CUBIC directly into congestion avoidance.
310        cubic.ssthresh = cubic.window;
311        cubic.recovery_start_time = Some(now);
312        let window = cubic.window;
313
314        // After ten days without a congestion event, w_cubic exceeds u64::MAX.
315        // Before this fix, computing the window increment overflowed.
316        let later = now + Duration::from_secs(10 * 24 * 60 * 60);
317        cubic.on_ack(later, later, BASE_DATAGRAM_SIZE, false, &rtt);
318
319        assert_eq!(cubic.window, window + BASE_DATAGRAM_SIZE);
320    }
321}
quinn_proto/congestion/cubic.rs

quinn_proto/congestion/
cubic.rs
