[rustc.git] / vendor / crossbeam-channel / tests / after.rs

//! Tests for the after channel flavor.

#[macro_use]
extern crate crossbeam_channel;
extern crate crossbeam_utils;
extern crate rand;

use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::thread;
use std::time::{Duration, Instant};

use crossbeam_channel::{after, Select, TryRecvError};
use crossbeam_utils::thread::scope;

fn ms(ms: u64) -> Duration {
    Duration::from_millis(ms)
}

#[test]
fn fire() {
    let start = Instant::now();
    let r = after(ms(50));

    assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
    thread::sleep(ms(100));

    let fired = r.try_recv().unwrap();
    assert!(start < fired);
    assert!(fired - start >= ms(50));

    let now = Instant::now();
    assert!(fired < now);
    assert!(now - fired >= ms(50));

    assert_eq!(r.try_recv(), Err(TryRecvError::Empty));

    select! {
        recv(r) -> _ => panic!(),
        default => {}
    }

    select! {
        recv(r) -> _ => panic!(),
        recv(after(ms(200))) -> _ => {}
    }
}

#[test]
fn capacity() {
    const COUNT: usize = 10;

    for i in 0..COUNT {
        let r = after(ms(i as u64));
        assert_eq!(r.capacity(), Some(1));
    }
}

#[test]
fn len_empty_full() {
    let r = after(ms(50));

    assert_eq!(r.len(), 0);
    assert_eq!(r.is_empty(), true);
    assert_eq!(r.is_full(), false);

    thread::sleep(ms(100));

    assert_eq!(r.len(), 1);
    assert_eq!(r.is_empty(), false);
    assert_eq!(r.is_full(), true);

    r.try_recv().unwrap();

    assert_eq!(r.len(), 0);
    assert_eq!(r.is_empty(), true);
    assert_eq!(r.is_full(), false);
}

#[test]
fn try_recv() {
    let r = after(ms(200));
    assert!(r.try_recv().is_err());

    thread::sleep(ms(100));
    assert!(r.try_recv().is_err());

    thread::sleep(ms(200));
    assert!(r.try_recv().is_ok());
    assert!(r.try_recv().is_err());

    thread::sleep(ms(200));
    assert!(r.try_recv().is_err());
}

#[test]
fn recv() {
    let start = Instant::now();
    let r = after(ms(50));

    let fired = r.recv().unwrap();
    assert!(start < fired);
    assert!(fired - start >= ms(50));

    let now = Instant::now();
    assert!(fired < now);
    assert!(now - fired < fired - start);

    assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
}

#[test]
fn recv_timeout() {
    let start = Instant::now();
    let r = after(ms(200));

    assert!(r.recv_timeout(ms(100)).is_err());
    let now = Instant::now();
    assert!(now - start >= ms(100));
    assert!(now - start <= ms(150));

    let fired = r.recv_timeout(ms(200)).unwrap();
    assert!(fired - start >= ms(200));
    assert!(fired - start <= ms(250));

    assert!(r.recv_timeout(ms(200)).is_err());
    let now = Instant::now();
    assert!(now - start >= ms(400));
    assert!(now - start <= ms(450));

    assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
}

#[test]
fn recv_two() {
    let r1 = after(ms(50));
    let r2 = after(ms(50));

    scope(|scope| {
        scope.spawn(|_| {
            select! {
                recv(r1) -> _ => {}
                recv(r2) -> _ => {}
            }
        });
        scope.spawn(|_| {
            select! {
                recv(r1) -> _ => {}
                recv(r2) -> _ => {}
            }
        });
    })
    .unwrap();
}

#[test]
fn recv_race() {
    select! {
        recv(after(ms(50))) -> _ => {}
        recv(after(ms(100))) -> _ => panic!(),
    }

    select! {
        recv(after(ms(100))) -> _ => panic!(),
        recv(after(ms(50))) -> _ => {}
    }
}

#[test]
fn stress_default() {
    const COUNT: usize = 10;

    for _ in 0..COUNT {
        select! {
            recv(after(ms(0))) -> _ => {}
            default => panic!(),
        }
    }

    for _ in 0..COUNT {
        select! {
            recv(after(ms(100))) -> _ => panic!(),
            default => {}
        }
    }
}

