Dump the session table on SIGUSR1

Split the receive loop into a select over the control socket and SIGUSR1, and factor the session table formatting into session_report().
This commit is contained in:
ktims committed 2026-10-22 17:49:12 -07:00
1 parent eeeef5fb73
commit 8321d01ffb
1 file changed
+35 -41
+35 -41
View File
@@ -21,14 +21,15 @@ use std::{
use tokio::sync::{oneshot, Mutex};
use env_logger::Env;
use log::{debug, info, warn};
use log::{debug, info, log_enabled, warn};
use nom_derive::Parse;
use rand::prelude::*;
use tokio::io;
use tokio::net::UdpSocket;
use tokio::signal::unix::{signal, SignalKind};
use tokio::sync::mpsc;
use tokio::task::{self, JoinHandle};
use tokio::task::JoinHandle;
use tokio::time;
use atomic::Atomic;
@@ -1135,21 +1136,35 @@ impl Bfdd {
.unwrap();
}
}
async fn run(mut self) -> io::Result<()> {
info!("rust-bfd starting up");
self.start_peers().await?;
info!(
"configured sessions: {}",
async fn session_report(&self) -> String {
future::join_all(
self.sessions_by_discr
.values()
.map(|session| async { session.get_stats().await.unwrap().to_string() })
.map(|session| async { session.get_stats().await.unwrap().to_string() }),
)
.await
.join("\n\t")
);
}
async fn handle_datagram(&self, data: &[u8], addr: SocketAddr) {
let Ok((_leftover, packet)) = BfdPacket::parse(data) else {
warn!("Failed to parse packet");
return;
};
debug!("rx packet ({:?}): {:?}", addr.ip(), packet);
// If the Length field is greater than the payload of the encapsulating protocol, the packet MUST be
// discarded.
if packet.flags.length() as usize > data.len() {
warn!("discarding packet with length > payload");
return;
}
self.rx_packet(addr, packet).await
}
async fn run(mut self) -> io::Result<()> {
info!("rust-bfd starting up");
self.start_peers().await?;
info!("configured sessions: {}", self.session_report().await);
let rx_task = task::spawn(async move {
let mut sigusr1 = signal(SignalKind::user_defined1())?;
let mut buf = [0; 1024];
loop {
// TODO: All received BFD Control packets that are demultiplexed to the session MUST be discarded if
@@ -1157,43 +1172,22 @@ impl Bfdd {
//
// TODO: Maybe it is okay to receive in the session-specific socket instead of this global one? But
// the RFC mentions that we must demux based on discriminator only.
match self.control_socket.recv_from(&mut buf).await {
tokio::select! {
result = self.control_socket.recv_from(&mut buf) => match result {
Ok((len, addr)) => {
if let Ok((_leftover, packet)) = BfdPacket::parse(&buf[..len]) {
debug!("rx packet ({:?}): {:?}", addr.ip(), packet);
// If the Length field is greater than the payload of the encapsulating protocol, the packet MUST be
// discarded.
if packet.flags.length() as usize > len {
warn!("discarding packet with length > payload");
continue;
self.handle_datagram(&buf[..len], addr).await;
if log_enabled!(log::Level::Debug) {
debug!("configured sessions: {}", self.session_report().await);
}
self.rx_packet(addr, packet).await
} else {
warn!("Failed to parse packet");
}
debug!(
"configured sessions: {}",
future::join_all(self.sessions_by_discr.values().map(
|session| async { session.get_stats().await.unwrap().to_string() }
))
.await
.join("\n\t")
);
}
Err(e) => {
warn!("Error receiving from socket {e:?}");
continue;
Err(e) => warn!("Error receiving from socket {e:?}"),
},
_ = sigusr1.recv() => {
info!("SIGUSR1 received");
info!("configured sessions: {}", self.session_report().await);
}
}
}
});
tokio::join!(rx_task)
.0
.expect("Unable to join on the receive thread");
Ok(())
}
}