From 214ffef6f3eb6ce1ea4b3ec2ad0afee008658849 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Oct 2022 16:20:24 +0200 Subject: [PATCH] Use proper error handling instead of lurking panics everywhere --- src/baseliner.rs | 9 +- src/config.rs | 35 +++--- src/error.rs | 98 ---------------- src/main.rs | 133 +++++++--------------- src/netlink.rs | 119 +++++++++++++------- src/pinger.rs | 29 +++-- src/pinger_icmp.rs | 37 +++---- src/pinger_icmp_ts.rs | 36 +++--- src/ratecontroller.rs | 227 +++++++++++++++++++++++--------------- src/reflector_selector.rs | 2 +- src/utils.rs | 16 ++- 11 files changed, 349 insertions(+), 392 deletions(-) delete mode 100644 src/error.rs diff --git a/src/baseliner.rs b/src/baseliner.rs index ac35e5e..32174fe 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -1,5 +1,6 @@ use crate::pinger::PingReply; -use crate::{Config, Utils}; +use crate::utils::Utils; +use crate::Config; use log::info; use std::collections::HashMap; use std::net::IpAddr; @@ -22,7 +23,7 @@ pub struct Baseliner { } impl Baseliner { - pub fn run(&self) { + pub fn run(&self) -> anyhow::Result<()> { /* * 135 seconds to decay to 50% for the slow factor and * 0.4 seconds to decay to 50% for the fast factor. @@ -35,7 +36,7 @@ impl Baseliner { let fast_factor = Utils::ewma_factor(self.config.tick_interval, 0.4); loop { - let time_data = self.stats_receiver.recv().unwrap(); + let time_data = self.stats_receiver.recv()?; let mut owd_baseline_map = self.owd_baseline.lock().unwrap(); let mut owd_recent_map = self.owd_recent.lock().unwrap(); @@ -84,6 +85,8 @@ impl Baseliner { "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection", time_data.reflector ); + // If reselection is disabled this would trigger an error + // so just ignore the result let _ = self.reselect_trigger.send(true); } else { owd_baseline.down_ewma = owd_baseline.down_ewma * slow_factor diff --git a/src/config.rs b/src/config.rs index c792807..4845ebb 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,10 +1,9 @@ -use crate::error::{ConfigParseError, InvalidMeasurementTypeError, MissingConfigError}; +use anyhow::Result; #[cfg(feature = "uci")] use log::warn; use log::Level; #[cfg(feature = "uci")] use rust_uci::Uci; -use std::error::Error; use std::fs::File; use std::io::BufRead; use std::net::IpAddr; @@ -12,6 +11,18 @@ use std::path::Path; use std::str::FromStr; use std::{env, io}; +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum ConfigError { + #[error("Invalid measurement type")] + InvalidMeasurementType(String), + #[error("Couldn't parse value for key: `{0}`: invalid value")] + ParseError(String), + #[error("No config value found for key: `{0}`")] + MissingValue(String), +} + fn read_lines

(filename: P) -> io::Result>> where P: AsRef, @@ -29,7 +40,7 @@ pub enum MeasurementType { } impl FromStr for MeasurementType { - type Err = InvalidMeasurementTypeError; + type Err = ConfigError; fn from_str(s: &str) -> Result { return match s.to_lowercase().as_str() { @@ -37,9 +48,7 @@ impl FromStr for MeasurementType { "icmp-timestamps" => Ok(MeasurementType::ICMPTimestamps), "ntp" => Ok(MeasurementType::NTP), "tcp-timestamps" => Ok(MeasurementType::TCPTimestamps), - &_ => Err(InvalidMeasurementTypeError { - type_: s.to_string(), - }), + &_ => Err(ConfigError::InvalidMeasurementType(s.to_string())), }; } } @@ -73,7 +82,7 @@ pub(crate) struct Config { } impl Config { - pub fn new() -> Result> { + pub fn new() -> Result { Ok(Self { // Network section download_base_kbits: Self::get::( @@ -180,22 +189,18 @@ impl Config { env_key: &str, uci_key: &str, default: Option, - ) -> Result> { + ) -> Result { return match Self::get_value(env_key, uci_key) { Some(val) => match val.parse::() { Ok(parsed_val) => Ok(parsed_val), // Ran into an compilation error while trying to return the // error as-is, so using my own error type to indicate something went wrong while parsing - Err(_) => Err(Box::new(ConfigParseError { - config_key: env_key.to_string(), - })), + Err(_) => Err(ConfigError::ParseError(env_key.to_string())), }, None => { return match default { Some(val) => Ok(val), - None => Err(Box::new(MissingConfigError { - config_key: env_key.to_string(), - })), + None => Err(ConfigError::MissingValue(env_key.to_string())), } } }; @@ -237,7 +242,7 @@ impl Config { None } - pub fn load_reflectors(&self) -> Result, Box> { + pub fn load_reflectors(&self) -> Result> { let lines = read_lines(self.reflector_list_file.clone())?; let mut reflectors: Vec = Vec::with_capacity(50); diff --git a/src/error.rs b/src/error.rs deleted file mode 100644 index d74cc26..0000000 --- a/src/error.rs +++ /dev/null @@ -1,98 +0,0 @@ -use std::error::Error; -use std::fmt; -use std::fmt::Display; - -#[derive(Default, Debug)] -pub struct ConfigParseError { - pub(crate) config_key: String, -} - -impl Error for ConfigParseError { - fn source(&self) -> Option<&(dyn Error + 'static)> { - None - } -} - -// Implement std::fmt::Display for AppError -impl Display for ConfigParseError { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!( - f, - "Couldn't parse value from string for key: {}", - self.config_key - ) // user-facing output - } -} - -#[derive(Default, Debug)] -pub struct InvalidMeasurementTypeError { - pub(crate) type_: String, -} - -impl Error for InvalidMeasurementTypeError { - fn source(&self) -> Option<&(dyn Error + 'static)> { - None - } -} - -impl Display for InvalidMeasurementTypeError { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "Couldn't find measurement type: {}", self.type_) - } -} - -#[derive(Default, Debug)] -pub struct MissingConfigError { - pub(crate) config_key: String, -} - -impl Error for MissingConfigError { - fn source(&self) -> Option<&(dyn Error + 'static)> { - None - } -} - -// Implement std::fmt::Display for AppError -impl Display for MissingConfigError { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!( - f, - "Couldn't find required configuration option: {}", - self.config_key - ) // user-facing output - } -} - -#[derive(Default, Debug)] -pub struct NoQdiscFoundError; - -impl Error for NoQdiscFoundError { - fn source(&self) -> Option<&(dyn Error + 'static)> { - None - } -} - -// Implement std::fmt::Display for AppError -impl Display for NoQdiscFoundError { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "Couldn't find a CAKE qdisc for interface") // user-facing output - } -} - -#[derive(Default, Debug)] -pub struct PingParseError { - pub(crate) msg: String, -} - -impl Error for PingParseError { - fn source(&self) -> Option<&(dyn Error + 'static)> { - None - } -} - -// Implement std::fmt::Display for AppError -impl Display for PingParseError { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "Couldn't parse received ping packet: {}", self.msg) // user-facing output - } -} diff --git a/src/main.rs b/src/main.rs index 161077d..59b682f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,7 +3,6 @@ extern crate core; mod baseliner; mod cake; mod config; -mod error; mod log; mod netlink; mod pinger; @@ -14,16 +13,15 @@ mod reflector_selector; mod utils; use crate::baseliner::{Baseliner, ReflectorStats}; -use ::log::{debug, error, info}; +use ::log::debug; use std::collections::HashMap; use std::net::IpAddr; -use std::process::ExitCode; use std::str::FromStr; use std::sync::mpsc::channel; use std::sync::{Arc, Mutex, RwLock}; use std::thread::sleep; use std::time::Duration; -use std::{panic, process, thread}; +use std::{process, thread}; use crate::config::{Config, MeasurementType}; use crate::netlink::Netlink; @@ -32,33 +30,15 @@ use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender}; use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender}; use crate::ratecontroller::{Ratecontroller, StatsDirection}; use crate::reflector_selector::ReflectorSelector; -use crate::utils::Utils; const VERSION: &str = env!("CARGO_PKG_VERSION"); -fn main() -> ExitCode { +fn main() -> anyhow::Result<()> { println!("Starting sqm-autorate version {}", VERSION); - let config = match Config::new() { - Ok(val) => val, - Err(e) => { - println!( - "Something went wrong while getting config: {}", - e.to_string() - ); - return ExitCode::FAILURE; - } - }; - - log::init(config.log_level).expect("Couldn't initialize logger"); - - let mut reflectors = match config.load_reflectors() { - Ok(refl) => refl, - Err(e) => { - info!("Couldn't load reflector list: {}", e.to_string()); - return ExitCode::FAILURE; - } - }; + let config = Config::new()?; + log::init(config.log_level)?; + let mut reflectors = config.load_reflectors()?; // The identifier field in ICMP is only 2 bytes // so take the last 2 bytes of the PID as the ID @@ -72,12 +52,12 @@ fn main() -> ExitCode { let reflector_pool_size = reflectors.len(); let default_reflectors = [ - IpAddr::from_str("9.9.9.9").unwrap(), - IpAddr::from_str("8.238.120.14").unwrap(), - IpAddr::from_str("74.82.42.42").unwrap(), - IpAddr::from_str("194.242.2.2").unwrap(), - IpAddr::from_str("208.67.222.222").unwrap(), - IpAddr::from_str("94.140.14.14").unwrap(), + IpAddr::from_str("9.9.9.9")?, + IpAddr::from_str("8.238.120.14")?, + IpAddr::from_str("74.82.42.42")?, + IpAddr::from_str("194.242.2.2")?, + IpAddr::from_str("208.67.222.222")?, + IpAddr::from_str("94.140.14.14")?, ]; match reflector_pool_size > 5 { @@ -117,66 +97,37 @@ fn main() -> ExitCode { stats_receiver: baseliner_stats_receiver, }; - let down_qdisc = match Netlink::qdisc_from_ifname(config.download_interface.as_str()) { - Ok(qdisc) => qdisc, - Err(e) => { - error!("Couldn't find download qdisc: {}", e); - return ExitCode::FAILURE; - } - }; - let up_qdisc = match Netlink::qdisc_from_ifname(config.upload_interface.as_str()) { - Ok(qdisc) => qdisc, - Err(e) => { - error!("Couldn't find upload qdisc: {}", e); - return ExitCode::FAILURE; - } - }; + let down_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?; + let up_qdisc = Netlink::qdisc_from_ifname(config.upload_interface.as_str())?; /* Set initial TC values to minimum * so there should be no initial bufferbloat to * fool the baseliner */ - Netlink::set_qdisc_rate(down_qdisc, config.download_min_kbits as u64) - .expect("Couldn't set ingress bandwidth"); - Netlink::set_qdisc_rate(up_qdisc, config.upload_min_kbits as u64) - .expect("Couldn't set egress bandwidth"); + Netlink::set_qdisc_rate(down_qdisc, config.download_min_kbits as u64)?; + Netlink::set_qdisc_rate(up_qdisc, config.upload_min_kbits as u64)?; sleep(Duration::new(0, 5e8 as u32)); let reflector_peers_lock_clone = reflector_peers_lock.clone(); - let receiver_handle = thread::Builder::new() - .name("receiver".to_string()) - .spawn(move || { - match pinger_receiver.listen( + let receiver_handle = thread::Builder::new().name("receiver".to_string()).spawn( + move || -> anyhow::Result<()> { + pinger_receiver.listen( id, config.measurement_type, reflector_peers_lock_clone, baseliner_stats_sender, - ) { - Ok(_) => {} - Err(e) => { - error!("Error occured in receiver thread: {}", e); - panic!(); - } - } - }) - .expect("Couldn't spawn ping receiver thread"); + ) + }, + )?; let baseliner_handle = thread::Builder::new() .name("baseliner".to_string()) - .spawn(move || baseliner.run()) - .expect("Couldn't spawn baseliner thread"); + .spawn(move || -> anyhow::Result<()> { baseliner.run() })?; let reflector_peers_lock_clone = reflector_peers_lock.clone(); - let sender_handle = thread::Builder::new() - .name("sender".to_string()) - .spawn(move || { - match pinger_sender.send(id, config.measurement_type, reflector_peers_lock_clone) { - Ok(_) => {} - Err(e) => { - error!("Error occured in sender thread: {}", e); - panic!(); - } - } - }) - .expect("Couldn't spawn ping sender thread"); + let sender_handle = thread::Builder::new().name("sender".to_string()).spawn( + move || -> anyhow::Result<()> { + pinger_sender.send(id, config.measurement_type, reflector_peers_lock_clone) + }, + )?; let mut threads = vec![receiver_handle, sender_handle, baseliner_handle]; @@ -190,22 +141,13 @@ fn main() -> ExitCode { }; let reselection_handle = thread::Builder::new() .name("reselection".to_string()) - .spawn(move || reflector_selector.run()) - .expect("Couldn't spawn reflector selector thread"); + .spawn(move || reflector_selector.run())?; threads.push(reselection_handle); } // Sleep 10 seconds before we start adjusting speeds sleep(Duration::new(10, 0)); - let ratecontroller = Ratecontroller { - config: config.clone(), - owd_baseline: owd_baseline.clone(), - owd_recent: owd_recent.clone(), - reflectors_lock: reflector_peers_lock.clone(), - reselect_trigger: reselect_sender.clone(), - }; - let dl_direction; let ul_direction; @@ -224,6 +166,16 @@ fn main() -> ExitCode { ul_direction = StatsDirection::TX; } + let ratecontroller = Ratecontroller::new( + config.clone(), + owd_baseline.clone(), + owd_recent.clone(), + reflector_peers_lock.clone(), + reselect_sender.clone(), + dl_direction, + ul_direction, + )?; + debug!( "Download direction: {}:{:?}", config.download_interface, dl_direction @@ -236,14 +188,13 @@ fn main() -> ExitCode { let ratecontroller_handle = thread::Builder::new() .name("ratecontroller".to_string()) - .spawn(move || ratecontroller.run(dl_direction, ul_direction)) - .expect("Couldn't spawn ratecontroller thread"); + .spawn(move || ratecontroller.run(dl_direction, ul_direction))?; threads.push(ratecontroller_handle); for thread in threads { - thread.join().unwrap(); + thread.join().expect("Error happened in thread")?; } - ExitCode::SUCCESS + Ok(()) } diff --git a/src/netlink.rs b/src/netlink.rs index 939e30b..668b19c 100644 --- a/src/netlink.rs +++ b/src/netlink.rs @@ -1,19 +1,49 @@ use crate::cake::TcaCake; -use crate::error::NoQdiscFoundError; use neli::consts::nl::{NlmF, NlmFFlags}; use neli::consts::rtnl::{Arphrd, IffFlags, Ifla, RtAddrFamily, Rtm, Tca}; use neli::consts::socket::NlFamily; -use neli::err::NlError; +use neli::err::{NlError, SerError}; use neli::nl::{NlPayload, Nlmsghdr}; use neli::rtnl::{Ifinfomsg, Rtattr, Tcmsg}; use neli::socket::NlSocketHandle; use neli::types::{Buffer, RtBuffer}; use serde::Deserialize; -use std::error::Error; +use std::io; +use thiserror::Error; use bincode::deserialize; +#[derive(Debug, Error)] +pub enum NetlinkError { + #[error("Couldn't deserialize to struct")] + Deserialization(#[from] Box), + + #[error("Couldn't find intreface `{0}`")] + InterfaceNotFound(String), + + #[error("Netlink interface error")] + NlInterfaceError(#[from] NlError), + + #[error("Netlink interface error")] + NlQdiscError(#[from] NlError), + + #[error("Couldn't find CAKE qdisc on interface `{0}`")] + NoQdiscFound(String), + + #[error("Couldn't find interface statistics: `{0}`")] + NoInterfaceStatsFound(String), + + #[error("Couldn't open Netlink socket")] + OpenSocket(#[from] io::Error), + + #[error("Serialization error")] + Serialization(#[from] SerError), + + #[error("Invalid Rtm type (expected {expected:?}, found {found:?})")] + WrongType { expected: Rtm, found: Rtm }, +} + #[derive(Deserialize, Copy, Clone, Default, Debug)] pub struct RtnlLinkStats64 { pub rx_packets: u64, @@ -43,7 +73,7 @@ pub struct RtnlLinkStats64 { } #[allow(dead_code)] -#[derive(Clone, Copy)] +#[derive(Clone, Copy, Debug)] pub struct Qdisc { ifindex: i32, handle: u32, @@ -55,13 +85,13 @@ pub struct Qdisc { pub struct Netlink {} impl Netlink { - fn nl_interface_get(socket: &mut NlSocketHandle, ifname: &str) -> Result<(), Box> { + fn nl_interface_get(socket: &mut NlSocketHandle, ifname: &str) -> Result<(), NetlinkError> { let mut attrs = RtBuffer::new(); const RTEXT_FILTER_VF: i32 = 1 << 0; - let attr_ifname = Rtattr::new(None, Ifla::Ifname, ifname).unwrap(); - let attr_ext_mask = Rtattr::new(None, Ifla::ExtMask, RTEXT_FILTER_VF).unwrap(); + let attr_ifname = Rtattr::new(None, Ifla::Ifname, ifname)?; + let attr_ext_mask = Rtattr::new(None, Ifla::ExtMask, RTEXT_FILTER_VF)?; attrs.push(attr_ifname); attrs.push(attr_ext_mask); @@ -88,28 +118,7 @@ impl Netlink { Ok(()) } - pub fn find_interface(ifname: &str) -> Result> { - let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); - - Self::nl_interface_get(&mut socket, ifname).expect("Wrong when sending intf"); - - for response in socket.iter(false) { - let header: Nlmsghdr = response?; - - if header.nl_type != Rtm::Newlink { - return Err(Box::new(NlError::msg("Netlink error retrieving link"))); - } - - if let NlPayload::Payload(p) = header.nl_payload { - return Ok(p.ifi_index); - } - } - - // we shouldn't reach here - Err(Box::new(NlError::msg("Netlink error retrieving link"))) - } - - pub fn get_interface_stats(ifname: &str) -> Result> { + pub fn find_interface(ifname: &str) -> Result { let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); Self::nl_interface_get(&mut socket, ifname)?; @@ -118,7 +127,34 @@ impl Netlink { let header: Nlmsghdr = response?; if header.nl_type != Rtm::Newlink { - return Err(Box::new(NlError::msg("Netlink error retrieving link"))); + return Err(NetlinkError::WrongType { + expected: Rtm::Newlink, + found: header.nl_type, + }); + } + + if let NlPayload::Payload(p) = header.nl_payload { + return Ok(p.ifi_index); + } + } + + // we shouldn't reach here + Err(NetlinkError::InterfaceNotFound(ifname.to_string())) + } + + pub fn get_interface_stats(ifname: &str) -> Result { + let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[])?; + + Self::nl_interface_get(&mut socket, ifname)?; + + for response in socket.iter(false) { + let header: Nlmsghdr = response?; + + if header.nl_type != Rtm::Newlink { + return Err(NetlinkError::WrongType { + expected: Rtm::Newlink, + found: header.nl_type, + }); } if let NlPayload::Payload(p) = header.nl_payload { @@ -126,7 +162,7 @@ impl Netlink { if attr.rta_type == Ifla::Stats64 { let buf = attr.rta_payload.as_ref(); - let stats: RtnlLinkStats64 = deserialize(buf).unwrap(); + let stats: RtnlLinkStats64 = deserialize(buf)?; return Ok(stats); } @@ -134,11 +170,11 @@ impl Netlink { } } - Err(Box::new(NlError::msg("Error getting stats"))) + Err(NetlinkError::NoInterfaceStatsFound(ifname.to_string())) } - pub fn qdisc_from_ifindex(ifindex: i32) -> Result> { - let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); + pub fn qdisc_from_ifindex(ifindex: i32) -> Result { + let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[])?; let tc_msg = Tcmsg::new( u8::from(RtAddrFamily::Unspecified), 0, @@ -157,14 +193,19 @@ impl Netlink { NlPayload::Payload(tc_msg), ); - socket.send(nlhdr)?; + if let Err(e) = socket.send(nlhdr) { + return Err(NetlinkError::Serialization(e)); + } for response in socket.iter(false) { let header: Nlmsghdr = response?; if let NlPayload::Payload(p) = header.nl_payload { if header.nl_type != Rtm::Newqdisc { - return Err(Box::new(NlError::msg("Netlink error retrieving qdisc"))); + return Err(NetlinkError::WrongType { + expected: Rtm::Newqdisc, + found: header.nl_type, + }); } if p.tcm_ifindex == ifindex { @@ -196,15 +237,15 @@ impl Netlink { } } - return Err(Box::new(NoQdiscFoundError)); + return Err(NetlinkError::NoQdiscFound(ifindex.to_string())); } - pub fn qdisc_from_ifname(ifname: &str) -> Result> { + pub fn qdisc_from_ifname(ifname: &str) -> Result { let ifindex = Netlink::find_interface(ifname)?; Netlink::qdisc_from_ifindex(ifindex) } - pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth_kbit: u64) -> Result<(), Box> { + pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth_kbit: u64) -> Result<(), NetlinkError> { let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[])?; let bandwidth = bandwidth_kbit * 1000 / 8; diff --git a/src/pinger.rs b/src/pinger.rs index 26e57a1..ac5fddd 100644 --- a/src/pinger.rs +++ b/src/pinger.rs @@ -1,17 +1,31 @@ use crate::MeasurementType; +use etherparse::ReadError; use log::{debug, error}; +use nix::errno::Errno; use nix::sys::socket::{ recvfrom, sendto, socket, AddressFamily, MsgFlags, SockFlag, SockProtocol, SockType, SockaddrIn, SockaddrIn6, SockaddrLike, }; -use std::error::Error; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6}; use std::os::unix::io::RawFd; use std::str::FromStr; use std::sync::mpsc::Sender; use std::sync::{Arc, RwLock}; -use std::thread; use std::time::Duration; +use std::{io, thread}; +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum PingError { + #[error("Couldn't parse number")] + InvalidNumber(#[from] io::Error), + #[error("Error parsing packet")] + InvalidPacket(#[from] ReadError), + #[error("Socket error")] + Socket(#[from] Errno), + #[error("Wrong ICMP identifier (expected {expected:?}, found {found:?})")] + WrongID { expected: u16, found: u16 }, +} #[allow(dead_code)] pub struct PingReply { @@ -58,7 +72,7 @@ pub trait PingListener { type_: MeasurementType, reflectors_lock: Arc>>, stats_sender: Sender, - ) -> Result<(), Box> { + ) -> anyhow::Result<()> { let socket = open_socket(type_)?; loop { @@ -98,12 +112,7 @@ pub trait PingListener { } } - fn parse_packet( - &self, - id: u16, - reflector: IpAddr, - buf: &[u8], - ) -> Result>; + fn parse_packet(&self, id: u16, reflector: IpAddr, buf: &[u8]) -> Result; } pub trait PingSender { @@ -112,7 +121,7 @@ pub trait PingSender { id: u16, type_: MeasurementType, reflectors_lock: Arc>>, - ) -> Result<(), Box> { + ) -> anyhow::Result<()> { let socket = open_socket(type_)?; let mut seq: u16 = 0; diff --git a/src/pinger_icmp.rs b/src/pinger_icmp.rs index 6f72dee..de8d711 100644 --- a/src/pinger_icmp.rs +++ b/src/pinger_icmp.rs @@ -1,8 +1,6 @@ -use std::error::Error; use std::net::IpAddr; -use crate::error::PingParseError; -use crate::pinger::{PingListener, PingReply, PingSender}; +use crate::pinger::{PingError, PingListener, PingReply, PingSender}; use byteorder::*; use etherparse::TransportSlice::{Icmpv4, Icmpv6}; use etherparse::{IcmpEchoHeader, Icmpv4Header, Icmpv4Type, SlicedPacket}; @@ -15,21 +13,17 @@ pub struct PingerICMPEchoSender {} impl PingListener for PingerICMPEchoListener { // Result: RTT, down time, up time - fn parse_packet( - &self, - id: u16, - reflector: IpAddr, - buf: &[u8], - ) -> Result> { + fn parse_packet(&self, id: u16, reflector: IpAddr, buf: &[u8]) -> Result { match SlicedPacket::from_ip(buf) { - Err(value) => println!("Err {:?}", value), + Err(err) => return Err(PingError::InvalidPacket(err)), Ok(value) => match value.transport { Some(Icmpv4(icmp)) => match icmp.icmp_type() { Icmpv4Type::EchoReply(echo) => { if echo.id != id { - return Err(Box::new(PingParseError { - msg: "Wrong ID".to_string(), - })); + return Err(PingError::WrongID { + expected: id, + found: echo.id, + }); } let time_sent = icmp @@ -57,17 +51,18 @@ impl PingListener for PingerICMPEchoListener { + (time_now.tv_nsec() as f64 / 1e9), }); } - _ => {} + _ => { + todo!() + } }, - Some(Icmpv6(_)) => {} - Some(_) => {} - None => {} + Some(Icmpv6(_)) => { + todo!() + } + _ => { + unimplemented!() + } }, } - - Err(Box::new(PingParseError { - msg: "Reached end of parsing function".to_string(), - })) } } diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs index fc6a130..96e47fb 100644 --- a/src/pinger_icmp_ts.rs +++ b/src/pinger_icmp_ts.rs @@ -1,13 +1,10 @@ -use std::error::Error; use std::net::IpAddr; -use crate::error::PingParseError; -use crate::pinger::{PingListener, PingReply, PingSender}; -use crate::Utils; +use crate::pinger::{PingError, PingListener, PingReply, PingSender}; +use crate::utils::Utils; use etherparse::icmpv4::TimestampMessage; use etherparse::TransportSlice::Icmpv4; use etherparse::{Icmpv4Header, Icmpv4Type, SlicedPacket}; -use log::warn; use nix::time::{clock_gettime, ClockId}; pub struct PingerICMPTimestampListener {} @@ -16,21 +13,17 @@ pub struct PingerICMPTimestampSender {} impl PingListener for PingerICMPTimestampListener { // Result: RTT, down time, up time - fn parse_packet( - &self, - id: u16, - reflector: IpAddr, - buf: &[u8], - ) -> Result> { + fn parse_packet(&self, id: u16, reflector: IpAddr, buf: &[u8]) -> Result { match SlicedPacket::from_ip(buf) { - Err(value) => warn!("Error parsing packet: {:?}", value), + Err(err) => return Err(PingError::InvalidPacket(err)), Ok(value) => match value.transport { Some(Icmpv4(icmp)) => match icmp.icmp_type() { Icmpv4Type::TimestampReply(reply) => { if reply.id != id { - return Err(Box::new(PingParseError { - msg: "Wrong ID".to_string(), - })); + return Err(PingError::WrongID { + expected: id, + found: reply.id, + }); } let time_now = clock_gettime(ClockId::CLOCK_REALTIME).unwrap(); @@ -59,16 +52,15 @@ impl PingListener for PingerICMPTimestampListener { + (time_now.tv_nsec() as f64 / 1e9), }); } - _ => {} + _ => { + unimplemented!() + } }, - Some(_) => {} - None => {} + _ => { + unimplemented!() + } }, } - - Err(Box::new(PingParseError { - msg: "Reached end of parsing function".to_string(), - })) } } diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index 5f9646b..6b75609 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -1,5 +1,6 @@ -use crate::netlink::Netlink; -use crate::{Config, ReflectorStats, Utils}; +use crate::netlink::{Netlink, NetlinkError, Qdisc}; +use crate::utils::Utils; +use crate::{Config, ReflectorStats}; use log::{debug, error, info, warn}; use rand::seq::SliceRandom; use rand::thread_rng; @@ -12,6 +13,16 @@ use std::sync::mpsc::Sender; use std::sync::{Arc, Mutex, RwLock}; use std::thread::sleep; use std::time::Duration; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum RatecontrolError { + #[error("Netlink error")] + Netlink(#[from] NetlinkError), + + #[error("Error sorting")] + Sorting, +} #[derive(Copy, Clone, Debug)] pub enum StatsDirection { @@ -19,34 +30,25 @@ pub enum StatsDirection { TX, } -fn f64_max(a: f64, b: f64) -> f64 { - *[a, b].iter().max_by(|a, b| a.total_cmp(b)).unwrap() -} +fn generate_initial_speeds(base_speed: f64, size: u32) -> Vec { + let mut rates = Vec::new(); -fn f64_min(a: f64, b: f64) -> f64 { - *[a, b].iter().min_by(|a, b| a.total_cmp(b)).unwrap() + for _ in 0..size { + rates.push((rand::thread_rng().next_u64() as f64 * 0.2 + 0.75) * base_speed); + } + + rates } fn get_interface_stats( config: Config, down_direction: StatsDirection, up_direction: StatsDirection, -) -> (i128, i128) { - let (down_rx, down_tx) = match Netlink::get_interface_stats(config.download_interface.as_str()) - { - Ok(stats) => (stats.rx_bytes as i128, stats.tx_bytes as i128), - Err(e) => { - error!("Error while retrieving stats: {}", e.to_string()); - (-1, -1) - } - }; - let (up_rx, up_tx) = match Netlink::get_interface_stats(config.upload_interface.as_str()) { - Ok(stats) => (stats.rx_bytes as i128, stats.tx_bytes as i128), - Err(e) => { - error!("Error while retrieving stats: {}", e.to_string()); - (-1, -1) - } - }; +) -> Result<(i128, i128), RatecontrolError> { + let down_stats = Netlink::get_interface_stats(config.download_interface.as_str())?; + let up_stats = Netlink::get_interface_stats(config.upload_interface.as_str())?; + let (down_rx, down_tx) = (down_stats.rx_bytes, down_stats.tx_bytes); + let (up_rx, up_tx) = (up_stats.rx_bytes, up_stats.tx_bytes); let rx_bytes = match down_direction { StatsDirection::RX => down_rx, @@ -58,42 +60,106 @@ fn get_interface_stats( StatsDirection::TX => up_tx, }; - (rx_bytes, tx_bytes) + Ok((rx_bytes.into(), tx_bytes.into())) +} + +#[derive(Clone, Debug)] +struct State { + current_bytes: i128, + current_rate: f64, + qdisc: Qdisc, + nrate: usize, + previous_bytes: i128, + safe_rates: Vec, + stats_direction: StatsDirection, } pub struct Ratecontroller { - pub(crate) config: Config, - pub(crate) owd_baseline: Arc>>, - pub(crate) owd_recent: Arc>>, - pub(crate) reflectors_lock: Arc>>, - pub(crate) reselect_trigger: Sender, + config: Config, + owd_baseline: Arc>>, + owd_recent: Arc>>, + reflectors_lock: Arc>>, + reselect_trigger: Sender, + + state_dl: State, + state_ul: State, } impl Ratecontroller { - pub fn run(&self, down_direction: StatsDirection, up_direction: StatsDirection) { + fn adjust_rate(&self) -> anyhow::Result<()> { + Ok(()) + } + + pub fn new( + config: Config, + owd_baseline: Arc>>, + owd_recent: Arc>>, + reflectors_lock: Arc>>, + reselect_trigger: Sender, + down_direction: StatsDirection, + up_direction: StatsDirection, + ) -> anyhow::Result { + let dl_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?; + let dl_safe_rates = + generate_initial_speeds(config.download_base_kbits, config.speed_hist_size); + let ul_qdisc = Netlink::qdisc_from_ifname(config.upload_interface.as_str())?; + let ul_safe_rates = + generate_initial_speeds(config.upload_base_kbits, config.speed_hist_size); + + Ok(Self { + config, + owd_baseline, + owd_recent, + reflectors_lock, + reselect_trigger, + state_dl: State { + current_bytes: 0, + current_rate: 0.0, + qdisc: dl_qdisc, + nrate: 0, + previous_bytes: 0, + safe_rates: dl_safe_rates, + stats_direction: down_direction, + }, + state_ul: State { + current_bytes: 0, + current_rate: 0.0, + qdisc: ul_qdisc, + nrate: 0, + previous_bytes: 0, + safe_rates: ul_safe_rates, + stats_direction: up_direction, + }, + }) + } + + pub fn run( + &self, + down_direction: StatsDirection, + up_direction: StatsDirection, + ) -> anyhow::Result<()> { let sleep_time_s = self.config.min_change_interval.floor() as u64; let sleep_time_ns = ((self.config.min_change_interval % 1.0) * 1e9) as u32; let sleep_time = Duration::new(sleep_time_s, sleep_time_ns); - let (start_s, _) = Utils::get_current_time(); - let (mut lastchg_s, mut lastchg_ns) = Utils::get_current_time(); + let (start_s, _) = Utils::get_current_time()?; + let (mut lastchg_s, mut lastchg_ns) = Utils::get_current_time()?; let mut lastchg_t = lastchg_s - start_s + lastchg_ns / 1e9; let mut lastdump_t = lastchg_t - 310.0; let mut rng = thread_rng(); - let down_qdisc = - Netlink::qdisc_from_ifname(self.config.download_interface.as_str()).unwrap(); - let up_qdisc = Netlink::qdisc_from_ifname(self.config.upload_interface.as_str()).unwrap(); + let down_qdisc = Netlink::qdisc_from_ifname(self.config.download_interface.as_str())?; + let up_qdisc = Netlink::qdisc_from_ifname(self.config.upload_interface.as_str())?; // set qdisc rates to 60% of base rate to make sure we start with sane baselines let mut cur_dl_rate: f64 = self.config.download_base_kbits * 0.6; let mut cur_ul_rate: f64 = self.config.upload_base_kbits * 0.6; - Netlink::set_qdisc_rate(down_qdisc, cur_dl_rate.floor() as u64).unwrap(); - Netlink::set_qdisc_rate(up_qdisc, cur_ul_rate.floor() as u64).unwrap(); + Netlink::set_qdisc_rate(down_qdisc, cur_dl_rate.round() as u64)?; + Netlink::set_qdisc_rate(up_qdisc, cur_ul_rate.round() as u64)?; let (mut prev_rx_bytes, mut prev_tx_bytes) = - get_interface_stats(self.config.clone(), down_direction, up_direction); + get_interface_stats(self.config.clone(), down_direction, up_direction)?; if prev_rx_bytes == -1 || prev_tx_bytes == -1 { panic!("Couldn't retrieve stats from interface."); @@ -101,18 +167,10 @@ impl Ratecontroller { let mut t_prev_bytes = lastchg_t; - let mut safe_dl_rates: Vec = Vec::new(); - let mut safe_ul_rates: Vec = Vec::new(); - for _ in 0..self.config.speed_hist_size { - safe_dl_rates.push( - (rand::thread_rng().next_u32() as f64 * 0.2 + 0.75) - * self.config.download_base_kbits, - ); - safe_ul_rates.push( - (rand::thread_rng().next_u32() as f64 * 0.2 + 0.75) - * self.config.download_base_kbits, - ); - } + let mut safe_dl_rates: Vec = + generate_initial_speeds(self.config.download_base_kbits, self.config.speed_hist_size); + let mut safe_ul_rates: Vec = + generate_initial_speeds(self.config.upload_base_kbits, self.config.speed_hist_size); let mut nrate_up = 0; let mut nrate_down = 0; @@ -126,35 +184,28 @@ impl Ratecontroller { speed_hist_fd_inner = File::options() .create(true) .write(true) - .open(self.config.speed_hist_file.as_str()) - .unwrap(); + .open(self.config.speed_hist_file.as_str())?; - speed_hist_fd_inner - .write("time,counter,upspeed,downspeed\n".as_bytes()) - .unwrap(); - speed_hist_fd_inner.flush().unwrap(); + speed_hist_fd_inner.write("time,counter,upspeed,downspeed\n".as_bytes())?; + speed_hist_fd_inner.flush()?; speed_hist_fd = Some(speed_hist_fd_inner); stats_fd_inner = File::options() .create(true) .write(true) - .open(self.config.stats_file.as_str()) - .unwrap(); + .open(self.config.stats_file.as_str())?; - stats_fd_inner - .write( - "times,timens,rxload,txload,deltadelaydown,deltadelayup,dlrate,uprate\n" - .as_bytes(), - ) - .unwrap(); - stats_fd_inner.flush().unwrap(); + stats_fd_inner.write( + "times,timens,rxload,txload,deltadelaydown,deltadelayup,dlrate,uprate\n".as_bytes(), + )?; + stats_fd_inner.flush()?; stats_fd = Some(stats_fd_inner); } loop { - let (mut now_s, now_ns) = Utils::get_current_time(); + let (mut now_s, now_ns) = Utils::get_current_time()?; let now_abstime = now_s + now_ns / 1e9; now_s = now_s - start_s; let now_t = now_s + now_ns / 1e9; @@ -221,7 +272,7 @@ impl Ratecontroller { next_ul_rate = cur_ul_rate; let (cur_rx_bytes, cur_tx_bytes) = - get_interface_stats(self.config.clone(), down_direction, up_direction); + get_interface_stats(self.config.clone(), down_direction, up_direction)?; if cur_rx_bytes == -1 || cur_tx_bytes == -1 { warn!("One or both Netlink stats could not be read. Skipping rate control algorithm"); } else if down_deltas.len() < 3 || up_deltas.len() < 3 { @@ -259,7 +310,7 @@ impl Ratecontroller { let max_dl = safe_dl_rates.iter().max_by(|a, b| a.total_cmp(b)).unwrap(); next_dl_rate = cur_dl_rate - * (1.0 + 0.1 * f64_max(0.0, 1.0 - cur_dl_rate / max_dl)) + * (1.0 + 0.1 * (1.0 - cur_dl_rate / max_dl).max(0.0)) + (self.config.download_base_kbits * 0.03); nrate_down = nrate_down + 1; nrate_down = nrate_down % self.config.speed_hist_size as usize; @@ -273,31 +324,31 @@ impl Ratecontroller { let max_ul = safe_ul_rates.iter().max_by(|a, b| a.total_cmp(b)).unwrap(); next_ul_rate = cur_ul_rate - * (1.0 + 0.1 * f64_max(0.0, 1.0 - cur_ul_rate / max_ul)) + * (1.0 + 0.1 * (1.0 - cur_ul_rate / max_ul).max(0.0)) + (self.config.download_base_kbits * 0.03); nrate_up = nrate_up + 1; nrate_up = nrate_up % self.config.speed_hist_size as usize; } if down_delta_stat > self.config.download_delay_ms { - if safe_dl_rates.len() > 0 { - next_dl_rate = f64_min( - 0.9 * cur_dl_rate * rx_load, - *safe_dl_rates.choose(&mut rng).unwrap(), - ) - } else { - next_dl_rate = 0.9 * cur_dl_rate * rx_load; + match safe_dl_rates.choose(&mut rng) { + Some(rnd_rate) => { + next_dl_rate = rnd_rate.min(0.9 * cur_dl_rate * rx_load); + } + None => { + next_dl_rate = 0.9 * cur_dl_rate * rx_load; + } } } if up_delta_stat > self.config.upload_delay_ms { - if safe_ul_rates.len() > 0 { - next_ul_rate = f64_min( - 0.9 * cur_ul_rate * tx_load, - *safe_ul_rates.choose(&mut rng).unwrap(), - ) - } else { - next_ul_rate = 0.9 * cur_ul_rate * tx_load; + match safe_ul_rates.choose(&mut rng) { + Some(rnd_rate) => { + next_ul_rate = rnd_rate.min(0.9 * cur_ul_rate * tx_load); + } + None => { + next_ul_rate = 0.9 * cur_ul_rate * tx_load; + } } } } @@ -307,8 +358,8 @@ impl Ratecontroller { prev_rx_bytes = cur_rx_bytes; prev_tx_bytes = cur_tx_bytes; - next_dl_rate = f64_max(self.config.download_min_kbits, next_dl_rate).floor(); - next_ul_rate = f64_max(self.config.upload_min_kbits, next_ul_rate).floor(); + next_dl_rate = next_dl_rate.max(self.config.download_min_kbits).round(); + next_ul_rate = next_ul_rate.max(self.config.upload_min_kbits).round(); if next_dl_rate != cur_dl_rate || next_ul_rate != cur_ul_rate { info!( @@ -318,17 +369,17 @@ impl Ratecontroller { } if next_dl_rate != cur_dl_rate { - Netlink::set_qdisc_rate(down_qdisc, next_dl_rate as u64).unwrap(); + Netlink::set_qdisc_rate(down_qdisc, next_dl_rate as u64)?; } if next_ul_rate != cur_ul_rate { - Netlink::set_qdisc_rate(up_qdisc, next_ul_rate as u64).unwrap(); + Netlink::set_qdisc_rate(up_qdisc, next_ul_rate as u64)?; } cur_dl_rate = next_dl_rate; cur_ul_rate = next_ul_rate; - (lastchg_s, lastchg_ns) = Utils::get_current_time(); + (lastchg_s, lastchg_ns) = Utils::get_current_time()?; if rx_load > 0.0 && tx_load > 0.0 diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index 8cb4406..7266526 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -18,7 +18,7 @@ pub struct ReflectorSelector { } impl ReflectorSelector { - pub fn run(&self) { + pub fn run(&self) -> anyhow::Result<()> { let mut selector_sleep_time = Duration::new(30, 0); let mut reselection_count = 0; let baseline_sleep_time = Duration::new( diff --git a/src/utils.rs b/src/utils.rs index e2776a6..6e28806 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,6 +1,14 @@ +use nix::errno::Errno; use nix::time::{clock_gettime, ClockId}; +use thiserror::Error; -pub(crate) struct Utils {} +#[derive(Debug, Error)] +pub enum UtilError { + #[error("libc error")] + Libc(#[from] Errno), +} + +pub struct Utils {} impl Utils { pub(crate) fn a_else_b(a: f64, b: f64) -> f64 { @@ -9,9 +17,9 @@ impl Utils { pub(crate) fn ewma_factor(tick: f64, dur: f64) -> f64 { ((0.5_f64).ln() / (dur / tick)).exp() } - pub(crate) fn get_current_time() -> (f64, f64) { - let time = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap(); - (time.tv_sec() as f64, time.tv_nsec() as f64) + pub(crate) fn get_current_time() -> Result<(f64, f64), UtilError> { + let time = clock_gettime(ClockId::CLOCK_MONOTONIC)?; + Ok((time.tv_sec() as f64, time.tv_nsec() as f64)) } #[cfg(target_endian = "big")]