1 //! Tests for the list channel flavor.
2 
3 #[macro_use]
4 extern crate crossbeam_channel;
5 extern crate crossbeam_utils;
6 extern crate rand;
7 
8 use std::any::Any;
9 use std::sync::atomic::AtomicUsize;
10 use std::sync::atomic::Ordering;
11 use std::thread;
12 use std::time::Duration;
13 
14 use crossbeam_channel::{unbounded, Receiver};
15 use crossbeam_channel::{RecvError, RecvTimeoutError, TryRecvError};
16 use crossbeam_channel::{SendError, SendTimeoutError, TrySendError};
17 use crossbeam_utils::thread::scope;
18 use rand::{thread_rng, Rng};
19 
ms(ms: u64) -> Duration20 fn ms(ms: u64) -> Duration {
21     Duration::from_millis(ms)
22 }
23 
24 #[test]
smoke()25 fn smoke() {
26     let (s, r) = unbounded();
27     s.try_send(7).unwrap();
28     assert_eq!(r.try_recv(), Ok(7));
29 
30     s.send(8).unwrap();
31     assert_eq!(r.recv(), Ok(8));
32 
33     assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
34     assert_eq!(r.recv_timeout(ms(1000)), Err(RecvTimeoutError::Timeout));
35 }
36 
37 #[test]
capacity()38 fn capacity() {
39     let (s, r) = unbounded::<()>();
40     assert_eq!(s.capacity(), None);
41     assert_eq!(r.capacity(), None);
42 }
43 
44 #[test]
len_empty_full()45 fn len_empty_full() {
46     let (s, r) = unbounded();
47 
48     assert_eq!(s.len(), 0);
49     assert_eq!(s.is_empty(), true);
50     assert_eq!(s.is_full(), false);
51     assert_eq!(r.len(), 0);
52     assert_eq!(r.is_empty(), true);
53     assert_eq!(r.is_full(), false);
54 
55     s.send(()).unwrap();
56 
57     assert_eq!(s.len(), 1);
58     assert_eq!(s.is_empty(), false);
59     assert_eq!(s.is_full(), false);
60     assert_eq!(r.len(), 1);
61     assert_eq!(r.is_empty(), false);
62     assert_eq!(r.is_full(), false);
63 
64     r.recv().unwrap();
65 
66     assert_eq!(s.len(), 0);
67     assert_eq!(s.is_empty(), true);
68     assert_eq!(s.is_full(), false);
69     assert_eq!(r.len(), 0);
70     assert_eq!(r.is_empty(), true);
71     assert_eq!(r.is_full(), false);
72 }
73 
74 #[test]
try_recv()75 fn try_recv() {
76     let (s, r) = unbounded();
77 
78     scope(|scope| {
79         scope.spawn(move |_| {
80             assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
81             thread::sleep(ms(1500));
82             assert_eq!(r.try_recv(), Ok(7));
83             thread::sleep(ms(500));
84             assert_eq!(r.try_recv(), Err(TryRecvError::Disconnected));
85         });
86         scope.spawn(move |_| {
87             thread::sleep(ms(1000));
88             s.send(7).unwrap();
89         });
90     })
91     .unwrap();
92 }
93 
94 #[test]
recv()95 fn recv() {
96     let (s, r) = unbounded();
97 
98     scope(|scope| {
99         scope.spawn(move |_| {
100             assert_eq!(r.recv(), Ok(7));
101             thread::sleep(ms(1000));
102             assert_eq!(r.recv(), Ok(8));
103             thread::sleep(ms(1000));
104             assert_eq!(r.recv(), Ok(9));
105             assert_eq!(r.recv(), Err(RecvError));
106         });
107         scope.spawn(move |_| {
108             thread::sleep(ms(1500));
109             s.send(7).unwrap();
110             s.send(8).unwrap();
111             s.send(9).unwrap();
112         });
113     })
114     .unwrap();
115 }
116 
117 #[test]
recv_timeout()118 fn recv_timeout() {
119     let (s, r) = unbounded::<i32>();
120 
121     scope(|scope| {
122         scope.spawn(move |_| {
123             assert_eq!(r.recv_timeout(ms(1000)), Err(RecvTimeoutError::Timeout));
124             assert_eq!(r.recv_timeout(ms(1000)), Ok(7));
125             assert_eq!(
126                 r.recv_timeout(ms(1000)),
127                 Err(RecvTimeoutError::Disconnected)
128             );
129         });
130         scope.spawn(move |_| {
131             thread::sleep(ms(1500));
132             s.send(7).unwrap();
133         });
134     })
135     .unwrap();
136 }
137 
138 #[test]
try_send()139 fn try_send() {
140     let (s, r) = unbounded();
141     for i in 0..1000 {
142         assert_eq!(s.try_send(i), Ok(()));
143     }
144 
145     drop(r);
146     assert_eq!(s.try_send(777), Err(TrySendError::Disconnected(777)));
147 }
148 
149 #[test]
send()150 fn send() {
151     let (s, r) = unbounded();
152     for i in 0..1000 {
153         assert_eq!(s.send(i), Ok(()));
154     }
155 
156     drop(r);
157     assert_eq!(s.send(777), Err(SendError(777)));
158 }
159 
160 #[test]
send_timeout()161 fn send_timeout() {
162     let (s, r) = unbounded();
163     for i in 0..1000 {
164         assert_eq!(s.send_timeout(i, ms(i as u64)), Ok(()));
165     }
166 
167     drop(r);
168     assert_eq!(
169         s.send_timeout(777, ms(0)),
170         Err(SendTimeoutError::Disconnected(777))
171     );
172 }
173 
174 #[test]
send_after_disconnect()175 fn send_after_disconnect() {
176     let (s, r) = unbounded();
177 
178     s.send(1).unwrap();
179     s.send(2).unwrap();
180     s.send(3).unwrap();
181 
182     drop(r);
183 
184     assert_eq!(s.send(4), Err(SendError(4)));
185     assert_eq!(s.try_send(5), Err(TrySendError::Disconnected(5)));
186     assert_eq!(
187         s.send_timeout(6, ms(0)),
188         Err(SendTimeoutError::Disconnected(6))
189     );
190 }
191 
192 #[test]
recv_after_disconnect()193 fn recv_after_disconnect() {
194     let (s, r) = unbounded();
195 
196     s.send(1).unwrap();
197     s.send(2).unwrap();
198     s.send(3).unwrap();
199 
200     drop(s);
201 
202     assert_eq!(r.recv(), Ok(1));
203     assert_eq!(r.recv(), Ok(2));
204     assert_eq!(r.recv(), Ok(3));
205     assert_eq!(r.recv(), Err(RecvError));
206 }
207 
208 #[test]
len()209 fn len() {
210     let (s, r) = unbounded();
211 
212     assert_eq!(s.len(), 0);
213     assert_eq!(r.len(), 0);
214 
215     for i in 0..50 {
216         s.send(i).unwrap();
217         assert_eq!(s.len(), i + 1);
218     }
219 
220     for i in 0..50 {
221         r.recv().unwrap();
222         assert_eq!(r.len(), 50 - i - 1);
223     }
224 
225     assert_eq!(s.len(), 0);
226     assert_eq!(r.len(), 0);
227 }
228 
229 #[test]
disconnect_wakes_receiver()230 fn disconnect_wakes_receiver() {
231     let (s, r) = unbounded::<()>();
232 
233     scope(|scope| {
234         scope.spawn(move |_| {
235             assert_eq!(r.recv(), Err(RecvError));
236         });
237         scope.spawn(move |_| {
238             thread::sleep(ms(1000));
239             drop(s);
240         });
241     })
242     .unwrap();
243 }
244 
245 #[test]
spsc()246 fn spsc() {
247     const COUNT: usize = 100_000;
248 
249     let (s, r) = unbounded();
250 
251     scope(|scope| {
252         scope.spawn(move |_| {
253             for i in 0..COUNT {
254                 assert_eq!(r.recv(), Ok(i));
255             }
256             assert_eq!(r.recv(), Err(RecvError));
257         });
258         scope.spawn(move |_| {
259             for i in 0..COUNT {
260                 s.send(i).unwrap();
261             }
262         });
263     })
264     .unwrap();
265 }
266 
267 #[test]
mpmc()268 fn mpmc() {
269     const COUNT: usize = 25_000;
270     const THREADS: usize = 4;
271 
272     let (s, r) = unbounded::<usize>();
273     let v = (0..COUNT).map(|_| AtomicUsize::new(0)).collect::<Vec<_>>();
274 
275     scope(|scope| {
276         for _ in 0..THREADS {
277             scope.spawn(|_| {
278                 for _ in 0..COUNT {
279                     let n = r.recv().unwrap();
280                     v[n].fetch_add(1, Ordering::SeqCst);
281                 }
282             });
283         }
284         for _ in 0..THREADS {
285             scope.spawn(|_| {
286                 for i in 0..COUNT {
287                     s.send(i).unwrap();
288                 }
289             });
290         }
291     })
292     .unwrap();
293 
294     assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
295 
296     for c in v {
297         assert_eq!(c.load(Ordering::SeqCst), THREADS);
298     }
299 }
300 
301 #[test]
stress_oneshot()302 fn stress_oneshot() {
303     const COUNT: usize = 10_000;
304 
305     for _ in 0..COUNT {
306         let (s, r) = unbounded();
307 
308         scope(|scope| {
309             scope.spawn(|_| r.recv().unwrap());
310             scope.spawn(|_| s.send(0).unwrap());
311         })
312         .unwrap();
313     }
314 }
315 
316 #[test]
stress_iter()317 fn stress_iter() {
318     const COUNT: usize = 100_000;
319 
320     let (request_s, request_r) = unbounded();
321     let (response_s, response_r) = unbounded();
322 
323     scope(|scope| {
324         scope.spawn(move |_| {
325             let mut count = 0;
326             loop {
327                 for x in response_r.try_iter() {
328                     count += x;
329                     if count == COUNT {
330                         return;
331                     }
332                 }
333                 request_s.send(()).unwrap();
334             }
335         });
336 
337         for _ in request_r.iter() {
338             if response_s.send(1).is_err() {
339                 break;
340             }
341         }
342     })
343     .unwrap();
344 }
345 
346 #[test]
stress_timeout_two_threads()347 fn stress_timeout_two_threads() {
348     const COUNT: usize = 100;
349 
350     let (s, r) = unbounded();
351 
352     scope(|scope| {
353         scope.spawn(|_| {
354             for i in 0..COUNT {
355                 if i % 2 == 0 {
356                     thread::sleep(ms(50));
357                 }
358                 s.send(i).unwrap();
359             }
360         });
361 
362         scope.spawn(|_| {
363             for i in 0..COUNT {
364                 if i % 2 == 0 {
365                     thread::sleep(ms(50));
366                 }
367                 loop {
368                     if let Ok(x) = r.recv_timeout(ms(10)) {
369                         assert_eq!(x, i);
370                         break;
371                     }
372                 }
373             }
374         });
375     })
376     .unwrap();
377 }
378 
379 #[test]
drops()380 fn drops() {
381     static DROPS: AtomicUsize = AtomicUsize::new(0);
382 
383     #[derive(Debug, PartialEq)]
384     struct DropCounter;
385 
386     impl Drop for DropCounter {
387         fn drop(&mut self) {
388             DROPS.fetch_add(1, Ordering::SeqCst);
389         }
390     }
391 
392     let mut rng = thread_rng();
393 
394     for _ in 0..100 {
395         let steps = rng.gen_range(0, 10_000);
396         let additional = rng.gen_range(0, 1000);
397 
398         DROPS.store(0, Ordering::SeqCst);
399         let (s, r) = unbounded::<DropCounter>();
400 
401         scope(|scope| {
402             scope.spawn(|_| {
403                 for _ in 0..steps {
404                     r.recv().unwrap();
405                 }
406             });
407 
408             scope.spawn(|_| {
409                 for _ in 0..steps {
410                     s.send(DropCounter).unwrap();
411                 }
412             });
413         })
414         .unwrap();
415 
416         for _ in 0..additional {
417             s.try_send(DropCounter).unwrap();
418         }
419 
420         assert_eq!(DROPS.load(Ordering::SeqCst), steps);
421         drop(s);
422         drop(r);
423         assert_eq!(DROPS.load(Ordering::SeqCst), steps + additional);
424     }
425 }
426 
427 #[test]
linearizable()428 fn linearizable() {
429     const COUNT: usize = 25_000;
430     const THREADS: usize = 4;
431 
432     let (s, r) = unbounded();
433 
434     scope(|scope| {
435         for _ in 0..THREADS {
436             scope.spawn(|_| {
437                 for _ in 0..COUNT {
438                     s.send(0).unwrap();
439                     r.try_recv().unwrap();
440                 }
441             });
442         }
443     })
444     .unwrap();
445 }
446 
447 #[test]
fairness()448 fn fairness() {
449     const COUNT: usize = 10_000;
450 
451     let (s1, r1) = unbounded::<()>();
452     let (s2, r2) = unbounded::<()>();
453 
454     for _ in 0..COUNT {
455         s1.send(()).unwrap();
456         s2.send(()).unwrap();
457     }
458 
459     let mut hits = [0usize; 2];
460     for _ in 0..COUNT {
461         select! {
462             recv(r1) -> _ => hits[0] += 1,
463             recv(r2) -> _ => hits[1] += 1,
464         }
465     }
466     assert!(hits.iter().all(|x| *x >= COUNT / hits.len() / 2));
467 }
468 
469 #[test]
fairness_duplicates()470 fn fairness_duplicates() {
471     const COUNT: usize = 10_000;
472 
473     let (s, r) = unbounded();
474 
475     for _ in 0..COUNT {
476         s.send(()).unwrap();
477     }
478 
479     let mut hits = [0usize; 5];
480     for _ in 0..COUNT {
481         select! {
482             recv(r) -> _ => hits[0] += 1,
483             recv(r) -> _ => hits[1] += 1,
484             recv(r) -> _ => hits[2] += 1,
485             recv(r) -> _ => hits[3] += 1,
486             recv(r) -> _ => hits[4] += 1,
487         }
488     }
489     assert!(hits.iter().all(|x| *x >= COUNT / hits.len() / 2));
490 }
491 
492 #[test]
recv_in_send()493 fn recv_in_send() {
494     let (s, r) = unbounded();
495     s.send(()).unwrap();
496 
497     select! {
498         send(s, assert_eq!(r.recv(), Ok(()))) -> _ => {}
499     }
500 }
501 
502 #[test]
channel_through_channel()503 fn channel_through_channel() {
504     const COUNT: usize = 1000;
505 
506     type T = Box<dyn Any + Send>;
507 
508     let (s, r) = unbounded::<T>();
509 
510     scope(|scope| {
511         scope.spawn(move |_| {
512             let mut s = s;
513 
514             for _ in 0..COUNT {
515                 let (new_s, new_r) = unbounded();
516                 let new_r: T = Box::new(Some(new_r));
517 
518                 s.send(new_r).unwrap();
519                 s = new_s;
520             }
521         });
522 
523         scope.spawn(move |_| {
524             let mut r = r;
525 
526             for _ in 0..COUNT {
527                 r = r
528                     .recv()
529                     .unwrap()
530                     .downcast_mut::<Option<Receiver<T>>>()
531                     .unwrap()
532                     .take()
533                     .unwrap()
534             }
535         });
536     })
537     .unwrap();
538 }
539