diff --git a/src/baseliner.rs b/src/baseliner.rs index 37f297c..9d2319f 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -5,12 +5,13 @@ use std::collections::HashMap; use std::net::IpAddr; use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, Mutex}; +use std::time::Instant; #[derive(Copy, Clone)] pub struct ReflectorStats { pub down_ewma: f64, pub up_ewma: f64, - pub last_receive_time_s: f64, + pub last_receive_time_s: Instant, } pub struct Baseliner { @@ -18,6 +19,7 @@ pub struct Baseliner { pub owd_baseline: Arc>>, pub owd_recent: Arc>>, pub reselect_trigger: Sender, + pub start_time: Instant, pub stats_receiver: Receiver, } @@ -64,8 +66,16 @@ impl Baseliner { .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 + if time_data + .last_receive_time_s + .duration_since(owd_baseline.last_receive_time_s) + .as_secs_f64() + > 30.0 + || time_data + .last_receive_time_s + .duration_since(owd_recent.last_receive_time_s) + .as_secs_f64() + > 30.0 { owd_baseline.down_ewma = time_data.down_time; owd_baseline.up_ewma = time_data.up_time; @@ -82,8 +92,9 @@ impl Baseliner { 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; + // mark the data as bad by setting the receive time to the time autorate was started + owd_baseline.last_receive_time_s = self.start_time; + owd_recent.last_receive_time_s = self.start_time; info!( "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection", time_data.reflector diff --git a/src/main.rs b/src/main.rs index d27042d..fcf0442 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,7 +14,7 @@ mod reflector_selector; mod time; use crate::baseliner::{Baseliner, ReflectorStats}; -use ::log::debug; +use ::log::{debug, info}; use std::collections::HashMap; use std::net::IpAddr; use std::str::FromStr; @@ -22,6 +22,7 @@ use std::sync::mpsc::channel; use std::sync::{Arc, Mutex, RwLock}; use std::thread::sleep; use std::time::Duration; +use std::time::Instant; use std::{process, thread}; use crate::config::{Config, MeasurementType}; @@ -40,6 +41,7 @@ fn main() -> anyhow::Result<()> { let config = Config::new()?; log::init(config.log_level)?; let mut reflectors = config.load_reflectors()?; + let start_t = Instant::now(); // The identifier field in ICMP is only 2 bytes // so take the last 2 bytes of the PID as the ID @@ -95,6 +97,7 @@ fn main() -> anyhow::Result<()> { owd_baseline: owd_baseline.clone(), owd_recent: owd_recent.clone(), reselect_trigger: reselect_sender.clone(), + start_time: start_t, stats_receiver: baseliner_stats_receiver, }; @@ -105,9 +108,21 @@ fn main() -> anyhow::Result<()> { * so there should be no initial bufferbloat to * fool the baseliner */ + info!( + "Setting shaper rates to minimum (D/L): {} / {}", + config.download_min_kbits, config.upload_min_kbits + ); Netlink::set_qdisc_rate(down_qdisc, config.download_min_kbits as u64)?; Netlink::set_qdisc_rate(up_qdisc, config.upload_min_kbits as u64)?; - sleep(Duration::new(0, 5e8 as u32)); + + // Sleep for a few seconds to give the shaper a chance + // to control the queue if load is heavy + let settle_sleep_time = Duration::new(2, 0); + info!( + "Sleeping for {} to give the shaper a chance to get in control if there's bloat", + settle_sleep_time.as_secs_f64() + ); + sleep(settle_sleep_time); let reflector_peers_lock_clone = reflector_peers_lock.clone(); let receiver_handle = thread::Builder::new().name("receiver".to_string()).spawn( diff --git a/src/pinger.rs b/src/pinger.rs index d8d737f..3aa8b78 100644 --- a/src/pinger.rs +++ b/src/pinger.rs @@ -6,7 +6,7 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6}; use std::str::FromStr; use std::sync::mpsc::Sender; use std::sync::{Arc, RwLock}; -use std::time::Duration; +use std::time::{Duration, Instant}; use std::{io, thread}; use thiserror::Error; @@ -36,7 +36,7 @@ pub struct PingReply { pub originate_timestamp: i64, pub receive_timestamp: i64, pub transmit_timestamp: i64, - pub last_receive_time_s: f64, + pub last_receive_time_s: Instant, } fn open_socket(type_: MeasurementType) -> io::Result { diff --git a/src/pinger_icmp.rs b/src/pinger_icmp.rs index fa4d4e2..b73b20e 100644 --- a/src/pinger_icmp.rs +++ b/src/pinger_icmp.rs @@ -1,4 +1,5 @@ use std::net::IpAddr; +use std::time::Instant; use crate::pinger::{PingError, PingListener, PingReply, PingSender}; use crate::time::Time; @@ -46,8 +47,7 @@ impl PingListener for PingerICMPEchoListener { originate_timestamp: 0, receive_timestamp: 0, transmit_timestamp: 0, - last_receive_time_s: clock.get_seconds() as f64 - + (clock.get_nanoseconds() as f64 / 1e9), + last_receive_time_s: Instant::now(), }) } type_ => Err(PingError::InvalidType(format!("{:?}", type_))), diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs index 7650fe6..7c0362b 100644 --- a/src/pinger_icmp_ts.rs +++ b/src/pinger_icmp_ts.rs @@ -6,6 +6,7 @@ use etherparse::TransportSlice::{Icmpv4, Icmpv6}; use etherparse::{Icmpv4Header, Icmpv4Type, SlicedPacket}; use rustix::thread::ClockId; use std::net::IpAddr; +use std::time::Instant; pub struct PingerICMPTimestampListener {} @@ -47,8 +48,7 @@ impl PingListener for PingerICMPTimestampListener { originate_timestamp: originate_timestamp as i64, receive_timestamp: receive_timestamp as i64, transmit_timestamp: transmit_timestamp as i64, - last_receive_time_s: time_now.get_seconds() as f64 - + (time_now.get_nanoseconds() as f64 / 1e9), + last_receive_time_s: Instant::now(), }) } type_ => Err(PingError::InvalidType(format!("{:?}", type_))), diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index 72dbf3a..450acfb 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -1,11 +1,9 @@ use crate::netlink::{Netlink, NetlinkError, Qdisc}; -use crate::time::Time; use crate::{Config, ReflectorStats}; use log::{debug, error, info, warn}; use rand::seq::SliceRandom; use rand::thread_rng; use rand::RngCore; -use rustix::thread::ClockId; use std::collections::HashMap; use std::fs::File; use std::io::Write; @@ -13,8 +11,12 @@ use std::net::IpAddr; use std::sync::mpsc::Sender; use std::sync::{Arc, Mutex, RwLock}; use std::thread::sleep; -use std::time::Duration; +use std::time::{Duration, Instant, SystemTime}; use thiserror::Error; +use time::format_description::FormatItem; +use time::formatting::Formattable; +use time::macros::format_description; +use time::OffsetDateTime; #[derive(Copy, Clone, Debug, PartialEq)] enum Direction { @@ -34,6 +36,18 @@ pub enum StatsDirection { TX, } +const DUMP_DATETIME_FORMAT: &[FormatItem] = format_description!( + "[year]-[month]-[day] [hour]:[minute]:[second] [offset_hour \ + sign:mandatory]:[offset_minute]:[offset_second]" +); + +fn time_format(dt: T, format: &(impl Formattable + ?Sized)) -> String +where + T: Into, +{ + dt.into().format(format).unwrap() +} + fn generate_initial_speeds(base_speed: f64, size: u32) -> Vec { let mut rates = Vec::new(); @@ -76,10 +90,9 @@ struct State { qdisc: Qdisc, load: f64, next_rate: f64, - now_t: f64, nrate: usize, previous_bytes: i128, - previous_bytes_t: f64, + prev_t: Instant, safe_rates: Vec, utilisation: f64, } @@ -114,6 +127,9 @@ impl Ratecontroller { ) }; + let now_t = Instant::now(); + let dur = now_t.duration_since(state.prev_t); + if !state.deltas.is_empty() { state.next_rate = state.current_rate; @@ -134,7 +150,7 @@ impl Ratecontroller { */ state.utilisation = (8.0 / 1000.0) * (state.current_bytes as f64 - state.previous_bytes as f64) - / (state.now_t - state.previous_bytes_t); + / dur.as_secs_f64(); state.load = state.utilisation / state.current_rate; if state.delta_stat > 0.0 @@ -170,21 +186,21 @@ impl Ratecontroller { } } - state.previous_bytes_t = state.now_t; - state.previous_bytes = state.current_bytes; - state.next_rate = state.next_rate.max(min_rate).round(); + state.previous_bytes = state.current_bytes; + state.prev_t = now_t; Ok(()) } - fn update_deltas(&mut self, now_abstime: f64) { + fn update_deltas(&mut self) { let state_dl = &mut self.state_dl; let state_ul = &mut self.state_ul; state_dl.deltas.clear(); state_ul.deltas.clear(); + let now_t = Instant::now(); let owd_baseline = self.owd_baseline.lock().unwrap(); let owd_recent = self.owd_recent.lock().unwrap(); let reflectors = self.reflectors_lock.read().unwrap(); @@ -193,8 +209,10 @@ impl Ratecontroller { // 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_interval + && now_t + .duration_since(owd_recent[reflector].last_receive_time_s) + .as_secs_f64() + < self.config.tick_interval * 2.0 { state_dl .deltas @@ -255,7 +273,6 @@ impl Ratecontroller { deltas: Vec::new(), load: 0.0, next_rate: 0.0, - now_t: 0.0, nrate: 0, qdisc: dl_qdisc, previous_bytes: cur_rx, @@ -270,7 +287,6 @@ impl Ratecontroller { deltas: Vec::new(), load: 0.0, next_rate: 0.0, - now_t: 0.0, nrate: 0, qdisc: ul_qdisc, previous_bytes: cur_tx, @@ -285,12 +301,8 @@ impl Ratecontroller { pub fn run(&mut self) -> anyhow::Result<()> { let sleep_time = Duration::from_secs_f64(self.config.min_change_interval); - let time = Time::new(ClockId::Monotonic); - let start_s = time.get_seconds() as f64; - let (mut lastchg_s, mut lastchg_ns) = - (time.get_seconds() as f64, time.get_nanoseconds() as f64); - let mut lastchg_t = lastchg_s - start_s + lastchg_ns / 1e9; - let mut lastdump_t = lastchg_t - 310.0; + let mut lastchg_t = Instant::now(); + let mut lastdump_t = Instant::now(); // set qdisc rates to 60% of base rate to make sure we start with sane baselines self.state_dl.current_rate = self.config.download_base_kbits * 0.6; @@ -336,14 +348,9 @@ impl Ratecontroller { loop { sleep(sleep_time); + let now_t = Instant::now(); - let now = Time::new(ClockId::Monotonic); - let (mut now_s, now_ns) = (now.get_seconds() as f64, now.get_nanoseconds() as f64); - let now_abstime = now_s + now_ns / 1e9; - now_s -= start_s; - let now_t = now_s + now_ns / 1e9; - - if now_t - lastchg_t > self.config.min_change_interval { + if now_t.duration_since(lastchg_t).as_secs_f64() > self.config.min_change_interval { // if it's been long enough, and the stats indicate needing to change speeds // change speeds here @@ -355,9 +362,7 @@ impl Ratecontroller { continue; } - self.state_dl.now_t = now_t; - self.state_ul.now_t = now_t; - self.update_deltas(now_abstime); + self.update_deltas(); self.calculate_rate(Direction::Down)?; self.calculate_rate(Direction::Up)?; @@ -381,16 +386,9 @@ impl Ratecontroller { self.state_dl.current_rate = self.state_dl.next_rate; self.state_ul.current_rate = self.state_ul.next_rate; - let lastchg = Clock::new(ClockId::Monotonic); - (lastchg_s, lastchg_ns) = ( - lastchg.get_seconds() as f64, - lastchg.get_nanoseconds() as f64, - ); - debug!( - "{},{},{},{},{},{},{},{}", - lastchg_s, - lastchg_ns, + "{},{},{},{},{},{},{}", + time_format(SystemTime::now(), DUMP_DATETIME_FORMAT), self.state_dl.load, self.state_ul.load, self.state_dl.delta_stat, @@ -402,9 +400,8 @@ impl Ratecontroller { if let Some(ref mut fd) = stats_fd { if let Err(e) = fd.write( format!( - "{},{},{},{},{},{},{},{}\n", - lastchg_s, - lastchg_ns, + "{},{},{},{},{},{},{}", + time_format(SystemTime::now(), DUMP_DATETIME_FORMAT), self.state_dl.load, self.state_ul.load, self.state_dl.delta_stat, @@ -418,17 +415,19 @@ impl Ratecontroller { } } - lastchg_s -= start_s; - lastchg_t = lastchg_s + lastchg_ns / 1e9; + lastchg_t = now_t; } if let Some(ref mut fd) = speed_hist_fd { - if now_t - lastdump_t > 300.0 { + if now_t.duration_since(lastdump_t).as_secs_f64() > 300.0 { for i in 0..self.config.speed_hist_size as usize { if let Err(e) = fd.write_all( format!( "{},{},{},{}\n", - now_t, i, self.state_ul.safe_rates[i], self.state_dl.safe_rates[i] + time_format(SystemTime::now(), DUMP_DATETIME_FORMAT), + i, + self.state_ul.safe_rates[i], + self.state_dl.safe_rates[i] ) .as_bytes(), ) { diff --git a/src/time.rs b/src/time.rs index c7e69fd..9df6662 100644 --- a/src/time.rs +++ b/src/time.rs @@ -16,14 +16,6 @@ impl Time { } } - pub fn get_seconds(&self) -> u64 { - self.time_s - } - - pub fn get_nanoseconds(&self) -> u64 { - self.time_ns - } - pub fn get_time_since_midnight(&self) -> i64 { (self.time_s as i64 % 86400 * 1000) + (self.time_ns as i64 / 1000000) }