From 3f908a2c8c1ec4585212d776fafe536ea17bf2b4 Mon Sep 17 00:00:00 2001 From: Frank Spitulski Date: Tue, 22 Jun 2021 13:38:52 -0700 Subject: [PATCH] feat(cca): cubic (#1122) --- .gitignore | 2 + quinn-proto/src/congestion.rs | 7 +- quinn-proto/src/congestion/cubic.rs | 264 +++++++++++++++++++++++++ quinn-proto/src/congestion/new_reno.rs | 11 +- quinn-proto/src/connection/mod.rs | 10 +- 5 files changed, 287 insertions(+), 7 deletions(-) create mode 100644 quinn-proto/src/congestion/cubic.rs diff --git a/.gitignore b/.gitignore index 435315408..29db52e05 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,5 @@ **/target/ **/*.rs.bk Cargo.lock + +.idea \ No newline at end of file diff --git a/quinn-proto/src/congestion.rs b/quinn-proto/src/congestion.rs index 97067c7dc..841caabe4 100644 --- a/quinn-proto/src/congestion.rs +++ b/quinn-proto/src/congestion.rs @@ -1,8 +1,11 @@ //! Logic for controlling the rate at which data is sent -use std::time::Instant; +use std::time::{Duration, Instant}; +mod cubic; mod new_reno; + +pub use cubic::{Cubic, CubicConfig}; pub use new_reno::{NewReno, NewRenoConfig}; /// Common interface for different congestion controllers @@ -11,7 +14,7 @@ pub trait Controller: Send { /// /// `app_limited` indicates whether the connection was blocked on outgoing /// application data prior to receiving these acknowledgements. - fn on_ack(&mut self, now: Instant, sent: Instant, bytes: u64, app_limited: bool); + fn on_ack(&mut self, now: Instant, sent: Instant, bytes: u64, app_limited: bool, rtt: Duration); /// Packets were deemed lost or marked congested /// diff --git a/quinn-proto/src/congestion/cubic.rs b/quinn-proto/src/congestion/cubic.rs new file mode 100644 index 000000000..ae76e783b --- /dev/null +++ b/quinn-proto/src/congestion/cubic.rs @@ -0,0 +1,264 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use super::{Controller, ControllerFactory}; +use std::cmp; + +/// CUBIC Constants. +/// +/// These are recommended value in RFC8312. +const BETA_CUBIC: f64 = 0.7; + +const C: f64 = 0.4; + +/// CUBIC State Variables. +/// +/// We need to keep those variables across the connection. +/// k, w_max, w_last_max is described in the RFC. +#[derive(Debug, Default, Clone)] +pub struct State { + k: f64, + + w_max: f64, + + w_last_max: f64, + + // Used in CUBIC fix (see on_packet_sent()) + last_sent_time: Option, + + // Store cwnd increment during congestion avoidance. + cwnd_inc: u64, +} + +/// CUBIC Functions. +/// +/// Note that these calculations are based on a count of cwnd as bytes, +/// not packets. +/// Unit of t (duration) and RTT are based on seconds (f64). +impl State { + // K = cbrt(w_max * (1 - beta_cubic) / C) (Eq. 2) + fn cubic_k(&self, max_datagram_size: u64) -> f64 { + let w_max = self.w_max / max_datagram_size as f64; + (w_max * (1.0 - BETA_CUBIC) / C).cbrt() + } + + // W_cubic(t) = C * (t - K)^3 - w_max (Eq. 1) + fn w_cubic(&self, t: Duration, max_datagram_size: u64) -> f64 { + let w_max = self.w_max / max_datagram_size as f64; + + (C * (t.as_secs_f64() - self.k).powi(3) + w_max) * max_datagram_size as f64 + } + + // W_est(t) = w_max * beta_cubic + 3 * (1 - beta_cubic) / (1 + beta_cubic) * + // (t / RTT) (Eq. 4) + fn w_est(&self, t: Duration, rtt: Duration, max_datagram_size: u64) -> f64 { + let w_max = self.w_max / max_datagram_size as f64; + (w_max * BETA_CUBIC + + 3.0 * (1.0 - BETA_CUBIC) / (1.0 + BETA_CUBIC) * t.as_secs_f64() / rtt.as_secs_f64()) + * max_datagram_size as f64 + } +} + +/// The RFC8312 congestion controller, as widely used for TCP +#[derive(Debug, Clone)] +pub struct Cubic { + config: Arc, + /// Maximum number of bytes in flight that may be sent. + window: u64, + /// Slow start threshold in bytes. When the congestion window is below ssthresh, the mode is + /// slow start and the window grows by the number of bytes acknowledged. + ssthresh: u64, + /// The time when QUIC first detects a loss, causing it to enter recovery. When a packet sent + /// after this time is acknowledged, QUIC exits recovery. + recovery_start_time: Option, + cubic_state: State, +} + +impl Cubic { + /// Construct a state using the given `config` and current time `now` + pub fn new(config: Arc, _now: Instant) -> Self { + Self { + window: config.initial_window, + ssthresh: u64::MAX, + recovery_start_time: None, + config, + cubic_state: Default::default(), + } + } +} + +impl Controller for Cubic { + fn on_ack( + &mut self, + now: Instant, + sent: Instant, + bytes: u64, + app_limited: bool, + rtt: Duration, + ) { + if app_limited + || self + .recovery_start_time + .map(|recovery_start_time| sent <= recovery_start_time) + .unwrap_or(false) + { + return; + } + + if self.window < self.ssthresh { + // Slow start + self.window += bytes; + } else { + // Congestion avoidance. + let ca_start_time; + + match self.recovery_start_time { + Some(t) => ca_start_time = t, + None => { + // When we come here without congestion_event() triggered, + // initialize congestion_recovery_start_time, w_max and k. + ca_start_time = now; + self.recovery_start_time = Some(now); + + self.cubic_state.w_max = self.window as f64; + self.cubic_state.k = 0.0; + } + } + + let t = now - ca_start_time; + + // w_cubic(t + rtt) + let w_cubic = self + .cubic_state + .w_cubic(t + rtt, self.config.max_datagram_size); + + // w_est(t) + let w_est = self + .cubic_state + .w_est(t, rtt, self.config.max_datagram_size); + + let mut cubic_cwnd = self.window; + + if w_cubic < w_est { + // TCP friendly region. + cubic_cwnd = cmp::max(cubic_cwnd, w_est as u64); + } else if cubic_cwnd < w_cubic as u64 { + // Concave region or convex region use same increment. + let cubic_inc = (w_cubic - cubic_cwnd as f64) / cubic_cwnd as f64 + * self.config.max_datagram_size as f64; + + cubic_cwnd += cubic_inc as u64; + } + + // Update the increment and increase cwnd by MSS. + self.cubic_state.cwnd_inc += cubic_cwnd - self.window; + + // cwnd_inc can be more than 1 MSS in the late stage of max probing. + // however RFC9002 §7.3.3 (Congestion Avoidance) limits + // the increase of cwnd to 1 max_datagram_size per cwnd acknowledged. + if self.cubic_state.cwnd_inc as u64 >= self.config.max_datagram_size { + self.window += self.config.max_datagram_size; + self.cubic_state.cwnd_inc = 0; + } + } + } + + fn on_congestion_event( + &mut self, + now: Instant, + sent: Instant, + _is_persistent_congestion: bool, + ) { + if self + .recovery_start_time + .map(|recovery_start_time| sent <= recovery_start_time) + .unwrap_or(false) + { + return; + } + + self.recovery_start_time = Some(now); + + // Fast convergence + if self.cubic_state.w_max < self.cubic_state.w_last_max { + self.cubic_state.w_last_max = self.cubic_state.w_max; + self.cubic_state.w_max = self.cubic_state.w_max as f64 * (1.0 + BETA_CUBIC) / 2.0; + } else { + self.cubic_state.w_last_max = self.cubic_state.w_max; + } + + self.cubic_state.w_max = self.window as f64; + self.ssthresh = (self.cubic_state.w_max * BETA_CUBIC) as u64; + self.ssthresh = cmp::max( + self.ssthresh, + self.config.max_datagram_size * self.config.minimum_window, + ); + self.window = self.ssthresh; + self.cubic_state.k = self.cubic_state.cubic_k(self.config.max_datagram_size); + + self.cubic_state.cwnd_inc = (self.cubic_state.cwnd_inc as f64 * BETA_CUBIC) as u64; + } + + fn window(&self) -> u64 { + self.window + } + + fn clone_box(&self) -> Box { + Box::new(self.clone()) + } + + fn initial_window(&self) -> u64 { + self.config.initial_window + } +} + +/// Configuration for the `Cubic` congestion controller +#[derive(Debug, Clone)] +pub struct CubicConfig { + max_datagram_size: u64, + initial_window: u64, + minimum_window: u64, +} + +impl CubicConfig { + /// The sender’s maximum UDP payload size. Does not include UDP or IP overhead. + /// + /// Used for calculating initial and minimum congestion windows. + pub fn max_datagram_size(&mut self, value: u64) -> &mut Self { + self.max_datagram_size = value; + self + } + + /// Default limit on the amount of outstanding data in bytes. + /// + /// Recommended value: `min(10 * max_datagram_size, max(2 * max_datagram_size, 14720))` + pub fn initial_window(&mut self, value: u64) -> &mut Self { + self.initial_window = value; + self + } + + /// Default minimum congestion window. + /// + /// Recommended value: `2 * max_datagram_size`. + pub fn minimum_window(&mut self, value: u64) -> &mut Self { + self.minimum_window = value; + self + } +} + +impl Default for CubicConfig { + fn default() -> Self { + const MAX_DATAGRAM_SIZE: u64 = 1232; + Self { + max_datagram_size: MAX_DATAGRAM_SIZE, + initial_window: 14720.max(2 * MAX_DATAGRAM_SIZE).min(10 * MAX_DATAGRAM_SIZE), + minimum_window: 2 * MAX_DATAGRAM_SIZE, + } + } +} + +impl ControllerFactory for Arc { + fn build(&self, now: Instant) -> Box { + Box::new(Cubic::new(self.clone(), now)) + } +} diff --git a/quinn-proto/src/congestion/new_reno.rs b/quinn-proto/src/congestion/new_reno.rs index fd9c46864..caee5d29c 100644 --- a/quinn-proto/src/congestion/new_reno.rs +++ b/quinn-proto/src/congestion/new_reno.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use super::{Controller, ControllerFactory}; @@ -33,7 +33,14 @@ impl NewReno { } impl Controller for NewReno { - fn on_ack(&mut self, _now: Instant, sent: Instant, bytes: u64, app_limited: bool) { + fn on_ack( + &mut self, + _now: Instant, + sent: Instant, + bytes: u64, + app_limited: bool, + _rtt: Duration, + ) { if app_limited || sent <= self.recovery_start_time { return; } diff --git a/quinn-proto/src/connection/mod.rs b/quinn-proto/src/connection/mod.rs index ded4ae56f..c6c65630c 100644 --- a/quinn-proto/src/connection/mod.rs +++ b/quinn-proto/src/connection/mod.rs @@ -1194,9 +1194,13 @@ where if info.ack_eliciting && self.path.challenge.is_none() { // Only pass ACKs to the congestion controller if we are not validating the current // path, so as to ignore any ACKs from older paths still coming in. - self.path - .congestion - .on_ack(now, info.time_sent, info.size.into(), self.app_limited); + self.path.congestion.on_ack( + now, + info.time_sent, + info.size.into(), + self.app_limited, + self.path.rtt.get(), + ); } // Update state for confirmed delivery of frames