Add average rate and intermediate rate statistics

Author: rmqtt

src/stats.rs

@@ -1,4 +1,5 @@
-use std::sync::atomic::{AtomicIsize, Ordering};
+use std::fmt;
+use std::fmt::Debug;
 
 use dashmap::DashMap;
 use once_cell::sync::OnceCell;
@@ -55,12 +56,23 @@ impl Stats {
 
     pub fn to_string(&self) -> String {
         let stats = Stats::instance();
+
         let conns = stats.conns.value();
         let conns_rate = stats.conns.rate();
+        let conns_avg_rate  = stats.conns.avg_rate();
+        let conns_medi_rate = stats.conns.medi_rate();
+
         let subs = stats.subs.value();
+        let subs_rate = stats.subs.rate();
+        let subs_avg_rate  = stats.subs.avg_rate();
+        let subs_medi_rate = stats.subs.medi_rate();
+
         let conn_fails = stats.conn_fails.value();
         let recvs = stats.recvs.value();
         let recvs_rate = stats.recvs.rate();
+        let recvs_avg_rate  = stats.recvs.avg_rate();
+        let recvs_medi_rate = stats.recvs.medi_rate();
+
         let sends = stats.sends.value();
         let sends_rate = stats.sends.rate();
         let closeds = stats.closeds.value();
@@ -69,65 +81,132 @@ impl Stats {
             .iter()
             .map(|entry| (entry.key().clone(), entry.value().value()))
             .collect::<Vec<(String, isize)>>();
-        format!("* Connecteds:{} {:0.2?}/s, conn_fails:{}, subs:{}, sends:{} {:0.2?}/s, recvs:{} {:0.2?}/s, closeds:{}, ifaddrs: {:?}, last err: {:?}",
-                conns, conns_rate, conn_fails, subs, sends, sends_rate, recvs, recvs_rate, closeds, ifaddrs, stats.last_err.write().take())
+        format!("* Connecteds:{} ({:0.2?}/s, {:0.2?}/s, {:0.2?}/s), conn_fails:{}, subs:{} ({:0.2?}/s, {:0.2?}/s, {:0.2?}/s), sends:{} {:0.2?}/s, recvs:{} ({:0.2?}/s, {:0.2?}/s, {:0.2?}/s), closeds:{}, ifaddrs: {:?}, last err: {:?}",
+                conns, conns_rate, conns_avg_rate, conns_medi_rate, conn_fails, subs, subs_rate, subs_avg_rate,subs_medi_rate, sends, sends_rate, recvs, recvs_rate, recvs_avg_rate, recvs_medi_rate, closeds, ifaddrs, stats.last_err.write().take())
     }
 }
 
-#[derive(Debug)]
-pub struct Counter(AtomicIsize, RwLock<DiscreteRateCounter>);
+pub type StartMillis = i64;
+pub type EndMillis = i64;
+
+pub struct Counter{
+    inner: RwLock<CounterInner>
+}
+
+pub struct CounterInner{
+    nums: isize,
+    rate: DiscreteRateCounter,
+    cost_times: Vec<(StartMillis, EndMillis)>,
+}
+
+impl Debug for Counter {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let inner = self.inner.read();
+        write!(f, "(nums: {}, rate: {:?}, cost_times_len: {})", inner.nums, inner.rate.rate(), inner.cost_times.len())
+    }
+}
 
 impl Counter {
     #[inline]
     fn new() -> Self {
-        Counter(
-            AtomicIsize::new(0),
-            RwLock::new(DiscreteRateCounter::new(10)),
-        )
+        Counter {
+            inner: RwLock::new(
+                CounterInner {
+                    nums: 0,
+                    rate: DiscreteRateCounter::new(100),
+                    cost_times: Vec::new()
+                })
+        }
     }
 
     #[inline]
     pub fn inc(&self) {
-        self.0.fetch_add(1, Ordering::SeqCst);
-        self.1.write().update();
+        let mut inner = self.inner.write();
+        inner.nums += 1;
+        inner.rate.update();
     }
 
     #[inline]
+    pub fn inc2(&self, start: StartMillis, end: EndMillis) {
+        let mut inner = self.inner.write();
+        inner.nums += 1;
+        inner.rate.update();
+        inner.cost_times.push((start, end));
+    }
+
     pub fn inc_limit(&self, limit: isize) -> bool {
-        let res = self
-            .0
-            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |x| {
-                if x < limit {
-                    Some(x + 1)
-                } else {
-                    None
-                }
-            });
-        if res.is_ok() {
-            self.1.write().update();
+        let mut inner = self.inner.write();
+        if inner.nums < limit {
+            inner.nums += 1;
+            inner.rate.update();
             true
-        } else {
+        }else{
             false
         }
     }
 
+    #[inline]
+    pub fn inc_limit_cost_times(&self, limit_cost_times: usize, start: StartMillis, end: EndMillis) {
+        let mut inner = self.inner.write();
+        inner.nums += 1;
+        inner.rate.update();
+        if inner.cost_times.len() > limit_cost_times {
+            inner.cost_times.remove(0);
+
+        }
+        inner.cost_times.push((start, end));
+    }
+
     #[inline]
     pub fn dec(&self) {
-        self.0.fetch_sub(1, Ordering::SeqCst);
+        self.inner.write().nums += 1;
     }
 
     #[inline]
     pub fn decs(&self, v: isize) {
-        self.0.fetch_sub(v, Ordering::SeqCst);
+        self.inner.write().nums += v;
     }
 
     #[inline]
     pub fn value(&self) -> isize {
-        self.0.load(Ordering::SeqCst)
+        self.inner.read().nums
     }
 
     #[inline]
     pub fn rate(&self) -> f64 {
-        self.1.read().rate()
+        self.inner.read().rate.rate()
+    }
+
+    #[inline]
+    pub fn avg_rate(&self) -> f64 {
+        let inner = self.inner.read();
+        if inner.cost_times.len() > 1 {
+            //average rate
+            let (first, _) = inner.cost_times.first().unwrap();
+            let (_, last) = inner.cost_times.last().unwrap();
+            //println!("avg_rate inner.cost_times.len(): {}, {:?}, first: {}, last: {}", inner.cost_times.len(), ((last - first) as f64 / 1000.0), first, last);
+            inner.cost_times.len() as f64 / ((last - first) as f64 / 1000.0)
+        }else{
+            1.0
+        }
+
+    }
+
+    #[inline]
+    pub fn medi_rate(&self) -> f64 {
+        let inner = self.inner.read();
+        if inner.cost_times.len() > 1 {
+            //intermediate rate
+            let len = inner.cost_times.len();
+            let start_idx = (len as f64 * 0.15) as usize;
+            let end_idx = len - start_idx - 1;
+            let (first, _) = inner.cost_times[start_idx];
+            let (_, last) = inner.cost_times[end_idx];
+            //println!("medi_rate inner.cost_times.len(): {}, {:?}", end_idx - start_idx, ((last - first) as f64 / 1000.0));
+            (end_idx - start_idx) as f64 / ((last - first) as f64 / 1000.0)
+        }else{
+            1.0
+        }
+
     }
 }

