Revert from a dashmap to a regular mutex for the flow container. Performance improved, and flow removal is less troublesome.

src/rust/lqosd/src/throughput_tracker/flow_data/flow_analysis/finished_flows.rs

@@ -1,13 +1,8 @@
 use super::{get_asn_name_and_country, FlowAnalysis};
 use crate::throughput_tracker::flow_data::FlowbeeRecipient;
-use dashmap::DashMap;
 use lqos_sys::flowbee_data::{FlowbeeData, FlowbeeKey};
 use once_cell::sync::Lazy;
-use std::{
-    collections::HashMap,
-    fmt::Debug,
-    sync::{Arc, Mutex},
-};
+use std::sync::{Arc, Mutex};
 
 struct TimeBuffer {
     buffer: Mutex<Vec<TimeEntry>>,

src/rust/lqosd/src/throughput_tracker/flow_data/flow_tracker.rs

@@ -1,13 +1,13 @@
 //! Provides a globally accessible vector of all flows. This is used to store
 //! all flows for the purpose of tracking and data-services.
 
-use dashmap::DashMap;
+use super::flow_analysis::FlowAnalysis;
 use lqos_sys::flowbee_data::{FlowbeeData, FlowbeeKey};
 use once_cell::sync::Lazy;
-use super::flow_analysis::FlowAnalysis;
+use std::{collections::HashMap, sync::Mutex};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct AsnId(pub u32);
 
-pub static ALL_FLOWS: Lazy<DashMap<FlowbeeKey, (FlowbeeData, FlowAnalysis)>> = Lazy::new(|| DashMap::new());
-
+pub static ALL_FLOWS: Lazy<Mutex<HashMap<FlowbeeKey, (FlowbeeData, FlowAnalysis)>>> =
+    Lazy::new(|| Mutex::new(HashMap::new()));

src/rust/lqosd/src/throughput_tracker/flow_data/mod.rs

@@ -13,7 +13,7 @@ use std::sync::{
     mpsc::{channel, Sender},
     Arc,
 };
-pub(crate) use flow_analysis::{setup_flow_analysis, lookup_asn_id, get_asn_name_and_country, FlowAnalysis};
+pub(crate) use flow_analysis::{setup_flow_analysis, get_asn_name_and_country, FlowAnalysis};
 
 
 trait FlowbeeRecipient {

src/rust/lqosd/src/throughput_tracker/mod.rs

@@ -1,24 +1,25 @@
+pub mod flow_data;
 mod heimdall_data;
 mod throughput_entry;
 mod tracking_data;
-pub mod flow_data;
 use crate::{
-    shaped_devices_tracker::{NETWORK_JSON, STATS_NEEDS_NEW_SHAPED_DEVICES, SHAPED_DEVICES}, stats::TIME_TO_POLL_HOSTS,
-    throughput_tracker::tracking_data::ThroughputTracker, long_term_stats::get_network_tree,
+    long_term_stats::get_network_tree,
+    shaped_devices_tracker::{NETWORK_JSON, SHAPED_DEVICES, STATS_NEEDS_NEW_SHAPED_DEVICES},
+    stats::TIME_TO_POLL_HOSTS,
+    throughput_tracker::tracking_data::ThroughputTracker,
 };
 pub use heimdall_data::get_flow_stats;
 use log::{info, warn};
 use lqos_bus::{BusResponse, FlowbeeProtocol, IpStats, TcHandle, TopFlowType, XdpPpingResult};
 use lqos_sys::flowbee_data::{FlowbeeData, FlowbeeKey};
 use lqos_utils::{unix_time::time_since_boot, XdpIpAddress};
-use lts_client::collector::{StatsUpdateMessage, ThroughputSummary, HostSummary};
+use lts_client::collector::{HostSummary, StatsUpdateMessage, ThroughputSummary};
 use once_cell::sync::Lazy;
 use tokio::{
     sync::mpsc::Sender,
     time::{Duration, Instant},
 };
-
-use self::flow_data::{get_asn_name_and_country, AsnId, FlowAnalysis, ALL_FLOWS};
+use self::flow_data::{get_asn_name_and_country, FlowAnalysis, ALL_FLOWS};
 
 const RETIRE_AFTER_SECONDS: u64 = 30;
 
@@ -32,43 +33,46 @@ pub static THROUGHPUT_TRACKER: Lazy<ThroughputTracker> = Lazy::new(ThroughputTra
 /// * `long_term_stats_tx` - an optional MPSC sender to notify the
 ///   collection thread that there is fresh data.
 pub async fn spawn_throughput_monitor(
-  long_term_stats_tx: Sender<StatsUpdateMessage>,
-  netflow_sender: std::sync::mpsc::Sender<(FlowbeeKey, (FlowbeeData, FlowAnalysis))>,
+    long_term_stats_tx: Sender<StatsUpdateMessage>,
+    netflow_sender: std::sync::mpsc::Sender<(FlowbeeKey, (FlowbeeData, FlowAnalysis))>,
 ) {
     info!("Starting the bandwidth monitor thread.");
     let interval_ms = 1000; // 1 second
     info!("Bandwidth check period set to {interval_ms} ms.");
-    tokio::spawn(throughput_task(interval_ms, long_term_stats_tx, netflow_sender));
+    tokio::spawn(throughput_task(
+        interval_ms,
+        long_term_stats_tx,
+        netflow_sender,
+    ));
 }
 
 async fn throughput_task(
-  interval_ms: u64, 
-  long_term_stats_tx: Sender<StatsUpdateMessage>,
-  netflow_sender: std::sync::mpsc::Sender<(FlowbeeKey, (FlowbeeData, FlowAnalysis))>
+    interval_ms: u64,
+    long_term_stats_tx: Sender<StatsUpdateMessage>,
+    netflow_sender: std::sync::mpsc::Sender<(FlowbeeKey, (FlowbeeData, FlowAnalysis))>,
 ) {
     // Obtain the flow timeout from the config, default to 30 seconds
     let timeout_seconds = if let Ok(config) = lqos_config::load_config() {
-      if let Some(flow_config) = config.flows {
-        flow_config.flow_timeout_seconds
-      } else {
-        30
-      }
+        if let Some(flow_config) = config.flows {
+            flow_config.flow_timeout_seconds
+        } else {
+            30
+        }
     } else {
-      30
+        30
     };
 
     // Obtain the netflow_enabled from the config, default to false
     let netflow_enabled = if let Ok(config) = lqos_config::load_config() {
-      if let Some(flow_config) = config.flows {
-        flow_config.netflow_enabled
-      } else {
-        false
-      }
+        if let Some(flow_config) = config.flows {
+            flow_config.netflow_enabled
+        } else {
+            false
+        }
     } else {
-      false
+        false
     };
 
-
     loop {
         let start = Instant::now();
 
@@ -76,34 +80,34 @@ async fn throughput_task(
         // the tokio runtime is not blocked.
         let my_netflow_sender = netflow_sender.clone();
         if let Err(e) = tokio::task::spawn_blocking(move || {
-
-          {
-              let net_json = NETWORK_JSON.read().unwrap();
-              net_json.zero_throughput_and_rtt();
-          } // Scope to end the lock
-          THROUGHPUT_TRACKER.copy_previous_and_reset_rtt();
-          THROUGHPUT_TRACKER.apply_new_throughput_counters();
-          THROUGHPUT_TRACKER.apply_flow_data(
-            timeout_seconds,
-            netflow_enabled,
-            my_netflow_sender.clone(),
-          );
-          THROUGHPUT_TRACKER.update_totals();
-          THROUGHPUT_TRACKER.next_cycle();
-          let duration_ms = start.elapsed().as_micros();
-          TIME_TO_POLL_HOSTS.store(duration_ms as u64, std::sync::atomic::Ordering::Relaxed);
-
-        }).await {
+            {
+                let net_json = NETWORK_JSON.read().unwrap();
+                net_json.zero_throughput_and_rtt();
+            } // Scope to end the lock
+            THROUGHPUT_TRACKER.copy_previous_and_reset_rtt();
+            THROUGHPUT_TRACKER.apply_new_throughput_counters();
+            THROUGHPUT_TRACKER.apply_flow_data(
+                timeout_seconds,
+                netflow_enabled,
+                my_netflow_sender.clone(),
+            );
+            THROUGHPUT_TRACKER.update_totals();
+            THROUGHPUT_TRACKER.next_cycle();
+            let duration_ms = start.elapsed().as_micros();
+            TIME_TO_POLL_HOSTS.store(duration_ms as u64, std::sync::atomic::Ordering::Relaxed);
+        })
+        .await
+        {
             log::error!("Error polling network. {e:?}");
         }
         tokio::spawn(submit_throughput_stats(long_term_stats_tx.clone()));
 
         let elapsed = start.elapsed();
         if elapsed.as_secs_f32() < 1.0 {
-          let sleep_duration = Duration::from_millis(interval_ms) - start.elapsed();
-          tokio::time::sleep(sleep_duration).await;
+            let sleep_duration = Duration::from_millis(interval_ms) - start.elapsed();
+            tokio::time::sleep(sleep_duration).await;
         } else {
-          log::error!("Throughput monitor thread is running behind. It took {elapsed} to poll the network.", elapsed=elapsed.as_secs_f32());
+            log::error!("Throughput monitor thread is running behind. It took {elapsed} to poll the network.", elapsed=elapsed.as_secs_f32());
         }
     }
 }
@@ -149,12 +153,15 @@ async fn submit_throughput_stats(long_term_stats_tx: Sender<StatsUpdateMessage>)
         })
         .collect();
 
-    let summary = Box::new((ThroughputSummary{
-        bits_per_second,
-        shaped_bits_per_second,
-        packets_per_second,
-        hosts,
-    }, get_network_tree()));
+    let summary = Box::new((
+        ThroughputSummary {
+            bits_per_second,
+            shaped_bits_per_second,
+            packets_per_second,
+            hosts,
+        },
+        get_network_tree(),
+    ));
 
     // Send the stats
     let result = long_term_stats_tx
@@ -199,141 +206,150 @@ type TopList = (XdpIpAddress, (u64, u64), (u64, u64), f32, TcHandle, String);
 
 pub fn top_n(start: u32, end: u32) -> BusResponse {
     let mut full_list: Vec<TopList> = {
-      let tp_cycle = THROUGHPUT_TRACKER.cycle.load(std::sync::atomic::Ordering::Relaxed);
-      THROUGHPUT_TRACKER.raw_data
-        .iter()
-        .filter(|v| !v.key().as_ip().is_loopback())
-        .filter(|d| retire_check(tp_cycle, d.most_recent_cycle))
-        .map(|te| {
-          (
-            *te.key(),
-            te.bytes_per_second,
-            te.packets_per_second,
-            te.median_latency().unwrap_or(0.0),
-            te.tc_handle,
-            te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
-          )
-        })
-        .collect()
+        let tp_cycle = THROUGHPUT_TRACKER
+            .cycle
+            .load(std::sync::atomic::Ordering::Relaxed);
+        THROUGHPUT_TRACKER
+            .raw_data
+            .iter()
+            .filter(|v| !v.key().as_ip().is_loopback())
+            .filter(|d| retire_check(tp_cycle, d.most_recent_cycle))
+            .map(|te| {
+                (
+                    *te.key(),
+                    te.bytes_per_second,
+                    te.packets_per_second,
+                    te.median_latency().unwrap_or(0.0),
+                    te.tc_handle,
+                    te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
+                )
+            })
+            .collect()
     };
     full_list.sort_by(|a, b| b.1 .0.cmp(&a.1 .0));
     let result = full_list
-      .iter()
-      .skip(start as usize)
-      .take((end as usize) - (start as usize))
-      .map(
-        |(
-          ip,
-          (bytes_dn, bytes_up),
-          (packets_dn, packets_up),
-          median_rtt,
-          tc_handle,
-          circuit_id,
-        )| IpStats {
-          ip_address: ip.as_ip().to_string(),
-          circuit_id: circuit_id.clone(),
-          bits_per_second: (bytes_dn * 8, bytes_up * 8),
-          packets_per_second: (*packets_dn, *packets_up),
-          median_tcp_rtt: *median_rtt,
-          tc_handle: *tc_handle,
-        },
-      )
-      .collect();
-    BusResponse::TopDownloaders(result)
-  }
-
-  pub fn worst_n(start: u32, end: u32) -> BusResponse {
-    let mut full_list: Vec<TopList> = {
-      let tp_cycle = THROUGHPUT_TRACKER.cycle.load(std::sync::atomic::Ordering::Relaxed);
-      THROUGHPUT_TRACKER.raw_data
         .iter()
-        .filter(|v| !v.key().as_ip().is_loopback())
-        .filter(|d| retire_check(tp_cycle, d.most_recent_cycle))
-        .filter(|te| te.median_latency().is_some())
-        .map(|te| {
-          (
-            *te.key(),
-            te.bytes_per_second,
-            te.packets_per_second,
-            te.median_latency().unwrap_or(0.0),
-            te.tc_handle,
-            te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
-          )
-        })
-        .collect()
+        .skip(start as usize)
+        .take((end as usize) - (start as usize))
+        .map(
+            |(
+                ip,
+                (bytes_dn, bytes_up),
+                (packets_dn, packets_up),
+                median_rtt,
+                tc_handle,
+                circuit_id,
+            )| IpStats {
+                ip_address: ip.as_ip().to_string(),
+                circuit_id: circuit_id.clone(),
+                bits_per_second: (bytes_dn * 8, bytes_up * 8),
+                packets_per_second: (*packets_dn, *packets_up),
+                median_tcp_rtt: *median_rtt,
+                tc_handle: *tc_handle,
+            },
+        )
+        .collect();
+    BusResponse::TopDownloaders(result)
+}
+
+pub fn worst_n(start: u32, end: u32) -> BusResponse {
+    let mut full_list: Vec<TopList> = {
+        let tp_cycle = THROUGHPUT_TRACKER
+            .cycle
+            .load(std::sync::atomic::Ordering::Relaxed);
+        THROUGHPUT_TRACKER
+            .raw_data
+            .iter()
+            .filter(|v| !v.key().as_ip().is_loopback())
+            .filter(|d| retire_check(tp_cycle, d.most_recent_cycle))
+            .filter(|te| te.median_latency().is_some())
+            .map(|te| {
+                (
+                    *te.key(),
+                    te.bytes_per_second,
+                    te.packets_per_second,
+                    te.median_latency().unwrap_or(0.0),
+                    te.tc_handle,
+                    te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
+                )
+            })
+            .collect()
     };
     full_list.sort_by(|a, b| b.3.partial_cmp(&a.3).unwrap());
     let result = full_list
-      .iter()
-      .skip(start as usize)
-      .take((end as usize) - (start as usize))
-      .map(
-        |(
-          ip,
-          (bytes_dn, bytes_up),
-          (packets_dn, packets_up),
-          median_rtt,
-          tc_handle,
-          circuit_id,
-        )| IpStats {
-          ip_address: ip.as_ip().to_string(),
-          circuit_id: circuit_id.clone(),
-          bits_per_second: (bytes_dn * 8, bytes_up * 8),
-          packets_per_second: (*packets_dn, *packets_up),
-          median_tcp_rtt: *median_rtt,
-          tc_handle: *tc_handle,
-        },
-      )
-      .collect();
-    BusResponse::WorstRtt(result)
-  }
-
-  pub fn best_n(start: u32, end: u32) -> BusResponse {
-    let mut full_list: Vec<TopList> = {
-      let tp_cycle = THROUGHPUT_TRACKER.cycle.load(std::sync::atomic::Ordering::Relaxed);
-      THROUGHPUT_TRACKER.raw_data
         .iter()
-        .filter(|v| !v.key().as_ip().is_loopback())
-        .filter(|d| retire_check(tp_cycle, d.most_recent_cycle))
-        .filter(|te| te.median_latency().is_some())
-        .map(|te| {
-          (
-            *te.key(),
-            te.bytes_per_second,
-            te.packets_per_second,
-            te.median_latency().unwrap_or(0.0),
-            te.tc_handle,
-            te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
-          )
-        })
-        .collect()
+        .skip(start as usize)
+        .take((end as usize) - (start as usize))
+        .map(
+            |(
+                ip,
+                (bytes_dn, bytes_up),
+                (packets_dn, packets_up),
+                median_rtt,
+                tc_handle,
+                circuit_id,
+            )| IpStats {
+                ip_address: ip.as_ip().to_string(),
+                circuit_id: circuit_id.clone(),
+                bits_per_second: (bytes_dn * 8, bytes_up * 8),
+                packets_per_second: (*packets_dn, *packets_up),
+                median_tcp_rtt: *median_rtt,
+                tc_handle: *tc_handle,
+            },
+        )
+        .collect();
+    BusResponse::WorstRtt(result)
+}
+
+pub fn best_n(start: u32, end: u32) -> BusResponse {
+    let mut full_list: Vec<TopList> = {
+        let tp_cycle = THROUGHPUT_TRACKER
+            .cycle
+            .load(std::sync::atomic::Ordering::Relaxed);
+        THROUGHPUT_TRACKER
+            .raw_data
+            .iter()
+            .filter(|v| !v.key().as_ip().is_loopback())
+            .filter(|d| retire_check(tp_cycle, d.most_recent_cycle))
+            .filter(|te| te.median_latency().is_some())
+            .map(|te| {
+                (
+                    *te.key(),
+                    te.bytes_per_second,
+                    te.packets_per_second,
+                    te.median_latency().unwrap_or(0.0),
+                    te.tc_handle,
+                    te.circuit_id.as_ref().unwrap_or(&String::new()).clone(),
+                )
+            })
+            .collect()
     };
     full_list.sort_by(|a, b| b.3.partial_cmp(&a.3).unwrap());
     full_list.reverse();
     let result = full_list
-      .iter()
-      .skip(start as usize)
-      .take((end as usize) - (start as usize))
-      .map(
-        |(
-          ip,
-          (bytes_dn, bytes_up),
-          (packets_dn, packets_up),
-          median_rtt,
-          tc_handle,
-          circuit_id,
-        )| IpStats {
-          ip_address: ip.as_ip().to_string(),
-          circuit_id: circuit_id.clone(),
-          bits_per_second: (bytes_dn * 8, bytes_up * 8),
-          packets_per_second: (*packets_dn, *packets_up),
-          median_tcp_rtt: *median_rtt,
-          tc_handle: *tc_handle,
-        },
-      )
-      .collect();
+        .iter()
+        .skip(start as usize)
+        .take((end as usize) - (start as usize))
+        .map(
+            |(
+                ip,
+                (bytes_dn, bytes_up),
+                (packets_dn, packets_up),
+                median_rtt,
+                tc_handle,
+                circuit_id,
+            )| IpStats {
+                ip_address: ip.as_ip().to_string(),
+                circuit_id: circuit_id.clone(),
+                bits_per_second: (bytes_dn * 8, bytes_up * 8),
+                packets_per_second: (*packets_dn, *packets_up),
+                median_tcp_rtt: *median_rtt,
+                tc_handle: *tc_handle,
+            },
+        )
+        .collect();
     BusResponse::BestRtt(result)
-  }  
+}
 
 pub fn xdp_pping_compat() -> BusResponse {
     let raw_cycle = THROUGHPUT_TRACKER
@@ -431,166 +447,170 @@ type FullList = (XdpIpAddress, (u64, u64), (u64, u64), f32, TcHandle, u64);
 pub fn all_unknown_ips() -> BusResponse {
     let boot_time = time_since_boot();
     if boot_time.is_err() {
-      warn!("The Linux system clock isn't available to provide time since boot, yet.");
-      warn!("This only happens immediately after a reboot.");
-      return BusResponse::NotReadyYet;
+        warn!("The Linux system clock isn't available to provide time since boot, yet.");
+        warn!("This only happens immediately after a reboot.");
+        return BusResponse::NotReadyYet;
     }
     let boot_time = boot_time.unwrap();
     let time_since_boot = Duration::from(boot_time);
-    let five_minutes_ago =
-      time_since_boot.saturating_sub(Duration::from_secs(300));
+    let five_minutes_ago = time_since_boot.saturating_sub(Duration::from_secs(300));
     let five_minutes_ago_nanoseconds = five_minutes_ago.as_nanos();
-  
+
     let mut full_list: Vec<FullList> = {
-      THROUGHPUT_TRACKER.raw_data
-        .iter()
-        .filter(|v| !v.key().as_ip().is_loopback())
-        .filter(|d| d.tc_handle.as_u32() == 0)
-        .filter(|d| d.last_seen as u128 > five_minutes_ago_nanoseconds)
-        .map(|te| {
-          (
-            *te.key(),
-            te.bytes,
-            te.packets,
-            te.median_latency().unwrap_or(0.0),
-            te.tc_handle,
-            te.most_recent_cycle,
-          )
-        })
-        .collect()
+        THROUGHPUT_TRACKER
+            .raw_data
+            .iter()
+            .filter(|v| !v.key().as_ip().is_loopback())
+            .filter(|d| d.tc_handle.as_u32() == 0)
+            .filter(|d| d.last_seen as u128 > five_minutes_ago_nanoseconds)
+            .map(|te| {
+                (
+                    *te.key(),
+                    te.bytes,
+                    te.packets,
+                    te.median_latency().unwrap_or(0.0),
+                    te.tc_handle,
+                    te.most_recent_cycle,
+                )
+            })
+            .collect()
     };
     full_list.sort_by(|a, b| b.5.partial_cmp(&a.5).unwrap());
     let result = full_list
-      .iter()
-      .map(
-        |(
-          ip,
-          (bytes_dn, bytes_up),
-          (packets_dn, packets_up),
-          median_rtt,
-          tc_handle,
-          _last_seen,
-        )| IpStats {
-          ip_address: ip.as_ip().to_string(),
-          circuit_id: String::new(),
-          bits_per_second: (bytes_dn * 8, bytes_up * 8),
-          packets_per_second: (*packets_dn, *packets_up),
-          median_tcp_rtt: *median_rtt,
-          tc_handle: *tc_handle,
-        },
-      )
-      .collect();
+        .iter()
+        .map(
+            |(
+                ip,
+                (bytes_dn, bytes_up),
+                (packets_dn, packets_up),
+                median_rtt,
+                tc_handle,
+                _last_seen,
+            )| IpStats {
+                ip_address: ip.as_ip().to_string(),
+                circuit_id: String::new(),
+                bits_per_second: (bytes_dn * 8, bytes_up * 8),
+                packets_per_second: (*packets_dn, *packets_up),
+                median_tcp_rtt: *median_rtt,
+                tc_handle: *tc_handle,
+            },
+        )
+        .collect();
     BusResponse::AllUnknownIps(result)
-  }
+}
 
-  /// For debugging: dump all active flows!
-  pub fn dump_active_flows() -> BusResponse {
-    let result: Vec<lqos_bus::FlowbeeData> = ALL_FLOWS.iter().map(|row| {
-      let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(row.value().1.asn_id.0);
+/// For debugging: dump all active flows!
+pub fn dump_active_flows() -> BusResponse {
+    let lock = ALL_FLOWS.lock().unwrap();
+    let result: Vec<lqos_bus::FlowbeeData> = lock
+        .iter()
+        .map(|(key, row)| {
+            let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(row.1.asn_id.0);
 
-      lqos_bus::FlowbeeData {
-        remote_ip: row.key().remote_ip.as_ip().to_string(),
-        local_ip: row.key().local_ip.as_ip().to_string(),
-        src_port: row.key().src_port,
-        dst_port: row.key().dst_port,
-        ip_protocol: FlowbeeProtocol::from(row.key().ip_protocol),
-        bytes_sent: row.value().0.bytes_sent,
-        packets_sent: row.value().0.packets_sent,
-        rate_estimate_bps: row.value().0.rate_estimate_bps,
-        retries: row.value().0.retries,
-        last_rtt: row.value().0.last_rtt,
-        end_status: row.value().0.end_status,
-        tos: row.value().0.tos,
-        flags: row.value().0.flags,
-        remote_asn: row.value().1.asn_id.0,
-        remote_asn_name,
-        remote_asn_country,
-        analysis: row.value().1.protocol_analysis.to_string(),
-      }
-    }).collect();
+            lqos_bus::FlowbeeData {
+                remote_ip: key.remote_ip.as_ip().to_string(),
+                local_ip: key.local_ip.as_ip().to_string(),
+                src_port: key.src_port,
+                dst_port: key.dst_port,
+                ip_protocol: FlowbeeProtocol::from(key.ip_protocol),
+                bytes_sent: row.0.bytes_sent,
+                packets_sent: row.0.packets_sent,
+                rate_estimate_bps: row.0.rate_estimate_bps,
+                retries: row.0.retries,
+                last_rtt: row.0.last_rtt,
+                end_status: row.0.end_status,
+                tos: row.0.tos,
+                flags: row.0.flags,
+                remote_asn: row.1.asn_id.0,
+                remote_asn_name,
+                remote_asn_country,
+                analysis: row.1.protocol_analysis.to_string(),
+            }
+        })
+        .collect();
 
     BusResponse::AllActiveFlows(result)
-  }
+}
 
-  /// Count active flows
-  pub fn count_active_flows() -> BusResponse {
-    BusResponse::CountActiveFlows(ALL_FLOWS.len() as u64)
-  }
+/// Count active flows
+pub fn count_active_flows() -> BusResponse {
+    let lock = ALL_FLOWS.lock().unwrap();
+    BusResponse::CountActiveFlows(lock.len() as u64)
+}
 
-  /// Top Flows Report
-  pub fn top_flows(n: u32, flow_type: TopFlowType) -> BusResponse {
-    let mut table: Vec<(FlowbeeKey, (FlowbeeData, FlowAnalysis))> = ALL_FLOWS
-      .iter()
-      .map(|row| (
-        row.key().clone(),
-        row.value().clone(),
-      ))
-      .collect();
+/// Top Flows Report
+pub fn top_flows(n: u32, flow_type: TopFlowType) -> BusResponse {
+    let lock = ALL_FLOWS.lock().unwrap();
+    let mut table: Vec<(FlowbeeKey, (FlowbeeData, FlowAnalysis))> = lock
+        .iter()
+        .map(|(key, value)| (key.clone(), value.clone()))
+        .collect();
+    std::mem::drop(lock); // Early lock release
 
     match flow_type {
-      TopFlowType::RateEstimate => {
-        table.sort_by(|a, b| {
-          let a_total = a.1.0.rate_estimate_bps[0] + a.1.0.rate_estimate_bps[1];
-          let b_total = b.1.0.rate_estimate_bps[0] + b.1.0.rate_estimate_bps[1];
-          b_total.cmp(&a_total)
-        });
-      }
-      TopFlowType::Bytes => {
-        table.sort_by(|a, b| {
-          let a_total = a.1.0.bytes_sent[0] + a.1.0.bytes_sent[1];
-          let b_total = b.1.0.bytes_sent[0] + b.1.0.bytes_sent[1];
-          b_total.cmp(&a_total)
-        });
-      }
-      TopFlowType::Packets => {
-        table.sort_by(|a, b| {
-          let a_total = a.1.0.packets_sent[0] + a.1.0.packets_sent[1];
-          let b_total = b.1.0.packets_sent[0] + b.1.0.packets_sent[1];
-          b_total.cmp(&a_total)
-        });
-      }
-      TopFlowType::Drops => {
-        table.sort_by(|a, b| {
-          let a_total = a.1.0.retries[0] + a.1.0.retries[1];
-          let b_total = b.1.0.retries[0] + b.1.0.retries[1];
-          b_total.cmp(&a_total)
-        });
-      }
-      TopFlowType::RoundTripTime => {
-        table.sort_by(|a, b| {
-          let a_total = a.1.0.last_rtt[0] + a.1.0.last_rtt[1];
-          let b_total = b.1.0.last_rtt[0] + b.1.0.last_rtt[1];
-          b_total.cmp(&a_total)
-        });
-      }
+        TopFlowType::RateEstimate => {
+            table.sort_by(|a, b| {
+                let a_total = a.1 .0.rate_estimate_bps[0] + a.1 .0.rate_estimate_bps[1];
+                let b_total = b.1 .0.rate_estimate_bps[0] + b.1 .0.rate_estimate_bps[1];
+                b_total.cmp(&a_total)
+            });
+        }
+        TopFlowType::Bytes => {
+            table.sort_by(|a, b| {
+                let a_total = a.1 .0.bytes_sent[0] + a.1 .0.bytes_sent[1];
+                let b_total = b.1 .0.bytes_sent[0] + b.1 .0.bytes_sent[1];
+                b_total.cmp(&a_total)
+            });
+        }
+        TopFlowType::Packets => {
+            table.sort_by(|a, b| {
+                let a_total = a.1 .0.packets_sent[0] + a.1 .0.packets_sent[1];
+                let b_total = b.1 .0.packets_sent[0] + b.1 .0.packets_sent[1];
+                b_total.cmp(&a_total)
+            });
+        }
+        TopFlowType::Drops => {
+            table.sort_by(|a, b| {
+                let a_total = a.1 .0.retries[0] + a.1 .0.retries[1];
+                let b_total = b.1 .0.retries[0] + b.1 .0.retries[1];
+                b_total.cmp(&a_total)
+            });
+        }
+        TopFlowType::RoundTripTime => {
+            table.sort_by(|a, b| {
+                let a_total = a.1 .0.last_rtt[0] + a.1 .0.last_rtt[1];
+                let b_total = b.1 .0.last_rtt[0] + b.1 .0.last_rtt[1];
+                b_total.cmp(&a_total)
+            });
+        }
     }
 
     let result = table
-      .iter()
-      .take(n as usize)
-      .map(|(ip, flow)| {
-        let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(flow.1.asn_id.0);
-        lqos_bus::FlowbeeData {
-          remote_ip: ip.remote_ip.as_ip().to_string(),
-          local_ip: ip.local_ip.as_ip().to_string(),
-          src_port: ip.src_port,
-          dst_port: ip.dst_port,
-          ip_protocol: FlowbeeProtocol::from(ip.ip_protocol),
-          bytes_sent: flow.0.bytes_sent,
-          packets_sent: flow.0.packets_sent,
-          rate_estimate_bps: flow.0.rate_estimate_bps,
-          retries: flow.0.retries,
-          last_rtt: flow.0.last_rtt,
-          end_status: flow.0.end_status,
-          tos: flow.0.tos,
-          flags: flow.0.flags,
-          remote_asn: flow.1.asn_id.0,
-          remote_asn_name,
-          remote_asn_country,
-          analysis: flow.1.protocol_analysis.to_string(),
-        }
-      })
-      .collect();
+        .iter()
+        .take(n as usize)
+        .map(|(ip, flow)| {
+            let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(flow.1.asn_id.0);
+            lqos_bus::FlowbeeData {
+                remote_ip: ip.remote_ip.as_ip().to_string(),
+                local_ip: ip.local_ip.as_ip().to_string(),
+                src_port: ip.src_port,
+                dst_port: ip.dst_port,
+                ip_protocol: FlowbeeProtocol::from(ip.ip_protocol),
+                bytes_sent: flow.0.bytes_sent,
+                packets_sent: flow.0.packets_sent,
+                rate_estimate_bps: flow.0.rate_estimate_bps,
+                retries: flow.0.retries,
+                last_rtt: flow.0.last_rtt,
+                end_status: flow.0.end_status,
+                tos: flow.0.tos,
+                flags: flow.0.flags,
+                remote_asn: flow.1.asn_id.0,
+                remote_asn_name,
+                remote_asn_country,
+                analysis: flow.1.protocol_analysis.to_string(),
+            }
+        })
+        .collect();
 
     BusResponse::TopFlows(result)
-  }
+}

src/rust/lqosd/src/throughput_tracker/tracking_data.rs

@@ -1,6 +1,6 @@
 use std::{sync::atomic::AtomicU64, time::Duration};
 use crate::{shaped_devices_tracker::{SHAPED_DEVICES, NETWORK_JSON}, stats::{HIGH_WATERMARK_DOWN, HIGH_WATERMARK_UP}};
-use super::{flow_data::{lookup_asn_id, AsnId, FlowAnalysis, ALL_FLOWS}, throughput_entry::ThroughputEntry, RETIRE_AFTER_SECONDS};
+use super::{flow_data::{FlowAnalysis, ALL_FLOWS}, throughput_entry::ThroughputEntry, RETIRE_AFTER_SECONDS};
 use dashmap::DashMap;
 use lqos_bus::TcHandle;
 use lqos_sys::{flowbee_data::{FlowbeeData, FlowbeeKey}, iterate_flows, throughput_for_each};
@@ -195,8 +195,9 @@ impl ThroughputTracker {
           // This flow has expired. Add it to the list to be cleaned
           expired_keys.push(key.clone());          
         } else {
+          let mut lock = ALL_FLOWS.lock().unwrap();
           // We have a valid flow, so it needs to be tracked
-          if let Some(mut this_flow) = ALL_FLOWS.get_mut(&key) {
+          if let Some(this_flow) = lock.get_mut(&key) {
             this_flow.0.last_seen = data.last_seen;
             this_flow.0.bytes_sent = data.bytes_sent;
             this_flow.0.packets_sent = data.packets_sent;
@@ -210,7 +211,7 @@ impl ThroughputTracker {
             // Insert it into the map
             let flow_analysis = FlowAnalysis::new(&key);
 
-            ALL_FLOWS.insert(key.clone(), (data.clone(), flow_analysis));
+            lock.insert(key.clone(), (data.clone(), flow_analysis));
           }
 
           // TCP - we have RTT data? 6 is TCP
@@ -241,15 +242,16 @@ impl ThroughputTracker {
         if let Err(e) = ret {
           log::warn!("Failed to end flows: {:?}", e);
         }
+        let mut lock = ALL_FLOWS.lock().unwrap();
         for key in expired_keys {
           // Send it off to netperf for analysis if we are supporting doing so.
           if netflow_enabled {
-            if let Some(d) = ALL_FLOWS.get(&key) {
+            if let Some(d) = lock.get(&key) {
               let _ = sender.send((key.clone(), (d.0.clone(), d.1.clone())));
             }
           }
 
-          //ALL_FLOWS.remove(&key);
+          lock.remove(&key);
         }
       }
     }