#![feature(futures_api, async_await, await_macro, pin)]
use futures::channel::{mpsc, oneshot};
use futures::executor::{block_on, block_on_stream};
use futures::future::{FutureExt, poll_fn};
use futures::stream::{Stream, StreamExt};
use futures::sink::{Sink, SinkExt};
use futures::task::Poll;
use pin_utils::pin_mut;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
trait AssertSend: Send {}
impl AssertSend for mpsc::Sender<i32> {}
impl AssertSend for mpsc::Receiver<i32> {}
#[test]
fn send_recv() {
let (mut tx, rx) = mpsc::channel::<i32>(16);
block_on(tx.send(1)).unwrap();
drop(tx);
let v: Vec<_> = block_on(rx.collect());
assert_eq!(v, vec![1]);
}
#[test]
fn send_recv_no_buffer() {
block_on(poll_fn(move |cx| {
let (tx, rx) = mpsc::channel::<i32>(0);
pin_mut!(tx, rx);
assert!(tx.reborrow().poll_flush(cx).is_ready());
assert!(tx.reborrow().poll_ready(cx).is_ready());
assert!(tx.reborrow().start_send(1).is_ok());
assert!(tx.reborrow().poll_ready(cx).is_pending());
assert!(tx.reborrow().start_send(0).unwrap_err().is_full());
assert!(tx.reborrow().poll_ready(cx).is_pending());
assert_eq!(rx.reborrow().poll_next(cx), Poll::Ready(Some(1)));
assert!(tx.reborrow().poll_ready(cx).is_ready());
assert!(tx.reborrow().poll_ready(cx).is_ready());
assert!(tx.reborrow().start_send(2).is_ok());
assert!(tx.reborrow().poll_ready(cx).is_pending());
assert_eq!(rx.reborrow().poll_next(cx), Poll::Ready(Some(2)));
assert!(tx.reborrow().poll_ready(cx).is_ready());
Poll::Ready(())
}));
}
#[test]
fn send_shared_recv() {
let (mut tx1, rx) = mpsc::channel::<i32>(16);
let mut rx = block_on_stream(rx);
let mut tx2 = tx1.clone();
block_on(tx1.send(1)).unwrap();
assert_eq!(rx.next(), Some(1));
block_on(tx2.send(2)).unwrap();
assert_eq!(rx.next(), Some(2));
}
#[test]
fn send_recv_threads() {
let (mut tx, rx) = mpsc::channel::<i32>(16);
let t = thread::spawn(move|| {
block_on(tx.send(1)).unwrap();
});
let v: Vec<_> = block_on(rx.take(1).collect());
assert_eq!(v, vec![1]);
t.join().unwrap();
}
#[test]
fn send_recv_threads_no_capacity() {
let (mut tx, rx) = mpsc::channel::<i32>(0);
let mut rx = block_on_stream(rx);
let (readytx, readyrx) = mpsc::channel::<()>(2);
let mut readyrx = block_on_stream(readyrx);
let t = thread::spawn(move || {
let mut readytx = readytx.sink_map_err(|_| panic!());
let (send_res_1, send_res_2) = block_on(tx.send(1).join(readytx.send(())));
send_res_1.unwrap();
send_res_2.unwrap();
block_on(tx.send(2).join(readytx.send(())));
});
readyrx.next();
assert_eq!(rx.next(), Some(1));
readyrx.next();
drop(readyrx);
assert_eq!(rx.next(), Some(2));
drop(rx);
t.join().unwrap();
}
#[test]
fn recv_close_gets_none() {
let (mut tx, mut rx) = mpsc::channel::<i32>(10);
block_on(poll_fn(move |cx| {
rx.close();
assert_eq!(rx.poll_next_unpin(cx), Poll::Ready(None));
match tx.poll_ready(cx) {
Poll::Pending | Poll::Ready(Ok(_)) => panic!(),
Poll::Ready(Err(e)) => assert!(e.is_disconnected()),
};
drop(&tx);
Poll::Ready(())
}));
}
#[test]
fn tx_close_gets_none() {
let (_, mut rx) = mpsc::channel::<i32>(10);
block_on(poll_fn(move |cx| {
assert_eq!(rx.poll_next_unpin(cx), Poll::Ready(None));
assert_eq!(rx.poll_next_unpin(cx), Poll::Ready(None));
Poll::Ready(())
}));
}
#[test]
fn stress_shared_unbounded() {
const AMT: u32 = 10000;
const NTHREADS: u32 = 8;
let (tx, rx) = mpsc::unbounded::<i32>();
let t = thread::spawn(move|| {
let result: Vec<_> = block_on(rx.collect());
assert_eq!(result.len(), (AMT * NTHREADS) as usize);
for item in result {
assert_eq!(item, 1);
}
});
for _ in 0..NTHREADS {
let tx = tx.clone();
thread::spawn(move|| {
for _ in 0..AMT {
tx.unbounded_send(1).unwrap();
}
});
}
drop(tx);
t.join().ok().unwrap();
}
#[test]
fn stress_shared_bounded_hard() {
const AMT: u32 = 10000;
const NTHREADS: u32 = 8;
let (tx, rx) = mpsc::channel::<i32>(0);
let t = thread::spawn(move|| {
let result: Vec<_> = block_on(rx.collect());
assert_eq!(result.len(), (AMT * NTHREADS) as usize);
for item in result {
assert_eq!(item, 1);
}
});
for _ in 0..NTHREADS {
let mut tx = tx.clone();
thread::spawn(move || {
for _ in 0..AMT {
block_on(tx.send(1)).unwrap();
}
});
}
drop(tx);
t.join().unwrap();
}
#[test]
fn stress_receiver_multi_task_bounded_hard() {
const AMT: usize = 10_000;
const NTHREADS: u32 = 2;
let (mut tx, rx) = mpsc::channel::<usize>(0);
let rx = Arc::new(Mutex::new(rx));
let n = Arc::new(AtomicUsize::new(0));
let mut th = vec![];
for _ in 0..NTHREADS {
let rx = rx.clone();
let n = n.clone();
let t = thread::spawn(move || {
let mut i = 0;
loop {
i += 1;
let mut rx = rx.lock().unwrap();
if i % 5 == 0 {
let item = block_on(rx.next());
if item.is_none() {
break;
}
n.fetch_add(1, Ordering::Relaxed);
} else {
let n = n.clone();
let f = poll_fn(move |cx| {
let r = match rx.poll_next_unpin(cx) {
Poll::Ready(Some(_)) => {
n.fetch_add(1, Ordering::Relaxed);
false
}
Poll::Ready(None) => true,
Poll::Pending => false,
};
Poll::Ready(r)
});
if block_on(f) {
break;
}
}
}
});
th.push(t);
}
for i in 0..AMT {
block_on(tx.send(i)).unwrap();
}
drop(tx);
for t in th {
t.join().unwrap();
}
assert_eq!(AMT, n.load(Ordering::Relaxed));
}
#[test]
fn stress_drop_sender() {
fn list() -> impl Stream<Item=i32> {
let (tx, rx) = mpsc::channel(1);
thread::spawn(move || {
block_on(send_one_two_three(tx));
});
rx
}
for _ in 0..10000 {
let v: Vec<_> = block_on(list().collect());
assert_eq!(v, vec![1, 2, 3]);
}
}
async fn send_one_two_three(mut tx: mpsc::Sender<i32>) {
for i in 1..=3 {
await!(tx.send(i)).unwrap();
}
}
fn stress_close_receiver_iter() {
let (tx, rx) = mpsc::unbounded();
let mut rx = block_on_stream(rx);
let (unwritten_tx, unwritten_rx) = std::sync::mpsc::channel();
let th = thread::spawn(move || {
for i in 1.. {
if let Err(_) = tx.unbounded_send(i) {
unwritten_tx.send(i).expect("unwritten_tx");
return;
}
}
});
assert_eq!(Some(1), rx.next());
rx.close();
for i in 2.. {
match rx.next() {
Some(r) => assert!(i == r),
None => {
let unwritten = unwritten_rx.recv().expect("unwritten_rx");
assert_eq!(unwritten, i);
th.join().unwrap();
return;
}
}
}
}
#[test]
fn stress_close_receiver() {
for _ in 0..10000 {
stress_close_receiver_iter();
}
}
async fn stress_poll_ready_sender(mut sender: mpsc::Sender<u32>, count: u32) {
for i in (1..=count).rev() {
await!(sender.send(i)).unwrap();
}
}
#[test]
fn stress_poll_ready() {
const AMT: u32 = 1000;
const NTHREADS: u32 = 8;
fn stress(capacity: usize) {
let (tx, rx) = mpsc::channel(capacity);
let mut threads = Vec::new();
for _ in 0..NTHREADS {
let sender = tx.clone();
threads.push(thread::spawn(move || {
block_on(stress_poll_ready_sender(sender, AMT))
}));
}
drop(tx);
let result: Vec<_> = block_on(rx.collect());
assert_eq!(result.len() as u32, AMT * NTHREADS);
for thread in threads {
thread.join().unwrap();
}
}
stress(0);
stress(1);
stress(8);
stress(16);
}
#[test]
fn try_send_1() {
const N: usize = 3000;
let (mut tx, rx) = mpsc::channel(0);
let t = thread::spawn(move || {
for i in 0..N {
loop {
if tx.try_send(i).is_ok() {
break
}
}
}
});
let result: Vec<_> = block_on(rx.collect());
for (i, j) in result.into_iter().enumerate() {
assert_eq!(i, j);
}
t.join().unwrap();
}
#[test]
fn try_send_2() {
let (mut tx, rx) = mpsc::channel(0);
let mut rx = block_on_stream(rx);
tx.try_send("hello").unwrap();
let (readytx, readyrx) = oneshot::channel::<()>();
let th = thread::spawn(move || {
block_on(poll_fn(|cx| {
assert!(tx.poll_ready(cx).is_pending());
Poll::Ready(())
}));
drop(readytx);
block_on(tx.send("goodbye")).unwrap();
});
drop(block_on(readyrx));
assert_eq!(rx.next(), Some("hello"));
assert_eq!(rx.next(), Some("goodbye"));
assert_eq!(rx.next(), None);
th.join().unwrap();
}
#[test]
fn try_send_fail() {
let (mut tx, rx) = mpsc::channel(0);
let mut rx = block_on_stream(rx);
tx.try_send("hello").unwrap();
assert!(tx.try_send("fail").is_err());
assert_eq!(rx.next(), Some("hello"));
tx.try_send("goodbye").unwrap();
drop(tx);
assert_eq!(rx.next(), Some("goodbye"));
assert_eq!(rx.next(), None);
}
#[test]
fn try_send_recv() {
let (mut tx, mut rx) = mpsc::channel(1);
tx.try_send("hello").unwrap();
tx.try_send("hello").unwrap();
tx.try_send("hello").unwrap_err(); rx.try_next().unwrap();
rx.try_next().unwrap();
rx.try_next().unwrap_err(); tx.try_send("hello").unwrap();
rx.try_next().unwrap();
rx.try_next().unwrap_err(); }