Fix: Reduce internal bus load and TreeGuard ticker latency

src/rust/lqos_bus/src/bus/unix_socket_server.rs

@@ -16,6 +16,10 @@ use tracing::{debug, error, info, warn};
 use super::BUS_SOCKET_DIRECTORY;
 use super::protocol::{decode_session_cbor, encode_reply_cbor, read_frame, write_frame};
 
+fn dropped_reply_response_count(reply: &BusReply) -> usize {
+    reply.responses.len()
+}
+
 /// Implements a Tokio-friendly server using Unix Sockets and the bus protocol.
 /// Requests are handled and then forwarded to the handler.
 pub struct UnixSocketServer {}
@@ -111,8 +115,11 @@ impl UnixSocketServer {
                 if let Some((reply_channel, msg)) = ret {
                   let mut response = BusReply { responses: Vec::with_capacity(8) };
                   handle_bus_requests(&[msg], &mut response.responses);
-                  if let Err(e) = reply_channel.send(response) {
-                      warn!("Unable to send response back to client: {:?}", e);
+                  if let Err(reply) = reply_channel.send(response) {
+                      warn!(
+                          dropped_response_count = dropped_reply_response_count(&reply),
+                          "Unable to send response back to client; receiver dropped"
+                      );
                   }
                 }
               },
@@ -226,3 +233,18 @@ pub enum UnixSocketServerError {
     #[error("Unable to write to socket")]
     WriteFail,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::dropped_reply_response_count;
+    use crate::{BusReply, BusResponse};
+
+    #[test]
+    fn dropped_reply_summary_only_counts_responses() {
+        let reply = BusReply {
+            responses: vec![BusResponse::Ack, BusResponse::Ack, BusResponse::Ack],
+        };
+
+        assert_eq!(dropped_reply_response_count(&reply), 3);
+    }
+}

src/rust/lqos_stormguard/src/lib.rs

@@ -11,6 +11,7 @@
 
 use lqos_bakery::BakeryCommands;
 use lqos_bus::StormguardDebugEntry;
+use lqos_config::NetworkJsonTransport;
 use lqos_queue_tracker::QUEUE_STRUCTURE_CHANGED_STORMGUARD;
 use parking_lot::Mutex;
 use std::time::Duration;
@@ -36,6 +37,7 @@ pub static STORMGUARD_DEBUG: Mutex<Vec<StormguardDebugEntry>> = Mutex::new(Vec::
 /// nothing to do.
 pub async fn start_stormguard(
     bakery: crossbeam_channel::Sender<BakeryCommands>,
+    network_map_provider: fn() -> Vec<(usize, NetworkJsonTransport)>,
 ) -> anyhow::Result<()> {
     let _ = tokio::time::sleep(Duration::from_secs(1)).await;
 
@@ -90,9 +92,12 @@ pub async fn start_stormguard(
         if let (Some(cfg), Some(tracker)) = (&config, &mut site_state_tracker) {
             let (active_ping_sample, active_ping_updated) = active_ping.latest();
             // Update all the ring buffers
-            tracker
-                .read_new_tick_data(cfg, active_ping_sample, active_ping_updated)
-                .await;
+            tracker.read_new_tick_data(
+                cfg,
+                active_ping_sample,
+                active_ping_updated,
+                network_map_provider(),
+            );
 
             // Check for state changes
             tracker.check_state(cfg);

src/rust/lqos_stormguard/src/site_state.rs

@@ -21,7 +21,8 @@ use crate::site_state::stormguard_state::StormguardState;
 use crate::{MOVING_AVERAGE_BUFFER_SIZE, READING_ACCUMULATOR_SIZE};
 use crossbeam_channel::Sender;
 use lqos_bakery::BakeryCommands;
-use lqos_bus::{BusRequest, BusResponse, StormguardDebugDirection, StormguardDebugEntry, TcHandle};
+use lqos_bus::{StormguardDebugDirection, StormguardDebugEntry, TcHandle};
+use lqos_config::NetworkJsonTransport;
 use lqos_queue_tracker::QUEUE_STRUCTURE;
 use std::collections::{HashMap, HashSet};
 use std::time::{Duration, Instant};
@@ -185,63 +186,53 @@ impl SiteStateTracker {
         }
     }
 
-    pub async fn read_new_tick_data(
+    pub fn read_new_tick_data(
         &mut self,
         config: &StormguardConfig,
         active_ping_sample: Option<TimedRtt>,
         active_ping_updated: bool,
+        all_nodes: Vec<(usize, NetworkJsonTransport)>,
     ) {
-        let requests = vec![BusRequest::GetFullNetworkMap];
-        let Ok(responses) = lqos_bus::bus_request(requests).await else {
-            info!("Failed to get lqosd stats");
-            return;
-        };
-
         for site in self.sites.values_mut() {
             site.current_throughput = (0.0, 0.0);
             site.clear_tick_rtt_state();
         }
 
-        for response in responses {
-            let BusResponse::NetworkMap(all_nodes) = response else {
+        for (_, node_info) in all_nodes {
+            let Some(target) = self.sites.get_mut(&node_info.name) else {
                 continue;
             };
-            for (_, node_info) in all_nodes {
-                let Some(target) = self.sites.get_mut(&node_info.name) else {
-                    continue;
-                };
 
-                // Record throughput (Mbps)
-                let down_mbps = (node_info.current_throughput.0 as f64 * 8.0) / 1_000_000.0;
-                let up_mbps = (node_info.current_throughput.1 as f64 * 8.0) / 1_000_000.0;
-                target.throughput_down.add(down_mbps);
-                target.throughput_up.add(up_mbps);
-                target.current_throughput = (down_mbps, up_mbps);
+            // Record throughput (Mbps)
+            let down_mbps = (node_info.current_throughput.0 as f64 * 8.0) / 1_000_000.0;
+            let up_mbps = (node_info.current_throughput.1 as f64 * 8.0) / 1_000_000.0;
+            target.throughput_down.add(down_mbps);
+            target.throughput_up.add(up_mbps);
+            target.current_throughput = (down_mbps, up_mbps);
 
-                // Retransmits (as a percentage of TCP packets)
-                let retransmits_down = if node_info.current_tcp_packets.0 > 0 {
-                    node_info.current_retransmits.0 as f64 / node_info.current_tcp_packets.0 as f64
-                } else {
-                    0.0
-                };
-                let retransmits_up = if node_info.current_tcp_packets.1 > 0 {
-                    node_info.current_retransmits.1 as f64 / node_info.current_tcp_packets.1 as f64
-                } else {
-                    0.0
-                };
-                target.retransmits_down.add(retransmits_down);
-                target.retransmits_up.add(retransmits_up);
+            // Retransmits (as a percentage of TCP packets)
+            let retransmits_down = if node_info.current_tcp_packets.0 > 0 {
+                node_info.current_retransmits.0 as f64 / node_info.current_tcp_packets.0 as f64
+            } else {
+                0.0
+            };
+            let retransmits_up = if node_info.current_tcp_packets.1 > 0 {
+                node_info.current_retransmits.1 as f64 / node_info.current_tcp_packets.1 as f64
+            } else {
+                0.0
+            };
+            target.retransmits_down.add(retransmits_down);
+            target.retransmits_up.add(retransmits_up);
 
-                // Round-Trip Time
-                if !node_info.rtts.is_empty() {
-                    let mut my_round_trip_times = node_info.rtts.clone();
-                    my_round_trip_times.sort_by(|a, b| a.total_cmp(b));
-                    let samples = my_round_trip_times.len();
-                    let mut idx = ((samples as f32) * 0.9).floor() as usize;
-                    idx = idx.min(samples.saturating_sub(1));
-                    let p90 = my_round_trip_times[idx] as f64;
-                    target.record_passive_rtt_sample(p90);
-                }
+            // Round-Trip Time
+            if !node_info.rtts.is_empty() {
+                let mut my_round_trip_times = node_info.rtts.clone();
+                my_round_trip_times.sort_by(|a, b| a.total_cmp(b));
+                let samples = my_round_trip_times.len();
+                let mut idx = ((samples as f32) * 0.9).floor() as usize;
+                idx = idx.min(samples.saturating_sub(1));
+                let p90 = my_round_trip_times[idx] as f64;
+                target.record_passive_rtt_sample(p90);
             }
         }
 

src/rust/lqosd/src/main.rs

@@ -336,7 +336,12 @@ fn main() -> Result<()> {
                         Err(e) => error!("Insight control channel failed to start: {:#}", e),
                     }
 
-                    match lqos_stormguard::start_stormguard(bakery_sender_for_async).await {
+                    match lqos_stormguard::start_stormguard(
+                        bakery_sender_for_async,
+                        shaped_devices_tracker::full_network_map_snapshot,
+                    )
+                    .await
+                    {
                         Ok(_) => info!("StormGuard started successfully"),
                         Err(e) => error!("StormGuard failed to start: {:#}", e),
                     }

src/rust/lqosd/src/node_manager/local_api/flow_map.rs

@@ -1,4 +1,18 @@
 use crate::throughput_tracker::flow_data;
+
 pub fn flow_map_data() -> Vec<(f64, f64, String, u64, f32)> {
     flow_data::RECENT_FLOWS.lat_lon_endpoints()
 }
+
+pub fn endpoints_by_country_data() -> Vec<(
+    String,
+    lqos_utils::units::DownUpOrder<u64>,
+    [f32; 2],
+    String,
+)> {
+    flow_data::RECENT_FLOWS.country_summary()
+}
+
+pub fn endpoint_latlon_data() -> Vec<(f64, f64, String, u64, f32)> {
+    flow_data::RECENT_FLOWS.lat_lon_endpoints()
+}

src/rust/lqosd/src/node_manager/ws/ticker.rs

@@ -143,7 +143,7 @@ async fn one_second_cadence(
             ),
             ticker_with_timeout(
                 "endpoint_latlon",
-                endpoint_latlon::endpoint_latlon(channels.clone(), bus_tx.clone())
+                endpoint_latlon::endpoint_latlon(channels.clone())
             ),
             ticker_with_timeout(
                 "tree_summary",

src/rust/lqosd/src/node_manager/ws/ticker/endpoint_latlon.rs

@@ -1,15 +1,11 @@
 use std::sync::Arc;
 
+use crate::node_manager::local_api::flow_map;
 use crate::node_manager::ws::messages::WsResponse;
 use crate::node_manager::ws::publish_subscribe::PubSub;
 use crate::node_manager::ws::published_channels::PublishedChannels;
-use lqos_bus::{BusReply, BusRequest, BusResponse};
-use tokio::sync::mpsc::Sender;
 
-pub async fn endpoint_latlon(
-    channels: Arc<PubSub>,
-    bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
-) {
+pub async fn endpoint_latlon(channels: Arc<PubSub>) {
     if !channels
         .is_channel_alive(PublishedChannels::EndpointLatLon)
         .await
@@ -17,29 +13,10 @@ pub async fn endpoint_latlon(
         return;
     }
 
-    let (tx, rx) = tokio::sync::oneshot::channel::<BusReply>();
-    let request = BusRequest::CurrentEndpointLatLon;
-    if let Err(e) = bus_tx.send((tx, request)).await {
-        tracing::warn!("EndpointLatLon: failed to send request to bus: {:?}", e);
-        return;
-    }
-    let replies = match rx.await {
-        Ok(r) => r,
-        Err(e) => {
-            tracing::warn!(
-                "EndpointLatLon: failed to receive response from bus: {:?}",
-                e
-            );
-            return;
-        }
+    let message = WsResponse::EndpointLatLon {
+        data: flow_map::endpoint_latlon_data(),
     };
-
-    for reply in replies.responses.into_iter() {
-        if let BusResponse::CurrentLatLon(points) = reply {
-            let message = WsResponse::EndpointLatLon { data: points };
-            channels
-                .send(PublishedChannels::EndpointLatLon, message)
-                .await;
-        }
-    }
+    channels
+        .send(PublishedChannels::EndpointLatLon, message)
+        .await;
 }

src/rust/lqosd/src/node_manager/ws/ticker/flow_endpoints.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use crate::node_manager::local_api::flow_map;
 use crate::node_manager::ws::messages::{EtherProtocolsData, WsResponse};
 use crate::node_manager::ws::publish_subscribe::PubSub;
 use crate::node_manager::ws::published_channels::PublishedChannels;
@@ -8,7 +9,7 @@ use tokio::sync::mpsc::Sender;
 
 pub async fn endpoints_by_country(
     channels: Arc<PubSub>,
-    bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
+    _bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
 ) {
     if !channels
         .is_channel_alive(PublishedChannels::EndpointsByCountry)
@@ -17,30 +18,12 @@ pub async fn endpoints_by_country(
         return;
     }
 
-    let (tx, rx) = tokio::sync::oneshot::channel::<BusReply>();
-    let request = BusRequest::CurrentEndpointsByCountry;
-    if let Err(e) = bus_tx.send((tx, request)).await {
-        tracing::warn!("EndpointsByCountry: failed to send request to bus: {:?}", e);
-        return;
-    }
-    let replies = match rx.await {
-        Ok(r) => r,
-        Err(e) => {
-            tracing::warn!(
-                "EndpointsByCountry: failed to receive throughput from bus: {:?}",
-                e
-            );
-            return;
-        }
+    let message = WsResponse::EndpointsByCountry {
+        data: flow_map::endpoints_by_country_data(),
     };
-    for reply in replies.responses.into_iter() {
-        if let BusResponse::CurrentEndpointsByCountry(countries) = reply {
-            let message = WsResponse::EndpointsByCountry { data: countries };
-            channels
-                .send(PublishedChannels::EndpointsByCountry, message)
-                .await;
-        }
-    }
+    channels
+        .send(PublishedChannels::EndpointsByCountry, message)
+        .await;
 }
 
 pub async fn ether_protocols(

src/rust/lqosd/src/shaped_devices_tracker/mod.rs

@@ -150,17 +150,17 @@ pub fn get_one_network_map_layer(parent_idx: usize) -> BusResponse {
     }
 }
 
-pub fn get_full_network_map() -> BusResponse {
+pub fn full_network_map_snapshot() -> Vec<(usize, NetworkJsonTransport)> {
     let nj = NETWORK_JSON.read();
-    let data = {
-        nj.get_nodes_when_ready()
-            .iter()
-            .enumerate()
-            .map(|(i, n)| (i, n.clone_to_transit()))
-            .collect::<Vec<(usize, NetworkJsonTransport)>>()
-    };
+    nj.get_nodes_when_ready()
+        .iter()
+        .enumerate()
+        .map(|(i, n)| (i, n.clone_to_transit()))
+        .collect()
+}
 
-    BusResponse::NetworkMap(data)
+pub fn get_full_network_map() -> BusResponse {
+    BusResponse::NetworkMap(full_network_map_snapshot())
 }
 
 pub fn get_top_n_root_queues(n_queues: usize) -> BusResponse {

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

@@ -1212,7 +1212,6 @@ pub fn current_endpoints_by_country() -> BusResponse {
 /// Current endpoint lat/lon
 pub fn current_lat_lon() -> BusResponse {
     let summary = flow_data::RECENT_FLOWS.lat_lon_endpoints();
-    debug!("CurrentLatLon payload: {:?}", summary);
     BusResponse::CurrentLatLon(summary)
 }
 

src/rust/lqosd/src/treeguard/actor.rs

@@ -23,12 +23,15 @@ use lqos_overrides::{NetworkAdjustment, OverrideFile, OverrideLayer, OverrideSto
 use lqos_utils::hash_to_i64;
 use lqos_utils::units::DownUpOrder;
 use lqos_utils::unix_time::{time_since_boot, unix_now};
+use parking_lot::RwLock;
 use std::collections::VecDeque;
 use std::sync::OnceLock;
 use std::time::{Duration, Instant};
 use tracing::{debug, warn};
 
 static TREEGUARD_SENDER: OnceLock<Sender<TreeguardCommand>> = OnceLock::new();
+static TREEGUARD_STATUS_CACHE: OnceLock<RwLock<TreeguardStatusData>> = OnceLock::new();
+static TREEGUARD_ACTIVITY_CACHE: OnceLock<RwLock<Vec<TreeguardActivityEntry>>> = OnceLock::new();
 
 const ACTIVITY_RING_CAPACITY: usize = 200;
 const UTIL_EWMA_ALPHA: f64 = 0.1;
@@ -60,6 +63,9 @@ pub(crate) fn start_treeguard_actor(
         return Ok(());
     }
 
+    let _ = TREEGUARD_STATUS_CACHE.set(RwLock::new(empty_status_snapshot()));
+    let _ = TREEGUARD_ACTIVITY_CACHE.set(RwLock::new(Vec::new()));
+
     let (tx, rx) = crossbeam_channel::bounded::<TreeguardCommand>(64);
     let _ = TREEGUARD_SENDER.set(tx);
 
@@ -70,6 +76,14 @@ pub(crate) fn start_treeguard_actor(
     Ok(())
 }
 
+pub(crate) fn cached_status_snapshot() -> Option<TreeguardStatusData> {
+    Some(TREEGUARD_STATUS_CACHE.get()?.read().clone())
+}
+
+pub(crate) fn cached_activity_snapshot() -> Option<Vec<TreeguardActivityEntry>> {
+    Some(TREEGUARD_ACTIVITY_CACHE.get()?.read().clone())
+}
+
 /// Requests a status snapshot from the TreeGuard actor.
 ///
 /// This function is not pure: it sends a message to the TreeGuard actor thread.
@@ -110,18 +124,9 @@ fn treeguard_actor_loop(
 ) {
     debug!("TreeGuard actor started");
 
-    let mut status = TreeguardStatusData {
-        enabled: false,
-        dry_run: true,
-        cpu_max_pct: None,
-        managed_nodes: 0,
-        managed_circuits: 0,
-        virtualized_nodes: 0,
-        fq_codel_circuits: 0,
-        last_action_summary: None,
-        warnings: Vec::new(),
-    };
+    let mut status = empty_status_snapshot();
     let mut activity: VecDeque<TreeguardActivityEntry> = VecDeque::new();
+    update_cached_snapshots(&status, &activity);
 
     let mut runtime_state = TreeguardRuntimeState::default();
 
@@ -143,6 +148,7 @@ fn treeguard_actor_loop(
                     &mut tick_seconds,
                     &mut runtime_state,
                 );
+                update_cached_snapshots(&status, &activity);
             }
             Err(crossbeam_channel::RecvTimeoutError::Disconnected) => {
                 warn!("TreeGuard actor command channel disconnected; exiting actor");
@@ -171,6 +177,32 @@ fn handle_command(
     }
 }
 
+fn empty_status_snapshot() -> TreeguardStatusData {
+    TreeguardStatusData {
+        enabled: false,
+        dry_run: true,
+        cpu_max_pct: None,
+        managed_nodes: 0,
+        managed_circuits: 0,
+        virtualized_nodes: 0,
+        fq_codel_circuits: 0,
+        last_action_summary: None,
+        warnings: Vec::new(),
+    }
+}
+
+fn update_cached_snapshots(
+    status: &TreeguardStatusData,
+    activity: &VecDeque<TreeguardActivityEntry>,
+) {
+    if let Some(cache) = TREEGUARD_STATUS_CACHE.get() {
+        *cache.write() = status.clone();
+    }
+    if let Some(cache) = TREEGUARD_ACTIVITY_CACHE.get() {
+        *cache.write() = activity.iter().cloned().rev().collect();
+    }
+}
+
 #[derive(Default)]
 struct TreeguardRuntimeState {
     link_states: FxHashMap<String, LinkState>,

src/rust/lqosd/src/treeguard/status.rs

@@ -9,6 +9,10 @@ use lqos_config::load_config;
 /// This function is not pure: it sends a request to the TreeGuard actor, and may fall back to
 /// reading the current configuration from disk (via the config cache).
 pub async fn treeguard_status_snapshot() -> TreeguardStatusData {
+    if let Some(snapshot) = actor::cached_status_snapshot() {
+        return snapshot;
+    }
+
     if let Some(snapshot) = actor::request_status_snapshot().await {
         return snapshot;
     }
@@ -71,5 +75,9 @@ pub async fn treeguard_status_snapshot() -> TreeguardStatusData {
 /// This function is not pure: it sends a request to the TreeGuard actor.
 /// If the actor isn't available, it returns an empty list.
 pub async fn treeguard_activity_snapshot() -> Vec<TreeguardActivityEntry> {
+    if let Some(snapshot) = actor::cached_activity_snapshot() {
+        return snapshot;
+    }
+
     actor::request_activity_snapshot().await.unwrap_or_default()
 }