feat(stats): add the counters shared by the capture and network threads

src/lib.rs

@@ -5,3 +5,4 @@ pub mod config;
 pub mod hyperhdr;
 pub mod pacing;
 pub mod slot;
+pub mod stats;

src/stats.rs

@@ -0,0 +1,119 @@
+//! Counters shared by the capture and network threads, logged periodically.
+
+use std::fmt;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+
+#[derive(Debug, Default)]
+pub struct Stats {
+    pub frames_received: AtomicU64,
+    pub frames_sent: AtomicU64,
+    pub keepalives: AtomicU64,
+    /// Frames replaced by a newer one before they were sent.
+    pub superseded: AtomicU64,
+    pub corrupted: AtomicU64,
+    pub bounds_errors: AtomicU64,
+    pub reconnects: AtomicU64,
+    pub bytes_sent: AtomicU64,
+    pub error_replies: AtomicU64,
+    convert_count: AtomicU64,
+    convert_ns_total: AtomicU64,
+    convert_ns_max: AtomicU64,
+}
+
+pub fn inc(counter: &AtomicU64) {
+    counter.fetch_add(1, Ordering::Relaxed);
+}
+
+pub fn add(counter: &AtomicU64, value: u64) {
+    counter.fetch_add(value, Ordering::Relaxed);
+}
+
+impl Stats {
+    pub fn record_convert(&self, took: Duration) {
+        let ns = u64::try_from(took.as_nanos()).unwrap_or(u64::MAX);
+        inc(&self.convert_count);
+        add(&self.convert_ns_total, ns);
+        self.convert_ns_max.fetch_max(ns, Ordering::Relaxed);
+    }
+
+    /// Returns the counters accumulated since the previous call and resets them.
+    pub fn take(&self) -> Snapshot {
+        let take = |c: &AtomicU64| c.swap(0, Ordering::Relaxed);
+        let convert_count = take(&self.convert_count);
+        let convert_ns_total = take(&self.convert_ns_total);
+        Snapshot {
+            frames_received: take(&self.frames_received),
+            frames_sent: take(&self.frames_sent),
+            keepalives: take(&self.keepalives),
+            superseded: take(&self.superseded),
+            corrupted: take(&self.corrupted),
+            bounds_errors: take(&self.bounds_errors),
+            reconnects: take(&self.reconnects),
+            bytes_sent: take(&self.bytes_sent),
+            error_replies: take(&self.error_replies),
+            convert_avg: Duration::from_nanos(
+                convert_ns_total.checked_div(convert_count).unwrap_or(0),
+            ),
+            convert_max: Duration::from_nanos(take(&self.convert_ns_max)),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Snapshot {
+    pub frames_received: u64,
+    pub frames_sent: u64,
+    pub keepalives: u64,
+    pub superseded: u64,
+    pub corrupted: u64,
+    pub bounds_errors: u64,
+    pub reconnects: u64,
+    pub bytes_sent: u64,
+    pub error_replies: u64,
+    pub convert_avg: Duration,
+    pub convert_max: Duration,
+}
+
+impl fmt::Display for Snapshot {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "received={} sent={} keepalive={} superseded={} corrupted={} bounds_errors={} \
+             reconnects={} bytes_sent={} error_replies={} convert_avg={:?} convert_max={:?}",
+            self.frames_received,
+            self.frames_sent,
+            self.keepalives,
+            self.superseded,
+            self.corrupted,
+            self.bounds_errors,
+            self.reconnects,
+            self.bytes_sent,
+            self.error_replies,
+            self.convert_avg,
+            self.convert_max,
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn take_resets() {
+        let s = Stats::default();
+        inc(&s.frames_sent);
+        add(&s.bytes_sent, 100);
+        s.record_convert(Duration::from_micros(100));
+        s.record_convert(Duration::from_micros(300));
+        let snap = s.take();
+        assert_eq!(snap.frames_sent, 1);
+        assert_eq!(snap.bytes_sent, 100);
+        assert_eq!(snap.convert_avg, Duration::from_micros(200));
+        assert_eq!(snap.convert_max, Duration::from_micros(300));
+        let snap = s.take();
+        assert_eq!(snap.frames_sent, 0);
+        assert_eq!(snap.convert_avg, Duration::ZERO);
+    }
+}