src/v3.rs

@@ -1,3 +1,4 @@
+use std::borrow::Cow;
 use futures::channel::mpsc;
 use futures::SinkExt;
 use futures::StreamExt;
@@ -17,6 +18,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::task::spawn_local;
 use uuid::Uuid;
+use crate::stats::{EndMillis, StartMillis};
 
 use super::connector::ConnectorFactory;
 use super::Stats;
@@ -139,8 +141,8 @@ impl Client {
         self.msg_tx.read().as_ref().cloned()
     }
 
-    fn subs_inc(&self) {
-        Stats::instance().subs.inc();
+    fn subs_inc(&self, start: StartMillis, end: EndMillis) {
+        Stats::instance().subs.inc2(start, end);
         self.subs.fetch_add(1, Ordering::SeqCst);
     }
 
@@ -233,15 +235,16 @@ impl Client {
                 let connect_enable = mgr.on_connect.fire(client.clone());
 
                 if connect_enable {
+                    let start = chrono::Local::now().timestamp_millis();
                     match builder.connect().await {
                         Ok(c) => {
                             mgr.on_connected.fire(client.clone());
 
                             let sink = c.sink();
 
                             client.set_sink(sink.clone());
-
-                            Stats::instance().conns.inc();
+                            let end = chrono::Local::now().timestamp_millis();
+                            Stats::instance().conns.inc2(start, end);
 
                             //subscribe
                             if opts.sub_switch {
@@ -306,21 +309,23 @@ impl Client {
         client_id: String,
     ) {
         'subscribe: loop {
+            let start = chrono::Local::now().timestamp_millis();
             match sink
                 .subscribe()
                 .topic_filter(sub_topic.clone(), qos)
                 .send()
                 .await
             {
                 Ok(rets) => {
+                    let end = chrono::Local::now().timestamp_millis();
                     for ret in rets {
                         if let SubscribeReturnCode::Failure = ret {
                             log::debug!("{:?} subscribe failure", client_id);
                             Stats::instance().set_last_err("subscribe failure".into());
                             time::sleep(Duration::from_secs(5)).await;
                             continue 'subscribe;
                         } else {
-                            c.subs_inc();
+                            c.subs_inc(start, end);
                         }
                     }
                     c.mgr
@@ -370,11 +375,13 @@ impl Client {
             );
 
             let payload = if let Some(payload) = &opts.pub_payload {
-                ntex::util::Bytes::from(payload.clone())
+                Cow::Borrowed(payload)
             } else {
-                ntex::util::Bytes::from("0".repeat(opts.pub_payload_len))
+                Cow::Owned("0".repeat(opts.pub_payload_len))
             };
 
+            let payload = format!("{{ \"data\": \"{}\", \"create_time\": {} }}", payload.as_str(), chrono::Local::now().timestamp_millis());
+            let payload = ntex::util::Bytes::from(payload);
             if sink.is_open() {
                 let packet_id = c.gen_packet_id();
 
@@ -428,7 +435,12 @@ impl Client {
             .start(
                 move |control: v3::client::ControlMessage<()>| match control {
                     v3::client::ControlMessage::Publish(publish) => {
-                        Stats::instance().recvs.inc();
+                        let now = chrono::Local::now().timestamp_millis();
+                        let payload = serde_json::from_slice::<serde_json::Value>(publish.packet().payload.as_ref()).unwrap();
+                        let create_time = payload.as_object().and_then(|obj|{
+                            obj.get("create_time").map(|v|v.as_i64())
+                        }).unwrap().unwrap();
+                        Stats::instance().recvs.inc_limit_cost_times((Stats::instance().conns.value() * 2) as usize, create_time, now);
                         client
                             .mgr
                             .on_message