use std::cmp;
use std::time::Instant;
use crate::packet;
use crate::recovery;
use crate::recovery::Acked;
use crate::recovery::CongestionControlOps;
use crate::recovery::Recovery;
use crate::recovery::Sent;
pub static RENO: CongestionControlOps = CongestionControlOps {
on_init,
reset,
on_packet_sent,
on_packets_acked,
congestion_event,
collapse_cwnd,
checkpoint,
rollback,
has_custom_pacing,
debug_fmt,
};
pub fn on_init(_r: &mut Recovery) {}
pub fn reset(_r: &mut Recovery) {}
pub fn on_packet_sent(r: &mut Recovery, sent_bytes: usize, _now: Instant) {
r.bytes_in_flight += sent_bytes;
}
fn on_packets_acked(
r: &mut Recovery, packets: &mut Vec<Acked>, epoch: packet::Epoch,
now: Instant,
) {
for pkt in packets.drain(..) {
on_packet_acked(r, &pkt, epoch, now);
}
}
fn on_packet_acked(
r: &mut Recovery, packet: &Acked, epoch: packet::Epoch, now: Instant,
) {
r.bytes_in_flight = r.bytes_in_flight.saturating_sub(packet.size);
if r.in_congestion_recovery(packet.time_sent) {
return;
}
if r.app_limited {
return;
}
if r.congestion_window < r.ssthresh {
r.bytes_acked_sl += packet.size;
if r.hystart.in_css(epoch) {
r.congestion_window += r.hystart.css_cwnd_inc(r.max_datagram_size);
} else {
r.congestion_window += r.max_datagram_size;
}
if r.hystart.on_packet_acked(epoch, packet, r.latest_rtt, now) {
r.ssthresh = r.congestion_window;
}
} else {
r.bytes_acked_ca += packet.size;
if r.bytes_acked_ca >= r.congestion_window {
r.bytes_acked_ca -= r.congestion_window;
r.congestion_window += r.max_datagram_size;
}
}
}
fn congestion_event(
r: &mut Recovery, _lost_bytes: usize, largest_lost_pkt: &Sent,
epoch: packet::Epoch, now: Instant,
) {
let time_sent = largest_lost_pkt.time_sent;
if !r.in_congestion_recovery(time_sent) {
r.congestion_recovery_start_time = Some(now);
r.congestion_window = (r.congestion_window as f64 *
recovery::LOSS_REDUCTION_FACTOR)
as usize;
r.congestion_window = cmp::max(
r.congestion_window,
r.max_datagram_size * recovery::MINIMUM_WINDOW_PACKETS,
);
r.bytes_acked_ca = (r.congestion_window as f64 *
recovery::LOSS_REDUCTION_FACTOR) as usize;
r.ssthresh = r.congestion_window;
if r.hystart.in_css(epoch) {
r.hystart.congestion_event();
}
}
}
pub fn collapse_cwnd(r: &mut Recovery) {
r.congestion_window = r.max_datagram_size * recovery::MINIMUM_WINDOW_PACKETS;
r.bytes_acked_sl = 0;
r.bytes_acked_ca = 0;
if r.hystart.enabled() {
r.hystart.reset();
}
}
fn checkpoint(_r: &mut Recovery) {}
fn rollback(_r: &mut Recovery) -> bool {
true
}
fn has_custom_pacing() -> bool {
false
}
fn debug_fmt(_r: &Recovery, _f: &mut std::fmt::Formatter) -> std::fmt::Result {
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use smallvec::smallvec;
use std::time::Duration;
#[test]
fn reno_init() {
let mut cfg = crate::Config::new(crate::PROTOCOL_VERSION).unwrap();
cfg.set_cc_algorithm(recovery::CongestionControlAlgorithm::Reno);
let r = Recovery::new(&cfg);
assert!(r.cwnd() > 0);
assert_eq!(r.bytes_in_flight, 0);
}
#[test]
fn reno_send() {
let mut cfg = crate::Config::new(crate::PROTOCOL_VERSION).unwrap();
cfg.set_cc_algorithm(recovery::CongestionControlAlgorithm::Reno);
let mut r = Recovery::new(&cfg);
let now = Instant::now();
r.on_packet_sent_cc(1000, now);
assert_eq!(r.bytes_in_flight, 1000);
}
#[test]
fn reno_slow_start() {
let mut cfg = crate::Config::new(crate::PROTOCOL_VERSION).unwrap();
cfg.set_cc_algorithm(recovery::CongestionControlAlgorithm::Reno);
let mut r = Recovery::new(&cfg);
let now = Instant::now();
let p = recovery::Sent {
pkt_num: 0,
frames: smallvec![],
time_sent: now,
time_acked: None,
time_lost: None,
size: r.max_datagram_size,
ack_eliciting: true,
in_flight: true,
delivered: 0,
delivered_time: std::time::Instant::now(),
first_sent_time: std::time::Instant::now(),
is_app_limited: false,
tx_in_flight: 0,
lost: 0,
has_data: false,
};
for _ in 0..r.initial_congestion_window_packets {
r.on_packet_sent_cc(p.size, now);
}
let cwnd_prev = r.cwnd();
let mut acked = vec![Acked {
pkt_num: p.pkt_num,
time_sent: p.time_sent,
size: p.size,
delivered: 0,
delivered_time: now,
first_sent_time: now,
is_app_limited: false,
tx_in_flight: 0,
lost: 0,
rtt: Duration::ZERO,
}];
r.on_packets_acked(&mut acked, packet::Epoch::Application, now);
assert_eq!(r.cwnd(), cwnd_prev + p.size);
}
#[test]
fn reno_slow_start_multi_acks() {
let mut cfg = crate::Config::new(crate::PROTOCOL_VERSION).unwrap();
cfg.set_cc_algorithm(recovery::CongestionControlAlgorithm::Reno);
let mut r = Recovery::new(&cfg);
let now = Instant::now();
let p = recovery::Sent {
pkt_num: 0,
frames: smallvec![],
time_sent: now,
time_acked: None,
time_lost: None,
size: r.max_datagram_size,
ack_eliciting: true,
in_flight: true,
delivered: 0,
delivered_time: std::time::Instant::now(),
first_sent_time: std::time::Instant::now(),
is_app_limited: false,
tx_in_flight: 0,
lost: 0,
has_data: false,
};
for _ in 0..r.initial_congestion_window_packets {
r.on_packet_sent_cc(p.size, now);
}
let cwnd_prev = r.cwnd();
let mut acked = vec![
Acked {
pkt_num: p.pkt_num,
time_sent: p.time_sent,
size: p.size,
delivered: 0,
delivered_time: now,
first_sent_time: now,
is_app_limited: false,
tx_in_flight: 0,
lost: 0,
rtt: Duration::ZERO,
},
Acked {
pkt_num: p.pkt_num,
time_sent: p.time_sent,
size: p.size,
delivered: 0,
delivered_time: now,
first_sent_time: now,
is_app_limited: false,
tx_in_flight: 0,
lost: 0,
rtt: Duration::ZERO,
},
Acked {
pkt_num: p.pkt_num,
time_sent: p.time_sent,
size: p.size,
delivered: 0,
delivered_time: now,
first_sent_time: now,
is_app_limited: false,
tx_in_flight: 0,
lost: 0,
rtt: Duration::ZERO,
},
];
r.on_packets_acked(&mut acked, packet::Epoch::Application, now);
assert_eq!(r.cwnd(), cwnd_prev + p.size * 3);
}
#[test]
fn reno_congestion_event() {
let mut cfg = crate::Config::new(crate::PROTOCOL_VERSION).unwrap();
cfg.set_cc_algorithm(recovery::CongestionControlAlgorithm::Reno);
let mut r = Recovery::new(&cfg);
let prev_cwnd = r.cwnd();
let now = Instant::now();
let p = recovery::Sent {
pkt_num: 0,
frames: smallvec![],
time_sent: now,
time_acked: None,
time_lost: None,
size: r.max_datagram_size,
ack_eliciting: true,
in_flight: true,
delivered: 0,
delivered_time: std::time::Instant::now(),
first_sent_time: std::time::Instant::now(),
is_app_limited: false,
has_data: false,
tx_in_flight: 0,
lost: 0,
};
r.congestion_event(
r.max_datagram_size,
&p,
packet::Epoch::Application,
now,
);
assert_eq!(prev_cwnd / 2, r.cwnd());
}
#[test]
fn reno_congestion_avoidance() {
let mut cfg = crate::Config::new(crate::PROTOCOL_VERSION).unwrap();
cfg.set_cc_algorithm(recovery::CongestionControlAlgorithm::Reno);
let mut r = Recovery::new(&cfg);
let now = Instant::now();
let prev_cwnd = r.cwnd();
r.on_packet_sent_cc(20000, now);
let p = recovery::Sent {
pkt_num: 0,
frames: smallvec![],
time_sent: now,
time_acked: None,
time_lost: None,
size: r.max_datagram_size,
ack_eliciting: true,
in_flight: true,
delivered: 0,
delivered_time: std::time::Instant::now(),
first_sent_time: std::time::Instant::now(),
is_app_limited: false,
has_data: false,
tx_in_flight: 0,
lost: 0,
};
r.congestion_event(
r.max_datagram_size,
&p,
packet::Epoch::Application,
now,
);
let cur_cwnd =
(prev_cwnd as f64 * recovery::LOSS_REDUCTION_FACTOR) as usize;
assert_eq!(r.cwnd(), cur_cwnd);
let rtt = Duration::from_millis(100);
let mut acked = vec![Acked {
pkt_num: 0,
time_sent: now + rtt,
size: 8000,
delivered: 0,
delivered_time: now,
first_sent_time: now,
is_app_limited: false,
tx_in_flight: 0,
lost: 0,
rtt: Duration::ZERO,
}];
r.update_rtt(rtt, Duration::from_millis(0), now);
r.on_packets_acked(&mut acked, packet::Epoch::Application, now + rtt * 2);
assert_eq!(r.cwnd(), cur_cwnd + r.max_datagram_size);
}
}