ratecontroller: use durations instead of manually handling time

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<Mutex<HashMap<IpAddr, ReflectorStats>>>,
     pub owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
     pub reselect_trigger: Sender<bool>,
+    pub start_time: Instant,
     pub stats_receiver: Receiver<PingReply>,
 }
 
@@ -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
+            if time_data
-                || time_data.last_receive_time_s - owd_recent.last_receive_time_s > 30.0
+                .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;
+                // mark the data as bad by setting the receive time to the time autorate was started
-                owd_recent.last_receive_time_s = time_data.last_receive_time_s - 60.0;
+                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

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(

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<Socket> {

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
+                            last_receive_time_s: Instant::now(),
-                                + (clock.get_nanoseconds() as f64 / 1e9),
                         })
                     }
                     type_ => Err(PingError::InvalidType(format!("{:?}", type_))),

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
+                            last_receive_time_s: Instant::now(),
-                                + (time_now.get_nanoseconds() as f64 / 1e9),
                         })
                     }
                     type_ => Err(PingError::InvalidType(format!("{:?}", type_))),

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<T>(dt: T, format: &(impl Formattable + ?Sized)) -> String
+where
+    T: Into<OffsetDateTime>,
+{
+    dt.into().format(format).unwrap()
+}
+
 fn generate_initial_speeds(base_speed: f64, size: u32) -> Vec<f64> {
     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<f64>,
     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_t
-                    > now_abstime - 2.0 * self.config.tick_interval
+                    .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 mut lastchg_t = Instant::now();
-        let start_s = time.get_seconds() as f64;
+        let mut lastdump_t = Instant::now();
-        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;
 
         // 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);
+            if now_t.duration_since(lastchg_t).as_secs_f64() > self.config.min_change_interval {
-            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 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.update_deltas();
-                self.state_ul.now_t = now_t;
-                self.update_deltas(now_abstime);
                 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,
+                    time_format(SystemTime::now(), DUMP_DATETIME_FORMAT),
-                    lastchg_ns,
                     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,
+                            time_format(SystemTime::now(), DUMP_DATETIME_FORMAT),
-                            lastchg_ns,
                             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 = now_t;
-                lastchg_t = lastchg_s + lastchg_ns / 1e9;
             }
 
             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(),
                         ) {

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)
     }