diff --git a/Cargo.lock b/Cargo.lock index 2ede7f6..995cc3e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -109,10 +109,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" [[package]] -name = "clang-sys" -version = "1.3.3" +name = "cfg_aliases" +version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a050e2153c5be08febd6734e29298e844fdb0fa21aeddd63b4eb7baa106c69b" +checksum = "fd16c4719339c4530435d38e511904438d07cce7950afa3718a84ac36c10e89e" + +[[package]] +name = "clang-sys" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa2e27ae6ab525c3d369ded447057bca5438d86dc3a68f6faafb8269ba82ebf3" dependencies = [ "glob", "libc", @@ -145,9 +151,9 @@ dependencies = [ [[package]] name = "either" -version = "1.7.0" +version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f107b87b6afc2a64fd13cac55fe06d6c8859f12d4b14cbcdd2c67d0976781be" +checksum = "90e5c1c8368803113bf0c9584fc495a58b86dc8a29edbf8fe877d21d9507e797" [[package]] name = "env_logger" @@ -203,6 +209,21 @@ version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4" +[[package]] +name = "instant" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a5bbe824c507c5da5956355e86a746d82e0e1464f65d862cc5e71da70e94b2c" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "itoa" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c8af84674fe1f223a982c933a0ee1086ac4d4052aa0fb8060c12c6ad838e754" + [[package]] name = "lazy_static" version = "1.4.0" @@ -217,9 +238,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" [[package]] name = "libc" -version = "0.2.132" +version = "0.2.133" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8371e4e5341c3a96db127eb2465ac681ced4c433e01dd0e938adbef26ba93ba5" +checksum = "c0f80d65747a3e43d1596c7c5492d95d5edddaabd45a7fcdb02b95f644164966" [[package]] name = "libloading" @@ -241,6 +262,16 @@ dependencies = [ "cmake", ] +[[package]] +name = "lock_api" +version = "0.4.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "435011366fe56583b16cf956f9df0095b405b82d76425bc8981c0e22e60ec4df" +dependencies = [ + "autocfg", + "scopeguard", +] + [[package]] name = "log" version = "0.4.17" @@ -312,6 +343,40 @@ dependencies = [ "version_check", ] +[[package]] +name = "num_threads" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2819ce041d2ee131036f4fc9d6ae7ae125a3a40e97ba64d04fe799ad9dabbb44" +dependencies = [ + "libc", +] + +[[package]] +name = "parking_lot" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99" +dependencies = [ + "instant", + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d76e8e1493bcac0d2766c42737f34458f1c8c50c0d23bcb24ea953affb273216" +dependencies = [ + "cfg-if", + "instant", + "libc", + "redox_syscall", + "smallvec", + "winapi", +] + [[package]] name = "peeking_take_while" version = "0.1.2" @@ -332,9 +397,9 @@ checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872" [[package]] name = "proc-macro2" -version = "1.0.43" +version = "1.0.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0a2ca2c61bc9f3d74d2886294ab7b9853abd9c1ad903a3ac7815c58989bb7bab" +checksum = "7bd7356a8122b6c4a24a82b278680c73357984ca2fc79a0f9fa6dea7dced7c58" dependencies = [ "unicode-ident", ] @@ -371,13 +436,22 @@ dependencies = [ [[package]] name = "rand_core" -version = "0.6.3" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d34f1408f55294453790c48b2f1ebbb1c5b4b7563eb1f418bcfcfdbb06ebb4e7" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ "getrandom", ] +[[package]] +name = "redox_syscall" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb5a58c1855b4b6819d59012155603f0b22ad30cad752600aadfcb695265519a" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.6.0" @@ -413,19 +487,25 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" [[package]] -name = "serde" -version = "1.0.144" +name = "scopeguard" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0f747710de3dcd43b88c9168773254e809d8ddbdf9653b84e2554ab219f17860" +checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd" + +[[package]] +name = "serde" +version = "1.0.145" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.144" +version = "1.0.145" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94ed3a816fb1d101812f83e789f888322c34e291f894f19590dc310963e87a00" +checksum = "81fa1584d3d1bcacd84c277a0dfe21f5b0f6accf4a23d04d4c6d61f1af522b4c" dependencies = [ "proc-macro2", "quote", @@ -438,6 +518,12 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "43b2853a4d09f215c24cc5489c992ce46052d359b5109343cbafbf26bc62f8a3" +[[package]] +name = "smallvec" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fd0db749597d91ff862fd1d55ea87f7855a744a8425a64695b6fca237d1dad1" + [[package]] name = "sqm-autorate-rust" version = "0.1.0" @@ -445,12 +531,42 @@ dependencies = [ "bincode", "byteorder", "etherparse", - "libc", + "log", "neli", "nix", "rand", "rust-uci", "serde", + "static_init", + "time", +] + +[[package]] +name = "static_init" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a2a1c578e98c1c16fc3b8ec1328f7659a500737d7a0c6d625e73e830ff9c1f6" +dependencies = [ + "bitflags", + "cfg_aliases", + "libc", + "parking_lot", + "parking_lot_core", + "static_init_macro", + "winapi", +] + +[[package]] +name = "static_init_macro" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70a2595fc3aa78f2d0e45dd425b22282dd863273761cc77780914b2cf3003acf" +dependencies = [ + "cfg_aliases", + "memchr", + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -461,9 +577,9 @@ checksum = "8ea5119cdb4c55b55d432abb513a0429384878c15dde60cc77b1c99de1a95a6a" [[package]] name = "syn" -version = "1.0.99" +version = "1.0.100" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "58dbef6ec655055e20b86b15a8cc6d439cca19b667537ac6a1369572d151ab13" +checksum = "52205623b1b0f064a4e71182c3b18ae902267282930c6d5462c91b859668426e" dependencies = [ "proc-macro2", "quote", @@ -489,16 +605,34 @@ dependencies = [ ] [[package]] -name = "unicode-ident" -version = "1.0.3" +name = "time" +version = "0.3.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4f5b37a154999a8f3f98cc23a628d850e154479cd94decf3414696e12e31aaf" +checksum = "3c3f9a28b618c3a6b9251b6908e9c99e04b9e5c02e6581ccbb67d59c34ef7f9b" +dependencies = [ + "itoa", + "libc", + "num_threads", + "time-macros", +] + +[[package]] +name = "time-macros" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42657b1a6f4d817cda8e7a0ace261fe0cc946cf3a80314390b22cc61ae080792" + +[[package]] +name = "unicode-ident" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dcc811dc4066ac62f84f11307873c4850cb653bfa9b1719cee2bd2204a4bc5dd" [[package]] name = "unicode-width" -version = "0.1.9" +version = "0.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3ed742d4ea2bd1176e236172c8429aaf54486e7ac098db29ffe6529e0ce50973" +checksum = "c0edd1e5b14653f783770bce4a4dabb4a5108a5370a5f5d8cfe8710c361f6c8b" [[package]] name = "vec_map" diff --git a/Cargo.toml b/Cargo.toml index 05aedc6..ec99865 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,15 +16,29 @@ strip = "symbols" bincode = "1.3.3" byteorder = "1.4.3" etherparse = "0.12.0" -libc = "0.2.132" neli = "0.6.2" -nix = { version = "0.25.0", features = ["net", "socket", "time"] } rand = "0.8.5" rust-uci = { version = "0.1.5", optional = true } -serde = { version = "1.0.144", features = ["derive"] } +static_init = "1.0.3" + +[dependencies.log] +features = ["std"] +version = "0.4.17" + +[dependencies.nix] +features = ["net", "socket", "time"] +version = "0.25.0" + +[dependencies.serde] +features = ["derive"] +version = "1.0.144" + +[dependencies.time] +features = ["formatting", "macros"] +version = "0.3.14" [features] uci = ["dep:rust-uci"] [patch.crates-io] -nix = { path = "../nix" } \ No newline at end of file +nix = { path = "../nix" } diff --git a/src/baseliner.rs b/src/baseliner.rs index bd6ce91..ac35e5e 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -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 ); diff --git a/src/config.rs b/src/config.rs index 5afec6d..9b2eeef 100644 --- a/src/config.rs +++ b/src/config.rs @@ -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 { - 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> { + Ok(Self { + // Network section + download_base_kbits: Self::get::( + "SQMA_DOWNLOAD_BASE_KBITS", + "sqm-autorate.@network[0].download_base_kbits", + None, + )?, + download_interface: Self::get::( + "SQMA_DOWNLOAD_INTERFACE", + "sqm-autorate.@network[0].download_interface", + None, + )?, + download_min_kbits: Self::get::( + "SQMA_DOWNLOAD_MIN_KBITS", + "sqm-autorate.@network[0].download_min_kbits", + None, + )?, + upload_base_kbits: Self::get::( + "SQMA_UPLOAD_BASE_KBITS", + "sqm-autorate.@network[0].upload_base_kbits", + None, + )?, + upload_interface: Self::get::( + "SQMA_UPLOAD_INTERFACE", + "sqm-autorate.@network[0].upload_interface", + None, + )?, + upload_min_kbits: Self::get::( + "SQMA_UPLOAD_MIN_KBITS", + "sqm-autorate.@network[0].upload_min_kbits", + None, + )?, + // Output section + log_level: Self::get::( + "SQMA_LOG_LEVEL", + "sqm-autorate.@output[0].log_level", + Some(Level::Error), + )?, + speed_hist_file: Self::get::( + "SQMA_SPEED_HIST_FILE", + "sqm-autorate.@output[0].speed_hist_file", + Some("/tmp/sqm-speedhist.csv".parse()?), + )?, + stats_file: Self::get::( + "SQMA_STATS_FILE", + "sqm-autorate.@output[0].stats_file", + Some("/tmp/sqm-autorate.csv".parse()?), + )?, + suppress_statistics: Self::get::( + "SQMA_SUPPRESS_STATISTICS", + "sqm-autorate.@output[0].suppress_statistics", + Some(false), + )?, + // Advanced section + download_delay_ms: Self::get::( + "SQMA_DOWNLOAD_DELAY_MS", + "sqm-autorate.@advanced_settings[0].download_delay_ms", + Some(15.0), + )?, + high_load_level: Self::get::( + "SQMA_HIGH_LOAD_LEVEL", + "sqm-autorate.@advanced_settings[0].high_load_level", + Some(0.8), + )?, + min_change_interval: Self::get::( + "SQMA_MIN_CHANGE_INTERVAL", + "sqm-autorate.@advanced_settings[0].min_change_interval", + Some(0.5), + )?, + num_reflectors: Self::get::( + "SQMA_NUM_REFLECTORS", + "sqm-autorate.@advanced_settings[0].num_reflectors", + Some(5), + )?, + reflector_list_file: Self::get::( + "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::( + "SQMA_SPEED_HIST_SIZE", + "sqm-autorate.@advanced_settings[0].speed_hist_size", + Some(100), + )?, + tick_interval: Self::get::( + "SQMA_TICK_INTERVAL", + "sqm-autorate.@advanced_settings[0].tick_interval", + Some(0.5), + )?, + upload_delay_ms: Self::get::( + "SQMA_UPLOAD_DELAY_MS", + "sqm-autorate.@advanced_settings[0].upload_delay_ms", + Some(15.0), + )?, }) } + fn get( + env_key: &str, + uci_key: &str, + default: Option, + ) -> Result> { + return match Self::get_value(env_key, uci_key, None) { + Ok(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(e) => { + return match default { + Some(val) => Ok(val), + None => Err(e), + } + } + }; + } + + fn get_value( + env_key: &str, + uci_key: &str, + default: Option, + ) -> Result> { + let mut err: Option>; + + 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, Box> { + 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 { - None + fn get_from_uci(_: &str) -> Result, Box> { + Ok(None) } pub fn load_reflectors(&self) -> Result, Box> { diff --git a/src/error.rs b/src/error.rs index 5a4e242..86d9a6a 100644 --- a/src/error.rs +++ b/src/error.rs @@ -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 } } diff --git a/src/log.rs b/src/log.rs new file mode 100644 index 0000000..61292b5 --- /dev/null +++ b/src/log.rs @@ -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(dt: T, format: &(impl Formattable + ?Sized)) -> String +where + T: Into, +{ + 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)) +} diff --git a/src/main.rs b/src/main.rs index 89db261..75b55e6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -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::::new())); - let mut owd_recent = Arc::new(Mutex::new(HashMap::::new())); - let mut reflector_peers_lock = Arc::new(Mutex::new(Vec::::new())); + let owd_baseline = Arc::new(Mutex::new(HashMap::::new())); + let owd_recent = Arc::new(Mutex::new(HashMap::::new())); + let reflector_peers_lock = Arc::new(Mutex::new(Vec::::new())); let mut reflector_pool = Vec::::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(); } diff --git a/src/netlink.rs b/src/netlink.rs index 2642221..3d5cc03 100644 --- a/src/netlink.rs +++ b/src/netlink.rs @@ -137,7 +137,7 @@ impl Netlink { Err(Box::new(NlError::msg("Error getting stats"))) } - pub fn find_qdisc(ifindex: i32) -> Result> { + pub fn qdisc_from_ifindex(ifindex: i32) -> Result> { 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> { + 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> { 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(); diff --git a/src/pinger.rs b/src/pinger.rs index a1aac9d..760c614 100644 --- a/src/pinger.rs +++ b/src/pinger.rs @@ -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 { 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>>, stats_sender: Sender, - ) { - let socket = open_socket(type_); + ) -> Result<(), Box> { + let socket = open_socket(type_)?; loop { let mut v: Vec = 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>>) { - let socket = open_socket(type_); + fn send( + &mut self, + type_: SocketType, + reflectors_lock: Arc>>, + ) -> Result<(), Box> { + let socket = open_socket(type_)?; let mut seq: u16 = 0; let tick_duration_ms: u16 = 500; diff --git a/src/pinger_icmp.rs b/src/pinger_icmp.rs index f7e505c..2b4e2e3 100644 --- a/src/pinger_icmp.rs +++ b/src/pinger_icmp.rs @@ -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 {})) } } diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index e5fc775..845a44e 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -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 = Vec::new(); let mut safe_ul_rates: Vec = 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 = None; + let mut speed_hist_fd_inner: File; + let mut stats_fd: Option = 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); } } diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index 158a781..a526e3c 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -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>>, pub(crate) owd_recent: Arc>>, pub(crate) reflector_peers_lock: Arc>>, pub(crate) reflector_pool: Vec, @@ -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() );