From 8fbafa455abd3b7188bb1cf2532d087db3f139e8 Mon Sep 17 00:00:00 2001 From: Keenan Tims Date: Thu, 22 Oct 2026 18:14:05 -0700 Subject: [PATCH] per-peer stats --- src/bin/rust-bfdd.rs | 42 +++++--- src/lib.rs | 239 +++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 262 insertions(+), 19 deletions(-) diff --git a/src/bin/rust-bfdd.rs b/src/bin/rust-bfdd.rs index 90f7018..7b478df 100644 --- a/src/bin/rust-bfdd.rs +++ b/src/bin/rust-bfdd.rs @@ -3,7 +3,7 @@ use futures::future; use rust_bfd::config::{Config, ConfigError, ResolvedConfig, ResolvedPeer}; use rust_bfd::events::{EventMessageSink, ScriptHookSink, StateChangeEvent}; use rust_bfd::{bfd::*, events::EventMessage}; -use rust_bfd::{set_ttl_or_hop_limit, BfdSessionStats}; +use rust_bfd::{set_ttl_or_hop_limit, AtomicInstant, AtomicIntervalStats, BfdSessionStats}; use std::{ collections::{HashMap, HashSet}, error::Error, @@ -124,8 +124,10 @@ struct BfdSessionState { detection_time: Atomic, // mutable by rx thread poll_mode: Atomic, // mutable by rx thread - control_packets_rx: Atomic, - control_packets_tx: Atomic, + rx_stats: AtomicIntervalStats, // reset each time the session enters Up + tx_stats: AtomicIntervalStats, // reset each time the session enters Up + last_up: AtomicInstant, + last_down: AtomicInstant, last_state_change: Mutex, last_remote_diag: Atomic, } @@ -221,11 +223,19 @@ impl BfdSessionState { fn set_detection_time(&self, value: BfdInterval) { self.detection_time.store(value, ORDERING) } - fn control_packets_rx(&self) -> u64 { - self.control_packets_rx.load(ORDERING) + fn record_rx(&self) { + self.rx_stats.record(Instant::now()); } - fn control_packets_tx(&self) -> u64 { - self.control_packets_tx.load(ORDERING) + fn record_tx(&self) { + self.tx_stats.record(Instant::now()); + } + fn mark_up(&self) { + self.rx_stats.reset(); + self.tx_stats.reset(); + self.last_up.store(Instant::now()); + } + fn mark_down(&self) { + self.last_down.store(Instant::now()); } fn last_state_change(&self) -> Instant { *self.last_state_change.try_lock().unwrap() @@ -266,8 +276,10 @@ impl From<&BfdSessionState> for BfdSessionStats { remote_discr: state.remote_discr(), state: state.session_state(), last_diag: state.last_remote_diag(), - control_packets_rx: state.control_packets_rx(), - control_packets_tx: state.control_packets_tx(), + rx: state.rx_stats.snapshot(), + tx: state.tx_stats.snapshot(), + last_up: state.last_up.load(), + last_down: state.last_down.load(), last_change: state.last_state_change(), detect_time: state.detection_time(), base_interval: BfdInterval::from_secs(0), // TODO: This is a hack and will be patched up in the session @@ -483,8 +495,10 @@ impl BfdSession { detection_time: Atomic::new(BfdInterval::from_secs(0)), poll_mode: Atomic::new(false), - control_packets_rx: Atomic::new(0), - control_packets_tx: Atomic::new(0), + rx_stats: AtomicIntervalStats::new(), + tx_stats: AtomicIntervalStats::new(), + last_up: AtomicInstant::new(), + last_down: AtomicInstant::new(), last_state_change: Mutex::new(Instant::now()), last_remote_diag: Atomic::new(BfdDiagnostic::None), }, @@ -514,6 +528,7 @@ impl BfdSession { *self.state.last_state_change.lock().await = Instant::now(); match (self.state.session_state(), new_state) { (BfdState::Up, BfdState::AdminDown | BfdState::Down | BfdState::Init) => { + self.state.mark_down(); if let Some(watchdog) = self.rx_watchdog.as_mut() { watchdog.send(WatchdogReset::Terminate).await.unwrap(); }; @@ -530,6 +545,7 @@ impl BfdSession { self.update_transmit_interval().await; } (BfdState::AdminDown | BfdState::Down | BfdState::Init, BfdState::Up) => { + self.state.mark_up(); self.start_watchdog().await; self.state.set_local_diag(BfdDiagnostic::None); // This doesn't seem to be required by the spec but seems sensible and at least FRR does it. // Advertise the configured interval now, but per section 6.8.3 keep transmitting at the old (slower) @@ -601,7 +617,7 @@ impl BfdSession { }; let socket = self.state.control_sock.clone(); let dest = self.state.peer_addr; - self.state.control_packets_tx.fetch_add(1, ORDERING); + self.state.record_tx(); debug!("tx packet: {:?}", packet); if let Err(e) = socket .send_to( @@ -674,7 +690,7 @@ impl BfdSession { if let Some(watchdog) = self.rx_watchdog.as_ref() { let _ = watchdog.send(WatchdogReset::ReceivedPacket).await; } - self.state.control_packets_rx.fetch_add(1, ORDERING); + self.state.record_rx(); let received_state = match p.flags.state() { Err(_) => { warn!("Invalid state, discarding"); diff --git a/src/lib.rs b/src/lib.rs index 5da552c..aecbe6c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,7 +6,8 @@ pub mod events; use std::{ fmt::{self, Display}, net::IpAddr, - time::Instant, + sync::atomic::{AtomicU64, Ordering}, + time::{Duration, Instant}, }; use log::debug; @@ -15,6 +16,157 @@ use tokio::net::UdpSocket; use crate::bfd::*; +/// A snapshot of the packet count and inter-packet interval statistics for one direction of a session. +#[derive(Debug, Clone, Copy, Default)] +pub struct IntervalStats { + count: u64, + min: Option, + max: Option, + mean: Option, +} +impl IntervalStats { + pub fn count(&self) -> u64 { + self.count + } + pub fn min(&self) -> Option { + self.min + } + pub fn max(&self) -> Option { + self.max + } + pub fn mean(&self) -> Option { + self.mean + } +} +impl Display for IntervalStats { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let ms = |d: Option| { + d.map_or("-".to_string(), |d| { + format!("{:.0}", d.as_secs_f64() * 1000.0) + }) + }; + write!( + f, + "{} pkts, interval min/avg/max {}/{}/{} ms", + self.count, + ms(self.min), + ms(self.mean), + ms(self.max) + ) + } +} + +/// An optional `Instant` that can be shared between threads, stored as nanoseconds since its construction. +/// Instants earlier than the construction time are clamped to it. +#[derive(Debug)] +pub struct AtomicInstant { + epoch: Instant, + nanos: AtomicU64, +} +impl AtomicInstant { + const NONE: u64 = u64::MAX; + + pub fn new() -> Self { + Self { + epoch: Instant::now(), + nanos: AtomicU64::new(Self::NONE), + } + } + pub fn load(&self) -> Option { + match self.nanos.load(Ordering::Relaxed) { + Self::NONE => None, + n => Some(self.epoch + Duration::from_nanos(n)), + } + } + pub fn store(&self, value: Instant) { + self.swap(value); + } + pub fn swap(&self, value: Instant) -> Option { + let n = u64::try_from(value.saturating_duration_since(self.epoch).as_nanos()) + .unwrap_or(Self::NONE - 1) + .min(Self::NONE - 1); + match self.nanos.swap(n, Ordering::Relaxed) { + Self::NONE => None, + prev => Some(self.epoch + Duration::from_nanos(prev)), + } + } + pub fn clear(&self) { + self.nanos.store(Self::NONE, Ordering::Relaxed); + } +} +impl Default for AtomicInstant { + fn default() -> Self { + Self::new() + } +} + +/// Lock-free accumulator for packet counts and inter-packet intervals. Updates are not atomic as a group, so it +/// is intended for a single writer; a concurrent reader may observe a mix of old and new values. +#[derive(Debug)] +pub struct AtomicIntervalStats { + count: AtomicU64, + last: AtomicInstant, + min_nanos: AtomicU64, + max_nanos: AtomicU64, + total_nanos: AtomicU64, +} +impl AtomicIntervalStats { + pub fn new() -> Self { + Self { + count: AtomicU64::new(0), + last: AtomicInstant::new(), + min_nanos: AtomicU64::new(u64::MAX), + max_nanos: AtomicU64::new(0), + total_nanos: AtomicU64::new(0), + } + } + pub fn record(&self, now: Instant) { + self.count.fetch_add(1, Ordering::Relaxed); + if let Some(prev) = self.last.swap(now) { + let nanos = + u64::try_from(now.saturating_duration_since(prev).as_nanos()).unwrap_or(u64::MAX); + self.min_nanos.fetch_min(nanos, Ordering::Relaxed); + self.max_nanos.fetch_max(nanos, Ordering::Relaxed); + self.total_nanos.fetch_add(nanos, Ordering::Relaxed); + } + } + pub fn reset(&self) { + self.count.store(0, Ordering::Relaxed); + self.last.clear(); + self.min_nanos.store(u64::MAX, Ordering::Relaxed); + self.max_nanos.store(0, Ordering::Relaxed); + self.total_nanos.store(0, Ordering::Relaxed); + } + pub fn snapshot(&self) -> IntervalStats { + let count = self.count.load(Ordering::Relaxed); + let intervals = count.saturating_sub(1); + if intervals == 0 { + return IntervalStats { + count, + ..Default::default() + }; + } + let total = self.total_nanos.load(Ordering::Relaxed); + IntervalStats { + count, + min: Some(Duration::from_nanos(self.min_nanos.load(Ordering::Relaxed))), + max: Some(Duration::from_nanos(self.max_nanos.load(Ordering::Relaxed))), + mean: Some(Duration::from_nanos(total / intervals)), + } + } +} +impl Default for AtomicIntervalStats { + fn default() -> Self { + Self::new() + } +} + +fn fmt_ago(t: Option) -> String { + t.map_or("never".to_string(), |t| { + format!("{}s ago", t.elapsed().as_secs()) + }) +} + #[derive(Debug)] pub struct BfdSessionStats { pub local_ip: IpAddr, @@ -23,8 +175,11 @@ pub struct BfdSessionStats { pub remote_discr: BfdDiscriminator, pub state: BfdState, pub last_diag: BfdDiagnostic, - pub control_packets_rx: u64, - pub control_packets_tx: u64, + /// Packet statistics since the session last entered Up + pub rx: IntervalStats, + pub tx: IntervalStats, + pub last_up: Option, + pub last_down: Option, pub last_change: Instant, pub detect_time: BfdInterval, pub base_interval: BfdInterval, @@ -40,9 +195,12 @@ impl Display for BfdSessionStats { writeln!( f, "{:<20} {:<9} {:>6} {:>6}", - self.remote_ip, self.state, self.control_packets_rx, self.control_packets_tx + self.remote_ip, + self.state, + self.rx.count(), + self.tx.count() )?; - write!( + writeln!( f, " LocalDiscr:{} RemoteDiscr:{} DetectTime: {} TxTime: {} Diag:{} Last: {}s", self.local_discr, @@ -51,6 +209,14 @@ impl Display for BfdSessionStats { self.base_interval, self.last_diag, Instant::now().duration_since(self.last_change).as_secs() + )?; + writeln!(f, " Rx: {}", self.rx)?; + writeln!(f, " Tx: {}", self.tx)?; + write!( + f, + " Last Up: {} Last Down: {}", + fmt_ago(self.last_up), + fmt_ago(self.last_down) ) } } @@ -70,7 +236,68 @@ pub fn set_ttl_or_hop_limit(sock: &UdpSocket, hop_limit: u8) -> Result<(), std:: debug!("Got {} after setting {}", res, i32_hop_limit); Ok(()) } - } else { + } else { panic!("UdpSocket with no address family ?!") } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn interval_stats_empty() { + let s = AtomicIntervalStats::new().snapshot(); + assert_eq!(s.count(), 0); + assert_eq!((s.min(), s.max(), s.mean()), (None, None, None)); + } + + #[test] + fn interval_stats_single_packet_has_no_intervals() { + let stats = AtomicIntervalStats::new(); + stats.record(Instant::now()); + let s = stats.snapshot(); + assert_eq!(s.count(), 1); + assert_eq!((s.min(), s.max(), s.mean()), (None, None, None)); + } + + #[test] + fn interval_stats_min_max_mean() { + let stats = AtomicIntervalStats::new(); + let t0 = Instant::now(); + for ms in [0, 100, 400, 500] { + stats.record(t0 + Duration::from_millis(ms)); + } + let s = stats.snapshot(); + assert_eq!(s.count(), 4); + assert_eq!(s.min(), Some(Duration::from_millis(100))); + assert_eq!(s.max(), Some(Duration::from_millis(300))); + assert_eq!(s.mean().unwrap().as_millis(), 166); + } + + #[test] + fn interval_stats_reset() { + let stats = AtomicIntervalStats::new(); + let t0 = Instant::now(); + stats.record(t0); + stats.record(t0 + Duration::from_millis(100)); + stats.reset(); + let s = stats.snapshot(); + assert_eq!(s.count(), 0); + assert_eq!((s.min(), s.max(), s.mean()), (None, None, None)); + stats.record(t0 + Duration::from_millis(900)); + stats.record(t0 + Duration::from_millis(1000)); + assert_eq!(stats.snapshot().max(), Some(Duration::from_millis(100))); + } + + #[test] + fn atomic_instant_roundtrip() { + let a = AtomicInstant::new(); + assert_eq!(a.load(), None); + let t = Instant::now() + Duration::from_secs(5); + assert_eq!(a.swap(t), None); + assert_eq!(a.load(), Some(t)); + a.clear(); + assert_eq!(a.load(), None); + } +}