#[test]
fn select() {
    const THREADS: usize = 4;
    const COUNT: usize = 1000;
    const TIMEOUT_MS: u64 = 100;

    let v = (0..COUNT)
        .map(|i| after(ms(i as u64 / TIMEOUT_MS / 2)))
        .collect::<Vec<_>>();
    let hits = AtomicUsize::new(0);

    scope(|scope| {
        for _ in 0..THREADS {
            scope.spawn(|_| {
                let v: Vec<&_> = v.iter().collect();

                loop {
                    let timeout = after(ms(TIMEOUT_MS));
                    let mut sel = Select::new();
                    for r in &v {
                        sel.recv(r);
                    }
                    let oper_timeout = sel.recv(&timeout);

                    let oper = sel.select();
                    match oper.index() {
                        i if i == oper_timeout => {
                            oper.recv(&timeout).unwrap();
                            break;
                        }
                        i => {
                            oper.recv(&v[i]).unwrap();
                            hits.fetch_add(1, Ordering::SeqCst);
                        }
                    }
                }
            });
        }
    })
    .unwrap();

    assert_eq!(hits.load(Ordering::SeqCst), COUNT);
}

#[test]
fn ready() {
    const THREADS: usize = 4;
    const COUNT: usize = 1000;
    const TIMEOUT_MS: u64 = 100;

    let v = (0..COUNT)
        .map(|i| after(ms(i as u64 / TIMEOUT_MS / 2)))
        .collect::<Vec<_>>();
    let hits = AtomicUsize::new(0);

    scope(|scope| {
        for _ in 0..THREADS {
            scope.spawn(|_| {
                let v: Vec<&_> = v.iter().collect();

                loop {
                    let timeout = after(ms(TIMEOUT_MS));
                    let mut sel = Select::new();
                    for r in &v {
                        sel.recv(r);
                    }
                    let oper_timeout = sel.recv(&timeout);

                    loop {
                        let i = sel.ready();
                        if i == oper_timeout {
                            timeout.try_recv().unwrap();
                            return;
                        } else if v[i].try_recv().is_ok() {
                            hits.fetch_add(1, Ordering::SeqCst);
                            break;
                        }
                    }
                }
            });
        }
    })
    .unwrap();

    assert_eq!(hits.load(Ordering::SeqCst), COUNT);
}

#[test]
fn stress_clone() {
    const RUNS: usize = 1000;
    const THREADS: usize = 10;
    const COUNT: usize = 50;

    for i in 0..RUNS {
        let r = after(ms(i as u64));

        scope(|scope| {
            for _ in 0..THREADS {
                scope.spawn(|_| {
                    let r = r.clone();
                    let _ = r.try_recv();

                    for _ in 0..COUNT {
                        drop(r.clone());
                        thread::yield_now();
                    }
                });
            }
        })
        .unwrap();
    }
}

#[test]
fn fairness() {
    const COUNT: usize = 1000;

    for &dur in &[0, 1] {
        let mut hits = [0usize; 2];

        for _ in 0..COUNT {
            select! {
                recv(after(ms(dur))) -> _ => hits[0] += 1,
                recv(after(ms(dur))) -> _ => hits[1] += 1,
            }
        }

        assert!(hits.iter().all(|x| *x >= COUNT / hits.len() / 2));
    }
}

