From b147006a6983ad1419252c5ea1c41577f16b2b0d Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sat, 17 Sep 2022 14:39:27 +0200 Subject: [PATCH] partly working implementation --- Cargo.lock | 421 ++++++++++++++++++++++++++++---------- Cargo.toml | 9 +- src/baseliner.rs | 113 ++++++++++ src/config.rs | 83 +++++++- src/error.rs | 37 ++++ src/main.rs | 204 ++++++++++++------ src/netlink.rs | 308 ++++++++++++++-------------- src/pinger.rs | 111 +++++++--- src/pinger_icmp.rs | 261 +++++++---------------- src/pinger_icmp_ts.rs | 204 ++++++++---------- src/ratecontroller.rs | 314 ++++++++++++++++++++++++++++ src/reflector_selector.rs | 129 ++++++++++++ src/utils.rs | 26 +++ 13 files changed, 1542 insertions(+), 678 deletions(-) create mode 100644 src/baseliner.rs create mode 100644 src/error.rs create mode 100644 src/ratecontroller.rs create mode 100644 src/reflector_selector.rs create mode 100644 src/utils.rs diff --git a/Cargo.lock b/Cargo.lock index 72b353a..976e14a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,13 +4,39 @@ version = 3 [[package]] name = "aho-corasick" -version = "0.7.18" +version = "0.7.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e37cfd5e7657ada45f742d6e99ca5788580b5c529dc78faf11ece6dc702656f" +checksum = "b4f55bd91a0978cbfd91c457a164bab8b4001c833b7f323132c0a4e1922dd44e" dependencies = [ "memchr", ] +[[package]] +name = "ansi_term" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d52a9bb7ec0cf484c551830a7ce27bd20d67eac647e1befb56b0be4ee39a55d2" +dependencies = [ + "winapi", +] + +[[package]] +name = "arrayvec" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8da52d66c7071e2e3fa2a1e5c6d088fec47b593032b254f5e980de8ea54454d6" + +[[package]] +name = "atty" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9b39be18770d11421cdb1b9947a45dd3f37e93092cbf377614828a319d5fee8" +dependencies = [ + "hermit-abi", + "libc", + "winapi", +] + [[package]] name = "autocfg" version = "1.1.0" @@ -26,6 +52,29 @@ dependencies = [ "serde", ] +[[package]] +name = "bindgen" +version = "0.58.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f8523b410d7187a43085e7e064416ea32ded16bd0a4e6fc025e21616d01258f" +dependencies = [ + "bitflags", + "cexpr", + "clang-sys", + "clap", + "env_logger", + "lazy_static", + "lazycell", + "log", + "peeking_take_while", + "proc-macro2", + "quote", + "regex", + "rustc-hash", + "shlex", + "which", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -44,13 +93,28 @@ version = "0.1.0" dependencies = [ "bincode", "byteorder", + "etherparse", "libc", "neli", "nix", - "pnet", - "pnet_transport", + "rand", + "rust-uci", "serde", - "serde_bytes", +] + +[[package]] +name = "cc" +version = "1.0.73" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fff2a6927b3bb87f9595d67196a70493f627687a71d87a0d692242c33f58c11" + +[[package]] +name = "cexpr" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f4aedb84272dbe89af497cf81375129abda4fc0a9e7c5d317498c15cc30c0d27" +dependencies = [ + "nom", ] [[package]] @@ -59,12 +123,80 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +[[package]] +name = "clang-sys" +version = "1.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a050e2153c5be08febd6734e29298e844fdb0fa21aeddd63b4eb7baa106c69b" +dependencies = [ + "glob", + "libc", + "libloading", +] + +[[package]] +name = "clap" +version = "2.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0610544180c38b88101fecf2dd634b174a62eef6946f84dfc6a7127512b381c" +dependencies = [ + "ansi_term", + "atty", + "bitflags", + "strsim", + "textwrap", + "unicode-width", + "vec_map", +] + +[[package]] +name = "cmake" +version = "0.1.48" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8ad8cef104ac57b68b89df3208164d228503abbdce70f6880ffa3d970e7443a" +dependencies = [ + "cc", +] + [[package]] name = "either" version = "1.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f107b87b6afc2a64fd13cac55fe06d6c8859f12d4b14cbcdd2c67d0976781be" +[[package]] +name = "env_logger" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a19187fea3ac7e84da7dacf48de0c45d63c6a76f9490dae389aead16c243fce3" +dependencies = [ + "atty", + "humantime", + "log", + "regex", + "termcolor", +] + +[[package]] +name = "etherparse" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bcb08c4aab4e2985045305551e67126b43f1b6b136bc4e1cd87fb0327877a611" +dependencies = [ + "arrayvec", +] + +[[package]] +name = "getrandom" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4eb1a864a501629691edf6c15a593b7a51eebaa1e8468e9ddc623de7c9b58ec6" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + [[package]] name = "glob" version = "0.3.0" @@ -72,20 +204,58 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b919933a397b79c37e33b77bb2aa3dc8eb6e165ad809e58ff75bc7db2e34574" [[package]] -name = "ipnetwork" -version = "0.19.0" +name = "hermit-abi" +version = "0.1.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1f84f1612606f3753f205a4e9a2efd6fe5b4c573a6269b2cc6c3003d44a0d127" +checksum = "62b467343b94ba476dcb2500d242dadbb39557df889310ac77c5d99100aaac33" dependencies = [ - "serde", + "libc", ] +[[package]] +name = "humantime" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4" + +[[package]] +name = "lazy_static" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" + +[[package]] +name = "lazycell" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" + [[package]] name = "libc" version = "0.2.132" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8371e4e5341c3a96db127eb2465ac681ced4c433e01dd0e938adbef26ba93ba5" +[[package]] +name = "libloading" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "efbc0f03f9a775e9f6aed295c6a1ba2253c5757a9e03d55c6caa46a681abcddd" +dependencies = [ + "cfg-if", + "winapi", +] + +[[package]] +name = "libuci-sys" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "544c17966d9131e80b3616783974a31b526197ca5b56948bb130eeb6c04fbaf1" +dependencies = [ + "bindgen", + "cmake", +] + [[package]] name = "log" version = "0.4.17" @@ -148,10 +318,20 @@ dependencies = [ ] [[package]] -name = "no-std-net" -version = "0.6.0" +name = "nom" +version = "5.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43794a0ace135be66a25d3ae77d41b91615fb68ae937f904090203e81f755b65" +checksum = "ffb4262d26ed83a1c0a33a38fe2bb15797329c85770da05e6b828ddb782627af" +dependencies = [ + "memchr", + "version_check", +] + +[[package]] +name = "peeking_take_while" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099" [[package]] name = "pin-utils" @@ -160,93 +340,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] -name = "pnet" -version = "0.31.0" +name = "ppv-lite86" +version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0caaf5b11fd907ff15cf14a4477bfabca4b37ab9e447a4f8dead969a59cdafad" -dependencies = [ - "pnet_base", - "pnet_datalink", - "pnet_packet", - "pnet_transport", -] - -[[package]] -name = "pnet_base" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f9d3a993d49e5fd5d4d854d6999d4addca1f72d86c65adf224a36757161c02b6" -dependencies = [ - "no-std-net", -] - -[[package]] -name = "pnet_datalink" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e466faf03a98ad27f6e15cd27a2b7cc89e73e640a43527742977bc503c37f8aa" -dependencies = [ - "ipnetwork", - "libc", - "pnet_base", - "pnet_sys", - "winapi", -] - -[[package]] -name = "pnet_macros" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48dd52a5211fac27e7acb14cfc9f30ae16ae0e956b7b779c8214c74559cef4c3" -dependencies = [ - "proc-macro2", - "quote", - "regex", - "syn", -] - -[[package]] -name = "pnet_macros_support" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89de095dc7739349559913aed1ef6a11e73ceade4897dadc77c5e09de6740750" -dependencies = [ - "pnet_base", -] - -[[package]] -name = "pnet_packet" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc3b5111e697c39c8b9795b9fdccbc301ab696699e88b9ea5a4e4628978f495f" -dependencies = [ - "glob", - "pnet_base", - "pnet_macros", - "pnet_macros_support", -] - -[[package]] -name = "pnet_sys" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "328e231f0add6d247d82421bf3790b4b33b39c8930637f428eef24c4c6a90805" -dependencies = [ - "libc", - "winapi", -] - -[[package]] -name = "pnet_transport" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff597185e6f1f5671b3122e4dba892a1c73e17c17e723d7669bd9299cbe7f124" -dependencies = [ - "libc", - "pnet_base", - "pnet_packet", - "pnet_sys", -] +checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872" [[package]] name = "proc-macro2" @@ -266,6 +363,36 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "rand" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" +dependencies = [ + "libc", + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d34f1408f55294453790c48b2f1ebbb1c5b4b7563eb1f418bcfcfdbb06ebb4e7" +dependencies = [ + "getrandom", +] + [[package]] name = "regex" version = "1.6.0" @@ -283,6 +410,23 @@ version = "0.6.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a3f87b73ce11b1619a3c6332f45341e0047173771e8b8b73f87bfeefb7b56244" +[[package]] +name = "rust-uci" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c664edb3ec081c1f2746960b04861354bbeaafdbb962fb8789d6ded5153ac3bb" +dependencies = [ + "libc", + "libuci-sys", + "log", +] + +[[package]] +name = "rustc-hash" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" + [[package]] name = "serde" version = "1.0.144" @@ -292,15 +436,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "serde_bytes" -version = "0.11.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfc50e8183eeeb6178dcb167ae34a8051d63535023ae38b5d8d12beae193d37b" -dependencies = [ - "serde", -] - [[package]] name = "serde_derive" version = "1.0.144" @@ -312,6 +447,18 @@ dependencies = [ "syn", ] +[[package]] +name = "shlex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43b2853a4d09f215c24cc5489c992ce46052d359b5109343cbafbf26bc62f8a3" + +[[package]] +name = "strsim" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ea5119cdb4c55b55d432abb513a0429384878c15dde60cc77b1c99de1a95a6a" + [[package]] name = "syn" version = "1.0.99" @@ -323,12 +470,63 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "termcolor" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bab24d30b911b2376f3a13cc2cd443142f0c81dda04c118693e35b3835757755" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "textwrap" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d326610f408c7a4eb6f51c37c330e496b08506c9457c9d34287ecc38809fb060" +dependencies = [ + "unicode-width", +] + [[package]] name = "unicode-ident" version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c4f5b37a154999a8f3f98cc23a628d850e154479cd94decf3414696e12e31aaf" +[[package]] +name = "unicode-width" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ed742d4ea2bd1176e236172c8429aaf54486e7ac098db29ffe6529e0ce50973" + +[[package]] +name = "vec_map" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1bddf1187be692e79c5ffeab891132dfb0f236ed36a43c7ed39f1165ee20191" + +[[package]] +name = "version_check" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" + +[[package]] +name = "wasi" +version = "0.11.0+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" + +[[package]] +name = "which" +version = "3.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d011071ae14a2f6671d0b74080ae0cd8ebf3a6f8c9589a2cd45f23126fe29724" +dependencies = [ + "libc", +] + [[package]] name = "winapi" version = "0.3.9" @@ -345,6 +543,15 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" +[[package]] +name = "winapi-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178" +dependencies = [ + "winapi", +] + [[package]] name = "winapi-x86_64-pc-windows-gnu" version = "0.4.0" diff --git a/Cargo.toml b/Cargo.toml index 49a0e9b..f478170 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,13 +15,16 @@ strip = "symbols" [dependencies] 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"] } -pnet = "0.31.0" -pnet_transport = "0.31.0" +rand = "0.8.5" +rust-uci = { version = "0.1.5", optional = true } serde = { version = "1.0.144", features = ["derive"] } -serde_bytes = "0.11" + +[features] +uci = ["dep:rust-uci"] [patch.crates-io] nix = { path = "../nix" } \ No newline at end of file diff --git a/src/baseliner.rs b/src/baseliner.rs new file mode 100644 index 0000000..bd6ce91 --- /dev/null +++ b/src/baseliner.rs @@ -0,0 +1,113 @@ +use crate::pinger::PingReply; +use crate::{Config, Utils}; +use std::collections::HashMap; +use std::net::IpAddr; +use std::sync::mpsc::{Receiver, Sender}; +use std::sync::{Arc, Mutex}; + +#[derive(Copy, Clone)] +pub struct ReflectorStats { + pub(crate) down_ewma: f64, + pub(crate) up_ewma: f64, + pub(crate) last_receive_time_s: f64, +} + +pub struct Baseliner { + pub(crate) config: Config, + pub(crate) owd_baseline: Arc>>, + pub(crate) owd_recent: Arc>>, + pub(crate) reselect_trigger: Sender, + pub(crate) stats_receiver: Receiver, +} + +impl Baseliner { + pub fn run(&self) { + /* + * 135 seconds to decay to 50% for the slow factor and + * 0.4 seconds to decay to 50% for the fast factor. + * The fast one can be adjusted to tune, try anything from 0.01 to 3.0 to get more or less sensitivity + * with more sensitivity we respond faster to bloat, but are at risk from triggering due to lag spikes that + * 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); + + loop { + let time_data = self.stats_receiver.recv().unwrap(); + + 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 { + 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 { + 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_baseline = owd_baseline_map + .entry(time_data.reflector) + .or_insert(owd_baseline_new); + + let mut owd_recent = owd_recent_map + .entry(time_data.reflector) + .or_insert(owd_recent_new); + + if time_data.last_receive_time_s - owd_baseline.last_receive_time_s > 30.0 + || time_data.last_receive_time_s - owd_recent.last_receive_time_s > 30.0 + { + owd_baseline.down_ewma = time_data.down_time; + owd_baseline.up_ewma = time_data.up_time; + owd_baseline.last_receive_time_s = time_data.last_receive_time_s; + owd_recent.down_ewma = time_data.down_time; + owd_recent.up_ewma = time_data.up_time; + owd_recent.last_receive_time_s = time_data.last_receive_time_s; + } + + owd_baseline.last_receive_time_s = time_data.last_receive_time_s; + owd_recent.last_receive_time_s = time_data.last_receive_time_s; + + // if this reflection is more than 5 seconds higher than baseline... mark it no good and trigger a reselection + if time_data.up_time > owd_baseline.up_ewma + 5000.0 + || time_data.down_time > owd_baseline.down_ewma + 5000.0 + { + 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; + let _ = self.reselect_trigger.send(true); + } else { + owd_baseline.down_ewma = owd_baseline.down_ewma * slow_factor + + (1.0 - slow_factor) * time_data.down_time; + owd_baseline.up_ewma = + owd_baseline.up_ewma * slow_factor + (1.0 - slow_factor) * time_data.up_time; + + owd_recent.down_ewma = + owd_recent.down_ewma * fast_factor + (1.0 - fast_factor) * time_data.down_time; + owd_recent.up_ewma = + owd_recent.up_ewma * fast_factor + (1.0 - fast_factor) * time_data.up_time; + + if owd_baseline.down_ewma > owd_recent.down_ewma { + owd_baseline.down_ewma = owd_recent.down_ewma; + } + + if owd_baseline.up_ewma > owd_recent.up_ewma { + owd_baseline.up_ewma = owd_recent.up_ewma; + } + } + + println!( + "Reflector {} up baseline = {} down baseline = {}", + time_data.reflector, owd_baseline.up_ewma, owd_baseline.down_ewma + ); + println!( + "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 e4e850f..5afec6d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,17 +1,90 @@ -use std::env; +#[cfg(feature = "uci")] +use rust_uci::Uci; +use std::error::Error; +use std::fs::File; +use std::io::BufRead; +use std::net::IpAddr; +use std::path::Path; +use std::str::FromStr; +use std::{env, io}; + +fn read_lines

