feat(threads): Add graceful shutdown checks to worker loops

src/baseliner.rs

@@ -1,3 +1,4 @@
+use crate::SHUTDOWN;
 use crate::Config;
 use crate::metrics::{Metric, MetricsSender};
 use crate::pinger::PingReply;
@@ -5,6 +6,7 @@ use crate::util::MutexExt;
 use log::info;
 use std::collections::HashMap;
 use std::net::IpAddr;
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::{Receiver, Sender};
 use std::sync::{Arc, Mutex};
 use std::time::Instant;
@@ -45,6 +47,10 @@ impl Baseliner {
         let fast_factor = ewma_factor(self.config.tick_interval, 0.4);
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Baseliner shutting down");
+                return Ok(());
+            }
             let time_data = self.stats_rx.recv()?;
 
             let mut owd_baseline_map = self.owd_baseline.lock_anyhow()?;

src/pinger.rs

@@ -1,10 +1,12 @@
+use crate::SHUTDOWN;
 use crate::MeasurementType;
 use crate::metrics::{Metric, MetricsSender};
 use crate::util::RwLockExt;
 use icmp_socket2::socket::IcmpSocket;
 use icmp_socket2::{IcmpSocket4, Icmpv4Packet};
-use log::{debug, warn};
+use log::{debug, info, warn};
 use std::net::{IpAddr, Ipv4Addr};
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::Sender;
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
@@ -58,6 +60,10 @@ pub trait PingListener {
         let socket = &mut open_socket(type_)?;
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Ping listener shutting down");
+                return Ok(());
+            }
             let (pkt, sender) = match socket.rcv_from() {
                 Ok(val) => val,
                 Err(_) => continue,
@@ -130,6 +136,10 @@ pub trait PingSender {
         let tick_duration_ms: u16 = (tick_interval * 1000.0) as u16;
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Ping sender shutting down");
+                return Ok(());
+            }
             // Clone the reflectors vec and drop the read lock immediately —
             // holding it for the entire tick would starve the reflector selector's write lock.
             let reflectors_unlocked = reflectors_lock.read_anyhow()?;

src/ratecontroller.rs

@@ -1,3 +1,4 @@
+use crate::SHUTDOWN;
 use crate::metrics::{Metric, MetricsSender};
 use crate::netlink::{Netlink, NetlinkError, Qdisc};
 use crate::time::Time;
@@ -9,6 +10,7 @@ use std::collections::HashMap;
 use std::fs::File;
 use std::io::Write;
 use std::net::IpAddr;
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::Sender;
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread::sleep;
@@ -326,6 +328,10 @@ impl Ratecontroller {
         }
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Rate controller shutting down");
+                return Ok(());
+            }
             sleep(sleep_time);
             let now_t = Instant::now();
 

src/reflector_selector.rs

@@ -1,9 +1,11 @@
+use crate::SHUTDOWN;
 use crate::metrics::{Metric, MetricsSender};
 use crate::util::{MutexExt, RwLockExt};
 use crate::{Config, ReflectorStats};
 use log::{debug, info};
 use std::collections::HashMap;
 use std::net::IpAddr;
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::Receiver;
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread::sleep;
@@ -28,6 +30,10 @@ impl ReflectorSelector {
         sleep(baseline_sleep_time);
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Reflector selector shutting down");
+                return Ok(());
+            }
             /*
              * 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,