Use proper error handling instead of lurking panics everywhere

This commit is contained in:
Nils Andreas Svee committed 2022-10-04 16:20:24 +02:00
1 parent 999987d157
commit 214ffef6f3
11 files changed
+349 -392

No files matched your search

+6 -3
View File
@@ -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
+20 -15
View File
@@ -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<P>(filename: P) -> io::Result<io::Lines<io::BufReader<File>>>
where
P: AsRef<Path>,
@@ -29,7 +40,7 @@ pub enum MeasurementType {
}
impl FromStr for MeasurementType {
type Err = InvalidMeasurementTypeError;
type Err = ConfigError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
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<Self, Box<dyn Error>> {
pub fn new() -> Result<Self> {
Ok(Self {
// Network section
download_base_kbits: Self::get::<f64>(
@@ -180,22 +189,18 @@ impl Config {
env_key: &str,
uci_key: &str,
default: Option<T>,
) -> Result<T, Box<dyn Error>> {
) -> Result<T, ConfigError> {
return match Self::get_value(env_key, uci_key) {
Some(val) => match val.parse::<T>() {
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<Vec<IpAddr>, Box<dyn Error>> {
pub fn load_reflectors(&self) -> Result<Vec<IpAddr>> {
let lines = read_lines(self.reflector_list_file.clone())?;
let mut reflectors: Vec<IpAddr> = Vec::with_capacity(50);
-98
View File
@@ -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
}
}
+42 -91
View File
@@ -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(())
}
+80 -39
View File
@@ -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<bincode::ErrorKind>),
#[error("Couldn't find intreface `{0}`")]
InterfaceNotFound(String),
#[error("Netlink interface error")]
NlInterfaceError(#[from] NlError<Rtm, Ifinfomsg>),
#[error("Netlink interface error")]
NlQdiscError(#[from] NlError<Rtm, Tcmsg>),
#[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<dyn Error>> {
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<i32, Box<dyn Error>> {
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<Rtm, Ifinfomsg> = 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<RtnlLinkStats64, Box<dyn Error>> {
pub fn find_interface(ifname: &str) -> Result<i32, NetlinkError> {
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<Rtm, Ifinfomsg> = 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<RtnlLinkStats64, NetlinkError> {
let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[])?;
Self::nl_interface_get(&mut socket, ifname)?;
for response in socket.iter(false) {
let header: Nlmsghdr<Rtm, Ifinfomsg> = 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<Qdisc, Box<dyn Error>> {
let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap();
pub fn qdisc_from_ifindex(ifindex: i32) -> Result<Qdisc, NetlinkError> {
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<Rtm, Tcmsg> = 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<Qdisc, Box<dyn Error>> {
pub fn qdisc_from_ifname(ifname: &str) -> Result<Qdisc, NetlinkError> {
let ifindex = Netlink::find_interface(ifname)?;
Netlink::qdisc_from_ifindex(ifindex)
}
pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth_kbit: u64) -> Result<(), Box<dyn Error>> {
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;
+19 -10
View File
@@ -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<RwLock<Vec<IpAddr>>>,
stats_sender: Sender<PingReply>,
) -> Result<(), Box<dyn Error>> {
) -> 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<PingReply, Box<dyn Error>>;
fn parse_packet(&self, id: u16, reflector: IpAddr, buf: &[u8]) -> Result<PingReply, PingError>;
}
pub trait PingSender {
@@ -112,7 +121,7 @@ pub trait PingSender {
id: u16,
type_: MeasurementType,
reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
) -> Result<(), Box<dyn Error>> {
) -> anyhow::Result<()> {
let socket = open_socket(type_)?;
let mut seq: u16 = 0;
+16 -21
View File
@@ -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<PingReply, Box<dyn Error>> {
fn parse_packet(&self, id: u16, reflector: IpAddr, buf: &[u8]) -> Result<PingReply, PingError> {
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(),
}))
}
}
+14 -22
View File
@@ -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<PingReply, Box<dyn Error>> {
fn parse_packet(&self, id: u16, reflector: IpAddr, buf: &[u8]) -> Result<PingReply, PingError> {
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(),
}))
}
}
+139 -88
View File
@@ -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<f64> {
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<f64>,
stats_direction: StatsDirection,
}
pub struct Ratecontroller {
pub(crate) config: Config,
pub(crate) owd_baseline: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
pub(crate) owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
pub(crate) reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
pub(crate) reselect_trigger: Sender<bool>,
config: Config,
owd_baseline: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
reselect_trigger: Sender<bool>,
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<Mutex<HashMap<IpAddr, ReflectorStats>>>,
owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
reselect_trigger: Sender<bool>,
down_direction: StatsDirection,
up_direction: StatsDirection,
) -> anyhow::Result<Self> {
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<f64> = Vec::new();
let mut safe_ul_rates: Vec<f64> = 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<f64> =
generate_initial_speeds(self.config.download_base_kbits, self.config.speed_hist_size);
let mut safe_ul_rates: Vec<f64> =
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
+1 -1
View File
@@ -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(
+12 -4
View File
@@ -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")]