Lts cake import2

src/rust/lts_client/src/collector/collation/mod.rs

@@ -6,9 +6,11 @@ use self::min_max::{MinMaxAvgPair, MinMaxAvg};
 pub(crate) use session_buffer::{StatsSession, SESSION_BUFFER};
 use lqos_utils::unix_time::unix_now;
 use tokio::sync::mpsc::Sender;
-use std::{collections::HashMap, net::IpAddr};
+use std::{collections::HashMap, net::IpAddr, sync::atomic::AtomicU64};
 use super::{HostSummary, NetworkTreeEntry};
 
+static QUEUE_STATS_COUNTER: AtomicU64 = AtomicU64::new(0);
+
 pub(crate) async fn collate_stats(comm_tx: Sender<SenderChannelMessage>) {
  let timestamp = unix_now().unwrap_or(0);
  if timestamp == 0 {
@@ -116,7 +118,12 @@ pub(crate) async fn collate_stats(comm_tx: Sender<SenderChannelMessage>) {
  let (cpu, ram) = system_stats::get_cpu_ram().await;
 
  // Obtain queue stats
- let cake_stats = super::update_cake_stats().await;
+ let last_cake = QUEUE_STATS_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+ let cake_stats = if last_cake % 10 == 0 {
+ super::update_cake_stats().await
+ } else {
+ None
+ };
 
 
  // Add to the submissions queue

src/rust/lts_client/src/collector/quick_drops/retriever.rs

@@ -93,8 +93,15 @@ impl AsyncQueueReader {
  if let Some(serde_json::Value::Number(drops)) = map.get("drops") {
  stats.drops = drops.as_u64().unwrap_or(0);
  }
- if let Some(serde_json::Value::Number(marks)) = map.get("ecn_mark") {
- stats.marks = marks.as_u64().unwrap_or(0);
+ if let Some(serde_json::Value::Array(tins)) = map.get("tins") {
+ stats.marks = 0;
+ for tin in tins.iter() {
+ if let Some(tin) = tin.as_object() {
+ if let Some(serde_json::Value::Number(marks)) = tin.get("ecn_mark") {
+ stats.marks += marks.as_u64().unwrap_or(0);
+ }
+ }
+ }
  }
  result.push(stats);
  }

src/rust/lts_client/src/collector/quick_drops/stats_diff.rs

@@ -1,7 +1,7 @@
+use super::CakeStats;
 use lqos_config::load_config;
-use tokio::sync::Mutex;
 use once_cell::sync::Lazy;
-use super::CakeStats;
+use tokio::sync::Mutex;
 
 static CAKE_TRACKER: Lazy<Mutex<CakeTracker>> = Lazy::new(|| Mutex::new(CakeTracker::new()));
 
@@ -35,10 +35,7 @@ impl CakeTracker {
  } else {
  let out_reader = super::AsyncQueueReader::new(outbound);
  let in_reader = super::AsyncQueueReader::new(inbound);
- let (up, down) = tokio::join!(
- out_reader.run(),
- in_reader.run(),
- );
+ let (up, down) = tokio::join!(out_reader.run(), in_reader.run(),);
  if let (Ok(Some(up)), Ok(Some(down))) = (up, down) {
  return self.read_up_down(up, down);
  }
@@ -47,21 +44,39 @@ impl CakeTracker {
  None
  }
 
- fn read_up_down(&mut self, up: Vec<CakeStats>, down: Vec<CakeStats>) -> Option<(Vec<CakeStats>, Vec<CakeStats>)> {
+ fn read_up_down(
+ &mut self,
+ mut up: Vec<CakeStats>,
+ mut down: Vec<CakeStats>,
+ ) -> Option<(Vec<CakeStats>, Vec<CakeStats>)> {
  if self.prev.is_none() {
  self.prev = Some((up, down));
  None
  } else {
  // Delta time
  if let Some((down, up)) = &mut self.current {
  down.iter_mut().for_each(|d| {
- if let Some(prev) = self.prev.as_ref().unwrap().0.iter().find(|p| p.circuit_id == d.circuit_id) {
+ if let Some(prev) = self
+ .prev
+ .as_ref()
+ .unwrap()
+ .0
+ .iter()
+ .find(|p| p.circuit_id == d.circuit_id)
+ {
  d.drops = d.drops.saturating_sub(prev.drops);
  d.marks = d.marks.saturating_sub(prev.marks);
  }
  });
  up.iter_mut().for_each(|d| {
- if let Some(prev) = self.prev.as_ref().unwrap().1.iter().find(|p| p.circuit_id == d.circuit_id) {
+ if let Some(prev) = self
+ .prev
+ .as_ref()
+ .unwrap()
+ .1
+ .iter()
+ .find(|p| p.circuit_id == d.circuit_id)
+ {
  d.drops = d.drops.saturating_sub(prev.drops);
  d.marks = d.marks.saturating_sub(prev.marks);
  }
@@ -71,7 +86,15 @@ impl CakeTracker {
  // Advance the previous
  self.prev = self.current.take();
 
- Some((up, down))
+ // Remove all zeroes
+ up.retain(|u| u.drops > 0 || u.marks > 0);
+ down.retain(|d| d.drops > 0 || d.marks > 0);
+
+ if up.is_empty() && down.is_empty() {
+ None
+ } else {
+ Some((up, down))
+ }
  }
  }
-}
+}