#[test]
fn fairness_duplicates() {
    const COUNT: usize = 1000;

    for &dur in &[0, 1] {
        let mut hits = [0usize; 5];

        for _ in 0..COUNT {
            let r = after(ms(dur));
            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));
    }
}
Commit	Line	Data
dfeec247 XL	1	//! Tests for the after channel flavor.
	2
	3	#[macro_use]
	4	extern crate crossbeam_channel;
	5	extern crate crossbeam_utils;
	6	extern crate rand;
	7
	8	use std::sync::atomic::AtomicUsize;
	9	use std::sync::atomic::Ordering;
	10	use std::thread;
	11	use std::time::{Duration, Instant};
	12
	13	use crossbeam_channel::{after, Select, TryRecvError};
	14	use crossbeam_utils::thread::scope;
	15
	16	fn ms(ms: u64) -> Duration {
	17	Duration::from_millis(ms)
	18	}
	19
	20	#[test]
	21	fn fire() {
	22	let start = Instant::now();
	23	let r = after(ms(50));
	24
	25	assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
	26	thread::sleep(ms(100));
	27
	28	let fired = r.try_recv().unwrap();
	29	assert!(start < fired);
	30	assert!(fired - start >= ms(50));
	31
	32	let now = Instant::now();
	33	assert!(fired < now);
	34	assert!(now - fired >= ms(50));
	35
	36	assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
	37
	38	select! {
	39	recv(r) -> _ => panic!(),
	40	default => {}
	41	}
	42
	43	select! {
	44	recv(r) -> _ => panic!(),
	45	recv(after(ms(200))) -> _ => {}
	46	}
	47	}
	48
	49	#[test]
	50	fn capacity() {
	51	const COUNT: usize = 10;
	52
	53	for i in 0..COUNT {
	54	let r = after(ms(i as u64));
	55	assert_eq!(r.capacity(), Some(1));
	56	}
	57	}
	58
	59	#[test]
	60	fn len_empty_full() {
	61	let r = after(ms(50));
	62
	63	assert_eq!(r.len(), 0);
	64	assert_eq!(r.is_empty(), true);
65	assert_eq!(r.is_full(), false);
66
67	thread::sleep(ms(100));
68
69	assert_eq!(r.len(), 1);
70	assert_eq!(r.is_empty(), false);
71	assert_eq!(r.is_full(), true);
72
73	r.try_recv().unwrap();
74
75	assert_eq!(r.len(), 0);
76	assert_eq!(r.is_empty(), true);
77	assert_eq!(r.is_full(), false);
78	}
79
80	#[test]
81	fn try_recv() {
82	let r = after(ms(200));
83	assert!(r.try_recv().is_err());
84
85	thread::sleep(ms(100));
86	assert!(r.try_recv().is_err());
87
88	thread::sleep(ms(200));
89	assert!(r.try_recv().is_ok());
90	assert!(r.try_recv().is_err());
91
92	thread::sleep(ms(200));
93	assert!(r.try_recv().is_err());
94	}
95
96	#[test]
97	fn recv() {
98	let start = Instant::now();
99	let r = after(ms(50));
100
101	let fired = r.recv().unwrap();
102	assert!(start < fired);
103	assert!(fired - start >= ms(50));
104
105	let now = Instant::now();
106	assert!(fired < now);
107	assert!(now - fired < fired - start);
108
109	assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
110	}
111
112	#[test]
113	fn recv_timeout() {
114	let start = Instant::now();
115	let r = after(ms(200));
116
117	assert!(r.recv_timeout(ms(100)).is_err());
118	let now = Instant::now();
119	assert!(now - start >= ms(100));
120	assert!(now - start <= ms(150));
121
122	let fired = r.recv_timeout(ms(200)).unwrap();
123	assert!(fired - start >= ms(200));
124	assert!(fired - start <= ms(250));
125
126	assert!(r.recv_timeout(ms(200)).is_err());
127	let now = Instant::now();
128	assert!(now - start >= ms(400));
129	assert!(now - start <= ms(450));
130
131	assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
132	}
133
134	#[test]
135	fn recv_two() {
136	let r1 = after(ms(50));
137	let r2 = after(ms(50));
138
139	scope(\|scope\| {
140	scope.spawn(\|_\| {
141	select! {
142	recv(r1) -> _ => {}
143	recv(r2) -> _ => {}
144	}
145	});
146	scope.spawn(\|_\| {
147	select! {
148	recv(r1) -> _ => {}
149	recv(r2) -> _ => {}
150	}
151	});
74b04a01 XL	152	})
74b04a01 XL	153	.unwrap();
dfeec247 XL	154	}
	155
	156	#[test]
	157	fn recv_race() {
	158	select! {
	159	recv(after(ms(50))) -> _ => {}
	160	recv(after(ms(100))) -> _ => panic!(),
	161	}
	162
	163	select! {
	164	recv(after(ms(100))) -> _ => panic!(),
	165	recv(after(ms(50))) -> _ => {}
	166	}
	167	}
	168
	169	#[test]
	170	fn stress_default() {
	171	const COUNT: usize = 10;
	172
	173	for _ in 0..COUNT {
	174	select! {
	175	recv(after(ms(0))) -> _ => {}
	176	default => panic!(),
	177	}
	178	}
	179
	180	for _ in 0..COUNT {
	181	select! {
	182	recv(after(ms(100))) -> _ => panic!(),
	183	default => {}
	184	}
	185	}
	186	}
	187
	188	#[test]
	189	fn select() {
	190	const THREADS: usize = 4;
	191	const COUNT: usize = 1000;
	192	const TIMEOUT_MS: u64 = 100;
	193
	194	let v = (0..COUNT)
	195	.map(\|i\| after(ms(i as u64 / TIMEOUT_MS / 2)))
	196	.collect::<Vec<_>>();
	197	let hits = AtomicUsize::new(0);
	198
	199	scope(\|scope\| {
	200	for _ in 0..THREADS {
	201	scope.spawn(\|_\| {
	202	let v: Vec<&_> = v.iter().collect();
	203
	204	loop {
	205	let timeout = after(ms(TIMEOUT_MS));
	206	let mut sel = Select::new();
	207	for r in &v {
	208	sel.recv(r);
	209	}
	210	let oper_timeout = sel.recv(&timeout);
	211
	212	let oper = sel.select();
	213	match oper.index() {
	214	i if i == oper_timeout => {
	215	oper.recv(&timeout).unwrap();
	216	break;
	217	}
218	i => {
219	oper.recv(&v[i]).unwrap();
220	hits.fetch_add(1, Ordering::SeqCst);
221	}
222	}
223	}
224	});
225	}
74b04a01 XL	226	})
74b04a01 XL	227	.unwrap();
dfeec247 XL	228
	229	assert_eq!(hits.load(Ordering::SeqCst), COUNT);
	230	}
	231
	232	#[test]
	233	fn ready() {
	234	const THREADS: usize = 4;
	235	const COUNT: usize = 1000;
	236	const TIMEOUT_MS: u64 = 100;
	237
	238	let v = (0..COUNT)
	239	.map(\|i\| after(ms(i as u64 / TIMEOUT_MS / 2)))
	240	.collect::<Vec<_>>();
	241	let hits = AtomicUsize::new(0);
	242
	243	scope(\|scope\| {
	244	for _ in 0..THREADS {
	245	scope.spawn(\|_\| {
	246	let v: Vec<&_> = v.iter().collect();
	247
	248	loop {
	249	let timeout = after(ms(TIMEOUT_MS));
	250	let mut sel = Select::new();
	251	for r in &v {
	252	sel.recv(r);
	253	}
	254	let oper_timeout = sel.recv(&timeout);
	255
	256	loop {
	257	let i = sel.ready();
	258	if i == oper_timeout {
	259	timeout.try_recv().unwrap();
	260	return;
	261	} else if v[i].try_recv().is_ok() {
	262	hits.fetch_add(1, Ordering::SeqCst);
	263	break;
	264	}
	265	}
	266	}
	267	});
	268	}
74b04a01 XL	269	})
74b04a01 XL	270	.unwrap();
dfeec247 XL	271
	272	assert_eq!(hits.load(Ordering::SeqCst), COUNT);
	273	}
	274
	275	#[test]
	276	fn stress_clone() {
	277	const RUNS: usize = 1000;
	278	const THREADS: usize = 10;
	279	const COUNT: usize = 50;
	280
	281	for i in 0..RUNS {
	282	let r = after(ms(i as u64));
	283
	284	scope(\|scope\| {
	285	for _ in 0..THREADS {
	286	scope.spawn(\|_\| {
	287	let r = r.clone();
	288	let _ = r.try_recv();
	289
	290	for _ in 0..COUNT {
	291	drop(r.clone());
	292	thread::yield_now();
	293	}
	294	});
	295	}
74b04a01 XL	296	})
74b04a01 XL	297	.unwrap();
dfeec247 XL	298	}
	299	}
	300
	301	#[test]
	302	fn fairness() {
	303	const COUNT: usize = 1000;
	304
	305	for &dur in &[0, 1] {
	306	let mut hits = [0usize; 2];
	307
	308	for _ in 0..COUNT {
	309	select! {
	310	recv(after(ms(dur))) -> _ => hits[0] += 1,
	311	recv(after(ms(dur))) -> _ => hits[1] += 1,
	312	}
	313	}
	314
	315	assert!(hits.iter().all(\|x\| *x >= COUNT / hits.len() / 2));
	316	}
	317	}
	318
	319	#[test]
	320	fn fairness_duplicates() {
	321	const COUNT: usize = 1000;
	322
	323	for &dur in &[0, 1] {
	324	let mut hits = [0usize; 5];
	325
	326	for _ in 0..COUNT {
	327	let r = after(ms(dur));
	328	select! {
	329	recv(r) -> _ => hits[0] += 1,
	330	recv(r) -> _ => hits[1] += 1,
	331	recv(r) -> _ => hits[2] += 1,
	332	recv(r) -> _ => hits[3] += 1,
	333	recv(r) -> _ => hits[4] += 1,
	334	}
	335	}
	336
	337	assert!(hits.iter().all(\|x\| *x >= COUNT / hits.len() / 2));
	338	}
	339	}