Don't call Instant::now() on each iteration

Author: HKalbasi

core/Cargo.toml

@@ -16,7 +16,6 @@ byteorder = "1.4.3"
 chrono = "0.4"
 colored = "2"
 cpu-time = "1.0.0"
-crossbeam-channel = "0.5.8"
 csv = "1.2.1"
 ctrlc = { version = "3.2.5", features = ["termination"] }
 dns-parser = { git = "https://github.com/stanford-esrg/dns-parser" }

core/src/conntrack/mod.rs

@@ -177,9 +177,13 @@ where
     }
 
     /// Checks for and removes inactive connections.
-    pub(crate) fn check_inactive(&mut self, subscription: &Subscription<T::Subscribed>) {
+    pub(crate) fn check_inactive(
+        &mut self,
+        subscription: &Subscription<T::Subscribed>,
+        now: Instant,
+    ) {
         self.timerwheel
-            .check_inactive(&mut self.table, subscription);
+            .check_inactive(&mut self.table, subscription, now);
     }
 }
 

core/src/conntrack/timerwheel.rs

@@ -1,20 +1,19 @@
 use crate::conntrack::{Conn, ConnId};
 use crate::subscription::{Subscription, Trackable};
 
-use crossbeam_channel::{tick, Receiver};
 use hashlink::linked_hash_map::LinkedHashMap;
 use hashlink::linked_hash_map::RawEntryMut;
 use std::collections::VecDeque;
 use std::time::{Duration, Instant};
 
 /// Tracks inactive connection expiration.
 pub(super) struct TimerWheel {
-    /// Period to check for inactive connections (in milliseconds).
-    period: usize,
+    /// Period to check for inactive connections.
+    period: Duration,
     /// Start time of the `TimerWheel`.
     start_ts: Instant,
-    /// Timeout ticker, fires every `period` milliseconds.
-    ticker: Receiver<Instant>,
+    /// Previous check time of the `TimerWheel`.
+    prev_ts: Instant,
     /// Index of the next bucket to expire.
     next_bucket: usize,
     /// List of timers.
@@ -29,11 +28,11 @@ impl TimerWheel {
             panic!("Timeout check period must be smaller than maximum inactivity timeout")
         }
         let start_ts = Instant::now();
-        let ticker = tick(Duration::from_millis(timeout_resolution as u64));
+        let period = Duration::from_millis(timeout_resolution as u64);
         TimerWheel {
-            period: timeout_resolution,
+            period,
             start_ts,
-            ticker,
+            prev_ts: start_ts,
             next_bucket: 0,
             timers: vec![VecDeque::new(); max_timeout / timeout_resolution],
         }
@@ -48,7 +47,8 @@ impl TimerWheel {
         inactivity_window: usize,
     ) {
         let current_time = (last_seen_ts - self.start_ts).as_millis() as usize;
-        let timer_index = ((current_time + inactivity_window) / self.period) % self.timers.len();
+        let period = self.period.as_millis() as usize;
+        let timer_index = ((current_time + inactivity_window) / period) % self.timers.len();
         log::debug!("Inserting into index: {}, {:?}", timer_index, current_time);
         self.timers[timer_index].push_back(conn_id.to_owned());
     }
@@ -59,9 +59,11 @@ impl TimerWheel {
         &mut self,
         table: &mut LinkedHashMap<ConnId, Conn<T>>,
         subscription: &Subscription<T::Subscribed>,
+        now: Instant,
     ) {
         let table_len = table.len();
-        if let Ok(now) = self.ticker.try_recv() {
+        if now - self.prev_ts >= self.period {
+            self.prev_ts = now;
             let nb_removed = self.remove_inactive(now, table, subscription);
             log::debug!(
                 "expired: {} ({})",
@@ -83,7 +85,7 @@ impl TimerWheel {
         table: &mut LinkedHashMap<ConnId, Conn<T>>,
         subscription: &Subscription<T::Subscribed>,
     ) -> usize {
-        let period = self.period;
+        let period = self.period.as_millis() as usize;
         let nb_buckets = self.timers.len();
         let mut not_expired: Vec<(usize, ConnId)> = vec![];
         let check_time = (now - self.start_ts).as_millis() as usize / period * period;

core/src/lcore/monitor.rs

@@ -14,7 +14,6 @@ use std::time::{Duration, Instant};
 
 use anyhow::{bail, Result};
 use chrono::Local;
-use crossbeam_channel::{tick, Receiver};
 use csv::Writer;
 use serde::Serialize;
 
@@ -81,7 +80,7 @@ impl Monitor {
                         port_wtrs.insert(*port_id, wtr);
                     }
                     return Some(Logger {
-                        ticker: tick(Duration::from_millis(log_cfg.interval)),
+                        interval: Duration::from_millis(log_cfg.interval),
                         path,
                         port_wtrs,
                         keywords: log_cfg.port_stats.clone(),
@@ -119,7 +118,12 @@ impl Monitor {
         let mut prev_rx = init_rx;
         let mut prev_ts = init_ts;
         let mut init = true;
-        let mut ticker = tokio::time::interval(Duration::from_millis(1000));
+        let mut display_ticker = tokio::time::interval(Duration::from_millis(1000));
+
+        let mut logger_ticker = self
+            .logger
+            .as_ref()
+            .map(|logger| tokio::time::interval(logger.interval));
         // Add a small delay to allow workers to start polling for packets
         tokio::time::sleep(Duration::from_millis(1000)).await;
         while self.is_running.load(Ordering::Relaxed) {
@@ -130,7 +134,7 @@ impl Monitor {
             }
 
             if let Some(display) = &self.display {
-                ticker.tick().await;
+                display_ticker.tick().await;
                 let curr_ts = Instant::now();
                 let delta = curr_ts - prev_ts;
                 match AggRxStats::collect(&self.ports, &display.keywords) {
@@ -160,11 +164,10 @@ impl Monitor {
             }
 
             if let Some(logger) = &mut self.logger {
-                if logger.ticker.try_recv().is_ok() {
-                    match logger.log_stats(init_ts.elapsed()) {
-                        Ok(_) => (),
-                        Err(error) => log::error!("Monitor log error: {}", error),
-                    }
+                logger_ticker.as_mut().unwrap().tick().await;
+                match logger.log_stats(init_ts.elapsed()) {
+                    Ok(_) => (),
+                    Err(error) => log::error!("Monitor log error: {}", error),
                 }
             }
         }
@@ -209,7 +212,7 @@ impl Display {
 
 #[derive(Debug)]
 struct Logger {
-    ticker: Receiver<Instant>,
+    interval: Duration,
     path: PathBuf,
     port_wtrs: HashMap<PortId, Writer<std::fs::File>>,
     keywords: Vec<String>,

core/src/lcore/rx_core.rs

@@ -12,6 +12,7 @@ use crate::subscription::*;
 
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
+use std::time::Instant;
 
 use itertools::Itertools;
 
@@ -95,12 +96,18 @@ where
         log::debug!("{:#?}", registry);
         let mut conn_table = ConnTracker::<S::Tracked>::new(config, registry, self.id);
 
+        let mut now = Instant::now();
+
         while self.is_running.load(Ordering::Relaxed) {
             for rxqueue in self.rxqueues.iter() {
                 let mbufs: Vec<Mbuf> = self.rx_burst(rxqueue, 32);
                 if mbufs.is_empty() {
                     IDLE_CYCLES.inc();
 
+                    if IDLE_CYCLES.get() & 1023 == 512 {
+                        now = Instant::now();
+                    }
+
                     #[cfg(feature = "prometheus")]
                     if IDLE_CYCLES.get() & 1023 == 0 && self.is_prometheus_enabled {
                         crate::stats::update_thread_local_stats(self.id);
@@ -133,7 +140,7 @@ where
                     }
                 }
             }
-            conn_table.check_inactive(&self.subscription);
+            conn_table.check_inactive(&self.subscription, now);
         }
 
         // // Deliver remaining data in table from unfinished connections