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                cubic_cwnd += cubic_inc as u64;
150            }
151
152            // Update the increment and increase cwnd by MSS.
153            self.cubic_state.cwnd_inc += cubic_cwnd - self.window;
154
155            // cwnd_inc can be more than 1 MSS in the late stage of max probing.
156            // however RFC9002 §7.3.3 (Congestion Avoidance) limits
157            // the increase of cwnd to 1 max_datagram_size per cwnd acknowledged.
158            if self.cubic_state.cwnd_inc >= self.current_mtu {
159                self.window += self.current_mtu;
160                self.cubic_state.cwnd_inc = 0;
161            }
162        }
163    }
164
165    fn on_congestion_event(
166        &mut self,
167        now: Instant,
168        sent: Instant,
169        is_persistent_congestion: bool,
170        _lost_bytes: u64,
171    ) {
172        if self
173            .recovery_start_time
174            .map(|recovery_start_time| sent <= recovery_start_time)
175            .unwrap_or(false)
176        {
177            return;
178        }
179
180        self.recovery_start_time = Some(now);
181        let window = self.window as f64;
182
183        // Fast convergence lowers W_max first; the 0.7 loss reduction still
184        // applies to the old window, not to that already-reduced W_max.
185        // https://www.rfc-editor.org/rfc/rfc9438.html#section-4.7
186        // https://www.rfc-editor.org/rfc/rfc9438.html#section-4.6
187        if window < self.cubic_state.w_max {
188            self.cubic_state.w_max = window * (1.0 + BETA_CUBIC) / 2.0;
189        } else {
190            self.cubic_state.w_max = window;
191        }
192
193        self.ssthresh = cmp::max((window * BETA_CUBIC) as u64, self.minimum_window());
194        self.window = self.ssthresh;
195        self.cubic_state.k = self.cubic_state.cubic_k(self.current_mtu);
196
197        self.cubic_state.cwnd_inc = (self.cubic_state.cwnd_inc as f64 * BETA_CUBIC) as u64;
198
199        if is_persistent_congestion {
200            self.recovery_start_time = None;
201            self.cubic_state.w_max = self.window as f64;
202
203            // 4.7 Timeout - reduce ssthresh based on BETA_CUBIC
204            self.ssthresh = cmp::max(
205                (self.window as f64 * BETA_CUBIC) as u64,
206                self.minimum_window(),
207            );
208
209            self.cubic_state.cwnd_inc = 0;
210
211            self.window = self.minimum_window();
212        }
213    }
214
215    fn on_mtu_update(&mut self, new_mtu: u16) {
216        self.current_mtu = new_mtu as u64;
217        self.window = self.window.max(self.minimum_window());
218    }
219
220    fn window(&self) -> u64 {
221        self.window
222    }
223
224    fn metrics(&self) -> super::ControllerMetrics {
225        super::ControllerMetrics {
226            congestion_window: self.window(),
227            ssthresh: Some(self.ssthresh),
228            pacing_rate: None,
229        }
230    }
231
232    fn clone_box(&self) -> Box<dyn Controller> {
233        Box::new(self.clone())
234    }
235
236    fn initial_window(&self) -> u64 {
237        self.config.initial_window
238    }
239
240    fn into_any(self: Box<Self>) -> Box<dyn Any> {
241        self
242    }
243}
244
245/// Configuration for the `Cubic` congestion controller
246#[derive(Debug, Clone)]
247pub struct CubicConfig {
248    initial_window: u64,
249}
250
251impl CubicConfig {
252    /// Default limit on the amount of outstanding data in bytes.
253    ///
254    /// Recommended value: `min(10 * max_datagram_size, max(2 * max_datagram_size, 14720))`
255    pub fn initial_window(&mut self, value: u64) -> &mut Self {
256        self.initial_window = value;
257        self
258    }
259}
260
261impl Default for CubicConfig {
262    fn default() -> Self {
263        Self {
264            initial_window: 14720.clamp(2 * BASE_DATAGRAM_SIZE, 10 * BASE_DATAGRAM_SIZE),
265        }
266    }
267}
268
269impl ControllerFactory for CubicConfig {
270    fn build(self: Arc<Self>, now: Instant, current_mtu: u16) -> Box<dyn Controller> {
271        Box::new(Cubic::new(self, now, current_mtu))
272    }
273}
274
275#[cfg(test)]
276mod tests {
277    use super::*;
278
279    #[test]
280    fn fast_convergence_reduces_w_max_without_double_reducing_window() {
281        let now = Instant::now();
282        let config = Arc::new(CubicConfig::default());
283        let mut cubic = Cubic::new(config, now, BASE_DATAGRAM_SIZE as u16);
284        let window = 8 * BASE_DATAGRAM_SIZE;
285
286        cubic.window = window;
287        cubic.ssthresh = window;
288        cubic.cubic_state.w_max = 12.0 * BASE_DATAGRAM_SIZE as f64;
289
290        cubic.on_congestion_event(now, now + Duration::from_millis(1), false, 0);
291
292        assert_eq!(
293            cubic.cubic_state.w_max,
294            window as f64 * (1.0 + BETA_CUBIC) / 2.0
295        );
296        assert_eq!(cubic.ssthresh, (window as f64 * BETA_CUBIC) as u64);
297        assert_eq!(cubic.window, cubic.ssthresh);
298    }
299}
quinn_proto\congestion/cubic.rs

quinn_proto\congestion/
cubic.rs
