Merge pull request #2 from hansieodendaal/ho_add_buffer_stats_and_logs

Buffer dropped messages stats

Author: CjS77

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tari_broadcast_channel"
-version = "0.1.0"
+version = "0.2.0"
 authors = ["Filip Dulic <[email protected]>", "Vladan Popovic <[email protected]>", "Bojan Petrovic <[email protected]>", "The Tari Development Community"]
 description = "Bounded non-blocking single-producer-multi-consumer broadcast channel"
 license = "Apache-2.0/MIT"
@@ -14,6 +14,7 @@ edition = "2018"
 arc-swap = "0.4.2"
 futures = { version = "^0.3.1" }
 crossbeam-channel = "0.3.9"
+log = "0.4.6"
 
 [dev-dependencies]
 criterion = "0.3.0"

examples/async-simple.rs

@@ -2,8 +2,13 @@ use futures::{executor::block_on, stream, StreamExt};
 use tari_broadcast_channel::bounded;
 
 fn main() {
-    let (publisher, subscriber1) = bounded(10);
+    let (publisher, subscriber1) = bounded(10, 1);
     let subscriber2 = subscriber1.clone();
+    let subscriber3 = subscriber1.clone();
+
+    assert_eq!(subscriber1.get_receiver_id(), 10000);
+    assert_eq!(subscriber2.get_receiver_id(), 10001);
+    assert_eq!(subscriber3.get_receiver_id(), 10002);
 
     block_on(async move {
         stream::iter(1..15).map(Ok).forward(publisher).await.unwrap();

examples/raw-simple.rs

@@ -1,12 +1,23 @@
 use tari_broadcast_channel::raw_bounded;
 
 fn main() {
-    let (tx, rx) = raw_bounded(10);
+    let (tx, rx) = raw_bounded(10, 1);
     (1..15).for_each(|x| tx.broadcast(x).unwrap());
 
+    let rx2 = rx.clone();
+
     let received: Vec<i32> = rx.map(|x| *x).collect();
     // Test that only the last 10 elements are in the received list.
     let expected: Vec<i32> = (5..15).collect();
 
     assert_eq!(expected, received);
+
+    // Test messages discarded
+    assert_eq!(rx2.get_dropped_messages_state(), false);
+    rx2.try_recv().unwrap();
+    assert_eq!(rx2.get_dropped_messages_count(), 4);
+    assert_eq!(rx2.get_dropped_messages_state(), true);
+    rx2.try_recv().unwrap();
+    assert_eq!(rx2.get_dropped_messages_count(), 4);
+    assert_eq!(rx2.get_dropped_messages_state(), false);
 }

src/async_channel.rs

@@ -5,11 +5,22 @@ use futures::{
     Sink,
     Stream,
 };
+use log::*;
 use std::{pin::Pin, sync::Arc, task::Poll};
 
-pub fn bounded<T>(size: usize) -> (Publisher<T>, Subscriber<T>) {
-    let (sender, receiver) = raw_bounded(size);
+const LOG_TARGET: &str = "tari_broadcast_channel::async_channel";
+const ID_MULTIPLIER: usize = 10_000;
+
+pub fn bounded<T>(size: usize, receiver_id: usize) -> (Publisher<T>, Subscriber<T>) {
+    let (sender, receiver) = raw_bounded(size, receiver_id);
     let (waker, sleeper) = alarm();
+    debug!(
+        target: LOG_TARGET,
+        "Buffer created with ID '{}' ({}) and size {}, consecutive subscriptions add 1 to the ID.",
+        receiver_id * ID_MULTIPLIER,
+        receiver_id,
+        size
+    );
     (Publisher { sender, waker }, Subscriber { receiver, sleeper })
 }
 
@@ -62,17 +73,40 @@ impl<T> Stream for Subscriber<T> {
     fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
         self.sleeper.register(cx.waker());
         match self.receiver.try_recv() {
-            Ok(item) => Poll::Ready(Some(item)),
-            Err(error) => match error {
-                TryRecvError::Empty => Poll::Pending,
-                TryRecvError::Disconnected => Poll::Ready(None),
+            Ok(item) => {
+                if self.receiver.get_dropped_messages_state() {
+                    debug!(
+                        target: LOG_TARGET,
+                        "Subscriber with ID '{}' has {} buffer messages dropped.",
+                        self.receiver.get_id(),
+                        self.receiver.get_dropped_messages_count()
+                    );
+                }
+                Poll::Ready(Some(item))
+            },
+            Err(error) => {
+                trace!(
+                    target: LOG_TARGET,
+                    "Subscriber with ID '{}', receive error: '{}'",
+                    self.receiver.get_id(),
+                    error
+                );
+                match error {
+                    TryRecvError::Empty => Poll::Pending,
+                    TryRecvError::Disconnected => Poll::Ready(None),
+                }
             },
         }
     }
 }
 
 impl<T> Clone for Subscriber<T> {
     fn clone(&self) -> Self {
+        trace!(
+            target: LOG_TARGET,
+            "Subscriber with ID '{}' was cloned, new subscription.",
+            self.receiver.get_id()
+        );
         Self {
             receiver: self.receiver.clone(),
             sleeper: self.sleeper.clone(),
@@ -88,6 +122,22 @@ impl<T: Send> PartialEq for Subscriber<T> {
 
 impl<T: Send> Eq for Subscriber<T> {}
 
+impl<T> Subscriber<T> {
+    /// Returns the receiver id
+    pub fn get_receiver_id(&self) -> usize {
+        self.receiver.get_id()
+    }
+
+    /// Returns the amount of dropped messages for the receiver
+    pub fn get_dropped_messages_count(&self) -> usize {
+        self.receiver.get_dropped_messages_count()
+    }
+
+    /// Returns the receiver's dropped messgages state for the last message read
+    pub fn get_dropped_messages_state(&self) -> bool {
+        self.receiver.get_dropped_messages_state()
+    }
+}
 // Helper struct used by sync and async implementations to wake Tasks / Threads
 #[derive(Debug)]
 pub struct Waker {
@@ -154,15 +204,30 @@ pub fn alarm() -> (Waker, Sleeper) {
 
 #[cfg(test)]
 mod test {
+    use crate::async_channel;
     use futures::{executor::block_on, stream, StreamExt};
 
     #[test]
     fn channel() {
-        let (publisher, subscriber1) = super::bounded(10);
+        let (publisher1, subscriber1) = super::bounded(10, 1);
         let subscriber2 = subscriber1.clone();
+        let subscriber3 = subscriber1.clone();
+
+        assert_eq!(subscriber1.get_receiver_id(), 10000);
+        assert_eq!(subscriber2.get_receiver_id(), 10001);
+        assert_eq!(subscriber3.get_receiver_id(), 10002);
+
+        let (publisher2, subscriber4): (async_channel::Publisher<usize>, async_channel::Subscriber<usize>) =
+            super::bounded(10, 2);
+        let subscriber5 = subscriber4.clone();
+        let subscriber6 = subscriber4.clone();
+
+        assert_eq!(subscriber4.get_receiver_id(), 20000);
+        assert_eq!(subscriber5.get_receiver_id(), 20001);
+        assert_eq!(subscriber6.get_receiver_id(), 20002);
 
         block_on(async move {
-            stream::iter(1..15).map(|i| Ok(i)).forward(publisher).await.unwrap();
+            stream::iter(1..15).map(|i| Ok(i)).forward(publisher1).await.unwrap();
         });
 
         let received1: Vec<u32> = block_on(async { subscriber1.map(|x| *x).collect().await });
@@ -171,5 +236,12 @@ mod test {
         let expected = (5..15).collect::<Vec<u32>>();
         assert_eq!(received1, expected);
         assert_eq!(received2, expected);
+        // Test messages discarded
+        subscriber3.receiver.try_recv().unwrap();
+        assert_eq!(subscriber3.receiver.get_dropped_messages_state(), true);
+        assert_eq!(subscriber3.receiver.get_dropped_messages_count(), 4);
+        subscriber3.receiver.try_recv().unwrap();
+        assert_eq!(subscriber3.receiver.get_dropped_messages_state(), false);
+        assert_eq!(subscriber3.receiver.get_dropped_messages_count(), 4);
     }
 }

src/atomic_counter.rs

@@ -48,3 +48,9 @@ impl PartialEq for AtomicCounter {
 }
 
 impl Eq for AtomicCounter {}
+
+impl fmt::Display for AtomicCounter {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
+        write!(f, "{}", self.get())
+    }
+}