use core::time::Duration;
use crate::arq::FULL_ACK_PERIOD;
use crate::arq::rtt::INITIAL_RTT;
use crate::arq::seq::{seq_diff, seq_gt};
const RC_INTERVAL: Duration = FULL_ACK_PERIOD;
const S_BYTES: f64 = 1500.0;
const SLOW_START_PKT_SND_PERIOD_US: f64 = 1.0;
const INITIAL_CWND_SIZE: f64 = 16.0;
const INITIAL_LAST_DEC_PERIOD_US: f64 = 1.0;
const DEFAULT_MAX_CWND_SIZE: f64 = 8_000.0;
const LOSS_RATIO_TOLERANCE: f64 = 0.02;
const RATE_BACKOFF_FACTOR: f64 = 1.03;
const AVG_NAK_NUM_OLD_WEIGHT: f64 = 0.97;
const AVG_NAK_NUM_NEW_WEIGHT: f64 = 0.03;
const MAX_DEC_COUNT: u32 = 5;
const INC_SCALE: f64 = 0.0000015;
const US_PER_SEC: f64 = 1_000_000.0;
const LOSS_BANDWIDTH_FACTOR: f64 = 2.0;
const LINK_CAPACITY_CLAMP_DIVISOR: f64 = 9.0;
const BITS_PER_BYTE: f64 = 8.0;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Phase {
SlowStart,
CongestionAvoidance,
}
impl Phase {
pub fn name(&self) -> &'static str {
match self {
Phase::SlowStart => "SlowStart",
Phase::CongestionAvoidance => "CongestionAvoidance",
}
}
}
broadcast_common::impl_spec_display!(Phase);
#[derive(Debug, Clone, Copy)]
struct XorShift64(u64);
impl XorShift64 {
const DEFAULT_SEED: u64 = 0x9E37_79B9_7F4A_7C15;
fn new(seed: u64) -> Self {
XorShift64(if seed == 0 { Self::DEFAULT_SEED } else { seed })
}
fn next_unit(&mut self) -> f64 {
let mut x = self.0;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
self.0 = x;
let r = x.wrapping_mul(0x2545_F491_4F6C_DD1D);
(r >> 11) as f64 * (1.0 / (1u64 << 53) as f64)
}
}
fn fmin(a: f64, b: f64) -> f64 {
if a < b { a } else { b }
}
fn fmax(a: f64, b: f64) -> f64 {
if a > b { a } else { b }
}
fn fround(x: f64) -> f64 {
let truncated = x as i64 as f64;
let diff = x - truncated;
if diff >= 0.5 {
truncated + 1.0
} else if diff <= -0.5 {
truncated - 1.0
} else {
truncated
}
}
fn next_power_of_10(x: f64) -> f64 {
debug_assert!(x > 0.0, "next_power_of_10 is only defined for x > 0");
let mut p = 1.0_f64;
while p < x {
p *= 10.0;
}
while p / 10.0 >= x {
p /= 10.0;
}
p
}
#[derive(Debug, Clone)]
pub struct FileCc {
phase: Phase,
cwnd_size: f64,
max_cwnd_size: f64,
pkt_snd_period_us: f64,
last_rc_time: Option<Duration>,
last_ack_seqno: u32,
b_loss: bool,
last_dec_period_us: f64,
last_dec_seq: Option<u32>,
avg_nak_num: f64,
nak_count: u32,
dec_count: u32,
dec_random: f64,
receiving_rate_pps: u64,
est_link_capacity_pps: u64,
rtt: Duration,
max_bw_bytes_per_sec: Option<u64>,
rng: XorShift64,
}
impl FileCc {
pub fn new(initial_seqno: u32) -> Self {
FileCc {
phase: Phase::SlowStart,
cwnd_size: INITIAL_CWND_SIZE,
max_cwnd_size: DEFAULT_MAX_CWND_SIZE,
pkt_snd_period_us: SLOW_START_PKT_SND_PERIOD_US,
last_rc_time: None,
last_ack_seqno: initial_seqno,
b_loss: false,
last_dec_period_us: INITIAL_LAST_DEC_PERIOD_US,
last_dec_seq: None,
avg_nak_num: 0.0,
nak_count: 0,
dec_count: 0,
dec_random: 1.0,
receiving_rate_pps: 0,
est_link_capacity_pps: 0,
rtt: INITIAL_RTT,
max_bw_bytes_per_sec: None,
rng: XorShift64::new(XorShift64::DEFAULT_SEED),
}
}
pub fn phase(&self) -> Phase {
self.phase
}
pub fn cwnd_size(&self) -> f64 {
self.cwnd_size
}
pub fn pkt_snd_period_us(&self) -> f64 {
self.pkt_snd_period_us
}
pub fn pkt_snd_period(&self) -> Duration {
Duration::from_micros(self.pkt_snd_period_us as u64)
}
pub fn max_cwnd_size(&self) -> f64 {
self.max_cwnd_size
}
pub fn set_max_cwnd_size(&mut self, packets: f64) {
self.max_cwnd_size = packets;
}
pub fn max_bw_bytes_per_sec(&self) -> Option<u64> {
self.max_bw_bytes_per_sec
}
pub fn set_max_bw_bytes_per_sec(&mut self, max_bw: Option<u64>) {
self.max_bw_bytes_per_sec = max_bw;
}
pub fn receiving_rate_pps(&self) -> u64 {
self.receiving_rate_pps
}
pub fn est_link_capacity_pps(&self) -> u64 {
self.est_link_capacity_pps
}
pub fn rtt(&self) -> Duration {
self.rtt
}
pub fn b_loss(&self) -> bool {
self.b_loss
}
pub fn avg_nak_num(&self) -> f64 {
self.avg_nak_num
}
pub fn nak_count(&self) -> u32 {
self.nak_count
}
pub fn dec_count(&self) -> u32 {
self.dec_count
}
pub fn last_dec_seq(&self) -> Option<u32> {
self.last_dec_seq
}
pub fn last_dec_period_us(&self) -> f64 {
self.last_dec_period_us
}
pub fn on_ack(
&mut self,
now: Duration,
ack_seqno: u32,
receiving_rate_pps: u64,
est_link_capacity_pps: u64,
rtt: Duration,
) {
self.receiving_rate_pps = receiving_rate_pps;
self.est_link_capacity_pps = est_link_capacity_pps;
self.rtt = rtt;
if let Some(last) = self.last_rc_time
&& now.saturating_sub(last) < RC_INTERVAL
{
return;
}
self.last_rc_time = Some(now);
match self.phase {
Phase::SlowStart => self.on_ack_slow_start(ack_seqno),
Phase::CongestionAvoidance => self.on_ack_congestion_avoidance(),
}
}
fn on_ack_slow_start(&mut self, ack_seqno: u32) {
let delta = seq_diff(ack_seqno, self.last_ack_seqno);
self.cwnd_size += f64::from(delta);
self.last_ack_seqno = ack_seqno;
if self.cwnd_size > self.max_cwnd_size {
self.end_slow_start();
}
}
fn on_ack_congestion_avoidance(&mut self) {
self.cwnd_size = self.receiving_rate_pps as f64 * self.rtt_plus_rc_interval_us()
/ US_PER_SEC
+ INITIAL_CWND_SIZE;
if self.b_loss {
self.b_loss = false;
return;
}
let loss_bandwidth = LOSS_BANDWIDTH_FACTOR * (US_PER_SEC / self.last_dec_period_us);
let link_capacity = fmin(loss_bandwidth, self.est_link_capacity_pps as f64);
let mut b = link_capacity - US_PER_SEC / self.pkt_snd_period_us;
if self.pkt_snd_period_us > self.last_dec_period_us
&& (link_capacity / LINK_CAPACITY_CLAMP_DIVISOR) < b
{
b = link_capacity / LINK_CAPACITY_CLAMP_DIVISOR;
}
let inc = if b <= 0.0 {
1.0 / S_BYTES
} else {
let raw = next_power_of_10(b * S_BYTES * BITS_PER_BYTE) * INC_SCALE / S_BYTES;
fmax(raw, 1.0 / S_BYTES)
};
let rc_interval_us = RC_INTERVAL.as_micros() as f64;
self.pkt_snd_period_us = (self.pkt_snd_period_us * rc_interval_us)
/ (self.pkt_snd_period_us * inc + rc_interval_us);
if let Some(max_bw) = self.max_bw_bytes_per_sec {
let min_period_us = US_PER_SEC / (max_bw as f64 / S_BYTES);
if self.pkt_snd_period_us < min_period_us {
self.pkt_snd_period_us = min_period_us;
}
}
}
fn rtt_plus_rc_interval_us(&self) -> f64 {
self.rtt.as_micros() as f64 + RC_INTERVAL.as_micros() as f64
}
fn end_slow_start(&mut self) {
self.phase = Phase::CongestionAvoidance;
self.pkt_snd_period_us = if self.receiving_rate_pps > 0 {
US_PER_SEC / self.receiving_rate_pps as f64
} else {
self.cwnd_size / self.rtt_plus_rc_interval_us()
};
}
pub fn on_loss(&mut self, lost_seqno: u32, largest_sent_seqno: u32, loss_ratio: f64) {
match self.phase {
Phase::SlowStart => self.end_slow_start(),
Phase::CongestionAvoidance => {
self.on_loss_congestion_avoidance(lost_seqno, largest_sent_seqno, loss_ratio)
}
}
}
fn on_loss_congestion_avoidance(
&mut self,
lost_seqno: u32,
largest_sent_seqno: u32,
loss_ratio: f64,
) {
self.b_loss = true;
if loss_ratio < LOSS_RATIO_TOLERANCE {
self.last_dec_period_us = self.pkt_snd_period_us;
return;
}
let is_new_period = match self.last_dec_seq {
None => true,
Some(last) => seq_gt(lost_seqno, last),
};
if is_new_period {
self.last_dec_period_us = self.pkt_snd_period_us;
self.pkt_snd_period_us *= RATE_BACKOFF_FACTOR;
self.avg_nak_num = AVG_NAK_NUM_OLD_WEIGHT * self.avg_nak_num
+ AVG_NAK_NUM_NEW_WEIGHT * self.nak_count as f64;
self.nak_count = 1;
self.dec_count = 1;
self.last_dec_seq = Some(largest_sent_seqno);
self.dec_random = self.next_dec_random();
return;
}
if self.dec_count <= MAX_DEC_COUNT
&& self.nak_count as f64 == self.dec_count as f64 * self.dec_random
{
self.pkt_snd_period_us *= RATE_BACKOFF_FACTOR;
self.dec_count += 1;
self.nak_count += 1;
self.last_dec_seq = Some(largest_sent_seqno);
}
}
fn next_dec_random(&mut self) -> f64 {
let hi = fmax(self.avg_nak_num, 1.0);
let r = self.rng.next_unit();
let val = fround(1.0 + r * (hi - 1.0));
fmax(val, 1.0)
}
pub fn on_timeout(&mut self) {
if self.phase == Phase::SlowStart {
self.end_slow_start();
}
}
pub fn tick(&mut self, _now: Duration) {}
}
impl Default for FileCc {
fn default() -> Self {
FileCc::new(0)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn starts_in_slow_start_with_spec_initial_values() {
let cc = FileCc::new(0);
assert_eq!(cc.phase(), Phase::SlowStart);
assert_eq!(cc.cwnd_size(), 16.0);
assert_eq!(cc.pkt_snd_period_us(), 1.0);
assert_eq!(cc.max_cwnd_size(), 8_000.0);
assert!(!cc.b_loss());
assert_eq!(cc.avg_nak_num(), 0.0);
assert_eq!(cc.nak_count(), 0);
assert_eq!(cc.dec_count(), 0);
assert_eq!(cc.last_dec_seq(), None);
assert_eq!(cc.last_dec_period_us(), 1.0);
assert_eq!(cc.rtt(), Duration::from_millis(100));
assert_eq!(cc.max_bw_bytes_per_sec(), None);
}
#[test]
fn slow_start_cwnd_grows_by_ack_seqno_delta() {
let mut cc = FileCc::new(0);
cc.on_ack(
Duration::from_millis(10),
5,
0,
0,
Duration::from_millis(100),
);
assert_eq!(cc.cwnd_size(), 21.0); assert_eq!(
cc.pkt_snd_period_us(),
1.0,
"fixed at 1us during slow start"
);
}
#[test]
fn rate_control_gate_blocks_updates_within_rc_interval() {
let mut cc = FileCc::new(0);
cc.on_ack(
Duration::from_millis(10),
5,
0,
0,
Duration::from_millis(100),
);
assert_eq!(cc.cwnd_size(), 21.0);
cc.on_ack(
Duration::from_millis(11),
999,
0,
0,
Duration::from_millis(100),
);
assert_eq!(
cc.cwnd_size(),
21.0,
"gate must block an ACK inside RC_INTERVAL"
);
}
#[test]
fn loss_during_slow_start_transitions_to_congestion_avoidance() {
let mut cc = FileCc::new(0);
assert_eq!(cc.phase(), Phase::SlowStart);
cc.on_loss(10, 10, 0.9);
assert_eq!(cc.phase(), Phase::CongestionAvoidance);
}
#[test]
fn timeout_during_slow_start_transitions_to_congestion_avoidance() {
let mut cc = FileCc::new(0);
cc.on_timeout();
assert_eq!(cc.phase(), Phase::CongestionAvoidance);
}
#[test]
fn timeout_during_congestion_avoidance_is_a_no_op() {
let mut cc = FileCc::new(0);
cc.on_loss(10, 10, 0.9);
assert_eq!(cc.phase(), Phase::CongestionAvoidance);
let period = cc.pkt_snd_period_us();
cc.on_timeout();
assert_eq!(cc.pkt_snd_period_us(), period);
}
#[test]
fn cwnd_exceeding_max_ends_slow_start() {
let mut cc = FileCc::new(0);
cc.set_max_cwnd_size(20.0);
cc.on_ack(
Duration::from_millis(10),
10,
0,
0,
Duration::from_millis(100),
);
assert_eq!(cc.phase(), Phase::CongestionAvoidance);
}
#[test]
fn max_bw_clamp_floors_pkt_snd_period() {
let mut cc = FileCc::new(0);
cc.on_loss(1, 1, 0.9); cc.set_max_bw_bytes_per_sec(Some(1)); cc.on_ack(
Duration::from_millis(10),
1,
1_000,
1_000,
Duration::from_millis(50),
);
assert!(cc.pkt_snd_period_us() >= 1_500_000_000.0);
}
#[test]
fn next_power_of_10_matches_known_values() {
assert_eq!(next_power_of_10(1.0), 1.0);
assert_eq!(next_power_of_10(9.9), 10.0);
assert_eq!(next_power_of_10(10.0), 10.0);
assert_eq!(next_power_of_10(10.1), 100.0);
assert_eq!(next_power_of_10(100.0), 100.0);
assert_eq!(next_power_of_10(0.05), 0.1);
}
#[test]
fn xorshift_produces_values_in_unit_range() {
let mut rng = XorShift64::new(1);
for _ in 0..100 {
let v = rng.next_unit();
assert!((0.0..1.0).contains(&v), "value out of range: {v}");
}
}
#[test]
fn dec_random_is_one_when_avg_nak_num_is_zero() {
let mut cc = FileCc::new(0);
assert_eq!(cc.avg_nak_num(), 0.0);
assert_eq!(cc.next_dec_random(), 1.0);
}
#[test]
fn dec_random_is_integer_valued_once_avg_nak_num_exceeds_one() {
let mut cc = FileCc::new(0);
cc.avg_nak_num = 6.0;
for _ in 0..200 {
let v = cc.next_dec_random();
assert_eq!(
v,
v.round(),
"DecRandom must be a whole number (got {v}) so Step 4's \
NAKCount == DecCount * DecRandom gate is actually satisfiable"
);
assert!((1.0..=6.0).contains(&v), "DecRandom out of range: {v}");
}
}
#[test]
fn repeat_decrease_is_a_one_shot_per_period_once_dec_random_exceeds_one() {
let mut cc = FileCc::new(0);
cc.on_loss(10, 10, 0.9); cc.on_loss(20, 20, 0.9); assert_eq!(cc.dec_count(), 1);
cc.dec_random = 3.0;
let after_first = cc.pkt_snd_period_us();
for sent in 21..=60u32 {
cc.on_loss(15, sent, 0.9);
}
assert_eq!(
cc.dec_count(),
1,
"with the literal spec's Step 4 pseudocode, DecCount must stay \
frozen at 1 for the rest of the period once the immediate \
post-reset check (NAKCount==DecCount*DecRandom, i.e. 1==1*3) \
fails — this is the draft's own one-shot property, not a bug"
);
assert_eq!(
cc.pkt_snd_period_us(),
after_first,
"PKT_SND_PERIOD must not change further once Step 4 has gone \
silent for the rest of the period"
);
}
#[test]
fn repeat_decrease_fires_repeatedly_when_dec_random_rounds_to_one() {
let mut cc = FileCc::new(0);
cc.avg_nak_num = 1.2; cc.on_loss(10, 10, 0.9);
cc.on_loss(20, 20, 0.9);
assert_eq!(cc.dec_random, 1.0);
assert_eq!(cc.dec_count(), 1);
for sent in 21..=25u32 {
cc.on_loss(15, sent, 0.9);
}
assert_eq!(
cc.dec_count(),
6,
"DecRandom==1 must let Step 4 fire on every same-period NAK"
);
}
#[test]
fn default_matches_new_zero() {
let a = FileCc::default();
let b = FileCc::new(0);
assert_eq!(a.cwnd_size(), b.cwnd_size());
assert_eq!(a.phase(), b.phase());
}
}