//! Tests for the array channel flavor.
-extern crate crossbeam;
#[macro_use]
-extern crate crossbeam_channel as channel;
+extern crate crossbeam_channel;
+extern crate crossbeam_utils;
extern crate rand;
-mod wrappers;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::thread;
+use std::time::Duration;
-macro_rules! tests {
- ($channel:path) => {
- use std::sync::atomic::AtomicUsize;
- use std::sync::atomic::Ordering;
- use std::thread;
- use std::time::Duration;
+use crossbeam_channel::bounded;
+use crossbeam_channel::{RecvError, RecvTimeoutError, TryRecvError};
+use crossbeam_channel::{SendError, SendTimeoutError, TrySendError};
+use crossbeam_utils::thread::scope;
+use rand::{thread_rng, Rng};
- use $channel as channel;
- use crossbeam;
- use rand::{thread_rng, Rng};
+fn ms(ms: u64) -> Duration {
+ Duration::from_millis(ms)
+}
- fn ms(ms: u64) -> Duration {
- Duration::from_millis(ms)
- }
+#[test]
+fn smoke() {
+ let (s, r) = bounded(1);
+ s.send(7).unwrap();
+ assert_eq!(r.try_recv(), Ok(7));
- #[test]
- fn smoke() {
- let (s, r) = channel::bounded(1);
- s.send(7);
- assert_eq!(r.try_recv(), Some(7));
+ s.send(8).unwrap();
+ assert_eq!(r.recv(), Ok(8));
- s.send(8);
- assert_eq!(r.recv(), Some(8));
+ assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
+ assert_eq!(r.recv_timeout(ms(1000)), Err(RecvTimeoutError::Timeout));
+}
- assert_eq!(r.try_recv(), None);
- select! {
- recv(r) => panic!(),
- recv(channel::after(ms(1000))) => {}
- }
- }
+#[test]
+fn capacity() {
+ for i in 1..10 {
+ let (s, r) = bounded::<()>(i);
+ assert_eq!(s.capacity(), Some(i));
+ assert_eq!(r.capacity(), Some(i));
+ }
+}
- #[test]
- fn capacity() {
- for i in 1..10 {
- let (s, r) = channel::bounded::<()>(i);
- assert_eq!(s.capacity(), Some(i));
- assert_eq!(r.capacity(), Some(i));
- }
- }
+#[test]
+fn len_empty_full() {
+ let (s, r) = bounded(2);
+
+ assert_eq!(s.len(), 0);
+ assert_eq!(s.is_empty(), true);
+ assert_eq!(s.is_full(), false);
+ assert_eq!(r.len(), 0);
+ assert_eq!(r.is_empty(), true);
+ assert_eq!(r.is_full(), false);
+
+ s.send(()).unwrap();
+
+ assert_eq!(s.len(), 1);
+ assert_eq!(s.is_empty(), false);
+ assert_eq!(s.is_full(), false);
+ assert_eq!(r.len(), 1);
+ assert_eq!(r.is_empty(), false);
+ assert_eq!(r.is_full(), false);
+
+ s.send(()).unwrap();
+
+ assert_eq!(s.len(), 2);
+ assert_eq!(s.is_empty(), false);
+ assert_eq!(s.is_full(), true);
+ assert_eq!(r.len(), 2);
+ assert_eq!(r.is_empty(), false);
+ assert_eq!(r.is_full(), true);
+
+ r.recv().unwrap();
+
+ assert_eq!(s.len(), 1);
+ assert_eq!(s.is_empty(), false);
+ assert_eq!(s.is_full(), false);
+ assert_eq!(r.len(), 1);
+ assert_eq!(r.is_empty(), false);
+ assert_eq!(r.is_full(), false);
+}
- #[test]
- fn len_empty_full() {
- let (s, r) = channel::bounded(2);
+#[test]
+fn try_recv() {
+ let (s, r) = bounded(100);
+
+ scope(|scope| {
+ scope.spawn(move |_| {
+ assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
+ thread::sleep(ms(1500));
+ assert_eq!(r.try_recv(), Ok(7));
+ thread::sleep(ms(500));
+ assert_eq!(r.try_recv(), Err(TryRecvError::Disconnected));
+ });
+ scope.spawn(move |_| {
+ thread::sleep(ms(1000));
+ s.send(7).unwrap();
+ });
+ }).unwrap();
+}
- assert_eq!(s.len(), 0);
- assert_eq!(s.is_empty(), true);
- assert_eq!(s.is_full(), false);
- assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+#[test]
+fn recv() {
+ let (s, r) = bounded(100);
+
+ scope(|scope| {
+ scope.spawn(move |_| {
+ assert_eq!(r.recv(), Ok(7));
+ thread::sleep(ms(1000));
+ assert_eq!(r.recv(), Ok(8));
+ thread::sleep(ms(1000));
+ assert_eq!(r.recv(), Ok(9));
+ assert_eq!(r.recv(), Err(RecvError));
+ });
+ scope.spawn(move |_| {
+ thread::sleep(ms(1500));
+ s.send(7).unwrap();
+ s.send(8).unwrap();
+ s.send(9).unwrap();
+ });
+ }).unwrap();
+}
- s.send(());
+#[test]
+fn recv_timeout() {
+ let (s, r) = bounded::<i32>(100);
+
+ scope(|scope| {
+ scope.spawn(move |_| {
+ assert_eq!(r.recv_timeout(ms(1000)), Err(RecvTimeoutError::Timeout));
+ assert_eq!(r.recv_timeout(ms(1000)), Ok(7));
+ assert_eq!(
+ r.recv_timeout(ms(1000)),
+ Err(RecvTimeoutError::Disconnected)
+ );
+ });
+ scope.spawn(move |_| {
+ thread::sleep(ms(1500));
+ s.send(7).unwrap();
+ });
+ }).unwrap();
+}
- assert_eq!(s.len(), 1);
- assert_eq!(s.is_empty(), false);
- assert_eq!(s.is_full(), false);
- assert_eq!(r.len(), 1);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), false);
+#[test]
+fn try_send() {
+ let (s, r) = bounded(1);
+
+ scope(|scope| {
+ scope.spawn(move |_| {
+ assert_eq!(s.try_send(1), Ok(()));
+ assert_eq!(s.try_send(2), Err(TrySendError::Full(2)));
+ thread::sleep(ms(1500));
+ assert_eq!(s.try_send(3), Ok(()));
+ thread::sleep(ms(500));
+ assert_eq!(s.try_send(4), Err(TrySendError::Disconnected(4)));
+ });
+ scope.spawn(move |_| {
+ thread::sleep(ms(1000));
+ assert_eq!(r.try_recv(), Ok(1));
+ assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
+ assert_eq!(r.recv(), Ok(3));
+ });
+ }).unwrap();
+}
- s.send(());
+#[test]
+fn send() {
+ let (s, r) = bounded(1);
+
+ scope(|scope| {
+ scope.spawn(|_| {
+ s.send(7).unwrap();
+ thread::sleep(ms(1000));
+ s.send(8).unwrap();
+ thread::sleep(ms(1000));
+ s.send(9).unwrap();
+ thread::sleep(ms(1000));
+ s.send(10).unwrap();
+ });
+ scope.spawn(|_| {
+ thread::sleep(ms(1500));
+ assert_eq!(r.recv(), Ok(7));
+ assert_eq!(r.recv(), Ok(8));
+ assert_eq!(r.recv(), Ok(9));
+ });
+ }).unwrap();
+}
- assert_eq!(s.len(), 2);
- assert_eq!(s.is_empty(), false);
- assert_eq!(s.is_full(), true);
- assert_eq!(r.len(), 2);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), true);
+#[test]
+fn send_timeout() {
+ let (s, r) = bounded(2);
+
+ scope(|scope| {
+ scope.spawn(move |_| {
+ assert_eq!(s.send_timeout(1, ms(1000)), Ok(()));
+ assert_eq!(s.send_timeout(2, ms(1000)), Ok(()));
+ assert_eq!(
+ s.send_timeout(3, ms(500)),
+ Err(SendTimeoutError::Timeout(3))
+ );
+ thread::sleep(ms(1000));
+ assert_eq!(s.send_timeout(4, ms(1000)), Ok(()));
+ thread::sleep(ms(1000));
+ assert_eq!(s.send(5), Err(SendError(5)));
+ });
+ scope.spawn(move |_| {
+ thread::sleep(ms(1000));
+ assert_eq!(r.recv(), Ok(1));
+ thread::sleep(ms(1000));
+ assert_eq!(r.recv(), Ok(2));
+ assert_eq!(r.recv(), Ok(4));
+ });
+ }).unwrap();
+}
- r.recv().unwrap();
+#[test]
+fn send_after_disconnect() {
+ let (s, r) = bounded(100);
- assert_eq!(s.len(), 1);
- assert_eq!(s.is_empty(), false);
- assert_eq!(s.is_full(), false);
- assert_eq!(r.len(), 1);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), false);
- }
+ s.send(1).unwrap();
+ s.send(2).unwrap();
+ s.send(3).unwrap();
- #[test]
- fn recv() {
- let (s, r) = channel::bounded(100);
-
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- assert_eq!(r.recv(), Some(7));
- thread::sleep(ms(1000));
- assert_eq!(r.recv(), Some(8));
- thread::sleep(ms(1000));
- assert_eq!(r.recv(), Some(9));
- assert_eq!(r.recv(), None);
- });
- scope.spawn(move || {
- thread::sleep(ms(1500));
- s.send(7);
- s.send(8);
- s.send(9);
- });
- });
- }
+ drop(r);
- #[test]
- fn recv_timeout() {
- let (s, r) = channel::bounded::<i32>(100);
+ assert_eq!(s.send(4), Err(SendError(4)));
+ assert_eq!(s.try_send(5), Err(TrySendError::Disconnected(5)));
+ assert_eq!(
+ s.send_timeout(6, ms(500)),
+ Err(SendTimeoutError::Disconnected(6))
+ );
+}
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- select! {
- recv(r) => panic!(),
- recv(channel::after(ms(1000))) => {}
- }
- select! {
- recv(r, v) => assert_eq!(v, Some(7)),
- recv(channel::after(ms(1000))) => panic!(),
- }
- select! {
- recv(r, v) => assert_eq!(v, None),
- recv(channel::after(ms(1000))) => panic!(),
- }
- });
- scope.spawn(move || {
- thread::sleep(ms(1500));
- s.send(7);
- });
- });
- }
+#[test]
+fn recv_after_disconnect() {
+ let (s, r) = bounded(100);
- #[test]
- fn try_recv() {
- let (s, r) = channel::bounded(100);
-
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- assert_eq!(r.try_recv(), None);
- thread::sleep(ms(1500));
- assert_eq!(r.try_recv(), Some(7));
- thread::sleep(ms(500));
- assert_eq!(r.try_recv(), None);
- });
- scope.spawn(move || {
- thread::sleep(ms(1000));
- s.send(7);
- });
- });
- }
+ s.send(1).unwrap();
+ s.send(2).unwrap();
+ s.send(3).unwrap();
- #[test]
- fn send() {
- let (s, r) = channel::bounded(1);
-
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- s.send(7);
- thread::sleep(ms(1000));
- s.send(8);
- thread::sleep(ms(1000));
- s.send(9);
- thread::sleep(ms(1000));
- s.send(10);
- });
- scope.spawn(move || {
- thread::sleep(ms(1500));
- assert_eq!(r.recv(), Some(7));
- assert_eq!(r.recv(), Some(8));
- assert_eq!(r.recv(), Some(9));
- });
- });
- }
+ drop(s);
- #[test]
- fn send_timeout() {
- let (s, r) = channel::bounded(2);
+ assert_eq!(r.recv(), Ok(1));
+ assert_eq!(r.recv(), Ok(2));
+ assert_eq!(r.recv(), Ok(3));
+ assert_eq!(r.recv(), Err(RecvError));
+}
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- select! {
- send(s, 1) => {}
- recv(channel::after(ms(1000))) => panic!(),
- }
- select! {
- send(s, 2) => {}
- recv(channel::after(ms(1000))) => panic!(),
- }
- select! {
- send(s, 3) => panic!(),
- recv(channel::after(ms(500))) => {}
- }
- thread::sleep(ms(1000));
- select! {
- send(s, 4) => {}
- recv(channel::after(ms(1000))) => panic!(),
- }
- });
- scope.spawn(move || {
- thread::sleep(ms(1000));
- assert_eq!(r.recv(), Some(1));
- thread::sleep(ms(1000));
- assert_eq!(r.recv(), Some(2));
- assert_eq!(r.recv(), Some(4));
- });
- });
- }
+#[test]
+fn len() {
+ const COUNT: usize = 25_000;
+ const CAP: usize = 1000;
- #[test]
- fn try_send() {
- let (s, r) = channel::bounded(1);
+ let (s, r) = bounded(CAP);
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- select! {
- send(s, 1) => {}
- default => panic!(),
- }
- select! {
- send(s, 2) => panic!(),
- default => {}
- }
- thread::sleep(ms(1500));
- select! {
- send(s, 3) => {}
- default => panic!(),
- }
- thread::sleep(ms(500));
- select! {
- send(s, 4) => {}
- default => panic!(),
- }
- });
- scope.spawn(move || {
- thread::sleep(ms(1000));
- assert_eq!(r.try_recv(), Some(1));
- assert_eq!(r.try_recv(), None);
- assert_eq!(r.recv(), Some(3));
- });
- });
+ assert_eq!(s.len(), 0);
+ assert_eq!(r.len(), 0);
+
+ for _ in 0..CAP / 10 {
+ for i in 0..50 {
+ s.send(i).unwrap();
+ assert_eq!(s.len(), i + 1);
}
- #[test]
- fn recv_after_close() {
- let (s, r) = channel::bounded(100);
+ for i in 0..50 {
+ r.recv().unwrap();
+ assert_eq!(r.len(), 50 - i - 1);
+ }
+ }
- s.send(1);
- s.send(2);
- s.send(3);
+ assert_eq!(s.len(), 0);
+ assert_eq!(r.len(), 0);
- drop(s);
+ for i in 0..CAP {
+ s.send(i).unwrap();
+ assert_eq!(s.len(), i + 1);
+ }
- assert_eq!(r.recv(), Some(1));
- assert_eq!(r.recv(), Some(2));
- assert_eq!(r.recv(), Some(3));
- assert_eq!(r.recv(), None);
- }
+ for _ in 0..CAP {
+ r.recv().unwrap();
+ }
- #[test]
- fn len() {
- const COUNT: usize = 25_000;
- const CAP: usize = 1000;
+ assert_eq!(s.len(), 0);
+ assert_eq!(r.len(), 0);
- let (s, r) = channel::bounded(CAP);
+ scope(|scope| {
+ scope.spawn(|_| {
+ for i in 0..COUNT {
+ assert_eq!(r.recv(), Ok(i));
+ let len = r.len();
+ assert!(len <= CAP);
+ }
+ });
- assert_eq!(s.len(), 0);
- assert_eq!(r.len(), 0);
+ scope.spawn(|_| {
+ for i in 0..COUNT {
+ s.send(i).unwrap();
+ let len = s.len();
+ assert!(len <= CAP);
+ }
+ });
+ }).unwrap();
- for _ in 0..CAP / 10 {
- for i in 0..50 {
- s.send(i);
- assert_eq!(s.len(), i + 1);
- }
+ assert_eq!(s.len(), 0);
+ assert_eq!(r.len(), 0);
+}
- for i in 0..50 {
- r.recv().unwrap();
- assert_eq!(r.len(), 50 - i - 1);
- }
- }
+#[test]
+fn disconnect_wakes_sender() {
+ let (s, r) = bounded(1);
+
+ scope(|scope| {
+ scope.spawn(move |_| {
+ assert_eq!(s.send(()), Ok(()));
+ assert_eq!(s.send(()), Err(SendError(())));
+ });
+ scope.spawn(move |_| {
+ thread::sleep(ms(1000));
+ drop(r);
+ });
+ }).unwrap();
+}
- assert_eq!(s.len(), 0);
- assert_eq!(r.len(), 0);
+#[test]
+fn disconnect_wakes_receiver() {
+ let (s, r) = bounded::<()>(1);
- for i in 0..CAP {
- s.send(i);
- assert_eq!(s.len(), i + 1);
- }
+ scope(|scope| {
+ scope.spawn(move |_| {
+ assert_eq!(r.recv(), Err(RecvError));
+ });
+ scope.spawn(move |_| {
+ thread::sleep(ms(1000));
+ drop(s);
+ });
+ }).unwrap();
+}
+
+#[test]
+fn spsc() {
+ const COUNT: usize = 100_000;
+
+ let (s, r) = bounded(3);
- for _ in 0..CAP {
- r.recv().unwrap();
+ scope(|scope| {
+ scope.spawn(move |_| {
+ for i in 0..COUNT {
+ assert_eq!(r.recv(), Ok(i));
}
+ assert_eq!(r.recv(), Err(RecvError));
+ });
+ scope.spawn(move |_| {
+ for i in 0..COUNT {
+ s.send(i).unwrap();
+ }
+ });
+ }).unwrap();
+}
- assert_eq!(s.len(), 0);
- assert_eq!(r.len(), 0);
+#[test]
+fn mpmc() {
+ const COUNT: usize = 25_000;
+ const THREADS: usize = 4;
- crossbeam::scope(|scope| {
- scope.spawn(|| {
- for i in 0..COUNT {
- assert_eq!(r.recv(), Some(i));
- let len = r.len();
- assert!(len <= CAP);
- }
- });
+ let (s, r) = bounded::<usize>(3);
+ let v = (0..COUNT).map(|_| AtomicUsize::new(0)).collect::<Vec<_>>();
- scope.spawn(|| {
- for i in 0..COUNT {
- s.send(i);
- let len = s.len();
- assert!(len <= CAP);
- }
- });
+ scope(|scope| {
+ for _ in 0..THREADS {
+ scope.spawn(|_| {
+ for _ in 0..COUNT {
+ let n = r.recv().unwrap();
+ v[n].fetch_add(1, Ordering::SeqCst);
+ }
});
-
- assert_eq!(s.len(), 0);
- assert_eq!(r.len(), 0);
}
-
- #[test]
- fn close_wakes_receiver() {
- let (s, r) = channel::bounded::<()>(1);
-
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- assert_eq!(r.recv(), None);
- });
- scope.spawn(move || {
- thread::sleep(ms(1000));
- drop(s);
- });
+ for _ in 0..THREADS {
+ scope.spawn(|_| {
+ for i in 0..COUNT {
+ s.send(i).unwrap();
+ }
});
}
+ }).unwrap();
- #[test]
- fn spsc() {
- const COUNT: usize = 100_000;
+ for c in v {
+ assert_eq!(c.load(Ordering::SeqCst), THREADS);
+ }
+}
- let (s, r) = channel::bounded(3);
+#[test]
+fn stress_oneshot() {
+ const COUNT: usize = 10_000;
- crossbeam::scope(|scope| {
- scope.spawn(move || {
- for i in 0..COUNT {
- assert_eq!(r.recv(), Some(i));
- }
- assert_eq!(r.recv(), None);
- });
- scope.spawn(move || {
- for i in 0..COUNT {
- s.send(i);
- }
- });
- });
- }
+ for _ in 0..COUNT {
+ let (s, r) = bounded(1);
- #[test]
- fn mpmc() {
- const COUNT: usize = 25_000;
- const THREADS: usize = 4;
-
- let (s, r) = channel::bounded::<usize>(3);
- let v = (0..COUNT).map(|_| AtomicUsize::new(0)).collect::<Vec<_>>();
-
- crossbeam::scope(|scope| {
- for _ in 0..THREADS {
- scope.spawn(|| {
- for _ in 0..COUNT {
- let n = r.recv().unwrap();
- v[n].fetch_add(1, Ordering::SeqCst);
- }
- });
- }
- for _ in 0..THREADS {
- scope.spawn(|| {
- for i in 0..COUNT {
- s.send(i);
- }
- });
- }
- });
+ scope(|scope| {
+ scope.spawn(|_| r.recv().unwrap());
+ scope.spawn(|_| s.send(0).unwrap());
+ }).unwrap();
+ }
+}
- for c in v {
- assert_eq!(c.load(Ordering::SeqCst), THREADS);
+#[test]
+fn stress_iter() {
+ const COUNT: usize = 100_000;
+
+ let (request_s, request_r) = bounded(1);
+ let (response_s, response_r) = bounded(1);
+
+ scope(|scope| {
+ scope.spawn(move |_| {
+ let mut count = 0;
+ loop {
+ for x in response_r.try_iter() {
+ count += x;
+ if count == COUNT {
+ return;
+ }
+ }
+ request_s.send(()).unwrap();
}
- }
+ });
- #[test]
- fn stress_timeout_two_threads() {
- const COUNT: usize = 100;
-
- let (s, r) = channel::bounded(2);
-
- crossbeam::scope(|scope| {
- scope.spawn(|| {
- for i in 0..COUNT {
- if i % 2 == 0 {
- thread::sleep(ms(50));
- }
- loop {
- select! {
- send(s, i) => break,
- recv(channel::after(ms(10))) => {}
- }
- }
- }
- });
-
- scope.spawn(|| {
- for i in 0..COUNT {
- if i % 2 == 0 {
- thread::sleep(ms(50));
- }
- loop {
- select! {
- recv(r, v) => {
- assert_eq!(v, Some(i));
- break;
- }
- recv(channel::after(ms(10))) => {}
- }
- }
- }
- });
- });
+ for _ in request_r.iter() {
+ if response_s.send(1).is_err() {
+ break;
+ }
}
+ }).unwrap();
+}
- #[test]
- fn drops() {
- const RUNS: usize = 100;
+#[test]
+fn stress_timeout_two_threads() {
+ const COUNT: usize = 100;
- static DROPS: AtomicUsize = AtomicUsize::new(0);
+ let (s, r) = bounded(2);
- #[derive(Debug, PartialEq)]
- struct DropCounter;
+ scope(|scope| {
+ scope.spawn(|_| {
+ for i in 0..COUNT {
+ if i % 2 == 0 {
+ thread::sleep(ms(50));
+ }
+ loop {
+ if let Ok(()) = s.send_timeout(i, ms(10)) {
+ break;
+ }
+ }
+ }
+ });
- impl Drop for DropCounter {
- fn drop(&mut self) {
- DROPS.fetch_add(1, Ordering::SeqCst);
+ scope.spawn(|_| {
+ for i in 0..COUNT {
+ if i % 2 == 0 {
+ thread::sleep(ms(50));
+ }
+ loop {
+ if let Ok(x) = r.recv_timeout(ms(10)) {
+ assert_eq!(x, i);
+ break;
+ }
}
}
+ });
+ }).unwrap();
+}
- let mut rng = thread_rng();
+#[test]
+fn drops() {
+ const RUNS: usize = 100;
- for _ in 0..RUNS {
- let steps = rng.gen_range(0, 10_000);
- let additional = rng.gen_range(0, 50);
+ static DROPS: AtomicUsize = AtomicUsize::new(0);
- DROPS.store(0, Ordering::SeqCst);
- let (s, r) = channel::bounded::<DropCounter>(50);
+ #[derive(Debug, PartialEq)]
+ struct DropCounter;
- crossbeam::scope(|scope| {
- scope.spawn(|| {
- for _ in 0..steps {
- r.recv().unwrap();
- }
- });
+ impl Drop for DropCounter {
+ fn drop(&mut self) {
+ DROPS.fetch_add(1, Ordering::SeqCst);
+ }
+ }
- scope.spawn(|| {
- for _ in 0..steps {
- s.send(DropCounter);
- }
- });
- });
+ let mut rng = thread_rng();
- for _ in 0..additional {
- s.send(DropCounter);
- }
+ for _ in 0..RUNS {
+ let steps = rng.gen_range(0, 10_000);
+ let additional = rng.gen_range(0, 50);
- assert_eq!(DROPS.load(Ordering::SeqCst), steps);
- drop(s);
- drop(r);
- assert_eq!(DROPS.load(Ordering::SeqCst), steps + additional);
- }
- }
+ DROPS.store(0, Ordering::SeqCst);
+ let (s, r) = bounded::<DropCounter>(50);
- #[test]
- fn linearizable() {
- const COUNT: usize = 25_000;
- const THREADS: usize = 4;
-
- let (s, r) = channel::bounded(THREADS);
-
- crossbeam::scope(|scope| {
- for _ in 0..THREADS {
- scope.spawn(|| {
- for _ in 0..COUNT {
- s.send(0);
- r.try_recv().unwrap();
- }
- });
+ scope(|scope| {
+ scope.spawn(|_| {
+ for _ in 0..steps {
+ r.recv().unwrap();
}
});
+
+ scope.spawn(|_| {
+ for _ in 0..steps {
+ s.send(DropCounter).unwrap();
+ }
+ });
+ }).unwrap();
+
+ for _ in 0..additional {
+ s.send(DropCounter).unwrap();
}
- #[test]
- fn fairness() {
- const COUNT: usize = 10_000;
+ assert_eq!(DROPS.load(Ordering::SeqCst), steps);
+ drop(s);
+ drop(r);
+ assert_eq!(DROPS.load(Ordering::SeqCst), steps + additional);
+ }
+}
- let (s1, r1) = channel::bounded::<()>(COUNT);
- let (s2, r2) = channel::bounded::<()>(COUNT);
+#[test]
+fn linearizable() {
+ const COUNT: usize = 25_000;
+ const THREADS: usize = 4;
- for _ in 0..COUNT {
- s1.send(());
- s2.send(());
- }
+ let (s, r) = bounded(THREADS);
- let mut hits = [0usize; 2];
- for _ in 0..COUNT {
- select! {
- recv(r1) => hits[0] += 1,
- recv(r2) => hits[1] += 1,
+ scope(|scope| {
+ for _ in 0..THREADS {
+ scope.spawn(|_| {
+ for _ in 0..COUNT {
+ s.send(0).unwrap();
+ r.try_recv().unwrap();
}
- }
- assert!(hits.iter().all(|x| *x >= COUNT / hits.len() / 2));
+ });
}
+ }).unwrap();
+}
- #[test]
- fn fairness_duplicates() {
- const COUNT: usize = 10_000;
+#[test]
+fn fairness() {
+ const COUNT: usize = 10_000;
- let (s, r) = channel::bounded::<()>(COUNT);
+ let (s1, r1) = bounded::<()>(COUNT);
+ let (s2, r2) = bounded::<()>(COUNT);
- for _ in 0..COUNT {
- s.send(());
- }
+ for _ in 0..COUNT {
+ s1.send(()).unwrap();
+ s2.send(()).unwrap();
+ }
- let mut hits = [0usize; 5];
- for _ in 0..COUNT {
- select! {
- recv(r) => hits[0] += 1,
- recv(r) => hits[1] += 1,
- recv(r) => hits[2] += 1,
- recv(r) => hits[3] += 1,
- recv(r) => hits[4] += 1,
- }
- }
- assert!(hits.iter().all(|x| *x >= COUNT / hits.len() / 2));
+ let mut hits = [0usize; 2];
+ for _ in 0..COUNT {
+ select! {
+ recv(r1) -> _ => hits[0] += 1,
+ recv(r2) -> _ => hits[1] += 1,
}
+ }
+ assert!(hits.iter().all(|x| *x >= COUNT / hits.len() / 2));
+}
- #[test]
- fn recv_in_send() {
- let (s, _r) = channel::bounded(1);
- s.send(());
+#[test]
+fn fairness_duplicates() {
+ const COUNT: usize = 10_000;
- select! {
- send(s, panic!()) => panic!(),
- default => {}
- }
+ let (s, r) = bounded::<()>(COUNT);
- let (s, r) = channel::bounded(2);
- s.send(());
+ for _ in 0..COUNT {
+ s.send(()).unwrap();
+ }
- select! {
- send(s, assert_eq!(r.recv(), Some(()))) => {}
- }
+ let mut hits = [0usize; 5];
+ for _ in 0..COUNT {
+ select! {
+ recv(r) -> _ => hits[0] += 1,
+ recv(r) -> _ => hits[1] += 1,
+ recv(r) -> _ => hits[2] += 1,
+ recv(r) -> _ => hits[3] += 1,
+ recv(r) -> _ => hits[4] += 1,
}
- };
-}
-
-mod normal {
- tests!(wrappers::normal);
+ }
+ assert!(hits.iter().all(|x| *x >= COUNT / hits.len() / 2));
}
-mod cloned {
- tests!(wrappers::cloned);
-}
+#[test]
+fn recv_in_send() {
+ let (s, _r) = bounded(1);
+ s.send(()).unwrap();
-mod select {
- tests!(wrappers::select);
-}
+ #[allow(unreachable_code)]
+ {
+ select! {
+ send(s, panic!()) -> _ => panic!(),
+ default => {}
+ }
+ }
-mod select_spin {
- tests!(wrappers::select_spin);
-}
+ let (s, r) = bounded(2);
+ s.send(()).unwrap();
-mod select_multi {
- tests!(wrappers::select_multi);
+ select! {
+ send(s, assert_eq!(r.recv(), Ok(()))) -> _ => {}
+ }
}
projects / cargo.git / blobdiff