Merge branch 'master' into mpmc

# Conflicts:
#	benches/basic.rs
#	src/lib.rs

Author: Restioson

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "flume"
-version = "0.7.0"
+version = "0.7.1"
 authors = ["Joshua Barretto <[email protected]>"]
 edition = "2018"
 description = "A blazingly fast multi-producer channel"
@@ -21,6 +21,7 @@ futures = { version = "^0.3", default-features = false, optional = true }
 
 [dev-dependencies]
 crossbeam-channel = "0.4"
+crossbeam-utils = "0.7"
 criterion = "0.3.1"
 async-std = { version = "1.5", features = ["attributes"] }
 

benches/basic.rs

@@ -7,6 +7,7 @@ use std::{
     fmt::Debug,
 };
 use criterion::{Criterion, Bencher, black_box};
+use std::time::Instant;
 
 trait Sender: Clone + Send + Sized + 'static {
     type Item: Debug + Default;
@@ -264,6 +265,70 @@ fn test_robin_b<S: Sender>(b: &mut Bencher, thread_num: usize, msg_num: usize) {
     });
 }
 
+fn test_mpsc_bounded_no_wait<S: Sender>(b: &mut Bencher, thread_num: u64) {
+    b.iter_custom(|iters| {
+        let iters = iters * 1000;
+        let (mut tx, mut rx) = S::bounded(iters as usize);
+        let start = Instant::now();
+
+        crossbeam_utils::thread::scope(|scope| {
+            for _ in 0..thread_num {
+                let tx = tx.clone();
+                scope.spawn(move |_| {
+                    for i in 0..iters / thread_num {
+                        tx.send(Default::default());
+                    }
+                });
+            }
+
+            for i in 0..iters - ((iters / thread_num) * thread_num) {
+                tx.send(Default::default());
+            }
+
+            for _ in 0..iters {
+                black_box(rx.recv());
+            }
+        })
+            .unwrap();
+
+        start.elapsed()
+    })
+}
+
+fn test_mpsc_bounded<S: Sender>(b: &mut Bencher, bound: usize, thread_num: usize) {
+    b.iter_custom(|iters| {
+        let (mut tx, mut rx) = S::bounded(bound);
+        let start = Instant::now();
+
+        crossbeam_utils::thread::scope(|scope| {
+            let msgs = iters as usize * bound;
+
+            for _ in 0..thread_num {
+                let tx = tx.clone();
+                scope.spawn(move |_| {
+                    for _ in 0..msgs / thread_num as usize {
+                        tx.send(Default::default());
+                    }
+                });
+            }
+
+            scope.spawn(move |_| {
+                // Remainder
+                for _ in 0..msgs - (msgs / thread_num as usize * thread_num)  {
+                    tx.send(Default::default());
+                }
+            });
+
+            for _ in 0..msgs {
+                black_box(rx.recv());
+            }
+        })
+            .unwrap();
+
+        start.elapsed()
+    })
+}
+
 fn create(b: &mut Criterion) {
     b.bench_function("create-flume", |b| test_create::<flume::Sender<u32>>(b));
     b.bench_function("create-crossbeam", |b| test_create::<crossbeam_channel::Sender<u32>>(b));
@@ -351,11 +416,30 @@ fn robin_b_4t_1000m(b: &mut Criterion) {
     b.bench_function("robin-b-4t-1000m-std", |b| test_robin_b::<mpsc::Sender<u32>>(b, 4, 1000));
 }
 
+fn mpsc_bounded_no_wait_4t(b: &mut Criterion) {
+    b.bench_function("mpsc-bounded-no-wait-4t-flume", |b| test_mpsc_bounded_no_wait::<flume::Sender<u32>>(b, 4));
+    b.bench_function("mpsc-bounded-no-wait-4t-crossbeam", |b| test_mpsc_bounded_no_wait::<crossbeam_channel::Sender<u32>>(b, 4));
+    b.bench_function("mpsc-bounded-no-wait-4t-std", |b| test_mpsc_bounded_no_wait::<mpsc::Sender<u32>>(b, 4));
+}
+
+fn mpsc_bounded_4t(b: &mut Criterion) {
+    for bound in &[1, 10, 50, 10_000] {
+        let text = format!("mpsc-bounded-small-4t-{}m-", bound);
+        let bound = *bound;
+
+        b.bench_function(&format!("{}{}", text, "flume"), |b| test_mpsc_bounded::<flume::Sender<u32>>(b, bound, 4));
+        b.bench_function(&format!("{}{}", text, "crossbeam"), |b| test_mpsc_bounded::<crossbeam_channel::Sender<u32>>(b, bound, 4));
+        b.bench_function(&format!("{}{}", text, "std"), |b| test_mpsc_bounded::<mpsc::Sender<u32>>(b, bound, 4));
+    }
+}
+
 criterion_group!(
     compare,
     create,
     oneshot,
     inout,
+    mpsc_bounded_no_wait_4t,
+    mpsc_bounded_4t,
     hydra_1t_1000m,
     hydra_32t_1m,
     hydra_32t_1000m,