(filename: P) -> io::Result>> +where + P: AsRef, +{ + let file = File::open(filename)?; + Ok(io::BufReader::new(file).lines()) +} #[allow(dead_code)] +#[derive(Clone)] pub(crate) struct Config { pub(crate) download_interface: String, pub(crate) upload_interface: String, + + pub(crate) reflector_list_file: String, + + pub(crate) min_change_interval: f64, + pub(crate) num_reflectors: u8, + pub(crate) tick_duration: f64, } impl Config { pub fn new() -> Self { - 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"), + // 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, + }) + } + + #[cfg(not(feature = "uci"))] + fn load_from_uci() -> Option { + None + } + + pub fn load_reflectors(&self) -> Result, Box> { + let lines = read_lines(self.reflector_list_file.clone())?; + + let mut reflectors: Vec = Vec::with_capacity(50); + + let mut first = true; + + for line in lines { + if first { + first = false; + continue; + } + + let line = line?; + let columns: Vec<&str> = line.split(",").collect(); + reflectors.push(IpAddr::from_str(columns[0])?); + } + + Ok(reflectors) + } } diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..5a4e242 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,37 @@ +use std::error::Error; +use std::fmt; +use std::fmt::Display; + +#[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 + } +} + +// 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; + +impl Error for PingParseError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + // The compiler transparently casts `&sqlx::Error` into a `&dyn Error` + 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") // user-facing output + } +} diff --git a/src/main.rs b/src/main.rs index ffc7e11..89db261 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,97 +1,173 @@ +extern crate core; + +mod baseliner; mod cake; mod config; +mod error; mod netlink; mod pinger; mod pinger_icmp; mod pinger_icmp_ts; +mod ratecontroller; +mod reflector_selector; +mod utils; +use crate::baseliner::{Baseliner, ReflectorStats}; +use std::collections::HashMap; use std::net::IpAddr; +use std::process::ExitCode; use std::str::FromStr; -use std::{process, thread, time}; +use std::sync::mpsc::channel; +use std::sync::{Arc, Mutex}; +use std::thread::sleep; +use std::time::Duration; +use std::{process, thread}; use crate::config::Config; -use crate::netlink::{find_interface, find_qdisc, get_interface_stats, set_qdisc_rate}; -use crate::pinger::{PingListener, PingSender, Pinger}; -use crate::pinger_icmp::{PingerICMPEcho, PingerICMPEchoListener, PingerICMPEchoSender}; -use crate::pinger_icmp_ts::PingerICMPTimestamps; +use crate::netlink::Netlink; +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; -#[macro_export] -macro_rules! error { - ($a:expr,$b:expr) => { - match $a { - //Some(e) => return e, - Ok(e) => e, - Err(e) => { - println!("{}: {}", $b, $a); - } - } - }; -} - -fn main() { +fn main() -> ExitCode { println!("starting up"); let config = Config::new(); - test(&config); + let mut reflectors = match config.load_reflectors() { + Ok(refl) => refl, + Err(e) => { + println!("Couldn't load reflectors: {}", e.to_string()); + return ExitCode::FAILURE; + } + }; - let mut reflectors: Vec = Vec::new(); + // The identifier field in ICMP is only 4 bytes + // so take the last 4 bytes of the PID as the ID + let id = (process::id() & 0xFFFF) as u16; - reflectors.push(IpAddr::from_str("9.9.9.9").unwrap()); - reflectors.push(IpAddr::from_str("9.9.9.10").unwrap()); + // 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 mut reflector_pool = Vec::::new(); + let reflector_pool_size = reflectors.len(); - get_interface_stats(config.upload_interface.as_str()).unwrap(); + 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(), + ]; - /*let mut pinger_receiver = - PingerICMPTimestamps::new((process::id() & 0xFFFF) as u16, reflectors.clone()); - let mut pinger_sender = - PingerICMPTimestamps::new((process::id() & 0xFFFF) as u16, reflectors.clone());*/ + match reflector_pool_size > 5 { + true => { + let mut peers = reflector_peers_lock.lock().unwrap(); + peers.append(default_reflectors.to_vec().as_mut()); + reflector_pool.append(reflectors.as_mut()); + } + false => { + let mut peers = reflector_peers_lock.lock().unwrap(); + peers.append(default_reflectors.to_vec().as_mut()); + } + } - let mut pinger_receiver = - PingerICMPEchoListener::new((process::id() & 0xFFF) as u16, reflectors.clone()); + let (baseliner_stats_sender, baseliner_stats_receiver) = channel(); + let (reselect_sender, reselect_receiver) = channel(); - let mut pinger_sender = - PingerICMPEchoSender::new((process::id() & 0xFFF) as u16, reflectors.clone()); + //let mut pinger_receiver = PingerICMPTimestampListener::new(id); + //let mut pinger_sender = PingerICMPTimestampSender::new(id); + let mut pinger_receiver = PingerICMPEchoListener::new(id); + let mut pinger_sender = PingerICMPEchoSender::new(id); + let mut baseliner = Baseliner { + config: config.clone(), + owd_baseline: owd_baseline.clone(), + owd_recent: owd_recent.clone(), + reselect_trigger: reselect_sender.clone(), + 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(); + + /* Set initial TC values to minimum + * so there should be no initial bufferbloat to + * fool the baseliner + */ + Netlink::set_qdisc_rate(down_qdisc, 5000).expect("Couldn't set ingress bandwidth"); + Netlink::set_qdisc_rate(up_qdisc, 1000).expect("Couldn't set egress bandwidth"); + 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 || pinger_receiver.listen()) - .unwrap(); + .spawn(move || { + pinger_receiver.listen( + SocketType::ICMP, + reflector_peers_lock_clone, + baseliner_stats_sender, + ) + }) + .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"); + let reflector_peers_lock_clone = reflector_peers_lock.clone(); let sender_handle = thread::Builder::new() .name("sender".to_string()) - .spawn(move || pinger_sender.send()) - .unwrap(); + .spawn(move || pinger_sender.send(SocketType::ICMP, reflector_peers_lock_clone)) + .expect("Couldn't spawn ping sender thread"); - let threads = vec![receiver_handle, sender_handle]; + let mut threads = vec![receiver_handle, sender_handle, baseliner_handle]; + + 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, + trigger_channel: reselect_receiver, + }; + let reselection_handle = thread::Builder::new() + .name("reselection".to_string()) + .spawn(move || reflector_selector.run()) + .expect("Couldn't spawn reflector selector thread"); + 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 ratecontroller_handle = thread::Builder::new() + .name("ratecontroller".to_string()) + .spawn(move || ratecontroller.run(StatsDirection::TX, StatsDirection::RX)) + .expect("Couldn't spawn ratecontroller thread"); + + threads.push(ratecontroller_handle); for thread in threads { + println!("thread: {}", thread.thread().name().unwrap()); thread.join().unwrap(); } -} -fn test(config: &Config) { - if true { - return; - } - - let start = time::Instant::now(); - let ifindex = - find_interface(config.upload_interface.as_str()).expect("Couldn't find interface"); - let find_if_time = time::Instant::now().duration_since(start); - println!("time get intf: {}", find_if_time.as_micros()); - - let start = time::Instant::now(); - let qdisc = find_qdisc(ifindex).expect("Couldn't find qdisc"); - let find_qdisc_time = time::Instant::now().duration_since(start); - - println!("time get qdisc: {}", find_qdisc_time.as_micros()); - - let start = time::Instant::now(); - set_qdisc_rate(qdisc, 1000 * 1000 / 8).unwrap(); - let set_qdisc_time = time::Instant::now().duration_since(start); - println!("time set qdisc: {}", set_qdisc_time.as_micros()); - - println!( - "time total: {}", - find_if_time.as_micros() + find_qdisc_time.as_micros() + set_qdisc_time.as_micros() - ); + ExitCode::SUCCESS } diff --git a/src/netlink.rs b/src/netlink.rs index a837b02..2642221 100644 --- a/src/netlink.rs +++ b/src/netlink.rs @@ -1,4 +1,5 @@ use crate::cake::TcaCake; +use crate::error::NoQdiscFoundError; use neli::consts::nl::{NlmF, NlmFFlags}; use neli::consts::rtnl::{Arphrd, IffFlags, Ifla, RtAddrFamily, Rtm, Tca}; @@ -10,27 +11,8 @@ use neli::socket::NlSocketHandle; use neli::types::{Buffer, RtBuffer}; use serde::Deserialize; use std::error::Error; -use std::fmt::Display; use bincode::deserialize; -use std::fmt; - -#[derive(Default, Debug)] -struct NoQdiscFoundError; - -impl Error for NoQdiscFoundError { - fn source(&self) -> Option<&(dyn Error + 'static)> { - // The compiler transparently casts `&sqlx::Error` into a `&dyn Error` - 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(Deserialize, Copy, Clone, Default, Debug)] pub struct RtnlLinkStats64 { @@ -61,6 +43,7 @@ pub struct RtnlLinkStats64 { } #[allow(dead_code)] +#[derive(Clone, Copy)] pub struct Qdisc { ifindex: i32, handle: u32, @@ -69,184 +52,189 @@ pub struct Qdisc { parent: u32, } -fn nl_interface_get(socket: &mut NlSocketHandle, ifname: &str) -> Result<(), Box> { - let mut attrs = RtBuffer::new(); +pub struct Netlink {} - const RTEXT_FILTER_VF: i32 = 1 << 0; +impl Netlink { + fn nl_interface_get(socket: &mut NlSocketHandle, ifname: &str) -> Result<(), Box> { + let mut attrs = RtBuffer::new(); - let attr_ifname = Rtattr::new(None, Ifla::Ifname, ifname).unwrap(); - let attr_ext_mask = Rtattr::new(None, Ifla::ExtMask, RTEXT_FILTER_VF).unwrap(); - attrs.push(attr_ifname); - attrs.push(attr_ext_mask); + const RTEXT_FILTER_VF: i32 = 1 << 0; - let if_msg = Ifinfomsg::new( - RtAddrFamily::Unspecified, - Arphrd::None, - -1, - IffFlags::empty(), - IffFlags::empty(), - attrs, - ); + let attr_ifname = Rtattr::new(None, Ifla::Ifname, ifname).unwrap(); + let attr_ext_mask = Rtattr::new(None, Ifla::ExtMask, RTEXT_FILTER_VF).unwrap(); + attrs.push(attr_ifname); + attrs.push(attr_ext_mask); - let nlhdr = Nlmsghdr::new( - None, - Rtm::Getlink, - NlmFFlags::new(&[NlmF::Request, NlmF::Ack]), - None, - None, - NlPayload::Payload(if_msg), - ); + let if_msg = Ifinfomsg::new( + RtAddrFamily::Unspecified, + Arphrd::None, + -1, + IffFlags::empty(), + IffFlags::empty(), + attrs, + ); - socket.send(nlhdr)?; + let nlhdr = Nlmsghdr::new( + None, + Rtm::Getlink, + NlmFFlags::new(&[NlmF::Request, NlmF::Ack]), + None, + None, + NlPayload::Payload(if_msg), + ); - Ok(()) -} + socket.send(nlhdr)?; -pub fn find_interface(ifname: &str) -> Result> { - let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); - - nl_interface_get(&mut socket, ifname).expect("Wrong when sending intf"); - - for response in socket.iter(false) { - let header: Nlmsghdr = response?; - - if header.nl_type != Rtm::Newlink { - return Err(Box::new(NlError::msg("Netlink error retrieving link"))); - } - - if let NlPayload::Payload(p) = header.nl_payload { - return Ok(p.ifi_index); - } + Ok(()) } - // we shouldn't reach here - Ok(-1) -} + pub fn find_interface(ifname: &str) -> Result> { + let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); -pub fn get_interface_stats(ifname: &str) -> Result> { - let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); + Self::nl_interface_get(&mut socket, ifname).expect("Wrong when sending intf"); - nl_interface_get(&mut socket, ifname).expect("Wrong when sending intf"); + for response in socket.iter(false) { + let header: Nlmsghdr = response?; - for response in socket.iter(false) { - let header: Nlmsghdr = response?; + if header.nl_type != Rtm::Newlink { + return Err(Box::new(NlError::msg("Netlink error retrieving link"))); + } - if header.nl_type != Rtm::Newlink { - return Err(Box::new(NlError::msg("Netlink error retrieving link"))); - } - - if let NlPayload::Payload(p) = header.nl_payload { - for attr in p.rtattrs.iter() { - if attr.rta_type == Ifla::Stats64 { - let buf = attr.rta_payload.as_ref(); - - let stats: RtnlLinkStats64 = deserialize(buf).unwrap(); - - return Ok(stats); - } + if let NlPayload::Payload(p) = header.nl_payload { + return Ok(p.ifi_index); } } + + // we shouldn't reach here + Ok(-1) } - Err(Box::new(NlError::msg("Error getting stats"))) -} + pub fn get_interface_stats(ifname: &str) -> Result> { + let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); -pub fn find_qdisc(ifindex: i32) -> Result> { - let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); - let tc_msg = Tcmsg::new( - u8::from(RtAddrFamily::Unspecified), - 0, - 0, - 0, - 0, - RtBuffer::new(), - ); + Self::nl_interface_get(&mut socket, ifname)?; - let nlhdr = Nlmsghdr::new( - None, - Rtm::Getqdisc, - NlmFFlags::new(&[NlmF::Request, NlmF::Dump]), - None, - None, - NlPayload::Payload(tc_msg), - ); + for response in socket.iter(false) { + let header: Nlmsghdr = response?; - socket.send(nlhdr)?; - - for response in socket.iter(false) { - let header: Nlmsghdr = response?; - - if let NlPayload::Payload(p) = header.nl_payload { - if header.nl_type != Rtm::Newqdisc { - return Err(Box::new(NlError::msg("Netlink error retrieving qdisc"))); + if header.nl_type != Rtm::Newlink { + return Err(Box::new(NlError::msg("Netlink error retrieving link"))); } - if p.tcm_ifindex == ifindex { - let mut _type = ""; - + if let NlPayload::Payload(p) = header.nl_payload { for attr in p.rtattrs.iter() { - if attr.rta_type == Tca::Kind { - let buff = attr.rta_payload.as_ref(); - _type = std::str::from_utf8(buff) - .expect("Found invalid UTF-8 string") - .trim_end_matches('\0'); // Null terminator is valid UTF-8, but breaks comparison, so we remove it + if attr.rta_type == Ifla::Stats64 { + let buf = attr.rta_payload.as_ref(); + + let stats: RtnlLinkStats64 = deserialize(buf).unwrap(); + + return Ok(stats); } } + } + } - let type_to_look_for = "cake"; + Err(Box::new(NlError::msg("Error getting stats"))) + } - if _type.eq(type_to_look_for) { - let qdisc = Qdisc { - ifindex: p.tcm_ifindex as i32, - handle: p.tcm_handle, - family: p.tcm_family, - refcnt: p.tcm_info, - parent: p.tcm_parent, - }; + pub fn find_qdisc(ifindex: i32) -> Result> { + let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); + let tc_msg = Tcmsg::new( + u8::from(RtAddrFamily::Unspecified), + 0, + 0, + 0, + 0, + RtBuffer::new(), + ); - return Ok(qdisc); + let nlhdr = Nlmsghdr::new( + None, + Rtm::Getqdisc, + NlmFFlags::new(&[NlmF::Request, NlmF::Dump]), + None, + None, + NlPayload::Payload(tc_msg), + ); + + socket.send(nlhdr)?; + + for response in socket.iter(false) { + let header: Nlmsghdr = response?; + + if let NlPayload::Payload(p) = header.nl_payload { + if header.nl_type != Rtm::Newqdisc { + return Err(Box::new(NlError::msg("Netlink error retrieving qdisc"))); + } + + if p.tcm_ifindex == ifindex { + let mut _type = ""; + + for attr in p.rtattrs.iter() { + if attr.rta_type == Tca::Kind { + let buff = attr.rta_payload.as_ref(); + _type = std::str::from_utf8(buff) + .expect("Found invalid UTF-8 string") + .trim_end_matches('\0'); // Null terminator is valid UTF-8, but breaks comparison, so we remove it + } + } + + let type_to_look_for = "cake"; + + if _type.eq(type_to_look_for) { + let qdisc = Qdisc { + ifindex: p.tcm_ifindex as i32, + handle: p.tcm_handle, + family: p.tcm_family, + refcnt: p.tcm_info, + parent: p.tcm_parent, + }; + + return Ok(qdisc); + } } } } + + return Err(Box::new(NoQdiscFoundError)); } - return Err(Box::new(NoQdiscFoundError)); -} + pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth: u64) -> Result<(), Box> { + let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); + let bandwidth = bandwidth * 1000 / 8; -pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth: u64) -> Result<(), Box> { - let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap(); + let mut attrs = RtBuffer::new(); - let mut attrs = RtBuffer::new(); + let attr_type = Rtattr::new(None, Tca::Kind, "cake").unwrap(); + let mut attr_options = Rtattr::new(None, Tca::Options, Buffer::from(Vec::new())).unwrap(); + attr_options.add_nested_attribute(&Rtattr::new( + None, + TcaCake::BaseRate64 as u16, + bandwidth, + )?)?; - let attr_type = Rtattr::new(None, Tca::Kind, "cake").unwrap(); - let mut attr_options = Rtattr::new(None, Tca::Options, Buffer::from(Vec::new())).unwrap(); - attr_options.add_nested_attribute(&Rtattr::new( - None, - TcaCake::BaseRate64 as u16, - bandwidth, - )?)?; + attrs.push(attr_type); + attrs.push(attr_options); - attrs.push(attr_type); - attrs.push(attr_options); + let tc_msg = Tcmsg::new( + u8::from(RtAddrFamily::Unspecified), + qdisc.ifindex, + 0, + qdisc.parent, + 0, + attrs, + ); - let tc_msg = Tcmsg::new( - u8::from(RtAddrFamily::Unspecified), - qdisc.ifindex, - 0, - qdisc.parent, - 0, - attrs, - ); + let nlhdr = Nlmsghdr::new( + None, + Rtm::Newqdisc, + NlmFFlags::new(&[NlmF::Request, NlmF::Ack]), + None, + None, + NlPayload::Payload(tc_msg), + ); - let nlhdr = Nlmsghdr::new( - None, - Rtm::Newqdisc, - NlmFFlags::new(&[NlmF::Request, NlmF::Ack]), - None, - None, - NlPayload::Payload(tc_msg), - ); - - socket.send(nlhdr)?; - Ok(()) + socket.send(nlhdr)?; + Ok(()) + } } diff --git a/src/pinger.rs b/src/pinger.rs index 72a2648..a1aac9d 100644 --- a/src/pinger.rs +++ b/src/pinger.rs @@ -4,68 +4,118 @@ use nix::sys::socket::{ }; 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, Mutex}; use std::thread; use std::time::Duration; +pub enum SocketType { + ICMP, + UDP, +} + +pub struct PingReply { + pub(crate) reflector: IpAddr, + pub(crate) seq: u16, + pub(crate) rtt: i64, + pub(crate) current_time: i64, + pub(crate) down_time: f64, + pub(crate) up_time: f64, + pub(crate) originate_timestamp: i64, + pub(crate) receive_timestamp: i64, + pub(crate) transmit_timestamp: i64, + pub(crate) last_receive_time_s: f64, +} + +fn open_socket(type_: SocketType) -> RawFd { + match type_ { + ICMP => { + socket( + AddressFamily::Inet, + SockType::Raw, + SockFlag::empty(), /* value */ + SockProtocol::ICMP, + ) + } + UDP => { + socket( + AddressFamily::Inet, + SockType::Datagram, + SockFlag::empty(), /* value */ + SockProtocol::Udp, + ) + } + } + .expect("Couldn't open socket") +} + pub trait PingListener { - fn new(id: u16, reflectors: Vec) -> Self; - + fn new(id: u16) -> Self; fn get_id(&self) -> u16; - fn get_reflectors(&self) -> Vec; - fn listen(&mut self) { - let socket = socket( - AddressFamily::Inet, - SockType::Raw, - SockFlag::empty(), /* value */ - SockProtocol::ICMP, - ) - .expect("Couldn't open socket"); + fn listen( + &mut self, + type_: SocketType, + reflectors_lock: Arc>>, + stats_sender: Sender, + ) { + let socket = open_socket(type_); loop { - let mut v: Vec = vec![0; 128]; - let buf = v.as_mut_slice(); + let mut v: Vec = vec![0; 40]; + let mut buf = v.as_mut_slice(); let (size, sender) = match recvfrom::(socket, buf) { Ok(val) => val, Err(_) => continue, }; - let (rtt, down_time, up_time) = match Self::parse_packet(buf, size) { + // etherparse doesn't like when the size in the header doesn't match the buffer + // so resize the buffer when actual packet size is known + buf = buf[..size].as_mut(); + + let addr_bytes = sender.expect("Should be an address here").ip(); + let addr = IpAddr::from(addr_bytes.to_be_bytes()); + + let reflectors = reflectors_lock.lock().unwrap(); + if !reflectors.contains(&addr) { + continue; + } + + let reply = match Self::parse_packet(addr, buf, size) { Ok(val) => val, Err(e) => { println!("Shit went to hell: {}", 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); + stats_sender.send(reply).unwrap(); } } - fn parse_packet(buf: &[u8], len: usize) -> Result<(i64, i64, i64), Box>; + fn parse_packet(reflector: IpAddr, buf: &[u8], len: usize) + -> Result>; } pub trait PingSender { - fn new(id: u16, reflectors: Vec) -> Self; - + fn new(id: u16) -> Self; fn get_id(&self) -> u16; - fn get_reflectors(&self) -> Vec; - fn send(&mut self) { - let socket = socket( - AddressFamily::Inet, - SockType::Raw, - SockFlag::empty(), /* value */ - SockProtocol::ICMP, - ) - .expect("Couldn't open socket"); + fn send(&mut self, type_: SocketType, reflectors_lock: Arc>>) { + let socket = open_socket(type_); let mut seq: u16 = 0; let tick_duration_ms: u16 = 500; let sleep_duration = Duration::from_millis(tick_duration_ms as u64); loop { - let reflectors = Self::get_reflectors(self); + let reflectors_unlocked = reflectors_lock.lock().unwrap(); + let reflectors = reflectors_unlocked.clone(); + drop(reflectors_unlocked); for reflector in reflectors.iter() { let addr: Box; @@ -101,10 +151,3 @@ pub trait PingSender { fn craft_packet(&self, seq: u16) -> Vec; } - -pub trait Pinger { - fn new(id: u16, reflectors: Vec) -> Self; - fn receive_loop(&mut self); - fn sender_loop(&mut self); - fn send_ping(&mut self, reflector: &IpAddr, id: u16, seq: u16) -> Result<(), Box>; -} diff --git a/src/pinger_icmp.rs b/src/pinger_icmp.rs index 08bab89..f7e505c 100644 --- a/src/pinger_icmp.rs +++ b/src/pinger_icmp.rs @@ -1,228 +1,121 @@ -use bincode::{deserialize, serialize}; use std::error::Error; use std::net::IpAddr; -use std::{thread, time}; -use crate::pinger::{PingListener, PingSender}; -use crate::Pinger; +use crate::pinger::{PingListener, PingReply, PingSender}; use byteorder::*; -use neli::err::NlError; +use etherparse::TransportSlice::{Icmpv4, Icmpv6}; +use etherparse::{IcmpEchoHeader, Icmpv4Header, Icmpv4Type, SlicedPacket}; use nix::sys::time::TimeValLike; use nix::time::{clock_gettime, ClockId}; -use pnet::packet::icmp::IcmpTypes::EchoReply; -use pnet::packet::ip::IpNextHeaderProtocols; -use pnet::packet::*; -use pnet_transport::TransportChannelType::Layer4; -use pnet_transport::TransportProtocol::Ipv4; -use pnet_transport::*; -use serde::{Deserialize, Serialize}; - -#[derive(Deserialize, Serialize, Clone, Default, Debug)] -//#[repr(C)] -pub struct ICMPEcho { - type_: u8, - code: u8, - checksum: u16, - identifier: u16, - sequence: u16, - #[serde(with = "serde_bytes")] - payload: Vec, -} - -pub struct PingerICMPEcho { - id: u16, - reflectors: Vec, - rx: TransportReceiver, - tx: TransportSender, -} pub struct PingerICMPEchoListener { id: u16, - reflectors: Vec, } pub struct PingerICMPEchoSender { id: u16, - reflectors: Vec, -} - -fn calculate_checksum(buffer: &mut [u8]) -> u16 { - let mut sum = 0u32; - for word in buffer.chunks(2) { - let mut part = u16::from(word[0]) << 8; - if word.len() > 1 { - part += u16::from(word[1]); - } - sum = sum.wrapping_add(u32::from(part)); - } - - while (sum >> 16) > 0 { - sum = (sum & 0xffff) + (sum >> 16); - } - - !sum as u16 -} - -impl Pinger for PingerICMPEcho { - fn new(id: u16, reflectors: Vec) -> Self { - let protocol = Layer4(Ipv4(IpNextHeaderProtocols::Icmp)); - - // Create a new transport channel, dealing with layer 4 packets on a test protocol - // It has a receive buffer of 4096 bytes. - let (tx, rx) = match transport_channel(4096, protocol) { - Ok((tx, rx)) => (tx, rx), - Err(e) => panic!( - "An error occurred when creating the transport channel: {}", - e - ), - }; - - PingerICMPEcho { - id, - reflectors, - rx, - tx, - } - } - - fn receive_loop(&mut self) { - let mut iter = icmp_packet_iter(&mut self.rx); - loop { - let (packet, sender) = match iter.next() { - Ok(res) => res, - Err(_) => continue, - }; - - let curr_time: u64 = clock_gettime(ClockId::CLOCK_MONOTONIC) - .unwrap() - .num_milliseconds() as u64; - - if packet.get_icmp_type() != EchoReply { - continue; - } - - let icmp_response = icmp::echo_reply::EchoReplyPacket::new(packet.packet()).unwrap(); - - if icmp_response.get_identifier() != self.id { - continue; - } - - let sent_time: u64 = icmp_response.payload().read_u64::().unwrap(); - let rtt = curr_time - sent_time; - - println!("Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Sent time: {:8} | RTT: {:8}", "ICMP", sender.to_string(), icmp_response.get_sequence_number(), curr_time, sent_time, rtt); - } - } - - fn sender_loop(&mut self) { - println!("My ID is: {}", self.id); - - let mut seq: u16 = 0; - let tick_duration_ms: u16 = 500; - let sleep_duration = time::Duration::from_millis(tick_duration_ms as u64); - - loop { - let reflectors = self.reflectors.clone(); - - for reflector in reflectors.iter() { - self.send_ping(reflector, self.id, seq) - .expect("TODO: panic message"); - } - - thread::sleep(sleep_duration); - - seq += 1; - - if seq > u16::MAX { - seq = 0; - } - } - } - - fn send_ping(&mut self, reflector: &IpAddr, id: u16, seq: u16) -> Result<(), Box> { - let time = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap(); - - let mut buf = [0u8; 8 + 56]; - - let mut packet = icmp::echo_request::MutableEchoRequestPacket::new(&mut buf).unwrap(); - - packet.populate(&icmp::echo_request::EchoRequest { - icmp_type: icmp::IcmpTypes::EchoRequest, - icmp_code: icmp::IcmpCode::new(0), - checksum: 0, - identifier: id, - sequence_number: seq, - payload: Vec::from(time.num_milliseconds().to_ne_bytes()), - }); - - packet.set_checksum(icmp::checksum( - &icmp::IcmpPacket::new(&packet.packet()).unwrap(), - )); - - self.tx.send_to(packet, *reflector).expect("Error sending"); - - Ok(()) - } } impl PingListener for PingerICMPEchoListener { - fn new(id: u16, reflectors: Vec) -> Self { - PingerICMPEchoListener { id, reflectors } + fn new(id: u16) -> Self { + PingerICMPEchoListener { id } } fn get_id(&self) -> u16 { self.id } - fn get_reflectors(&self) -> Vec { - self.reflectors.clone() - } + // Result: RTT, down time, up time + fn parse_packet( + reflector: IpAddr, + buf: &[u8], + len: usize, + ) -> Result> { + match SlicedPacket::from_ip(buf) { + Err(value) => println!("Err {:?}", value), + Ok(value) => match value.transport { + Some(Icmpv4(icmp)) => match icmp.icmp_type() { + Icmpv4Type::EchoReply(echo) => { + let time_sent = icmp + .payload() + .read_u64::() + .expect("Couldn't parse payload to time") + as i64; - fn parse_packet(buf: &[u8], len: usize) -> Result<(i64, i64, i64), Box> { - if len != 44 { - println!("len: {}", len); - return Err(Box::new(NlError::msg("Wrong length"))); + let time_now = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap(); + + let time_ms = time_now.num_milliseconds(); + + let rtt = time_ms - time_sent; + return Ok(PingReply { + reflector, + seq: echo.seq, + rtt, + current_time: time_ms, + down_time: (rtt / 2) as f64, + up_time: (rtt / 2) as f64, + originate_timestamp: 0, + receive_timestamp: 0, + transmit_timestamp: 0, + last_receive_time_s: time_now.tv_sec() as f64 + + (time_now.tv_nsec() as f64 / 1e9), + }); + } + _ => {} + }, + Some(Icmpv6(icmp)) => {} + Some(_) => {} + None => {} + }, } - let hdr: ICMPEcho = deserialize(buf).expect("Failed to parse ICMP packet"); - - todo!() + 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, + }) } } impl PingSender for PingerICMPEchoSender { - fn new(id: u16, reflectors: Vec) -> Self { - PingerICMPEchoSender { id, reflectors } + fn new(id: u16) -> Self { + PingerICMPEchoSender { id } } fn get_id(&self) -> u16 { self.id } - fn get_reflectors(&self) -> Vec { - self.reflectors.clone() - } - fn craft_packet(&self, seq: u16) -> Vec { let time = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap(); let time_u64: u64 = time.num_milliseconds() as u64; + let payload = time_u64.to_ne_bytes(); - let mut hdr = ICMPEcho { - type_: 8, - code: 0, - checksum: 0, - identifier: self.get_id(), - sequence: seq, - payload: time_u64.to_ne_bytes().to_vec(), - }; + // Construct a header with checksum based on the payload + let hdr = Icmpv4Header::with_checksum( + Icmpv4Type::EchoRequest(IcmpEchoHeader { + id: self.get_id(), + seq, + }), + &payload, + ); - println!("id: {}", hdr.identifier); - println!("seq: {}", hdr.sequence); + // Create a buffer to hold the result of header + payload + let mut result = Vec::::with_capacity(hdr.header_len() + payload.len()); - let mut tmp_buf_v = serialize(&hdr).unwrap(); - let mut tmp_buf = tmp_buf_v.as_mut_slice(); - hdr.checksum = calculate_checksum(tmp_buf).to_be(); + // Write the header to the buffer + hdr.write(&mut result).expect("Error writing packet"); - serialize(&hdr).unwrap() + // Write the payload to the buffer + result.append(&mut payload.to_vec()); + + result } } diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs index 5995836..d0bfaf2 100644 --- a/src/pinger_icmp_ts.rs +++ b/src/pinger_icmp_ts.rs @@ -1,151 +1,113 @@ -use bincode::{deserialize, serialize}; use std::error::Error; use std::net::IpAddr; -use std::{thread, time}; -use crate::Pinger; +use crate::error::PingParseError; +use crate::pinger::{PingListener, PingReply, PingSender}; +use crate::Utils; +use etherparse::icmpv4::TimestampMessage; +use etherparse::TransportSlice::Icmpv4; +use etherparse::{Icmpv4Header, Icmpv4Type, SlicedPacket}; use nix::time::{clock_gettime, ClockId}; -use pnet::packet::icmp::IcmpTypes::{Timestamp, TimestampReply}; -use pnet::packet::ip::IpNextHeaderProtocols; -use pnet::packet::*; -use pnet_transport::TransportChannelType::Layer4; -use pnet_transport::TransportProtocol::Ipv4; -use pnet_transport::*; -use serde::{Deserialize, Serialize}; -#[derive(Deserialize, Serialize, Copy, Clone, Default, Debug)] -pub struct ICMPTimestamp { - type_: u8, - code: u8, - checksum: u16, - identifier: u16, - sequence: u16, - originate_time: u32, - receive_time: u32, - transmit_time: u32, -} - -pub struct PingerICMPTimestamps { +pub struct PingerICMPTimestampListener { id: u16, - reflectors: Vec, - rx: TransportReceiver, - tx: TransportSender, } -impl Pinger for PingerICMPTimestamps { - fn new(id: u16, reflectors: Vec) -> Self { - let protocol = Layer4(Ipv4(IpNextHeaderProtocols::Icmp)); +pub struct PingerICMPTimestampSender { + id: u16, +} - // Create a new transport channel, dealing with layer 4 packets on a test protocol - // It has a receive buffer of 4096 bytes. - let (tx, rx) = match transport_channel(4096, protocol) { - Ok((tx, rx)) => (tx, rx), - Err(e) => panic!( - "An error occurred when creating the transport channel: {}", - e - ), - }; - - PingerICMPTimestamps { - id, - reflectors, - rx, - tx, - } +impl PingListener for PingerICMPTimestampListener { + fn new(id: u16) -> Self { + PingerICMPTimestampListener { id } } - fn receive_loop(&mut self) { - let mut iter = icmp_packet_iter(&mut self.rx); - loop { - let (packet, sender) = match iter.next() { - Ok(res) => res, - Err(_) => continue, - }; - - let time = clock_gettime(ClockId::CLOCK_REALTIME).unwrap(); - let time_since_midnight: i64 = - (time.tv_sec() % 86400 * 1000) + (time.tv_nsec() / 1000000); - - if packet.get_icmp_type() != TimestampReply { - continue; - } - - let hdr: ICMPTimestamp = deserialize(packet.packet()).unwrap(); - - if hdr.identifier != self.id { - println!("wrong id: {} != {}", hdr.identifier, self.id); - continue; - } - - let rtt: i64 = time_since_midnight - hdr.originate_time.to_be() as i64; - let dl_time: i64 = time_since_midnight - hdr.transmit_time.to_be() as i64; - let ul_time: i64 = hdr.receive_time.to_be() as i64 - hdr.originate_time.to_be() as i64; - - 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", sender.to_string(), hdr.sequence, time_since_midnight, hdr.originate_time.to_be(), hdr.receive_time.to_be(), hdr.transmit_time.to_be(), rtt, ul_time, dl_time); - } + fn get_id(&self) -> u16 { + self.id } - fn sender_loop(&mut self) { - println!("My ID is: {}", self.id); + // Result: RTT, down time, up time + fn parse_packet( + reflector: IpAddr, + buf: &[u8], + len: usize, + ) -> Result> { + match SlicedPacket::from_ip(buf) { + Err(value) => println!("Err {:?}", value), + Ok(value) => match value.transport { + Some(Icmpv4(icmp)) => match icmp.icmp_type() { + Icmpv4Type::TimestampReply(reply) => { + let time_now = clock_gettime(ClockId::CLOCK_REALTIME).unwrap(); + let time_since_midnight: i64 = (time_now.tv_sec() as i64 % 86400 * 1000) + + (time_now.tv_nsec() as i64 / 1000000); - let mut seq: u16 = 0; - let tick_duration_ms: u16 = 500; - let sleep_duration = time::Duration::from_millis(tick_duration_ms as u64); + let originate_timestamp = Utils::to_ne(reply.originate_timestamp); + let receive_timestamp = Utils::to_ne(reply.receive_timestamp); + let transmit_timestamp = Utils::to_ne(reply.transmit_timestamp); - loop { - let reflectors = self.reflectors.clone(); + let rtt: i64 = time_since_midnight - originate_timestamp as i64; + let dl_time: i64 = time_since_midnight - transmit_timestamp as i64; + let ul_time: i64 = receive_timestamp as i64 - originate_timestamp as i64; - for reflector in reflectors.iter() { - self.send_ping(reflector, self.id, seq) - .expect("TODO: panic message"); - } - - thread::sleep(sleep_duration); - - seq += 1; - - if seq > u16::MAX { - seq = 0; - } + return Ok(PingReply { + reflector, + seq: reply.seq, + rtt, + current_time: time_since_midnight, + down_time: dl_time as f64, + up_time: ul_time as f64, + originate_timestamp: originate_timestamp as i64, + receive_timestamp: receive_timestamp as i64, + transmit_timestamp: transmit_timestamp as i64, + last_receive_time_s: time_now.tv_sec() as f64 + + (time_now.tv_nsec() as f64 / 1e9), + }); + } + _ => {} + }, + Some(_) => {} + None => {} + }, } + + Err(Box::new(PingParseError {})) + } +} + +impl PingSender for PingerICMPTimestampSender { + fn new(id: u16) -> Self { + PingerICMPTimestampSender { id } } - fn send_ping(&mut self, reflector: &IpAddr, id: u16, seq: u16) -> Result<(), Box> { - let mut buf = [0u8; 8 + 56]; - - let mut packet = pnet::packet::icmp::MutableIcmpPacket::new(&mut buf).unwrap(); + fn get_id(&self) -> u16 { + self.id + } + fn craft_packet(&self, seq: u16) -> Vec { let time = clock_gettime(ClockId::CLOCK_REALTIME).unwrap(); let time_since_midnight: u32 = ((time.tv_sec() % 86400 * 1000) + (time.tv_nsec() / 1000000)) as u32; - let icmp = ICMPTimestamp { - type_: 0, - code: 0, - checksum: 0, - identifier: id, - sequence: seq, - originate_time: time_since_midnight.to_be(), - receive_time: 0, - transmit_time: 0, - }; + let payload: [u8; 0] = []; - let buf_v: Vec = serialize(&icmp).unwrap(); - let payload = &buf_v[4..]; + // Construct a header with checksum based on the payload + let hdr = Icmpv4Header::with_checksum( + Icmpv4Type::TimestampRequest(TimestampMessage { + id: self.id, + seq, + originate_timestamp: time_since_midnight, + receive_timestamp: 0, + transmit_timestamp: 0, + }), + &payload, + ); - packet.populate(&icmp::Icmp { - icmp_type: Timestamp, - icmp_code: icmp::IcmpCode::new(0), - checksum: 0, - payload: Vec::from(payload), - }); + // Create a buffer to hold the result of header + payload + let mut result = Vec::::with_capacity(hdr.header_len()); - packet.set_checksum(icmp::checksum( - &icmp::IcmpPacket::new(&packet.packet()).unwrap(), - )); + // Write the header to the buffer + hdr.write(&mut result).expect("Error writing packet"); - self.tx.send_to(packet, *reflector).expect("Error sending"); - - Ok(()) + result } } diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs new file mode 100644 index 0000000..20b2d30 --- /dev/null +++ b/src/ratecontroller.rs @@ -0,0 +1,314 @@ +use crate::netlink::Netlink; +use crate::{Config, ReflectorStats, Utils}; +use rand::seq::SliceRandom; +use rand::thread_rng; +use rand::RngCore; +use std::collections::HashMap; +use std::net::IpAddr; +use std::sync::mpsc::Sender; +use std::sync::{Arc, Mutex}; +use std::thread::sleep; +use std::time::Duration; + +#[derive(Copy, Clone)] +pub enum StatsDirection { + RX, + TX, +} + +fn f64_max(a: f64, b: f64) -> f64 { + *[a, b].iter().max_by(|a, b| a.total_cmp(b)).unwrap() +} + +fn f64_min(a: f64, b: f64) -> f64 { + *[a, b].iter().min_by(|a, b| a.total_cmp(b)).unwrap() +} + +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) => { + println!("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()); + (-1, -1) + } + }; + + let rx_bytes = match down_direction { + RX => down_rx, + TX => down_tx, + }; + + let tx_bytes = match up_direction { + RX => up_rx, + TX => up_tx, + }; + + (rx_bytes, tx_bytes) +} + +pub struct Ratecontroller { + pub(crate) config: Config, + pub(crate) owd_baseline: Arc>>, + pub(crate) owd_recent: Arc>>, + pub(crate) reflectors_lock: Arc>>, + pub(crate) reselect_trigger: Sender, +} + +impl Ratecontroller { + pub fn run(&self, down_direction: StatsDirection, up_direction: StatsDirection) { + 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, start_ns) = 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 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(); + + // 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; + 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(); + + let (mut prev_rx_bytes, mut prev_tx_bytes) = + 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."); + } + + let mut t_prev_bytes = lastchg_t; + + 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); + } + + let mut nrate_up = 0; + let mut nrate_down = 0; + + loop { + 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; + + if now_t - lastchg_t > self.config.min_change_interval { + let owd_baseline = self.owd_baseline.lock().unwrap(); + let owd_recent = self.owd_recent.lock().unwrap(); + let reflectors = self.reflectors_lock.lock().unwrap(); + + let mut down_delta_stat: f64; + let mut up_delta_stat: f64; + + if reflectors.len() > 0 { + let mut down_deltas = Vec::new(); + let mut up_deltas = Vec::new(); + let ( + mut next_dl_rate, + mut next_ul_rate, + mut rx_load, + mut tx_load, + mut up_utilisation, + mut down_utilisation, + ): (f64, f64, f64, f64, f64, f64); + + rx_load = -1.0; + tx_load = -1.0; + down_delta_stat = 0.0; + up_delta_stat = 0.0; + + for reflector in reflectors.iter() { + 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 + { + down_deltas.push( + owd_recent[reflector].down_ewma - owd_baseline[reflector].down_ewma, + ); + up_deltas.push( + owd_recent[reflector].up_ewma - owd_baseline[reflector].up_ewma, + ); + } + } + + if down_deltas.len() < 5 || up_deltas.len() < 5 { + // trigger reselection + let _ = self.reselect_trigger.send(true); + } + + next_dl_rate = cur_dl_rate; + next_ul_rate = cur_ul_rate; + + 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"); + } else if down_deltas.len() < 3 || up_deltas.len() < 3 { + next_dl_rate = min_dl_rate; + next_ul_rate = min_ul_rate; + } else { + down_deltas.sort_by(|a, b| a.partial_cmp(b).unwrap()); + up_deltas.sort_by(|a, b| a.partial_cmp(b).unwrap()); + + down_delta_stat = Utils::a_else_b(down_deltas[3], down_deltas[1]); + up_delta_stat = Utils::a_else_b(up_deltas[3], up_deltas[1]); + + if down_delta_stat > 0.0 && up_delta_stat > 0.0 { + /* + * TODO - find where the (8 / 1000) comes from and + * i. convert to a pre-computed factor + * ii. ideally, see if it can be defined in terms of constants, eg ticks per second and number of active reflectors + */ + down_utilisation = (8.0 / 1000.0) + * (cur_rx_bytes as f64 - prev_rx_bytes as f64) + / (now_t - t_prev_bytes); + rx_load = down_utilisation / cur_dl_rate; + up_utilisation = (8.0 / 1000.0) + * (cur_tx_bytes as f64 - prev_tx_bytes as f64) + / (now_t - t_prev_bytes); + tx_load = up_utilisation / cur_ul_rate; + next_ul_rate = cur_ul_rate; + next_dl_rate = cur_dl_rate; + + if down_delta_stat > 0.0 + && down_delta_stat < dl_max_delta_owd + && rx_load > 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); + nrate_down = nrate_down + 1; + nrate_down = nrate_down % histsize; + } + + if up_delta_stat > 0.0 + && up_delta_stat < dl_max_delta_owd + && tx_load > 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); + nrate_up = nrate_up + 1; + nrate_up = nrate_up % histsize; + } + + if down_delta_stat > dl_max_delta_owd { + 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; + } + } + + if up_delta_stat > ul_max_delta_owd { + 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; + } + } + } + } + + t_prev_bytes = now_t; + 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(); + + if next_dl_rate != cur_dl_rate || next_ul_rate != cur_ul_rate { + println!( + "next_ul_rate {} next_dl_rate {}", + next_ul_rate, next_dl_rate + ); + } + + if next_dl_rate != cur_dl_rate { + Netlink::set_qdisc_rate(down_qdisc, next_dl_rate as u64).unwrap(); + } + + if next_ul_rate != cur_ul_rate { + Netlink::set_qdisc_rate(up_qdisc, next_ul_rate as u64).unwrap(); + } + + cur_dl_rate = next_dl_rate; + cur_ul_rate = next_ul_rate; + + (lastchg_s, lastchg_ns) = Utils::get_current_time(); + + if rx_load > 0.0 + && tx_load > 0.0 + && down_delta_stat > 0.0 + && up_delta_stat > 0.0 + { + println!( + "{},{},{},{},{},{},{},{}", + lastchg_s, + lastchg_ns, + rx_load, + tx_load, + down_delta_stat, + up_delta_stat, + cur_dl_rate, + cur_ul_rate + ); + } + + lastchg_s = lastchg_s - start_s; + lastchg_t = lastchg_s + lastchg_ns / 1e9; + } + } + + sleep(sleep_time); + } + } +} diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs new file mode 100644 index 0000000..158a781 --- /dev/null +++ b/src/reflector_selector.rs @@ -0,0 +1,129 @@ +use crate::{Config, ReflectorStats}; +use rand::seq::SliceRandom; +use rand::{thread_rng, Rng}; +use std::collections::HashMap; +use std::net::IpAddr; +use std::sync::mpsc::Receiver; +use std::sync::{Arc, Mutex}; +use std::thread::sleep; +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, + pub(crate) trigger_channel: Receiver, +} + +impl ReflectorSelector { + pub fn run(&self) { + 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, + ); + + let mut rng = thread_rng(); + + // Initial wait of several seconds to allow some OWD data to build up + sleep(baseline_sleep_time); + + loop { + /* + * Selection is triggered either by some other thread triggering it through the channel, + * or it passes the timeout. In any case we don't care about the result of this function, + * so we ignore the result of it. + */ + let _ = self + .trigger_channel + .recv_timeout(selector_sleep_time) + .unwrap_or(true); + reselection_count += 1; + + // After 40 reselections, slow down to every 15 minutes + if reselection_count > 40 { + selector_sleep_time = Duration::new(15 * 60, 0); + } + + let mut next_peers: Vec = Vec::new(); + let mut reflectors_peers = self.reflector_peers_lock.lock().unwrap(); + + // Include all current peers + for reflector in reflectors_peers.iter() { + next_peers.push(*reflector); + } + + for i in 1..20 { + let next_candidate = self.reflector_pool.choose(&mut rng).unwrap(); + next_peers.push(*next_candidate); + } + + // Put all the pool members back into the peers for some re-baselining... + *reflectors_peers = next_peers.clone(); + + // Drop the MutexGuard explicitly, as Rust won't unlock the mutex by default + // until the guard goes out of scope + drop(reflectors_peers); + + // Wait for several seconds to allow all reflectors to be re-baselined + sleep(baseline_sleep_time); + + // Re-acquire the lock when we wake up again + reflectors_peers = self.reflector_peers_lock.lock().unwrap(); + 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); + } else { + println!( + "No data found from candidate reflector: {} - skipping", + peer.to_string() + ); + } + } + + // Sort the candidates table now by ascending RTT + candidates.sort_by(|a, b| a.1.cmp(&b.1)); + + // Now we will just limit the candidates down to 2 * num_reflectors + let mut num_reflectors = self.config.num_reflectors.clone(); + let candidate_pool_num = (2 * num_reflectors) as usize; + candidates = candidates[0..candidate_pool_num - 1].to_vec(); + + for (candidate, rtt) in candidates.iter() { + println!("Fastest candidate {}: {}", candidate, rtt); + } + + // Shuffle the deck so we avoid overwhelming good reflectors (Fisher-Yates) + for i in (1 as usize..candidates.len()).rev() { + let j = rng.gen_range(0..(i + 1)); + candidates.swap(i, j); + } + + if (candidates.len() as u8) < num_reflectors { + num_reflectors = candidates.len() as u8; + } + + let mut new_peers = Vec::new(); + for i in 1..num_reflectors { + new_peers.push(candidates[i as usize].0); + println!( + "New selected peer: {}", + candidates[i as usize].0.to_string() + ); + } + + *reflectors_peers = new_peers; + } + } +} diff --git a/src/utils.rs b/src/utils.rs new file mode 100644 index 0000000..e2776a6 --- /dev/null +++ b/src/utils.rs @@ -0,0 +1,26 @@ +use nix::time::{clock_gettime, ClockId}; + +pub(crate) struct Utils {} + +impl Utils { + pub(crate) fn a_else_b(a: f64, b: f64) -> f64 { + return if a > 0.0 { a } else { b }; + } + 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) + } + + #[cfg(target_endian = "big")] + pub(crate) fn to_ne(val: u64) -> u64 { + val.to_be() + } + + #[cfg(target_endian = "little")] + pub(crate) fn to_ne(val: u32) -> u32 { + val.to_le() + } +}