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<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