Fix periodic transmit jitter to follow RFC 5880 6.8.7
Jitter was added to a fixed 0.75x clock, so gaps ranged 0.5x-1.0x with a mean of 0.75x of the negotiated interval, below the required 75-100%. Each delay is now drawn independently from 75-100% of the interval (90-100% when DetectMult is 1). Adds statistical and end-to-end sender tests.
This commit is contained in:
ktims committed
1 parent
5e5da0c78c
commit
0c19b955f9
1 file changed
+104
-22
+104
-22
@@ -348,9 +348,9 @@ struct BfdPeriodicSenderHandle {
|
||||
}
|
||||
|
||||
impl BfdPeriodicSenderHandle {
|
||||
fn new(bfd: BfdSessionHandle, interval: BfdInterval) -> Self {
|
||||
fn new(bfd: BfdSessionHandle, interval: BfdInterval, detect_mult: u8) -> Self {
|
||||
let (tx, rx) = mpsc::channel(32);
|
||||
let mut periodic_sender = BfdPeriodicSender::new(rx, bfd, interval);
|
||||
let mut periodic_sender = BfdPeriodicSender::new(rx, bfd, interval, detect_mult);
|
||||
tokio::spawn(async move { periodic_sender.run().await });
|
||||
|
||||
Self { tx }
|
||||
@@ -361,6 +361,13 @@ struct BfdPeriodicSender {
|
||||
rx: mpsc::Receiver<PeriodicControlCommand>,
|
||||
bfd: BfdSessionHandle,
|
||||
cur_interval: BfdInterval,
|
||||
detect_mult: u8,
|
||||
}
|
||||
|
||||
fn jittered_delay(interval: time::Duration, detect_mult: u8) -> time::Duration {
|
||||
// https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.7
|
||||
let min_percent = if detect_mult == 1 { 90 } else { 75 };
|
||||
rand::thread_rng().gen_range(interval * min_percent / 100..=interval)
|
||||
}
|
||||
|
||||
impl BfdPeriodicSender {
|
||||
@@ -368,30 +375,28 @@ impl BfdPeriodicSender {
|
||||
rx: mpsc::Receiver<PeriodicControlCommand>,
|
||||
bfd: BfdSessionHandle,
|
||||
interval: BfdInterval,
|
||||
detect_mult: u8,
|
||||
) -> Self {
|
||||
Self {
|
||||
rx,
|
||||
bfd,
|
||||
cur_interval: interval,
|
||||
detect_mult,
|
||||
}
|
||||
}
|
||||
|
||||
fn next_delay(&self) -> time::Duration {
|
||||
jittered_delay(self.cur_interval.into(), self.detect_mult)
|
||||
}
|
||||
async fn run(&mut self) {
|
||||
let base_interval = self.cur_interval * 3 / 4;
|
||||
let mut clock = time::interval(base_interval.into());
|
||||
let mut running = true;
|
||||
let mut next_tx = time::Instant::now() + self.next_delay();
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = clock.tick(), if running => {
|
||||
let jitter = time::Duration::from_micros(
|
||||
rand::thread_rng()
|
||||
.gen_range(0..clock.period().as_micros() / 3)
|
||||
.try_into()
|
||||
.unwrap(),
|
||||
);
|
||||
time::sleep(jitter).await;
|
||||
// debug!("tick! sent periodic control packet after {}ms", (base_interval + jitter).as_millis());
|
||||
_ = time::sleep_until(next_tx), if running => {
|
||||
self.bfd.transmit_control_packet(false).await.unwrap();
|
||||
next_tx = time::Instant::now() + self.next_delay();
|
||||
},
|
||||
Some(cmd) = self.rx.recv() => {
|
||||
match cmd {
|
||||
@@ -400,14 +405,15 @@ impl BfdPeriodicSender {
|
||||
debug!("Stopping periodic packets");
|
||||
running= false}
|
||||
,
|
||||
PeriodicControlCommand::Start => { debug!("Starting periodic packets"); running = true },
|
||||
PeriodicControlCommand::Start => {
|
||||
debug!("Starting periodic packets");
|
||||
running = true;
|
||||
next_tx = time::Instant::now() + self.next_delay();
|
||||
},
|
||||
PeriodicControlCommand::SetMinInterval(i) => {
|
||||
let base_interval: Duration = (i * 3 / 4).into();
|
||||
debug!("Updating base interval to {}ms (jittering {} to {}ms)", base_interval.as_millis(), base_interval.as_millis(), base_interval.as_millis() + base_interval.as_millis() / 3);
|
||||
clock = time::interval_at(
|
||||
time::Instant::now() + base_interval,
|
||||
base_interval,
|
||||
);
|
||||
self.cur_interval = i;
|
||||
debug!("Updating transmit interval to {}ms", Duration::from(i).as_millis());
|
||||
next_tx = time::Instant::now() + self.next_delay();
|
||||
},
|
||||
PeriodicControlCommand::SendPollResponse => {
|
||||
debug!("Sending poll response");
|
||||
@@ -431,8 +437,11 @@ impl BfdSession {
|
||||
event_dispatcher: EventDispatcherHandle,
|
||||
) -> Self {
|
||||
let mut rng = rand::thread_rng();
|
||||
let periodic =
|
||||
BfdPeriodicSenderHandle::new(handle.clone(), time::Duration::from_secs(1).into());
|
||||
let periodic = BfdPeriodicSenderHandle::new(
|
||||
handle.clone(),
|
||||
time::Duration::from_secs(1).into(),
|
||||
peer.detect_mult,
|
||||
);
|
||||
let configured_desired_min_tx = peer.desired_min_tx;
|
||||
let slow_desired_min_tx =
|
||||
std::cmp::max(configured_desired_min_tx, BfdInterval::from_secs(1));
|
||||
@@ -1188,6 +1197,79 @@ async fn main() -> io::Result<()> {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn jitter_is_uniform_over_rfc_range() {
|
||||
// RFC 5880 6.8.7: each interval is reduced by a random 0-25% (0-10% when DetectMult is 1), so delays are
|
||||
// spread over [floor, interval] with a mean at the midpoint, not clustered at either end.
|
||||
let interval = time::Duration::from_millis(300);
|
||||
const SAMPLES: u32 = 20_000;
|
||||
for (mult, floor_ms, mean_ms) in [(3u8, 225.0, 262.5), (1, 270.0, 285.0)] {
|
||||
let delays: Vec<f64> = (0..SAMPLES)
|
||||
.map(|_| jittered_delay(interval, mult).as_secs_f64() * 1000.0)
|
||||
.collect();
|
||||
let mean = delays.iter().sum::<f64>() / f64::from(SAMPLES);
|
||||
let min = delays.iter().copied().fold(f64::MAX, f64::min);
|
||||
let max = delays.iter().copied().fold(f64::MIN, f64::max);
|
||||
let width = 300.0 - floor_ms;
|
||||
assert!(
|
||||
(mean - mean_ms).abs() < 1.0,
|
||||
"mult {mult}: mean {mean}ms, expected {mean_ms}ms"
|
||||
);
|
||||
assert!(
|
||||
min < floor_ms + width * 0.02,
|
||||
"mult {mult}: never approached the floor (min {min}ms)"
|
||||
);
|
||||
assert!(
|
||||
max > 300.0 - width * 0.02,
|
||||
"mult {mult}: never approached the interval (max {max}ms)"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn periodic_sender_gaps_respect_rfc_floor() {
|
||||
// Gaps are measured between actual sends, so jitter must not be layered onto a shorter fixed clock.
|
||||
// Only the lower bound is asserted: scheduler overshoot can legitimately stretch a gap.
|
||||
let interval = time::Duration::from_millis(40);
|
||||
let floor = interval * 3 / 4;
|
||||
let (tx, mut rx) = mpsc::channel(32);
|
||||
let sender = BfdPeriodicSenderHandle::new(BfdSessionHandle { tx }, interval.into(), 3);
|
||||
|
||||
let mut sends = Vec::new();
|
||||
while sends.len() < 40 {
|
||||
match rx.recv().await {
|
||||
Some(SessionControlCommand::TxControlPacket {
|
||||
poll_response: false,
|
||||
}) => sends.push(time::Instant::now()),
|
||||
other => panic!("unexpected command (channel open: {})", other.is_some()),
|
||||
}
|
||||
}
|
||||
sender.tx.send(PeriodicControlCommand::Quit).await.unwrap();
|
||||
|
||||
for pair in sends.windows(2) {
|
||||
let gap = pair[1] - pair[0];
|
||||
assert!(
|
||||
gap + time::Duration::from_millis(2) >= floor,
|
||||
"gap {gap:?} below {floor:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jitter_within_rfc_bounds() {
|
||||
let interval = time::Duration::from_millis(300);
|
||||
for (mult, floor_ms) in [(3, 225), (1, 270)] {
|
||||
for _ in 0..1000 {
|
||||
let d = jittered_delay(interval, mult);
|
||||
assert!(
|
||||
d >= time::Duration::from_millis(floor_ms),
|
||||
"{d:?} below floor for mult {mult}"
|
||||
);
|
||||
assert!(d <= interval, "{d:?} above interval");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repeated_bind_and_peer_flags() {
|
||||
let args = Args::try_parse_from([
|
||||
|
||||
Reference in new issue
Block a user