diff --git a/src/bin/rust-bfdd.rs b/src/bin/rust-bfdd.rs index b0e8749..90f7018 100644 --- a/src/bin/rust-bfdd.rs +++ b/src/bin/rust-bfdd.rs @@ -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,65 +1136,58 @@ impl Bfdd { .unwrap(); } } + async fn session_report(&self) -> String { + future::join_all( + self.sessions_by_discr + .values() + .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: {}", - future::join_all( - self.sessions_by_discr - .values() - .map(|session| async { session.get_stats().await.unwrap().to_string() }) - ) - .await - .join("\n\t") - ); + info!("configured sessions: {}", self.session_report().await); - let rx_task = task::spawn(async move { - let mut buf = [0; 1024]; - loop { - // TODO: All received BFD Control packets that are demultiplexed to the session MUST be discarded if - // the received TTL or Hop Limit is not equal to 255. Need raw sockets for this. - // - // 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 { + 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 + // the received TTL or Hop Limit is not equal to 255. Need raw sockets for this. + // + // 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. + 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.rx_packet(addr, packet).await - } else { - warn!("Failed to parse packet"); + self.handle_datagram(&buf[..len], addr).await; + if log_enabled!(log::Level::Debug) { + debug!("configured sessions: {}", self.session_report().await); } - - 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(()) + } } }