Lochnair-Labs/sqm-autorate-rust
2
0 Fork 0
Code Issues 5 Pull Requests 2 Actions 4 Packages Projects Releases Wiki Activity

Implement missing features, some cleanup

Browse Source
This commit is contained in:
Nils Andreas Svee committed 2022-09-28 00:25:35 +02:00
1 parent c06078fccb
commit e3b7b7e9ba
12 files changed
+704 -174
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/baseliner.rs
+11 -6
View File
@@ -1,5 +1,6 @@
use crate::pinger::PingReply;
use crate::{Config, Utils};
use log::info;
use std::collections::HashMap;
use std::net::IpAddr;
use std::sync::mpsc::{Receiver, Sender};
@@ -30,8 +31,8 @@ impl Baseliner {
* aren't bloat related, with less sensitivity (bigger numbers) we smooth through quick spikes
* but take longer to respond to real bufferbloat
*/
let slow_factor = Utils::ewma_factor(self.config.tick_duration, 135.0);
let fast_factor = Utils::ewma_factor(self.config.tick_duration, 0.4);
let slow_factor = Utils::ewma_factor(self.config.tick_interval, 135.0);
let fast_factor = Utils::ewma_factor(self.config.tick_interval, 0.4);
loop {
let time_data = self.stats_receiver.recv().unwrap();
@@ -39,13 +40,13 @@ impl Baseliner {
let mut owd_baseline_map = self.owd_baseline.lock().unwrap();
let mut owd_recent_map = self.owd_recent.lock().unwrap();
let mut owd_baseline_new = ReflectorStats {
let owd_baseline_new = ReflectorStats {
down_ewma: time_data.down_time,
up_ewma: time_data.down_time,
last_receive_time_s: time_data.last_receive_time_s,
};
let mut owd_recent_new = ReflectorStats {
let owd_recent_new = ReflectorStats {
down_ewma: time_data.down_time,
up_ewma: time_data.down_time,
last_receive_time_s: time_data.last_receive_time_s,
@@ -79,6 +80,10 @@ impl Baseliner {
{
owd_baseline.last_receive_time_s = time_data.last_receive_time_s - 60.0;
owd_recent.last_receive_time_s = time_data.last_receive_time_s - 60.0;
info!(
"Reflector {} has OWD > 5 seconds more than baseline, triggering reselection",
time_data.reflector
);
let _ = self.reselect_trigger.send(true);
} else {
owd_baseline.down_ewma = owd_baseline.down_ewma * slow_factor
@@ -100,11 +105,11 @@ impl Baseliner {
}
}
println!(
info!(
"Reflector {} up baseline = {} down baseline = {}",
time_data.reflector, owd_baseline.up_ewma, owd_baseline.down_ewma
);
println!(
info!(
"Reflector {} up recent = {} down recent = {}",
time_data.reflector, owd_recent.up_ewma, owd_recent.down_ewma
);
src/config.rs
+190 -35
View File
@@ -1,3 +1,5 @@
use crate::error::{ConfigParseError, MissingConfigError};
use log::Level;
#[cfg(feature = "uci")]
use rust_uci::Uci;
use std::error::Error;
@@ -16,55 +18,208 @@ where
Ok(io::BufReader::new(file).lines())
}
#[allow(dead_code)]
#[derive(Clone)]
#[derive(Clone, Debug)]
pub(crate) enum AutorateType {
ICMP = 1,
ICMPTimestamps,
NTP,
TCPTimestamps,
}
#[derive(Clone, Debug)]
pub(crate) struct Config {
// Network section
pub(crate) download_interface: String,
pub(crate) upload_interface: String,
pub(crate) download_base_kbits: f64,
pub(crate) download_min_kbits: f64,
pub(crate) upload_base_kbits: f64,
pub(crate) upload_min_kbits: f64,
pub(crate) reflector_list_file: String,
// Output section
pub(crate) log_level: Level,
pub(crate) speed_hist_file: String,
pub(crate) stats_file: String,
pub(crate) suppress_statistics: bool,
// Advanced section
pub(crate) download_delay_ms: f64,
pub(crate) high_load_level: f64,
pub(crate) min_change_interval: f64,
pub(crate) num_reflectors: u8,
pub(crate) tick_duration: f64,
pub(crate) reflector_list_file: String,
pub(crate) speed_hist_size: u32,
pub(crate) tick_interval: f64,
pub(crate) upload_delay_ms: f64,
}
impl Config {
pub fn new() -> Self {
// First try to load from UCI (if enabled), if nothing is returned, use the environment instead.
match Self::load_from_uci() {
Some(config) => config,
None => Self {
download_interface: env::var("DOWNLOAD_INTERFACE")
.expect("A download interface is required"),
upload_interface: env::var("UPLOAD_INTERFACE")
.expect("A upload interface is required"),
reflector_list_file: env::var("REFLECTOR_LIST_FILE")
.expect("A reflector list is required"),
min_change_interval: 0.5,
num_reflectors: 5,
tick_duration: 0.5,
},
}
}
#[cfg(feature = "uci")]
fn load_from_uci() -> Option<Self> {
let uci: Uci = Uci::new().expect("Couldn't create UCI instance");
Some(Self {
download_interface: "wan".to_string(),
upload_interface: "wan".to_string(),
reflector_list_file: "".to_string(),
min_change_interval: 0.5,
num_reflectors: 5,
tick_duration: 0.5,
pub fn new() -> Result<Self, Box<dyn Error>> {
Ok(Self {
// Network section
download_base_kbits: Self::get::<f64>(
"SQMA_DOWNLOAD_BASE_KBITS",
"sqm-autorate.@network[0].download_base_kbits",
None,
)?,
download_interface: Self::get::<String>(
"SQMA_DOWNLOAD_INTERFACE",
"sqm-autorate.@network[0].download_interface",
None,
)?,
download_min_kbits: Self::get::<f64>(
"SQMA_DOWNLOAD_MIN_KBITS",
"sqm-autorate.@network[0].download_min_kbits",
None,
)?,
upload_base_kbits: Self::get::<f64>(
"SQMA_UPLOAD_BASE_KBITS",
"sqm-autorate.@network[0].upload_base_kbits",
None,
)?,
upload_interface: Self::get::<String>(
"SQMA_UPLOAD_INTERFACE",
"sqm-autorate.@network[0].upload_interface",
None,
)?,
upload_min_kbits: Self::get::<f64>(
"SQMA_UPLOAD_MIN_KBITS",
"sqm-autorate.@network[0].upload_min_kbits",
None,
)?,
// Output section
log_level: Self::get::<Level>(
"SQMA_LOG_LEVEL",
"sqm-autorate.@output[0].log_level",
Some(Level::Error),
)?,
speed_hist_file: Self::get::<String>(
"SQMA_SPEED_HIST_FILE",
"sqm-autorate.@output[0].speed_hist_file",
Some("/tmp/sqm-speedhist.csv".parse()?),
)?,
stats_file: Self::get::<String>(
"SQMA_STATS_FILE",
"sqm-autorate.@output[0].stats_file",
Some("/tmp/sqm-autorate.csv".parse()?),
)?,
suppress_statistics: Self::get::<bool>(
"SQMA_SUPPRESS_STATISTICS",
"sqm-autorate.@output[0].suppress_statistics",
Some(false),
)?,
// Advanced section
download_delay_ms: Self::get::<f64>(
"SQMA_DOWNLOAD_DELAY_MS",
"sqm-autorate.@advanced_settings[0].download_delay_ms",
Some(15.0),
)?,
high_load_level: Self::get::<f64>(
"SQMA_HIGH_LOAD_LEVEL",
"sqm-autorate.@advanced_settings[0].high_load_level",
Some(0.8),
)?,
min_change_interval: Self::get::<f64>(
"SQMA_MIN_CHANGE_INTERVAL",
"sqm-autorate.@advanced_settings[0].min_change_interval",
Some(0.5),
)?,
num_reflectors: Self::get::<u8>(
"SQMA_NUM_REFLECTORS",
"sqm-autorate.@advanced_settings[0].num_reflectors",
Some(5),
)?,
reflector_list_file: Self::get::<String>(
"SQMA_REFLECTOR_LIST_FILE",
"sqm-autorate.@advanced_settings[0].reflector_list_file",
Some("/etc/sqm-autorate/reflectors-icmp.csv".parse()?),
)?,
speed_hist_size: Self::get::<u32>(
"SQMA_SPEED_HIST_SIZE",
"sqm-autorate.@advanced_settings[0].speed_hist_size",
Some(100),
)?,
tick_interval: Self::get::<f64>(
"SQMA_TICK_INTERVAL",
"sqm-autorate.@advanced_settings[0].tick_interval",
Some(0.5),
)?,
upload_delay_ms: Self::get::<f64>(
"SQMA_UPLOAD_DELAY_MS",
"sqm-autorate.@advanced_settings[0].upload_delay_ms",
Some(15.0),
)?,
})
}
fn get<T: std::str::FromStr>(
env_key: &str,
uci_key: &str,
default: Option<T>,
) -> Result<T, Box<dyn Error>> {
return match Self::get_value(env_key, uci_key, None) {
Ok(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(e) => {
return match default {
Some(val) => Ok(val),
None => Err(e),
}
}
};
}
fn get_value(
env_key: &str,
uci_key: &str,
default: Option<String>,
) -> Result<String, Box<dyn Error>> {
let mut err: Option<Box<dyn Error>>;
match env::var(env_key) {
Ok(val) => return Ok(val),
Err(e) => err = Some(Box::new(e)),
};
match Self::get_from_uci(uci_key) {
Ok(opt) => match opt {
Some(val) => return Ok(val),
None => {}
},
Err(e) => err = Some(e),
};
match default {
Some(val) => return Ok(val),
None => {
err = Some(Box::new(MissingConfigError {
config_key: env_key.to_string(),
}))
}
}
Err(err.unwrap())
}
#[cfg(feature = "uci")]
fn get_from_uci(key: &str) -> Result<Option<String>, Box<dyn Error>> {
let mut uci: Uci = Uci::new()?;
return match uci.get(key) {
Ok(val) => Ok(Some(val)),
Err(e) => Err(Box::new(e)),
};
}
#[cfg(not(feature = "uci"))]
fn load_from_uci() -> Option<Self> {
None
fn get_from_uci(_: &str) -> Result<Option<String>, Box<dyn Error>> {
Ok(None)
}
pub fn load_reflectors(&self) -> Result<Vec<IpAddr>, Box<dyn Error>> {
src/error.rs
+44 -2
View File
@@ -2,12 +2,55 @@ 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 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)> {
// The compiler transparently casts `&sqlx::Error` into a `&dyn Error`
None
}
}
@@ -24,7 +67,6 @@ pub struct PingParseError;
impl Error for PingParseError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
// The compiler transparently casts `&sqlx::Error` into a `&dyn Error`
None
}
}
src/log.rs
+55
View File
@@ -0,0 +1,55 @@
use static_init::dynamic;
use std::error::Error;
use std::fmt;
use std::str::FromStr;
use std::sync::{Arc, RwLock};
use std::time::SystemTime;
use log::{Level, LevelFilter, Metadata, Record, SetLoggerError};
use time::format_description::FormatItem;
use time::formatting::Formattable;
use time::macros::format_description;
use time::OffsetDateTime;
const LOG_DATETIME_FORMAT: &[FormatItem] = format_description!(
"[year]-[month]-[day] [hour]:[minute]:[second] [offset_hour \
sign:mandatory]:[offset_minute]:[offset_second]"
);
#[derive(Clone, Copy)]
pub struct SimpleLogger {
pub level: log::Level,
}
fn time_format<T>(dt: T, format: &(impl Formattable + ?Sized)) -> String
where
T: Into<OffsetDateTime>,
{
dt.into().format(format).unwrap()
}
impl log::Log for SimpleLogger {
fn enabled(&self, metadata: &Metadata) -> bool {
metadata.level() <= Level::Info
}
fn log(&self, record: &Record) {
if self.enabled(record.metadata()) {
println!(
"{} {:5} {}:{}: {}",
time_format(SystemTime::now(), &LOG_DATETIME_FORMAT),
record.level(),
record.file().unwrap(),
record.line().unwrap(),
record.args()
);
}
}
fn flush(&self) {}
}
pub fn init(level: Level) -> Result<(), SetLoggerError> {
log::set_boxed_logger(Box::new(SimpleLogger { level }))
.map(|()| log::set_max_level(LevelFilter::Trace))
}
src/main.rs
+59 -22
View File
@@ -4,6 +4,7 @@ mod baseliner;
mod cake;
mod config;
mod error;
mod log;
mod netlink;
mod pinger;
mod pinger_icmp;
@@ -13,7 +14,9 @@ mod reflector_selector;
mod utils;
use crate::baseliner::{Baseliner, ReflectorStats};
use ::log::{error, info, Level, LevelFilter};
use std::collections::HashMap;
use std::error::Error;
use std::net::IpAddr;
use std::process::ExitCode;
use std::str::FromStr;
@@ -21,25 +24,39 @@ use std::sync::mpsc::channel;
use std::sync::{Arc, Mutex};
use std::thread::sleep;
use std::time::Duration;
use std::{process, thread};
use std::{panic, process, thread};
use crate::config::Config;
use crate::netlink::Netlink;
use crate::log::SimpleLogger;
use crate::netlink::{Netlink, Qdisc};
use crate::pinger::{PingListener, PingSender, SocketType};
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 = "0.0.1";
fn main() -> ExitCode {
println!("starting up");
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;
}
};
crate::log::init(config.log_level).expect("Couldn't initialize logger");
let config = Config::new();
let mut reflectors = match config.load_reflectors() {
Ok(refl) => refl,
Err(e) => {
println!("Couldn't load reflectors: {}", e.to_string());
info!("Couldn't load reflector list: {}", e.to_string());
return ExitCode::FAILURE;
}
};
@@ -49,9 +66,9 @@ fn main() -> ExitCode {
let id = (process::id() & 0xFFFF) as u16;
// Create data structures shared by different threads
let mut owd_baseline = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new()));
let mut owd_recent = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new()));
let mut reflector_peers_lock = Arc::new(Mutex::new(Vec::<IpAddr>::new()));
let owd_baseline = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new()));
let owd_recent = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new()));
let reflector_peers_lock = Arc::new(Mutex::new(Vec::<IpAddr>::new()));
let mut reflector_pool = Vec::<IpAddr>::new();
let reflector_pool_size = reflectors.len();
@@ -84,7 +101,7 @@ fn main() -> ExitCode {
let mut pinger_receiver = PingerICMPEchoListener::new(id);
let mut pinger_sender = PingerICMPEchoSender::new(id);
let mut baseliner = Baseliner {
let baseliner = Baseliner {
config: config.clone(),
owd_baseline: owd_baseline.clone(),
owd_recent: owd_recent.clone(),
@@ -92,12 +109,20 @@ fn main() -> ExitCode {
stats_receiver: baseliner_stats_receiver,
};
let dl_intf = config.clone().download_interface;
let ul_intf = config.clone().upload_interface;
let down_ifindex = Netlink::find_interface(dl_intf.as_str()).unwrap();
let up_ifindex = Netlink::find_interface(ul_intf.as_str()).unwrap();
let down_qdisc = Netlink::find_qdisc(down_ifindex).unwrap();
let up_qdisc = Netlink::find_qdisc(up_ifindex).unwrap();
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;
}
};
/* Set initial TC values to minimum
* so there should be no initial bufferbloat to
@@ -111,11 +136,17 @@ fn main() -> ExitCode {
let receiver_handle = thread::Builder::new()
.name("receiver".to_string())
.spawn(move || {
pinger_receiver.listen(
match pinger_receiver.listen(
SocketType::ICMP,
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()
@@ -125,7 +156,15 @@ fn main() -> ExitCode {
let reflector_peers_lock_clone = reflector_peers_lock.clone();
let sender_handle = thread::Builder::new()
.name("sender".to_string())
.spawn(move || pinger_sender.send(SocketType::ICMP, reflector_peers_lock_clone))
.spawn(
move || match pinger_sender.send(SocketType::ICMP, reflector_peers_lock_clone) {
Ok(_) => {}
Err(e) => {
error!("Error occured in sender thread: {}", e);
panic!();
}
},
)
.expect("Couldn't spawn ping sender thread");
let mut threads = vec![receiver_handle, sender_handle, baseliner_handle];
@@ -133,7 +172,6 @@ fn main() -> ExitCode {
if reflector_pool_size > 5 {
let reflector_selector = ReflectorSelector {
config: config.clone(),
owd_baseline: owd_baseline.clone(),
owd_recent: owd_recent.clone(),
reflector_peers_lock: reflector_peers_lock.clone(),
reflector_pool,
@@ -159,13 +197,12 @@ fn main() -> ExitCode {
let ratecontroller_handle = thread::Builder::new()
.name("ratecontroller".to_string())
.spawn(move || ratecontroller.run(StatsDirection::TX, StatsDirection::RX))
.spawn(move || ratecontroller.run(StatsDirection::RX, StatsDirection::TX))
.expect("Couldn't spawn ratecontroller thread");
threads.push(ratecontroller_handle);
for thread in threads {
println!("thread: {}", thread.thread().name().unwrap());
thread.join().unwrap();
}
src/netlink.rs
+8 -3
View File
@@ -137,7 +137,7 @@ impl Netlink {
Err(Box::new(NlError::msg("Error getting stats")))
}
pub fn find_qdisc(ifindex: i32) -> Result<Qdisc, Box<dyn Error>> {
pub fn qdisc_from_ifindex(ifindex: i32) -> Result<Qdisc, Box<dyn Error>> {
let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap();
let tc_msg = Tcmsg::new(
u8::from(RtAddrFamily::Unspecified),
@@ -199,9 +199,14 @@ impl Netlink {
return Err(Box::new(NoQdiscFoundError));
}
pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth: u64) -> Result<(), Box<dyn Error>> {
pub fn qdisc_from_ifname(ifname: &str) -> Result<Qdisc, Box<dyn Error>> {
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>> {
let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap();
let bandwidth = bandwidth * 1000 / 8;
let bandwidth = bandwidth_kbit * 1000 / 8;
let mut attrs = RtBuffer::new();
src/pinger.rs
+18 -10
View File
@@ -1,3 +1,4 @@
use log::error;
use nix::sys::socket::{
recvfrom, sendto, socket, AddressFamily, MsgFlags, SockFlag, SockProtocol, SockType,
SockaddrIn, SockaddrIn6, SockaddrLike,
@@ -16,6 +17,7 @@ pub enum SocketType {
UDP,
}
#[allow(dead_code)]
pub struct PingReply {
pub(crate) reflector: IpAddr,
pub(crate) seq: u16,
@@ -29,9 +31,9 @@ pub struct PingReply {
pub(crate) last_receive_time_s: f64,
}
fn open_socket(type_: SocketType) -> RawFd {
fn open_socket(type_: SocketType) -> nix::Result<RawFd> {
match type_ {
ICMP => {
SocketType::ICMP => {
socket(
AddressFamily::Inet,
SockType::Raw,
@@ -39,7 +41,7 @@ fn open_socket(type_: SocketType) -> RawFd {
SockProtocol::ICMP,
)
}
UDP => {
SocketType::UDP => {
socket(
AddressFamily::Inet,
SockType::Datagram,
@@ -48,7 +50,6 @@ fn open_socket(type_: SocketType) -> RawFd {
)
}
}
.expect("Couldn't open socket")
}
pub trait PingListener {
@@ -60,8 +61,8 @@ pub trait PingListener {
type_: SocketType,
reflectors_lock: Arc<Mutex<Vec<IpAddr>>>,
stats_sender: Sender<PingReply>,
) {
let socket = open_socket(type_);
) -> Result<(), Box<dyn Error>> {
let socket = open_socket(type_)?;
loop {
let mut v: Vec<u8> = vec![0; 40];
@@ -87,12 +88,15 @@ pub trait PingListener {
let reply = match Self::parse_packet(addr, buf, size) {
Ok(val) => val,
Err(e) => {
println!("Shit went to hell: {}", e.to_string());
error!(
"Something went wrong while parsing packet: {}",
e.to_string()
);
continue;
}
};
println!("Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:8} | Received time: {:8} | Transmit time : {:8} | RTT: {:8} | UL time: {:5} | DL time: {:5}", "ICMP", addr.to_string(), reply.seq, reply.current_time, reply.originate_timestamp, reply.receive_timestamp, reply.transmit_timestamp, reply.rtt, reply.up_time, reply.down_time);
//debug!("Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:8} | Received time: {:8} | Transmit time : {:8} | RTT: {:8} | UL time: {:5} | DL time: {:5}", "ICMP", addr.to_string(), reply.seq, reply.current_time, reply.originate_timestamp, reply.receive_timestamp, reply.transmit_timestamp, reply.rtt, reply.up_time, reply.down_time);
stats_sender.send(reply).unwrap();
}
}
@@ -105,8 +109,12 @@ pub trait PingSender {
fn new(id: u16) -> Self;
fn get_id(&self) -> u16;
fn send(&mut self, type_: SocketType, reflectors_lock: Arc<Mutex<Vec<IpAddr>>>) {
let socket = open_socket(type_);
fn send(
&mut self,
type_: SocketType,
reflectors_lock: Arc<Mutex<Vec<IpAddr>>>,
) -> Result<(), Box<dyn Error>> {
let socket = open_socket(type_)?;
let mut seq: u16 = 0;
let tick_duration_ms: u16 = 500;
src/pinger_icmp.rs
+3 -13
View File
@@ -1,6 +1,7 @@
use std::error::Error;
use std::net::IpAddr;
use crate::error::PingParseError;
use crate::pinger::{PingListener, PingReply, PingSender};
use byteorder::*;
use etherparse::TransportSlice::{Icmpv4, Icmpv6};
@@ -63,24 +64,13 @@ impl PingListener for PingerICMPEchoListener {
}
_ => {}
},
Some(Icmpv6(icmp)) => {}
Some(Icmpv6(_)) => {}
Some(_) => {}
None => {}
},
}
Ok(PingReply {
reflector: IpAddr::from([0, 0, 0, 0]),
seq: 0,
rtt: 0,
current_time: 0,
down_time: 0.0,
up_time: 0.0,
originate_timestamp: 0,
receive_timestamp: 0,
transmit_timestamp: 0,
last_receive_time_s: 0.0,
})
Err(Box::new(PingParseError {}))
}
}
src/ratecontroller.rs
+127 -45
View File
@@ -1,9 +1,12 @@
use crate::netlink::Netlink;
use crate::{Config, ReflectorStats, Utils};
use log::{debug, error, info, warn};
use rand::seq::SliceRandom;
use rand::thread_rng;
use rand::RngCore;
use std::collections::HashMap;
use std::fs::File;
use std::io::Write;
use std::net::IpAddr;
use std::sync::mpsc::Sender;
use std::sync::{Arc, Mutex};
@@ -33,26 +36,26 @@ fn get_interface_stats(
{
Ok(stats) => (stats.rx_bytes as i128, stats.tx_bytes as i128),
Err(e) => {
println!("Error while retrieving stats: {}", e.to_string());
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) => {
println!("Error while retrieving stats: {}", e.to_string());
error!("Error while retrieving stats: {}", e.to_string());
(-1, -1)
}
};
let rx_bytes = match down_direction {
RX => down_rx,
TX => down_tx,
StatsDirection::RX => down_rx,
StatsDirection::TX => down_tx,
};
let tx_bytes = match up_direction {
RX => up_rx,
TX => up_tx,
StatsDirection::RX => up_rx,
StatsDirection::TX => up_tx,
};
(rx_bytes, tx_bytes)
@@ -79,26 +82,13 @@ impl Ratecontroller {
let mut rng = thread_rng();
let base_dl_rate = 45000.0;
let base_ul_rate = 15000.0;
let min_dl_rate = 5000.0;
let min_ul_rate = 1000.0;
let histsize = 100;
let dl_max_delta_owd = 15.0;
let ul_max_delta_owd = 15.0;
let high_load_level = 0.8;
let down_ifindex =
Netlink::find_interface(self.config.download_interface.as_str()).unwrap();
let up_ifindex = Netlink::find_interface(self.config.upload_interface.as_str()).unwrap();
let down_qdisc = Netlink::find_qdisc(down_ifindex).unwrap();
let up_qdisc = Netlink::find_qdisc(up_ifindex).unwrap();
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();
// set qdisc rates to 60% of base rate to make sure we start with sane baselines
let mut cur_dl_rate: f64 = base_dl_rate * 0.6;
let mut cur_ul_rate: f64 = base_ul_rate * 0.6;
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();
@@ -113,14 +103,56 @@ impl Ratecontroller {
let mut safe_dl_rates: Vec<f64> = Vec::new();
let mut safe_ul_rates: Vec<f64> = Vec::new();
for i in 0..histsize {
safe_dl_rates.push((rand::thread_rng().next_u32() as f64 * 0.2 + 0.75) * base_dl_rate);
safe_ul_rates.push((rand::thread_rng().next_u32() as f64 * 0.2 + 0.75) * base_dl_rate);
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 nrate_up = 0;
let mut nrate_down = 0;
let mut speed_hist_fd: Option<File> = None;
let mut speed_hist_fd_inner: File;
let mut stats_fd: Option<File> = None;
let mut stats_fd_inner: File;
if !self.config.suppress_statistics {
speed_hist_fd_inner = File::options()
.create(true)
.write(true)
.open(self.config.speed_hist_file.as_str())
.unwrap();
speed_hist_fd_inner
.write("time,counter,upspeed,downspeed\n".as_bytes())
.unwrap();
speed_hist_fd_inner.flush().unwrap();
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();
stats_fd_inner
.write(
"times,timens,rxload,txload,deltadelaydown,deltadelayup,dlrate,uprate\n"
.as_bytes(),
)
.unwrap();
stats_fd_inner.flush().unwrap();
stats_fd = Some(stats_fd_inner);
}
loop {
let (mut now_s, now_ns) = Utils::get_current_time();
let now_abstime = now_s + now_ns / 1e9;
@@ -128,6 +160,8 @@ impl Ratecontroller {
let now_t = now_s + now_ns / 1e9;
if now_t - lastchg_t > self.config.min_change_interval {
// if it's been long enough, and the stats indicate needing to change speeds
// change speeds here
let owd_baseline = self.owd_baseline.lock().unwrap();
let owd_recent = self.owd_recent.lock().unwrap();
let reflectors = self.reflectors_lock.lock().unwrap();
@@ -135,6 +169,8 @@ impl Ratecontroller {
let mut down_delta_stat: f64;
let mut up_delta_stat: f64;
// If we have no reflector peers to iterate over, don't attempt any rate changes.
// This will occur under normal operation when the reflector peers table is updated.
if reflectors.len() > 0 {
let mut down_deltas = Vec::new();
let mut up_deltas = Vec::new();
@@ -153,10 +189,11 @@ impl Ratecontroller {
up_delta_stat = 0.0;
for reflector in reflectors.iter() {
// only consider this data if it's less than 2 * tick_duration seconds old
if owd_baseline.contains_key(reflector)
&& owd_recent.contains_key(reflector)
&& owd_recent[reflector].last_receive_time_s
> now_abstime - 2.0 * self.config.tick_duration
> now_abstime - 2.0 * self.config.tick_interval
{
down_deltas.push(
owd_recent[reflector].down_ewma - owd_baseline[reflector].down_ewma,
@@ -164,11 +201,19 @@ impl Ratecontroller {
up_deltas.push(
owd_recent[reflector].up_ewma - owd_baseline[reflector].up_ewma,
);
debug!(
"Reflector: {} down_delay: {} up_delay: {}",
reflector,
down_deltas.last().unwrap(),
up_deltas.last().unwrap()
);
}
}
if down_deltas.len() < 5 || up_deltas.len() < 5 {
// trigger reselection
warn!("Not enough delta values, triggering reselection");
let _ = self.reselect_trigger.send(true);
}
@@ -178,10 +223,10 @@ impl Ratecontroller {
let (cur_rx_bytes, cur_tx_bytes) =
get_interface_stats(self.config.clone(), down_direction, up_direction);
if cur_rx_bytes == -1 || cur_tx_bytes == -1 {
println!("One or both Netlink stats could not be read. Skipping rate control algorithm");
warn!("One or both Netlink stats could not be read. Skipping rate control algorithm");
} else if down_deltas.len() < 3 || up_deltas.len() < 3 {
next_dl_rate = min_dl_rate;
next_ul_rate = min_ul_rate;
next_dl_rate = self.config.download_min_kbits;
next_ul_rate = self.config.upload_min_kbits;
} else {
down_deltas.sort_by(|a, b| a.partial_cmp(b).unwrap());
up_deltas.sort_by(|a, b| a.partial_cmp(b).unwrap());
@@ -207,34 +252,34 @@ impl Ratecontroller {
next_dl_rate = cur_dl_rate;
if down_delta_stat > 0.0
&& down_delta_stat < dl_max_delta_owd
&& rx_load > high_load_level
&& down_delta_stat < self.config.download_delay_ms
&& rx_load > self.config.high_load_level
{
safe_dl_rates[nrate_down] = (cur_dl_rate * rx_load).floor();
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))
+ (base_dl_rate * 0.03);
+ (self.config.download_base_kbits * 0.03);
nrate_down = nrate_down + 1;
nrate_down = nrate_down % histsize;
nrate_down = nrate_down % self.config.speed_hist_size as usize;
}
if up_delta_stat > 0.0
&& up_delta_stat < dl_max_delta_owd
&& tx_load > high_load_level
&& up_delta_stat < self.config.download_delay_ms
&& tx_load > self.config.high_load_level
{
safe_ul_rates[nrate_up] = (cur_ul_rate * tx_load).floor();
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))
+ (base_dl_rate * 0.03);
+ (self.config.download_base_kbits * 0.03);
nrate_up = nrate_up + 1;
nrate_up = nrate_up % histsize;
nrate_up = nrate_up % self.config.speed_hist_size as usize;
}
if down_delta_stat > dl_max_delta_owd {
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,
@@ -245,7 +290,7 @@ impl Ratecontroller {
}
}
if up_delta_stat > ul_max_delta_owd {
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,
@@ -262,11 +307,11 @@ impl Ratecontroller {
prev_rx_bytes = cur_rx_bytes;
prev_tx_bytes = cur_tx_bytes;
next_dl_rate = f64_max(min_dl_rate, next_dl_rate).floor();
next_ul_rate = f64_max(min_ul_rate, next_ul_rate).floor();
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();
if next_dl_rate != cur_dl_rate || next_ul_rate != cur_ul_rate {
println!(
info!(
"next_ul_rate {} next_dl_rate {}",
next_ul_rate, next_dl_rate
);
@@ -290,7 +335,7 @@ impl Ratecontroller {
&& down_delta_stat > 0.0
&& up_delta_stat > 0.0
{
println!(
debug!(
"{},{},{},{},{},{},{},{}",
lastchg_s,
lastchg_ns,
@@ -301,6 +346,25 @@ impl Ratecontroller {
cur_dl_rate,
cur_ul_rate
);
if let Some(ref mut fd) = stats_fd {
if let Err(e) = fd.write(
format!(
"{},{},{},{},{},{},{},{}\n",
lastchg_s,
lastchg_ns,
rx_load,
tx_load,
down_delta_stat,
up_delta_stat,
cur_dl_rate,
cur_ul_rate
)
.as_bytes(),
) {
warn!("Failed to write statistics: {}", e);
}
}
}
lastchg_s = lastchg_s - start_s;
@@ -308,6 +372,24 @@ impl Ratecontroller {
}
}
if let Some(ref mut fd) = speed_hist_fd {
if now_t - lastdump_t > 300.0 {
for i in 0..self.config.speed_hist_size as usize {
if let Err(e) = fd.write(
format!(
"{},{},{},{}\n",
now_t, i, safe_ul_rates[i], safe_dl_rates[i]
)
.as_bytes(),
) {
warn!("Failed to write speed history file: {}", e);
}
}
lastdump_t = now_t;
}
}
sleep(sleep_time);
}
}
src/reflector_selector.rs
+13 -10
View File
@@ -1,4 +1,5 @@
use crate::{Config, ReflectorStats};
use log::{debug, info};
use rand::seq::SliceRandom;
use rand::{thread_rng, Rng};
use std::collections::HashMap;
@@ -10,7 +11,6 @@ use std::time::Duration;
pub struct ReflectorSelector {
pub(crate) config: Config,
pub(crate) owd_baseline: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
pub(crate) owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
pub(crate) reflector_peers_lock: Arc<Mutex<Vec<IpAddr>>>,
pub(crate) reflector_pool: Vec<IpAddr>,
@@ -22,8 +22,8 @@ impl ReflectorSelector {
let mut selector_sleep_time = Duration::new(30, 0);
let mut reselection_count = 0;
let baseline_sleep_time = Duration::new(
(self.config.tick_duration * std::f64::consts::PI) as u64,
(((self.config.tick_duration * std::f64::consts::PI) % 1.0) * 1e9) as u32,
(self.config.tick_interval * std::f64::consts::PI) as u64,
(((self.config.tick_interval * std::f64::consts::PI) % 1.0) * 1e9) as u32,
);
let mut rng = thread_rng();
@@ -42,6 +42,7 @@ impl ReflectorSelector {
.recv_timeout(selector_sleep_time)
.unwrap_or(true);
reselection_count += 1;
info!("Starting reselection [#{}]", reselection_count);
// After 40 reselections, slow down to every 15 minutes
if reselection_count > 40 {
@@ -53,11 +54,13 @@ impl ReflectorSelector {
// Include all current peers
for reflector in reflectors_peers.iter() {
debug!("Current peer: {}", reflector.to_string());
next_peers.push(*reflector);
}
for i in 1..20 {
for _ in 1..20 {
let next_candidate = self.reflector_pool.choose(&mut rng).unwrap();
debug!("Next candidate: {}", next_candidate.to_string());
next_peers.push(*next_candidate);
}
@@ -68,6 +71,7 @@ impl ReflectorSelector {
// until the guard goes out of scope
drop(reflectors_peers);
debug!("Waiting for candidates to be baselined");
// Wait for several seconds to allow all reflectors to be re-baselined
sleep(baseline_sleep_time);
@@ -76,16 +80,15 @@ impl ReflectorSelector {
reflectors_peers.len();
let mut candidates = Vec::new();
let owd_baseline = self.owd_baseline.lock().unwrap();
let owd_recent = self.owd_recent.lock().unwrap();
for peer in next_peers {
if owd_recent.contains_key(&peer) {
let rtt = (owd_recent[&peer].down_ewma + owd_recent[&peer].up_ewma) as u64;
candidates.push((peer, rtt));
println!("Candidate reflector: {} RTT: {}", peer.to_string(), rtt);
info!("Candidate reflector: {} RTT: {}", peer.to_string(), rtt);
} else {
println!(
info!(
"No data found from candidate reflector: {} - skipping",
peer.to_string()
);
@@ -101,7 +104,7 @@ impl ReflectorSelector {
candidates = candidates[0..candidate_pool_num - 1].to_vec();
for (candidate, rtt) in candidates.iter() {
println!("Fastest candidate {}: {}", candidate, rtt);
info!("Fastest candidate {}: {}", candidate, rtt);
}
// Shuffle the deck so we avoid overwhelming good reflectors (Fisher-Yates)
@@ -115,9 +118,9 @@ impl ReflectorSelector {
}
let mut new_peers = Vec::new();
for i in 1..num_reflectors {
for i in 0..num_reflectors {
new_peers.push(candidates[i as usize].0);
println!(
info!(
"New selected peer: {}",
candidates[i as usize].0.to_string()
);