Track timeouts in seconds

Author: ComputerCraftr

src/main.rs

@@ -102,7 +102,7 @@ fn main() -> io::Result<()> {
 
     // Global application state
     let locked = Arc::new(AtomicBool::new(false));
-    let last_seen_ns = Arc::new(AtomicU64::new(0));
+    let last_seen_s = Arc::new(AtomicU64::new(0));
 
     let stats = Arc::new(Stats::new());
     let exit_code_set = Arc::new(AtomicU32::new(0));
@@ -136,7 +136,7 @@ fn main() -> io::Result<()> {
             let sock_mgrs_a = sock_mgrs.clone();
             let worker_id = worker_base;
             let locked_a = Arc::clone(&locked);
-            let last_seen_a = Arc::clone(&last_seen_ns);
+            let last_seen_a = Arc::clone(&last_seen_s);
             let stats_a = Arc::clone(&stats);
 
             thread::spawn(move || {
@@ -159,7 +159,7 @@ fn main() -> io::Result<()> {
             let sock_mgr_b = Arc::clone(&sock_mgr);
             let worker_id = worker_base + 1;
             let locked_b = Arc::clone(&locked);
-            let last_seen_b = Arc::clone(&last_seen_ns);
+            let last_seen_b = Arc::clone(&last_seen_s);
             let stats_b = Arc::clone(&stats);
 
             thread::spawn(move || {
@@ -181,7 +181,7 @@ fn main() -> io::Result<()> {
         let cfg_w = Arc::clone(&cfg);
         let sock_mgrs_w = sock_mgrs.clone();
         let locked_w = Arc::clone(&locked);
-        let last_seen_w = Arc::clone(&last_seen_ns);
+        let last_seen_w = Arc::clone(&last_seen_s);
         let exit_code_set_w = Arc::clone(&exit_code_set);
 
         thread::spawn(move || {

src/net/payload.rs

@@ -163,7 +163,7 @@ pub(crate) fn handle_payload_result(
     t_recv: Instant,
     cfg: &Config,
     stats: &Stats,
-    last_seen_ns: &AtomicU64,
+    last_seen_s: &AtomicU64,
     validated: &ValidatedPayload<'_>,
     send_res: &io::Result<bool>,
     sock_connected: bool,
@@ -172,7 +172,8 @@ pub(crate) fn handle_payload_result(
 ) {
     match send_res {
         Ok(res) => {
-            last_seen_ns.store(Stats::dur_ns(t_start, t_recv), AtomOrdering::Relaxed);
+            let last_seen = t_recv.saturating_duration_since(t_start).as_secs();
+            last_seen_s.store(last_seen, AtomOrdering::Relaxed);
             if cfg.stats_interval_mins != 0 {
                 let t_send = Instant::now();
                 stats.send_add(c2u, validated.len as u64, t_recv, t_send);

src/stats.rs

@@ -64,14 +64,9 @@ impl Stats {
         }
     }
 
-    #[inline]
-    pub fn dur_ns(start: Instant, end: Instant) -> u64 {
-        end.saturating_duration_since(start).as_nanos() as u64
-    }
-
     #[inline]
     pub fn send_add(&self, c2u: bool, bytes: u64, start: Instant, end: Instant) {
-        let lat_ns = Self::dur_ns(start, end);
+        let lat_ns = end.saturating_duration_since(start).as_nanos() as u64;
         let (atom_pkts, atom_bytes, atom_lat_sum_ns, atom_lat_max_ns, atom_bytes_max) = if c2u {
             (
                 &self.agg.c2u_pkts,

src/worker.rs

@@ -148,21 +148,18 @@ pub fn run_watchdog_thread(
     cfg: &Config,
     sock_mgrs: &[Arc<SocketManager>],
     locked: &AtomicBool,
-    last_seen_ns: &AtomicU64,
+    last_seen_s: &AtomicU64,
     exit_code_set: &AtomicU32,
 ) {
-    let timeout_ns = Duration::from_secs(cfg.timeout_secs)
-        .as_nanos()
-        .min(u128::from(u64::MAX)) as u64;
     let period = Duration::from_secs(1);
     loop {
         thread::sleep(period);
         if locked.load(AtomOrdering::Relaxed) {
             let now = Instant::now();
-            let now_ns = Stats::dur_ns(t_start, now);
-            let last_ns = last_seen_ns.load(AtomOrdering::Relaxed);
+            let now_s = now.saturating_duration_since(t_start).as_secs();
+            let last_s = last_seen_s.load(AtomOrdering::Relaxed);
 
-            if last_ns != 0 && now_ns.saturating_sub(last_ns) >= timeout_ns {
+            if last_s != 0 && now_s.saturating_sub(last_s) >= cfg.timeout_secs {
                 match cfg.on_timeout {
                     TimeoutAction::Drop => {
                         log_warn!(
@@ -191,7 +188,7 @@ pub fn run_watchdog_thread(
                             }
                         }
                         locked.store(false, AtomOrdering::Relaxed);
-                        last_seen_ns.store(0, AtomOrdering::Relaxed);
+                        last_seen_s.store(0, AtomOrdering::Relaxed);
                     }
                     _ => {
                         log_warn!(
@@ -213,7 +210,7 @@ pub fn run_upstream_to_client_thread(
     sock_mgr: &SocketManager,
     worker_id: usize,
     locked: &AtomicBool,
-    last_seen_ns: &AtomicU64,
+    last_seen_s: &AtomicU64,
     stats: &Stats,
 ) {
     const C2U: bool = false;
@@ -260,7 +257,7 @@ pub fn run_upstream_to_client_thread(
                         t_recv,
                         cfg,
                         stats,
-                        last_seen_ns,
+                        last_seen_s,
                         &validated,
                         &send_res,
                         handles.client_connected,
@@ -288,7 +285,7 @@ pub fn run_client_to_upstream_thread(
     all_sock_mgrs: &[Arc<SocketManager>],
     worker_id: usize,
     locked: &AtomicBool,
-    last_seen_ns: &AtomicU64,
+    last_seen_s: &AtomicU64,
     stats: &Stats,
 ) {
     const C2U: bool = true;
@@ -336,7 +333,7 @@ pub fn run_client_to_upstream_thread(
                             t_recv,
                             cfg,
                             stats,
-                            last_seen_ns,
+                            last_seen_s,
                             &validated,
                             &send_res,
                             handles.upstream_connected,
@@ -452,7 +449,7 @@ pub fn run_client_to_upstream_thread(
                             t_recv,
                             cfg,
                             stats,
-                            last_seen_ns,
+                            last_seen_s,
                             &validated,
                             &send_res,
                             handles.upstream_connected,
@@ -485,7 +482,7 @@ pub fn run_client_to_upstream_thread(
                             t_recv,
                             cfg,
                             stats,
-                            last_seen_ns,
+                            last_seen_s,
                             &validated,
                             &send_res,
                             handles.upstream_connected,