mod max_filter;
use crate::RttEstimator;
use crate::congestion::bbr3::max_filter::MaxFilter;
use crate::congestion::{Controller, ControllerFactory, ControllerMetrics};
use crate::{Duration, Instant};
use rand::{RngExt, SeedableRng};
use rand_pcg::Pcg32;
use std::any::Any;
use std::cmp::{max, min};
use std::collections::VecDeque;
use std::sync::Arc;
const MAX_BW_FILTER_LEN: usize = 2;
const EXTRA_ACKED_FILTER_LEN: usize = 10;
const ROUND_COUNT_WINDOW: u64 = 10;
const MIN_MAX_DATAGRAM_SIZE: u16 = 1200;
const MAX_DATAGRAM_SIZE: u64 = 65527;
const PACING_RATE_1_2MBPS: f64 = 1200.0 * 1000.0;
const PACING_RATE_24MBPS: f64 = 24000.0 * 1000.0;
const HIGH_PACE_MAX_QUANTUM: u64 = 64 * 1000;
const STARTUP_PACING_GAIN: f64 = 2.773;
const DEFAULT_PACING_GAIN: f64 = 1.0;
const PROBE_BW_DOWN_PACING_GAIN: f64 = 0.9;
const PROBE_BW_UP_PACING_GAIN: f64 = 1.25;
const PACING_MARGIN_PERCENT: f64 = 1.0;
const DEFAULT_CWND_GAIN: f64 = 2.0;
const DRAIN_PACING_GAIN: f64 = 1.0 / DEFAULT_CWND_GAIN;
const PROBE_BW_UP_CWND_GAIN: f64 = 2.25;
const PROBE_RTT_CWND_GAIN: f64 = 0.5;
const PROBE_RTT_DURATION_MS: u64 = 200;
const PROBE_RTT_INTERVAL_SEC: u64 = 5;
const LOSS_THRESH: f64 = 0.02;
const BETA: f64 = 0.7;
const HEADROOM: f64 = 0.15;
const MIN_RTT_FILTER_LEN: u64 = 10;
const FULL_BW_GROWTH: f64 = 1.25;
const MAX_FULL_BW_COUNT: u64 = 3;
const MAX_LONG_TERM_PROBE_UP_ROUNDS: u32 = 30;
const MAX_RENO_ROUNDS: u64 = 63;
const MIN_PROBE_WAIT_MS: u64 = 2000;
const MAX_ADDED_PROBE_WAIT_MS: u64 = 1000;
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
enum ProbeBwSubstate {
Down,
Cruise,
Refill,
Up,
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
enum BbrState {
Startup,
Drain,
ProbeBw(ProbeBwSubstate),
ProbeRtt,
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
enum AckPhase {
ProbeStarting,
ProbeStopping,
Refilling,
ProbeFeedback,
}
#[derive(Debug, Clone, Copy)]
struct BbrPacket {
delivered: u64,
delivered_time: Instant,
first_send_time: Instant,
send_time: Instant,
is_app_limited: bool,
tx_in_flight: u64,
packet_number: u64,
size: u16,
lost: u64,
acknowledged: bool,
stale: bool,
round_count: u64,
}
#[derive(Debug, Clone, Copy)]
struct BbrRateSample {
delivery_rate: f64,
is_app_limited: bool,
interval: Duration,
delivered: u64,
prior_delivered: u64,
prior_time: Instant,
send_elapsed: Duration,
ack_elapsed: Duration,
rtt: Duration,
tx_in_flight: u64,
newly_acked: u64,
newly_lost: u64,
lost: u64,
last_end_seq: u64,
last_packet: BbrPacket,
}
#[derive(Debug, Clone)]
pub struct Bbr3 {
smss: u64,
initial_cwnd: u64,
delivered: u64,
inflight: u64,
is_cwnd_limited: bool,
cycle_count: u64,
cwnd: u64,
pacing_rate: f64,
send_quantum: u64,
pacing_gain: f64,
default_pacing_gain: f64,
probe_bw_down_pacing_gain: f64,
probe_bw_up_pacing_gain: f64,
startup_pacing_gain: f64,
drain_pacing_gain: f64,
pacing_margin_percent: f64,
cwnd_gain: f64,
default_cwnd_gain: f64,
probe_rng: Pcg32,
probe_bw_up_cwnd_gain: f64,
probe_rtt_cwnd_gain: f64,
state: BbrState,
undo_state: BbrState,
round_count: u64,
round_start: bool,
next_round_delivered: u64,
idle_restart: bool,
min_pipe_cwnd: u64,
max_bw: f64,
bw_shortterm: f64,
undo_bw_shortterm: f64,
bw: f64,
min_rtt: Duration,
bdp: u64,
extra_acked: u64,
offload_budget: u64,
max_inflight: u64,
inflight_longterm: u64,
undo_inflight_longterm: u64,
inflight_shortterm: u64,
undo_inflight_shortterm: u64,
bw_latest: f64,
inflight_latest: u64,
max_bw_filter: MaxFilter,
extra_acked_interval_start: Option<Instant>,
extra_acked_delivered: u64,
extra_acked_filter: MaxFilter,
full_bw_reached: bool,
full_bw_now: bool,
full_bw: f64,
full_bw_count: u64,
min_rtt_stamp: Option<Instant>,
probe_rtt_duration: Duration,
probe_rtt_interval: Duration,
probe_rtt_min_delay: Duration,
probe_rtt_min_stamp: Option<Instant>,
probe_rtt_expired: bool,
delivered_time: Option<Instant>,
first_send_time: Option<Instant>,
app_limited: u64,
lost: u64,
srtt: Duration,
packets: VecDeque<BbrPacket>,
rs: Option<BbrRateSample>,
rounds_since_bw_probe: u64,
bw_probe_wait: Duration,
bw_probe_up_rounds: u32,
bw_probe_up_acks: u64,
probe_up_cnt: u64,
cycle_stamp: Option<Instant>,
ack_phase: AckPhase,
bw_probe_samples: bool,
loss_round_delivered: u64,
loss_in_round: bool,
probe_rtt_done_stamp: Option<Instant>,
probe_rtt_round_done: bool,
prior_cwnd: u64,
loss_round_start: bool,
drain_start_round: u64,
ack_eliciting_threshold: u64,
max_ack_delay: Duration,
}
impl Bbr3 {
fn new(config: Arc<Bbr3Config>, current_mtu: u16) -> Self {
let probe_rng: Pcg32;
if let Some(probe_seed) = config.probe_rng_seed {
probe_rng = Pcg32::from_seed(probe_seed);
} else {
probe_rng = Pcg32::from_rng(&mut rand::rng());
}
let smss = min(
max(MIN_MAX_DATAGRAM_SIZE, current_mtu) as u64,
MAX_DATAGRAM_SIZE,
);
let initial_cwnd = config.initial_window;
let startup_pacing_gain = config.startup_pacing_gain.unwrap_or(STARTUP_PACING_GAIN);
let default_pacing_gain = config.default_pacing_gain.unwrap_or(DEFAULT_PACING_GAIN);
let probe_bw_down_pacing_gain = config
.probe_bw_down_pacing_gain
.unwrap_or(PROBE_BW_DOWN_PACING_GAIN);
let probe_bw_up_pacing_gain = config
.probe_bw_up_pacing_gain
.unwrap_or(PROBE_BW_UP_PACING_GAIN);
let drain_pacing_gain = config.drain_pacing_gain.unwrap_or(DRAIN_PACING_GAIN);
let pacing_margin_percent = config
.pacing_margin_percent
.unwrap_or(PACING_MARGIN_PERCENT);
let default_cwnd_gain = config.default_cwnd_gain.unwrap_or(DEFAULT_CWND_GAIN);
let probe_bw_up_cwnd_gain = config
.probe_bw_up_cwnd_gain
.unwrap_or(PROBE_BW_UP_CWND_GAIN);
let probe_rtt_cwnd_gain = config.probe_rtt_cwnd_gain.unwrap_or(PROBE_RTT_CWND_GAIN);
let nominal_bandwidth = initial_cwnd as f64 / 0.001;
let pacing_rate = startup_pacing_gain * nominal_bandwidth;
Self {
smss,
initial_cwnd,
delivered: 0,
inflight: 0,
is_cwnd_limited: false,
cycle_count: 0,
cwnd: initial_cwnd,
pacing_rate,
send_quantum: 2 * smss, pacing_gain: startup_pacing_gain,
startup_pacing_gain,
default_pacing_gain,
probe_bw_down_pacing_gain,
probe_bw_up_pacing_gain,
drain_pacing_gain,
pacing_margin_percent,
cwnd_gain: default_cwnd_gain,
default_cwnd_gain,
probe_rng,
probe_bw_up_cwnd_gain,
state: BbrState::Startup,
undo_state: BbrState::Startup,
round_count: 0,
round_start: true,
next_round_delivered: 0,
idle_restart: false,
min_pipe_cwnd: 4 * smss, max_bw: 0.0,
bw_shortterm: f64::INFINITY,
undo_bw_shortterm: f64::INFINITY,
bw: 0.0,
min_rtt: Duration::from_secs(u64::MAX),
bdp: 0,
extra_acked: 0,
offload_budget: 0,
max_inflight: 0,
inflight_longterm: u64::MAX,
undo_inflight_longterm: u64::MAX,
inflight_shortterm: u64::MAX,
undo_inflight_shortterm: u64::MAX,
bw_latest: 0.0,
inflight_latest: 0,
max_bw_filter: MaxFilter::new(MAX_BW_FILTER_LEN as u64),
extra_acked_interval_start: None,
extra_acked_delivered: 0,
extra_acked_filter: MaxFilter::new(EXTRA_ACKED_FILTER_LEN as u64),
full_bw_reached: false,
full_bw_now: false,
full_bw: 0.0,
full_bw_count: 0,
min_rtt_stamp: None,
probe_rtt_cwnd_gain,
probe_rtt_duration: Duration::from_millis(PROBE_RTT_DURATION_MS),
probe_rtt_interval: Duration::from_secs(PROBE_RTT_INTERVAL_SEC),
probe_rtt_min_delay: Duration::ZERO,
probe_rtt_min_stamp: None,
probe_rtt_expired: false,
delivered_time: None,
first_send_time: None,
app_limited: 0,
lost: 0,
srtt: Duration::ZERO,
rs: None,
packets: VecDeque::new(),
rounds_since_bw_probe: 0,
bw_probe_wait: Duration::ZERO,
bw_probe_up_rounds: 0,
bw_probe_up_acks: 0,
probe_up_cnt: 0,
cycle_stamp: None,
ack_phase: AckPhase::ProbeStarting,
bw_probe_samples: false,
loss_round_delivered: 0,
loss_in_round: false,
probe_rtt_done_stamp: None,
probe_rtt_round_done: false,
prior_cwnd: 0,
loss_round_start: false,
drain_start_round: 0,
ack_eliciting_threshold: 1,
max_ack_delay: Duration::from_millis(25),
}
}
fn enter_startup(&mut self) {
self.state = BbrState::Startup;
self.pacing_gain = self.startup_pacing_gain;
self.cwnd_gain = self.default_cwnd_gain;
}
fn reset_full_bw(&mut self) {
self.full_bw = 0.0;
self.full_bw_count = 0;
self.full_bw_now = false;
}
fn note_loss(&mut self) {
if !self.loss_in_round {
self.loss_round_delivered = self.delivered;
}
self.save_state_upon_loss();
self.loss_in_round = true;
}
fn save_state_upon_loss(&mut self) {
self.undo_state = self.state;
self.undo_bw_shortterm = self.bw_shortterm;
self.undo_inflight_shortterm = self.inflight_shortterm;
self.undo_inflight_longterm = self.inflight_longterm;
}
fn inflight_at_loss(&mut self, packet_size: u64) -> u64 {
if let Some(rate_sample) = self.rs {
let inflight_prev = rate_sample.tx_in_flight.saturating_sub(packet_size);
let inflight_prev_threshold = LOSS_THRESH * inflight_prev as f64;
let lost_prev = rate_sample.lost.saturating_sub(packet_size);
let compared_loss = (inflight_prev_threshold.round() as u64) - lost_prev;
let lost_prefix = compared_loss as f64 / (1.0 - LOSS_THRESH);
let inflight_at_loss = inflight_prev + lost_prefix as u64;
return inflight_at_loss;
}
0
}
fn save_cwnd(&mut self) {
if !self.loss_in_round && self.state != BbrState::ProbeRtt {
self.prior_cwnd = self.cwnd;
} else {
self.prior_cwnd = max(self.prior_cwnd, self.cwnd);
}
}
fn restore_cwnd(&mut self) {
self.cwnd = max(self.cwnd, self.prior_cwnd);
}
fn probe_rtt_cwnd(&mut self) -> u64 {
let mut probe_rtt_cwnd = self.bdp_multiple(self.bw, self.probe_rtt_cwnd_gain);
probe_rtt_cwnd = max(probe_rtt_cwnd, self.min_pipe_cwnd);
probe_rtt_cwnd
}
fn bound_cwnd_for_probe_rtt(&mut self) {
if self.state == BbrState::ProbeRtt {
self.cwnd = min(self.cwnd, self.probe_rtt_cwnd());
}
}
fn target_inflight(&self) -> u64 {
min(self.bdp, self.cwnd)
}
fn handle_inflight_too_high(&mut self, now: Instant) {
self.bw_probe_samples = false;
if let Some(rate_sample) = self.rs
&& !rate_sample.is_app_limited
{
self.inflight_longterm = max(
rate_sample.tx_in_flight,
(self.target_inflight() as f64 * BETA) as u64,
);
}
if self.state == BbrState::ProbeBw(ProbeBwSubstate::Up) {
self.start_probe_bw_down(now);
}
}
fn is_inflight_too_high(&self) -> bool {
if let Some(rate_sample) = self.rs {
return rate_sample.lost as f64 > rate_sample.tx_in_flight as f64 * LOSS_THRESH;
}
false
}
fn check_startup_high_loss(&mut self) {
if self.full_bw_reached {
return;
}
if self.is_inflight_too_high() {
let mut new_inflight_hi = self.bdp.max(self.inflight_latest);
if let Some(rate_sample) = self.rs
&& new_inflight_hi < rate_sample.delivered
{
new_inflight_hi = rate_sample.delivered;
}
self.inflight_longterm = new_inflight_hi;
self.full_bw_reached = true;
}
}
fn enter_probe_bw(&mut self, now: Instant) {
self.cwnd_gain = self.default_cwnd_gain;
self.start_probe_bw_down(now);
}
fn pick_probe_wait(&mut self) {
self.rounds_since_bw_probe = self.probe_rng.random_bool(0.5) as u64;
self.bw_probe_wait = Duration::from_millis(
MIN_PROBE_WAIT_MS + self.probe_rng.random_range(0..=MAX_ADDED_PROBE_WAIT_MS),
);
}
fn has_elapsed_in_phase(&mut self, interval: Duration, now: Instant) -> bool {
if let Some(cycle_stamp) = self.cycle_stamp {
now > cycle_stamp.checked_add(interval).unwrap_or(cycle_stamp)
} else {
true
}
}
fn exit_probe_rtt(&mut self, now: Instant) {
self.reset_short_term_model();
if self.full_bw_reached {
self.start_probe_bw_down(now);
self.start_probe_bw_cruise();
} else {
self.enter_startup();
}
}
fn check_probe_rtt_done(&mut self, now: Instant) {
if let Some(probe_rtt_done_stamp) = self.probe_rtt_done_stamp
&& now > probe_rtt_done_stamp
{
self.probe_rtt_min_stamp = Some(now);
self.restore_cwnd();
self.exit_probe_rtt(now);
}
}
fn maybe_enter_probe_bw_refill(&mut self, now: Instant) -> bool {
if self.has_elapsed_in_phase(self.bw_probe_wait, now)
|| self.is_reno_coexistence_probe_time()
{
self.start_probe_bw_refill();
return true;
}
false
}
fn maybe_go_down(&mut self) -> bool {
if self.is_cwnd_limited && self.cwnd >= self.inflight_longterm {
self.reset_full_bw();
if let Some(rate_sample) = self.rs {
self.full_bw = rate_sample.delivery_rate;
}
} else if self.full_bw_now {
return true;
}
false
}
fn is_reno_coexistence_probe_time(&self) -> bool {
let reno_rounds = self.target_inflight();
let rounds = min(reno_rounds, MAX_RENO_ROUNDS);
self.rounds_since_bw_probe >= rounds
}
fn bdp_multiple(&mut self, bw: f64, gain: f64) -> u64 {
if self.min_rtt == Duration::from_secs(u64::MAX) {
return self.initial_cwnd;
}
self.bdp = (bw * self.min_rtt.as_secs_f64()).round() as u64;
(gain * self.bdp as f64) as u64
}
fn update_offload_budget(&mut self) {
let base = self.send_quantum;
let threshold_bytes = self.ack_eliciting_threshold.saturating_mul(self.smss);
let delay_bytes = (self.max_ack_delay.as_secs_f64() * self.max_bw).round() as u64;
let delayed_ack_term = min(threshold_bytes, delay_bytes);
self.offload_budget = base.saturating_add(delayed_ack_term);
}
fn quantization_budget(&mut self, inflight_cap: u64) -> u64 {
self.update_offload_budget();
let mut inflight_cap = max(inflight_cap, self.offload_budget);
inflight_cap = max(inflight_cap, self.min_pipe_cwnd);
if self.state == BbrState::ProbeBw(ProbeBwSubstate::Up) {
inflight_cap += 2 * self.smss;
}
inflight_cap
}
fn get_inflight(&mut self, gain: f64) -> u64 {
let inflight_cap = self.bdp_multiple(self.max_bw, gain);
self.quantization_budget(inflight_cap)
}
fn update_max_inflight(&mut self) {
let mut inflight_cap = self.bdp_multiple(self.max_bw, self.cwnd_gain);
inflight_cap += self.extra_acked;
self.max_inflight = self.quantization_budget(inflight_cap);
}
fn reset_congestion_signals(&mut self) {
self.loss_in_round = false;
self.bw_latest = 0.0;
self.inflight_latest = 0;
}
fn start_round(&mut self) {
self.next_round_delivered = self.delivered;
self.is_cwnd_limited = false;
}
fn update_round(&mut self, packet: BbrPacket) {
if packet.delivered >= self.next_round_delivered {
self.start_round();
self.round_count += 1;
self.rounds_since_bw_probe += 1;
self.round_start = true;
} else {
self.round_start = false;
}
}
fn start_probe_bw_down(&mut self, now: Instant) {
self.reset_congestion_signals();
self.probe_up_cnt = u64::MAX;
self.pick_probe_wait();
self.cycle_stamp = Some(now);
self.ack_phase = AckPhase::ProbeStopping;
self.start_round();
self.pacing_gain = self.probe_bw_down_pacing_gain;
self.cwnd_gain = self.default_cwnd_gain;
self.state = BbrState::ProbeBw(ProbeBwSubstate::Down);
}
fn inflight_with_headroom(&self) -> u64 {
if self.inflight_longterm == u64::MAX {
return u64::MAX;
}
let total_headroom = max(self.smss, (HEADROOM * self.inflight_longterm as f64) as u64);
if let Some(inflight_with_headroom) = self.inflight_longterm.checked_sub(total_headroom) {
max(inflight_with_headroom, self.min_pipe_cwnd)
} else {
self.min_pipe_cwnd
}
}
fn set_pacing_rate_with_gain(&mut self, gain: f64) {
let rate = gain * self.bw * (100.0 - self.pacing_margin_percent) / 100.0;
if self.full_bw_reached || rate > self.pacing_rate {
self.pacing_rate = rate;
}
}
fn raise_inflight_long_term_slope(&mut self) {
let growth_this_round = self
.smss
.checked_shl(self.bw_probe_up_rounds)
.unwrap_or(u64::MAX);
self.bw_probe_up_rounds = min(self.bw_probe_up_rounds + 1, MAX_LONG_TERM_PROBE_UP_ROUNDS);
self.probe_up_cnt = max(self.cwnd / growth_this_round, 1);
}
fn probe_inflight_long_term_upward(&mut self) {
if !self.is_cwnd_limited || self.cwnd < self.inflight_longterm {
return;
}
if let Some(rate_sample) = self.rs {
self.bw_probe_up_acks += rate_sample.newly_acked;
}
if self.bw_probe_up_acks >= self.probe_up_cnt && self.probe_up_cnt > 0 {
let delta = self.bw_probe_up_acks / self.probe_up_cnt;
self.bw_probe_up_acks -= delta * self.probe_up_cnt;
self.inflight_longterm += delta;
if self.round_start {
self.raise_inflight_long_term_slope();
}
}
}
fn advance_max_bw_filter(&mut self) {
self.cycle_count = self.cycle_count.saturating_add(1);
}
fn adapt_long_term_model(&mut self) {
if self.ack_phase == AckPhase::ProbeStarting && self.round_start {
self.ack_phase = AckPhase::ProbeFeedback;
}
if self.ack_phase == AckPhase::ProbeStopping
&& self.round_start
&& let BbrState::ProbeBw(_) = self.state
&& let Some(rate_sample) = self.rs
&& !rate_sample.is_app_limited
{
self.advance_max_bw_filter();
}
if !self.is_inflight_too_high() {
if self.inflight_longterm == u64::MAX {
return;
}
if let Some(rate_sample) = self.rs
&& rate_sample.tx_in_flight > self.inflight_longterm
{
self.inflight_longterm = rate_sample.tx_in_flight;
}
if self.state == BbrState::ProbeBw(ProbeBwSubstate::Up) {
self.probe_inflight_long_term_upward();
}
}
}
fn maybe_update_budget_and_time_to_cruise(&mut self) -> bool {
if self.inflight > self.inflight_with_headroom() {
return false;
}
if self.inflight > self.get_inflight(1.0) {
return false;
}
true
}
fn start_probe_bw_cruise(&mut self) {
self.state = BbrState::ProbeBw(ProbeBwSubstate::Cruise);
self.pacing_gain = self.default_pacing_gain;
self.cwnd_gain = self.default_cwnd_gain;
}
fn reset_short_term_model(&mut self) {
self.bw_shortterm = f64::INFINITY;
self.inflight_shortterm = u64::MAX;
}
fn init_lower_bounds(&mut self) {
if self.bw_shortterm == f64::INFINITY {
self.bw_shortterm = self.max_bw;
}
if self.inflight_shortterm == u64::MAX {
self.inflight_shortterm = self.cwnd;
}
}
fn loss_lower_bounds(&mut self) {
self.bw_shortterm = [self.bw_latest, BETA * self.bw_shortterm]
.iter()
.copied()
.fold(f64::NAN, f64::max);
self.inflight_shortterm = max(
self.inflight_latest,
(BETA * self.inflight_shortterm as f64) as u64,
);
}
fn bound_bw_for_model(&mut self) {
self.bw = [self.max_bw, self.bw_shortterm]
.iter()
.copied()
.fold(f64::NAN, f64::min);
}
fn start_probe_bw_refill(&mut self) {
self.reset_short_term_model();
self.bw_probe_up_rounds = 0;
self.bw_probe_up_acks = 0;
self.ack_phase = AckPhase::Refilling;
self.start_round();
self.cwnd_gain = self.default_cwnd_gain;
self.pacing_gain = self.default_pacing_gain;
self.state = BbrState::ProbeBw(ProbeBwSubstate::Refill);
}
fn start_probe_bw_up(&mut self) {
self.ack_phase = AckPhase::ProbeStarting;
self.start_round();
self.reset_full_bw();
if let Some(rate_sample) = self.rs {
self.full_bw = rate_sample.delivery_rate;
}
self.state = BbrState::ProbeBw(ProbeBwSubstate::Up);
self.pacing_gain = self.probe_bw_up_pacing_gain;
self.cwnd_gain = self.probe_bw_up_cwnd_gain;
self.raise_inflight_long_term_slope();
}
fn enter_probe_rtt(&mut self) {
self.state = BbrState::ProbeRtt;
self.pacing_gain = self.default_pacing_gain;
self.cwnd_gain = self.probe_rtt_cwnd_gain;
}
fn handle_restart_from_idle(&mut self, now: Instant) {
if self.inflight == 0 && self.app_limited != 0 {
self.idle_restart = true;
self.extra_acked_interval_start = Some(now);
match self.state {
BbrState::ProbeBw(_) => {
self.set_pacing_rate_with_gain(1.0);
}
BbrState::ProbeRtt => {
self.check_probe_rtt_done(now);
}
_ => {}
}
}
}
fn update_probe_bw_cycle_phase(&mut self, now: Instant) {
if !self.full_bw_reached {
return;
}
self.adapt_long_term_model();
let state = self.state;
match state {
BbrState::ProbeBw(ProbeBwSubstate::Down) => {
if self.maybe_enter_probe_bw_refill(now) {
return;
}
if self.maybe_update_budget_and_time_to_cruise() {
self.start_probe_bw_cruise();
}
}
BbrState::ProbeBw(ProbeBwSubstate::Cruise) if self.maybe_enter_probe_bw_refill(now) => {
}
BbrState::ProbeBw(ProbeBwSubstate::Refill) if self.round_start => {
self.bw_probe_samples = true;
self.start_probe_bw_up();
}
BbrState::ProbeBw(ProbeBwSubstate::Up) if self.maybe_go_down() => {
self.start_probe_bw_down(now);
}
_ => {}
}
}
fn update_latest_delivery_signals(&mut self) {
self.loss_round_start = false;
if let Some(rate_sample) = self.rs {
self.bw_latest = [self.bw_latest, rate_sample.delivery_rate]
.iter()
.copied()
.fold(f64::NAN, f64::max);
self.inflight_latest = max(self.inflight_latest, rate_sample.delivered);
if rate_sample.prior_delivered >= self.loss_round_delivered {
self.loss_round_delivered = self.delivered;
self.loss_round_start = true;
}
}
}
fn adapt_lower_bounds_from_congestion(&mut self) {
match self.state {
BbrState::ProbeBw(ProbeBwSubstate::Refill)
| BbrState::ProbeBw(ProbeBwSubstate::Up)
| BbrState::Startup => {}
_ => {
if self.loss_in_round {
self.init_lower_bounds();
self.loss_lower_bounds();
}
}
}
}
fn update_max_bw(&mut self, p: BbrPacket) {
self.update_round(p);
if let Some(rate_sample) = self.rs
&& rate_sample.delivery_rate > 0.0
&& (rate_sample.delivery_rate >= self.max_bw || !rate_sample.is_app_limited)
{
self.max_bw_filter
.update_max(self.cycle_count, rate_sample.delivery_rate.round() as u64);
self.max_bw = self.max_bw_filter.get_max() as f64;
}
}
fn update_congestion_signals(&mut self, p: BbrPacket) {
self.update_max_bw(p);
if !self.loss_round_start {
return;
}
self.adapt_lower_bounds_from_congestion();
self.loss_in_round = false;
}
fn update_ack_aggregation(&mut self, now: Instant) {
let interval;
if let Some(extra_acked_interval_start) = self.extra_acked_interval_start {
interval = now - extra_acked_interval_start;
} else {
interval = Duration::from_secs(0);
}
let mut expected_delivered = (self.bw * interval.as_secs_f64()) as u64;
if self.extra_acked_delivered <= expected_delivered {
self.extra_acked_delivered = 0;
self.extra_acked_interval_start = Some(now);
expected_delivered = 0;
}
if let Some(rate_sample) = self.rs {
self.extra_acked_delivered += rate_sample.newly_acked;
}
let mut extra = self
.extra_acked_delivered
.saturating_sub(expected_delivered);
extra = min(extra, self.cwnd);
if self.full_bw_reached {
self.extra_acked_filter.update_max(self.round_count, extra);
self.extra_acked = self.extra_acked_filter.get_max();
} else {
self.extra_acked = extra; }
}
fn check_full_bw_reached(&mut self) {
if self.full_bw_now || !self.round_start {
return;
}
if let Some(rate_sample) = self.rs {
if rate_sample.is_app_limited {
return;
}
if rate_sample.delivery_rate >= self.full_bw * FULL_BW_GROWTH {
self.reset_full_bw();
self.full_bw = rate_sample.delivery_rate;
return;
}
}
self.full_bw_count += 1;
self.full_bw_now = self.full_bw_count >= MAX_FULL_BW_COUNT;
if self.full_bw_now {
self.full_bw_reached = true;
}
}
fn enter_drain(&mut self) {
self.state = BbrState::Drain;
self.pacing_gain = self.drain_pacing_gain;
self.cwnd_gain = self.default_cwnd_gain;
self.drain_start_round = self.round_count;
}
fn check_startup_done(&mut self) {
self.check_startup_high_loss();
if self.state == BbrState::Startup && self.full_bw_reached {
self.enter_drain();
}
}
fn check_drain_done(&mut self, now: Instant) {
if self.state == BbrState::Drain
&& (self.inflight <= self.get_inflight(1.0)
|| self.round_count > self.drain_start_round + 3)
{
self.enter_probe_bw(now);
}
}
fn update_min_rtt(&mut self, now: Instant) {
if let Some(probe_rtt_min_stamp) = self.probe_rtt_min_stamp {
self.probe_rtt_expired = now
> probe_rtt_min_stamp
.checked_add(self.probe_rtt_interval)
.unwrap_or(probe_rtt_min_stamp);
} else {
self.probe_rtt_expired = true;
}
if let Some(rate_sample) = self.rs
&& rate_sample.rtt >= Duration::from_secs(0)
&& (rate_sample.rtt < self.probe_rtt_min_delay || self.probe_rtt_expired)
{
self.probe_rtt_min_delay = rate_sample.rtt;
self.probe_rtt_min_stamp = Some(now);
}
let min_rtt_expired;
if let Some(min_rtt_stamp) = self.min_rtt_stamp {
min_rtt_expired = now
> min_rtt_stamp
.checked_add(Duration::from_secs(MIN_RTT_FILTER_LEN))
.unwrap_or(min_rtt_stamp);
} else {
min_rtt_expired = true;
}
if self.probe_rtt_min_delay < self.min_rtt || min_rtt_expired {
self.min_rtt = self.probe_rtt_min_delay;
self.min_rtt_stamp = self.probe_rtt_min_stamp;
}
}
fn handle_probe_rtt(&mut self, now: Instant) {
if self.probe_rtt_done_stamp.is_none() && self.inflight <= self.probe_rtt_cwnd() {
self.probe_rtt_done_stamp =
Some(now.checked_add(self.probe_rtt_duration).unwrap_or(now));
self.probe_rtt_round_done = false;
self.start_round();
} else if self.probe_rtt_done_stamp.is_some() {
if self.round_start {
self.probe_rtt_round_done = true;
}
if self.probe_rtt_round_done {
self.check_probe_rtt_done(now);
}
}
}
fn check_probe_rtt(&mut self, now: Instant) {
match self.state {
BbrState::ProbeRtt => {
self.handle_probe_rtt(now);
}
_ => {
if self.probe_rtt_expired && !self.idle_restart {
self.enter_probe_rtt();
self.save_cwnd();
self.probe_rtt_done_stamp = None;
self.ack_phase = AckPhase::ProbeStopping;
self.start_round();
}
}
}
if let Some(rate_sample) = self.rs
&& rate_sample.delivered > 0
{
self.idle_restart = false;
}
}
fn advance_latest_delivery_signals(&mut self) {
if self.loss_round_start
&& let Some(rate_sample) = self.rs
{
self.bw_latest = rate_sample.delivery_rate;
self.inflight_latest = rate_sample.delivered;
}
}
fn update_model_and_state(&mut self, p: BbrPacket, now: Instant) {
self.update_latest_delivery_signals();
self.reset_congestion_signals();
self.update_congestion_signals(p);
self.update_ack_aggregation(now);
self.check_full_bw_reached();
self.check_startup_done();
self.check_drain_done(now);
self.update_probe_bw_cycle_phase(now);
self.update_min_rtt(now);
self.check_probe_rtt(now);
self.advance_latest_delivery_signals();
self.bound_bw_for_model();
}
fn set_pacing_rate(&mut self) {
self.set_pacing_rate_with_gain(self.pacing_gain);
}
fn set_send_quantum(&mut self) {
self.send_quantum = match self.pacing_rate {
rate if rate < PACING_RATE_1_2MBPS => MAX_DATAGRAM_SIZE,
rate if rate < PACING_RATE_24MBPS => 2 * MAX_DATAGRAM_SIZE,
_ => min((self.pacing_rate / 1000.0) as u64, HIGH_PACE_MAX_QUANTUM),
};
}
fn bound_cwnd_for_model(&mut self) {
let mut cap = u64::MAX;
match self.state {
BbrState::ProbeRtt => {
cap = self.inflight_with_headroom();
}
BbrState::ProbeBw(ProbeBwSubstate::Cruise) => {
cap = self.inflight_with_headroom();
}
BbrState::ProbeBw(_) => {
cap = self.inflight_longterm;
}
_ => {}
}
cap = min(cap, self.inflight_shortterm);
cap = max(cap, self.min_pipe_cwnd);
self.cwnd = min(self.cwnd, cap);
}
fn set_cwnd(&mut self) {
self.update_max_inflight();
if self.full_bw_reached {
if let Some(rate_sample) = self.rs {
self.cwnd = min(self.cwnd + rate_sample.newly_acked, self.max_inflight);
} else {
self.cwnd = min(self.cwnd, self.max_inflight);
}
} else if (self.cwnd < self.max_inflight || self.delivered < self.initial_cwnd)
&& let Some(rate_sample) = self.rs
{
self.cwnd += rate_sample.newly_acked;
}
self.cwnd = max(self.cwnd, self.min_pipe_cwnd);
self.bound_cwnd_for_probe_rtt();
self.bound_cwnd_for_model();
}
fn update_control_parameters(&mut self) {
self.set_pacing_rate();
self.set_send_quantum();
self.set_cwnd();
}
fn is_newest_packet(&self, send_time: Instant, end_seq: u64) -> bool {
if let Some(first_send_time) = self.first_send_time {
if send_time > first_send_time {
return true;
}
if let Some(rate_sample) = self.rs
&& end_seq > rate_sample.last_end_seq
{
return true;
}
}
false
}
fn process_lost_packet(&mut self, lost_bytes: u64, packet_index: usize, now: Instant) {
let p = self.packets[packet_index];
self.note_loss();
if !self.bw_probe_samples {
self.packets.remove(packet_index);
return;
}
if let Some(mut rate_sample) = self.rs {
rate_sample.newly_lost += lost_bytes;
rate_sample.tx_in_flight = p.tx_in_flight;
rate_sample.lost = self.lost.saturating_sub(p.lost);
rate_sample.is_app_limited = p.is_app_limited;
if self.is_inflight_too_high() {
rate_sample.tx_in_flight = self.inflight_at_loss(p.size as u64);
self.handle_inflight_too_high(now);
}
self.rs = Some(rate_sample);
}
self.packets.remove(packet_index);
}
}
impl Controller for Bbr3 {
fn on_packet_sent(&mut self, now: Instant, bytes: u16, pn: u64) {
if self.inflight == 0 {
self.first_send_time = Some(now);
self.delivered_time = Some(now);
}
let added_bytes = bytes as u64;
self.inflight += added_bytes;
self.packets.push_back(BbrPacket {
delivered: self.delivered,
delivered_time: self.delivered_time.unwrap_or(now),
first_send_time: self.first_send_time.unwrap_or(now),
send_time: now,
is_app_limited: self.app_limited != 0,
tx_in_flight: self.inflight,
packet_number: pn,
size: bytes,
lost: self.lost,
acknowledged: false,
stale: false,
round_count: self.round_count,
});
self.handle_restart_from_idle(now);
}
fn on_ack(
&mut self,
now: Instant,
sent: Instant,
bytes: u64,
pn: u64,
_app_limited: bool,
rtt: &RttEstimator,
) {
if let Some(mut rate_sample) = self.rs {
rate_sample.newly_acked += bytes;
self.rs = Some(rate_sample);
self.delivered += bytes;
self.delivered_time = Some(now);
}
let p_index_result = self.packets.binary_search_by_key(&pn, |p| p.packet_number);
let is_newest_packet = self.is_newest_packet(sent, pn);
if let Ok(p_index) = p_index_result
&& let Some(p) = self.packets.get_mut(p_index)
{
p.acknowledged = true;
if let Some(mut rate_sample) = self.rs {
rate_sample.rtt = now - p.send_time;
if is_newest_packet {
self.srtt = rtt.get();
rate_sample.prior_delivered = p.delivered;
rate_sample.prior_time = p.delivered_time;
rate_sample.is_app_limited = p.is_app_limited;
rate_sample.tx_in_flight = p.tx_in_flight;
rate_sample.send_elapsed = p.send_time - p.first_send_time;
rate_sample.ack_elapsed = self.delivered_time.unwrap_or(now) - p.delivered_time;
rate_sample.last_end_seq = pn;
self.first_send_time = Some(p.send_time);
rate_sample.last_packet = *p;
self.rs = Some(rate_sample);
self.update_model_and_state(rate_sample.last_packet, now);
self.update_control_parameters();
}
} else {
let rate_sample = BbrRateSample {
rtt: rtt.get(),
prior_time: p.delivered_time,
interval: Duration::ZERO,
delivery_rate: 0.0,
is_app_limited: p.is_app_limited,
delivered: 0,
prior_delivered: p.delivered,
tx_in_flight: p.tx_in_flight,
send_elapsed: p.send_time - p.first_send_time,
ack_elapsed: self.delivered_time.unwrap_or(now) - p.delivered_time,
newly_acked: bytes,
newly_lost: 0,
lost: 0,
last_end_seq: pn,
last_packet: *p,
};
self.rs = Some(rate_sample);
self.first_send_time = Some(p.send_time);
self.srtt = rate_sample.rtt;
self.update_model_and_state(rate_sample.last_packet, now);
self.update_control_parameters();
}
}
}
fn on_end_acks(
&mut self,
_now: Instant,
in_flight: u64,
app_limited: bool,
largest_packet_num_acked: Option<u64>,
) {
self.inflight = in_flight;
if let Some(largest_packet_num) = largest_packet_num_acked {
if self.app_limited != 0 && largest_packet_num > self.app_limited {
self.app_limited = 0;
} else if app_limited {
self.app_limited = self.app_limited.max(largest_packet_num);
}
self.packets.retain(|&p| !p.stale);
for p in self.packets.iter_mut() {
if p.acknowledged || self.round_count - p.round_count > ROUND_COUNT_WINDOW {
p.stale = true;
}
}
if let Some(mut rate_sample) = self.rs {
if rate_sample.prior_delivered == 0 {
return;
}
rate_sample.interval = max(rate_sample.send_elapsed, rate_sample.ack_elapsed);
rate_sample.delivered = self.delivered.saturating_sub(rate_sample.prior_delivered);
if rate_sample.interval < self.min_rtt
&& self.min_rtt != Duration::from_secs(u64::MAX)
{
return;
}
if rate_sample.interval != Duration::ZERO {
rate_sample.delivery_rate =
rate_sample.delivered as f64 / rate_sample.interval.as_secs_f64();
}
if rate_sample.delivered >= self.cwnd {
self.is_cwnd_limited = true;
}
self.rs = Some(rate_sample);
rate_sample.newly_acked = 0;
rate_sample.lost = 0;
rate_sample.newly_lost = 0;
self.rs = Some(rate_sample);
}
}
}
fn on_congestion_event(
&mut self,
now: Instant,
_sent: Instant,
is_persistent_congestion: bool,
is_ecn: bool,
lost_bytes: u64,
largest_lost_pn: u64,
) {
if is_ecn {
self.lost += lost_bytes;
let p_index_result = self
.packets
.binary_search_by_key(&largest_lost_pn, |p| p.packet_number);
if let Ok(p_index) = p_index_result {
self.process_lost_packet(lost_bytes, p_index, now);
}
if is_persistent_congestion {
self.cwnd = self.min_pipe_cwnd;
}
}
}
fn on_packet_lost(&mut self, lost_bytes: u16, pn: u64, now: Instant) {
let lost_bytes_64 = lost_bytes as u64;
self.lost += lost_bytes_64;
let p_index_result = self.packets.binary_search_by_key(&pn, |p| p.packet_number);
if let Ok(p_index) = p_index_result {
self.process_lost_packet(lost_bytes_64, p_index, now);
}
}
fn on_spurious_congestion_event(&mut self) {
self.loss_in_round = false;
self.reset_full_bw();
self.bw_shortterm = [self.bw_shortterm, self.undo_bw_shortterm]
.iter()
.copied()
.fold(f64::NAN, f64::max);
self.inflight_shortterm = max(self.inflight_shortterm, self.undo_inflight_shortterm);
self.inflight_longterm = max(self.inflight_longterm, self.undo_inflight_longterm);
if self.state != BbrState::ProbeRtt && self.state != self.undo_state {
if self.undo_state == BbrState::Startup {
self.enter_startup();
} else if self.undo_state == BbrState::ProbeBw(ProbeBwSubstate::Up) {
self.start_probe_bw_up();
}
}
}
fn on_mtu_update(&mut self, new_mtu: u16) {
self.smss = min(
max(MIN_MAX_DATAGRAM_SIZE, new_mtu) as u64,
MAX_DATAGRAM_SIZE,
);
self.set_cwnd();
}
fn on_ack_frequency_update(
&mut self,
ack_eliciting_threshold: u64,
requested_max_ack_delay: Duration,
) {
self.ack_eliciting_threshold = ack_eliciting_threshold;
self.max_ack_delay = requested_max_ack_delay;
}
fn window(&self) -> u64 {
self.cwnd
}
fn metrics(&self) -> ControllerMetrics {
ControllerMetrics {
congestion_window: self.window(),
ssthresh: None,
pacing_rate: Some(self.pacing_rate.round() as u64),
send_quantum: Some(self.send_quantum),
}
}
fn clone_box(&self) -> Box<dyn Controller> {
Box::new(self.clone())
}
fn initial_window(&self) -> u64 {
self.initial_cwnd
}
fn into_any(self: Box<Self>) -> Box<dyn Any> {
self
}
}
#[derive(Debug, Clone)]
pub struct Bbr3Config {
initial_window: u64,
probe_rng_seed: Option<[u8; 16]>,
startup_pacing_gain: Option<f64>,
default_pacing_gain: Option<f64>,
probe_bw_down_pacing_gain: Option<f64>,
probe_bw_up_pacing_gain: Option<f64>,
probe_bw_up_cwnd_gain: Option<f64>,
probe_rtt_cwnd_gain: Option<f64>,
drain_pacing_gain: Option<f64>,
pacing_margin_percent: Option<f64>,
default_cwnd_gain: Option<f64>,
}
impl Bbr3Config {
pub fn initial_window(&mut self, value: u64) -> &mut Self {
self.initial_window = value;
self
}
}
impl Default for Bbr3Config {
fn default() -> Self {
Self {
initial_window: 14720.clamp(2 * MAX_DATAGRAM_SIZE, 10 * MAX_DATAGRAM_SIZE),
probe_rng_seed: None,
startup_pacing_gain: None,
default_pacing_gain: None,
probe_bw_down_pacing_gain: None,
probe_bw_up_pacing_gain: None,
probe_bw_up_cwnd_gain: None,
probe_rtt_cwnd_gain: None,
drain_pacing_gain: None,
pacing_margin_percent: None,
default_cwnd_gain: None,
}
}
}
impl ControllerFactory for Bbr3Config {
fn build(self: Arc<Self>, _now: Instant, current_mtu: u16) -> Box<dyn Controller> {
Box::new(Bbr3::new(self, current_mtu))
}
}
#[cfg(test)]
mod test {
use super::*;
#[test]
fn test_probe_rng() {
let seed: [u8; 16] = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
let config = Bbr3Config {
initial_window: 14720.clamp(2 * MAX_DATAGRAM_SIZE, 10 * MAX_DATAGRAM_SIZE),
probe_rng_seed: Some(seed),
startup_pacing_gain: None,
default_pacing_gain: None,
probe_bw_down_pacing_gain: None,
probe_bw_up_pacing_gain: None,
probe_bw_up_cwnd_gain: None,
probe_rtt_cwnd_gain: None,
drain_pacing_gain: None,
pacing_margin_percent: None,
default_cwnd_gain: None,
};
let mut bbr3 = Bbr3::new(Arc::new(config), 2500);
bbr3.pick_probe_wait();
assert_eq!(bbr3.rounds_since_bw_probe, 1);
assert_eq!(bbr3.bw_probe_wait, Duration::from_millis(2652));
bbr3.pick_probe_wait();
assert_eq!(bbr3.rounds_since_bw_probe, 1);
assert_eq!(bbr3.bw_probe_wait, Duration::from_millis(2570));
}
}