diff --git a/src/bin/rust-bfdd.rs b/src/bin/rust-bfdd.rs index c29dce0..28870ba 100644 --- a/src/bin/rust-bfdd.rs +++ b/src/bin/rust-bfdd.rs @@ -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, 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, 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 = (0..SAMPLES) + .map(|_| jittered_delay(interval, mult).as_secs_f64() * 1000.0) + .collect(); + let mean = delays.iter().sum::() / 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([