2 files changed, 145 insertions, 0 deletions
diff --git a/tokio/src/sync/tests/loom_broadcast.rs b/tokio/src/sync/tests/loom_broadcast.rs
new file mode 100644
index 00000000..53f76a25
--- /dev/null
+++ b/tokio/src/sync/tests/loom_broadcast.rs
@@ -0,0 +1,144 @@
+use crate::sync::broadcast;
+use crate::sync::broadcast::RecvError::{Closed, Lagged};
+
+use loom::future::block_on;
+use loom::sync::Arc;
+use loom::thread;
+use tokio_test::{assert_err, assert_ok};
+
+// An `Arc` is used as the value in order to detect memory leaks.
+#[test]
+fn broadcast_two() {
+    loom::model(|| {
+        let (tx, mut rx1) = broadcast::channel::<Arc<&'static str>>(16);
+        let mut rx2 = tx.subscribe();
+
+        let th1 = thread::spawn(move || {
+            block_on(async {
+                let v = assert_ok!(rx1.recv().await);
+                assert_eq!(*v, "hello");
+
+                let v = assert_ok!(rx1.recv().await);
+                assert_eq!(*v, "world");
+
+                match assert_err!(rx1.recv().await) {
+                    Closed => {}
+                    _ => panic!(),
+                }
+            });
+        });
+
+        let th2 = thread::spawn(move || {
+            block_on(async {
+                let v = assert_ok!(rx2.recv().await);
+                assert_eq!(*v, "hello");
+
+                let v = assert_ok!(rx2.recv().await);
+                assert_eq!(*v, "world");
+
+                match assert_err!(rx2.recv().await) {
+                    Closed => {}
+                    _ => panic!(),
+                }
+            });
+        });
+
+        assert_ok!(tx.send(Arc::new("hello")));
+        assert_ok!(tx.send(Arc::new("world")));
+        drop(tx);
+
+        assert_ok!(th1.join());
+        assert_ok!(th2.join());
+    });
+}
+
+#[test]
+fn broadcast_wrap() {
+    loom::model(|| {
+        let (tx, mut rx1) = broadcast::channel(2);
+        let mut rx2 = tx.subscribe();
+
+        let th1 = thread::spawn(move || {
+            block_on(async {
+                let mut num = 0;
+
+                loop {
+                    match rx1.recv().await {
+                        Ok(_) => num += 1,
+                        Err(Closed) => break,
+                        Err(Lagged(n)) => {
+                            num += n as usize
+                        },
+                    }
+                }
+
+                assert_eq!(num, 3);
+            });
+        });
+
+        let th2 = thread::spawn(move || {
+            block_on(async {
+                let mut num = 0;
+
+                loop {
+                    match rx2.recv().await {
+                        Ok(_) => num += 1,
+                        Err(Closed) => break,
+                        Err(Lagged(n)) => {
+                            num += n as usize
+                        }
+                    }
+                }
+
+                assert_eq!(num, 3);
+            });
+        });
+
+        assert_ok!(tx.send("one"));
+        assert_ok!(tx.send("two"));
+        assert_ok!(tx.send("three"));
+
+        drop(tx);
+
+        assert_ok!(th1.join());
+        assert_ok!(th2.join());
+    });
+}
+
+#[test]
+fn drop_rx() {
+    loom::model(|| {
+        let (tx, mut rx1) = broadcast::channel(16);
+        let rx2 = tx.subscribe();
+
+        let th1 = thread::spawn(move || {
+            block_on(async {
+                let v = assert_ok!(rx1.recv().await);
+                assert_eq!(v, "one");
+
+                let v = assert_ok!(rx1.recv().await);
+                assert_eq!(v, "two");
+
+                let v = assert_ok!(rx1.recv().await);
+                assert_eq!(v, "three");
+
+                match assert_err!(rx1.recv().await) {
+                    Closed => {}
+                    _ => panic!(),
+                }
+            });
+        });
+
+        let th2 = thread::spawn(move || {
+            drop(rx2);
+        });
+
+        assert_ok!(tx.send("one"));
+        assert_ok!(tx.send("two"));
+        assert_ok!(tx.send("three"));
+        drop(tx);
+
+        assert_ok!(th1.join());
+        assert_ok!(th2.join());
+    });
+}
diff --git a/tokio/src/sync/tests/mod.rs b/tokio/src/sync/tests/mod.rs
index 0b50cc95..2ee140cb 100644
--- a/tokio/src/sync/tests/mod.rs
+++ b/tokio/src/sync/tests/mod.rs
@@ -5,6 +5,7 @@ cfg_not_loom! {
 
 cfg_loom! {
     mod loom_atomic_waker;
+    mod loom_broadcast;
     mod loom_list;
     mod loom_mpsc;
     mod loom_oneshot;