From 698f8c2f01ea549d77d7dc3338a12e04c11057b9 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 17 Apr 2024 14:02:58 +0200 Subject: Adding upstream version 1.64.0+dfsg1. Signed-off-by: Daniel Baumann --- vendor/futures-channel/tests/channel.rs | 66 +++ vendor/futures-channel/tests/mpsc-close.rs | 298 ++++++++++++++ vendor/futures-channel/tests/mpsc.rs | 630 +++++++++++++++++++++++++++++ vendor/futures-channel/tests/oneshot.rs | 252 ++++++++++++ 4 files changed, 1246 insertions(+) create mode 100644 vendor/futures-channel/tests/channel.rs create mode 100644 vendor/futures-channel/tests/mpsc-close.rs create mode 100644 vendor/futures-channel/tests/mpsc.rs create mode 100644 vendor/futures-channel/tests/oneshot.rs (limited to 'vendor/futures-channel/tests') diff --git a/vendor/futures-channel/tests/channel.rs b/vendor/futures-channel/tests/channel.rs new file mode 100644 index 000000000..5f01a8ef4 --- /dev/null +++ b/vendor/futures-channel/tests/channel.rs @@ -0,0 +1,66 @@ +use futures::channel::mpsc; +use futures::executor::block_on; +use futures::future::poll_fn; +use futures::sink::SinkExt; +use futures::stream::StreamExt; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::thread; + +#[test] +fn sequence() { + let (tx, rx) = mpsc::channel(1); + + let amt = 20; + let t = thread::spawn(move || block_on(send_sequence(amt, tx))); + let list: Vec<_> = block_on(rx.collect()); + let mut list = list.into_iter(); + for i in (1..=amt).rev() { + assert_eq!(list.next(), Some(i)); + } + assert_eq!(list.next(), None); + + t.join().unwrap(); +} + +async fn send_sequence(n: u32, mut sender: mpsc::Sender) { + for x in 0..n { + sender.send(n - x).await.unwrap(); + } +} + +#[test] +fn drop_sender() { + let (tx, mut rx) = mpsc::channel::(1); + drop(tx); + let f = poll_fn(|cx| rx.poll_next_unpin(cx)); + assert_eq!(block_on(f), None) +} + +#[test] +fn drop_rx() { + let (mut tx, rx) = mpsc::channel::(1); + block_on(tx.send(1)).unwrap(); + drop(rx); + assert!(block_on(tx.send(1)).is_err()); +} + +#[test] +fn drop_order() { + static DROPS: AtomicUsize = AtomicUsize::new(0); + let (mut tx, rx) = mpsc::channel(1); + + struct A; + + impl Drop for A { + fn drop(&mut self) { + DROPS.fetch_add(1, Ordering::SeqCst); + } + } + + block_on(tx.send(A)).unwrap(); + assert_eq!(DROPS.load(Ordering::SeqCst), 0); + drop(rx); + assert_eq!(DROPS.load(Ordering::SeqCst), 1); + assert!(block_on(tx.send(A)).is_err()); + assert_eq!(DROPS.load(Ordering::SeqCst), 2); +} diff --git a/vendor/futures-channel/tests/mpsc-close.rs b/vendor/futures-channel/tests/mpsc-close.rs new file mode 100644 index 000000000..81203d334 --- /dev/null +++ b/vendor/futures-channel/tests/mpsc-close.rs @@ -0,0 +1,298 @@ +use futures::channel::mpsc; +use futures::executor::block_on; +use futures::future::Future; +use futures::sink::SinkExt; +use futures::stream::StreamExt; +use futures::task::{Context, Poll}; +use std::pin::Pin; +use std::sync::{Arc, Weak}; +use std::thread; +use std::time::{Duration, Instant}; + +#[test] +fn smoke() { + let (mut sender, receiver) = mpsc::channel(1); + + let t = thread::spawn(move || while let Ok(()) = block_on(sender.send(42)) {}); + + // `receiver` needs to be dropped for `sender` to stop sending and therefore before the join. + block_on(receiver.take(3).for_each(|_| futures::future::ready(()))); + + t.join().unwrap() +} + +#[test] +fn multiple_senders_disconnect() { + { + let (mut tx1, mut rx) = mpsc::channel(1); + let (tx2, mut tx3, mut tx4) = (tx1.clone(), tx1.clone(), tx1.clone()); + + // disconnect, dropping and Sink::poll_close should all close this sender but leave the + // channel open for other senders + tx1.disconnect(); + drop(tx2); + block_on(tx3.close()).unwrap(); + + assert!(tx1.is_closed()); + assert!(tx3.is_closed()); + assert!(!tx4.is_closed()); + + block_on(tx4.send(5)).unwrap(); + assert_eq!(block_on(rx.next()), Some(5)); + + // dropping the final sender will close the channel + drop(tx4); + assert_eq!(block_on(rx.next()), None); + } + + { + let (mut tx1, mut rx) = mpsc::unbounded(); + let (tx2, mut tx3, mut tx4) = (tx1.clone(), tx1.clone(), tx1.clone()); + + // disconnect, dropping and Sink::poll_close should all close this sender but leave the + // channel open for other senders + tx1.disconnect(); + drop(tx2); + block_on(tx3.close()).unwrap(); + + assert!(tx1.is_closed()); + assert!(tx3.is_closed()); + assert!(!tx4.is_closed()); + + block_on(tx4.send(5)).unwrap(); + assert_eq!(block_on(rx.next()), Some(5)); + + // dropping the final sender will close the channel + drop(tx4); + assert_eq!(block_on(rx.next()), None); + } +} + +#[test] +fn multiple_senders_close_channel() { + { + let (mut tx1, mut rx) = mpsc::channel(1); + let mut tx2 = tx1.clone(); + + // close_channel should shut down the whole channel + tx1.close_channel(); + + assert!(tx1.is_closed()); + assert!(tx2.is_closed()); + + let err = block_on(tx2.send(5)).unwrap_err(); + assert!(err.is_disconnected()); + + assert_eq!(block_on(rx.next()), None); + } + + { + let (tx1, mut rx) = mpsc::unbounded(); + let mut tx2 = tx1.clone(); + + // close_channel should shut down the whole channel + tx1.close_channel(); + + assert!(tx1.is_closed()); + assert!(tx2.is_closed()); + + let err = block_on(tx2.send(5)).unwrap_err(); + assert!(err.is_disconnected()); + + assert_eq!(block_on(rx.next()), None); + } +} + +#[test] +fn single_receiver_drop_closes_channel_and_drains() { + { + let ref_count = Arc::new(0); + let weak_ref = Arc::downgrade(&ref_count); + + let (sender, receiver) = mpsc::unbounded(); + sender.unbounded_send(ref_count).expect("failed to send"); + + // Verify that the sent message is still live. + assert!(weak_ref.upgrade().is_some()); + + drop(receiver); + + // The sender should know the channel is closed. + assert!(sender.is_closed()); + + // Verify that the sent message has been dropped. + assert!(weak_ref.upgrade().is_none()); + } + + { + let ref_count = Arc::new(0); + let weak_ref = Arc::downgrade(&ref_count); + + let (mut sender, receiver) = mpsc::channel(1); + sender.try_send(ref_count).expect("failed to send"); + + // Verify that the sent message is still live. + assert!(weak_ref.upgrade().is_some()); + + drop(receiver); + + // The sender should know the channel is closed. + assert!(sender.is_closed()); + + // Verify that the sent message has been dropped. + assert!(weak_ref.upgrade().is_none()); + assert!(sender.is_closed()); + } +} + +// Stress test that `try_send()`s occurring concurrently with receiver +// close/drops don't appear as successful sends. +#[test] +fn stress_try_send_as_receiver_closes() { + const AMT: usize = 10000; + // To provide variable timing characteristics (in the hopes of + // reproducing the collision that leads to a race), we busy-re-poll + // the test MPSC receiver a variable number of times before actually + // stopping. We vary this countdown between 1 and the following + // value. + const MAX_COUNTDOWN: usize = 20; + // When we detect that a successfully sent item is still in the + // queue after a disconnect, we spin for up to 100ms to confirm that + // it is a persistent condition and not a concurrency illusion. + const SPIN_TIMEOUT_S: u64 = 10; + const SPIN_SLEEP_MS: u64 = 10; + struct TestRx { + rx: mpsc::Receiver>, + // The number of times to query `rx` before dropping it. + poll_count: usize, + } + struct TestTask { + command_rx: mpsc::Receiver, + test_rx: Option>>, + countdown: usize, + } + impl TestTask { + /// Create a new TestTask + fn new() -> (TestTask, mpsc::Sender) { + let (command_tx, command_rx) = mpsc::channel::(0); + ( + TestTask { + command_rx, + test_rx: None, + countdown: 0, // 0 means no countdown is in progress. + }, + command_tx, + ) + } + } + impl Future for TestTask { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + // Poll the test channel, if one is present. + if let Some(rx) = &mut self.test_rx { + if let Poll::Ready(v) = rx.poll_next_unpin(cx) { + let _ = v.expect("test finished unexpectedly!"); + } + self.countdown -= 1; + // Busy-poll until the countdown is finished. + cx.waker().wake_by_ref(); + } + // Accept any newly submitted MPSC channels for testing. + match self.command_rx.poll_next_unpin(cx) { + Poll::Ready(Some(TestRx { rx, poll_count })) => { + self.test_rx = Some(rx); + self.countdown = poll_count; + cx.waker().wake_by_ref(); + } + Poll::Ready(None) => return Poll::Ready(()), + Poll::Pending => {} + } + if self.countdown == 0 { + // Countdown complete -- drop the Receiver. + self.test_rx = None; + } + Poll::Pending + } + } + let (f, mut cmd_tx) = TestTask::new(); + let bg = thread::spawn(move || block_on(f)); + for i in 0..AMT { + let (mut test_tx, rx) = mpsc::channel(0); + let poll_count = i % MAX_COUNTDOWN; + cmd_tx.try_send(TestRx { rx, poll_count }).unwrap(); + let mut prev_weak: Option> = None; + let mut attempted_sends = 0; + let mut successful_sends = 0; + loop { + // Create a test item. + let item = Arc::new(()); + let weak = Arc::downgrade(&item); + match test_tx.try_send(item) { + Ok(_) => { + prev_weak = Some(weak); + successful_sends += 1; + } + Err(ref e) if e.is_full() => {} + Err(ref e) if e.is_disconnected() => { + // Test for evidence of the race condition. + if let Some(prev_weak) = prev_weak { + if prev_weak.upgrade().is_some() { + // The previously sent item is still allocated. + // However, there appears to be some aspect of the + // concurrency that can legitimately cause the Arc + // to be momentarily valid. Spin for up to 100ms + // waiting for the previously sent item to be + // dropped. + let t0 = Instant::now(); + let mut spins = 0; + loop { + if prev_weak.upgrade().is_none() { + break; + } + assert!( + t0.elapsed() < Duration::from_secs(SPIN_TIMEOUT_S), + "item not dropped on iteration {} after \ + {} sends ({} successful). spin=({})", + i, + attempted_sends, + successful_sends, + spins + ); + spins += 1; + thread::sleep(Duration::from_millis(SPIN_SLEEP_MS)); + } + } + } + break; + } + Err(ref e) => panic!("unexpected error: {}", e), + } + attempted_sends += 1; + } + } + drop(cmd_tx); + bg.join().expect("background thread join"); +} + +#[test] +fn unbounded_try_next_after_none() { + let (tx, mut rx) = mpsc::unbounded::(); + // Drop the sender, close the channel. + drop(tx); + // Receive the end of channel. + assert_eq!(Ok(None), rx.try_next().map_err(|_| ())); + // None received, check we can call `try_next` again. + assert_eq!(Ok(None), rx.try_next().map_err(|_| ())); +} + +#[test] +fn bounded_try_next_after_none() { + let (tx, mut rx) = mpsc::channel::(17); + // Drop the sender, close the channel. + drop(tx); + // Receive the end of channel. + assert_eq!(Ok(None), rx.try_next().map_err(|_| ())); + // None received, check we can call `try_next` again. + assert_eq!(Ok(None), rx.try_next().map_err(|_| ())); +} diff --git a/vendor/futures-channel/tests/mpsc.rs b/vendor/futures-channel/tests/mpsc.rs new file mode 100644 index 000000000..88cdef13d --- /dev/null +++ b/vendor/futures-channel/tests/mpsc.rs @@ -0,0 +1,630 @@ +use futures::channel::{mpsc, oneshot}; +use futures::executor::{block_on, block_on_stream}; +use futures::future::{poll_fn, FutureExt}; +use futures::pin_mut; +use futures::sink::{Sink, SinkExt}; +use futures::stream::{Stream, StreamExt}; +use futures::task::{Context, Poll}; +use futures_test::task::{new_count_waker, noop_context}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; + +trait AssertSend: Send {} +impl AssertSend for mpsc::Sender {} +impl AssertSend for mpsc::Receiver {} + +#[test] +fn send_recv() { + let (mut tx, rx) = mpsc::channel::(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() { + // Run on a task context + block_on(poll_fn(move |cx| { + let (tx, rx) = mpsc::channel::(0); + pin_mut!(tx, rx); + + assert!(tx.as_mut().poll_flush(cx).is_ready()); + assert!(tx.as_mut().poll_ready(cx).is_ready()); + + // Send first message + assert!(tx.as_mut().start_send(1).is_ok()); + assert!(tx.as_mut().poll_ready(cx).is_pending()); + + // poll_ready said Pending, so no room in buffer, therefore new sends + // should get rejected with is_full. + assert!(tx.as_mut().start_send(0).unwrap_err().is_full()); + assert!(tx.as_mut().poll_ready(cx).is_pending()); + + // Take the value + assert_eq!(rx.as_mut().poll_next(cx), Poll::Ready(Some(1))); + assert!(tx.as_mut().poll_ready(cx).is_ready()); + + // Send second message + assert!(tx.as_mut().poll_ready(cx).is_ready()); + assert!(tx.as_mut().start_send(2).is_ok()); + assert!(tx.as_mut().poll_ready(cx).is_pending()); + + // Take the value + assert_eq!(rx.as_mut().poll_next(cx), Poll::Ready(Some(2))); + assert!(tx.as_mut().poll_ready(cx).is_ready()); + + Poll::Ready(()) + })); +} + +#[test] +fn send_shared_recv() { + let (mut tx1, rx) = mpsc::channel::(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::(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::(0); + + let t = thread::spawn(move || { + block_on(tx.send(1)).unwrap(); + block_on(tx.send(2)).unwrap(); + }); + + let v: Vec<_> = block_on(rx.collect()); + assert_eq!(v, vec![1, 2]); + + t.join().unwrap(); +} + +#[test] +fn recv_close_gets_none() { + let (mut tx, mut rx) = mpsc::channel::(10); + + // Run on a task context + 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()), + }; + + Poll::Ready(()) + })); +} + +#[test] +fn tx_close_gets_none() { + let (_, mut rx) = mpsc::channel::(10); + + // Run on a task context + block_on(poll_fn(move |cx| { + assert_eq!(rx.poll_next_unpin(cx), Poll::Ready(None)); + Poll::Ready(()) + })); +} + +// #[test] +// fn spawn_sends_items() { +// let core = local_executor::Core::new(); +// let stream = unfold(0, |i| Some(ok::<_,u8>((i, i + 1)))); +// let rx = mpsc::spawn(stream, &core, 1); +// assert_eq!(core.run(rx.take(4).collect()).unwrap(), +// [0, 1, 2, 3]); +// } + +// #[test] +// fn spawn_kill_dead_stream() { +// use std::thread; +// use std::time::Duration; +// use futures::future::Either; +// use futures::sync::oneshot; +// +// // a stream which never returns anything (maybe a remote end isn't +// // responding), but dropping it leads to observable side effects +// // (like closing connections, releasing limited resources, ...) +// #[derive(Debug)] +// struct Dead { +// // when dropped you should get Err(oneshot::Canceled) on the +// // receiving end +// done: oneshot::Sender<()>, +// } +// impl Stream for Dead { +// type Item = (); +// type Error = (); +// +// fn poll(&mut self) -> Poll, Self::Error> { +// Ok(Poll::Pending) +// } +// } +// +// // need to implement a timeout for the test, as it would hang +// // forever right now +// let (timeout_tx, timeout_rx) = oneshot::channel(); +// thread::spawn(move || { +// thread::sleep(Duration::from_millis(1000)); +// let _ = timeout_tx.send(()); +// }); +// +// let core = local_executor::Core::new(); +// let (done_tx, done_rx) = oneshot::channel(); +// let stream = Dead{done: done_tx}; +// let rx = mpsc::spawn(stream, &core, 1); +// let res = core.run( +// Ok::<_, ()>(()) +// .into_future() +// .then(move |_| { +// // now drop the spawned stream: maybe some timeout exceeded, +// // or some connection on this end was closed by the remote +// // end. +// drop(rx); +// // and wait for the spawned stream to release its resources +// done_rx +// }) +// .select2(timeout_rx) +// ); +// match res { +// Err(Either::A((oneshot::Canceled, _))) => (), +// _ => { +// panic!("dead stream wasn't canceled"); +// }, +// } +// } + +#[test] +fn stress_shared_unbounded() { + const AMT: u32 = 10000; + const NTHREADS: u32 = 8; + let (tx, rx) = mpsc::unbounded::(); + + 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::(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(); +} + +#[allow(clippy::same_item_push)] +#[test] +fn stress_receiver_multi_task_bounded_hard() { + const AMT: usize = 10_000; + const NTHREADS: u32 = 2; + + let (mut tx, rx) = mpsc::channel::(0); + let rx = Arc::new(Mutex::new(Some(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_opt = rx.lock().unwrap(); + if let Some(rx) = &mut *rx_opt { + if i % 5 == 0 { + let item = block_on(rx.next()); + + if item.is_none() { + *rx_opt = None; + break; + } + + n.fetch_add(1, Ordering::Relaxed); + } else { + // Just poll + let n = n.clone(); + match rx.poll_next_unpin(&mut noop_context()) { + Poll::Ready(Some(_)) => { + n.fetch_add(1, Ordering::Relaxed); + } + Poll::Ready(None) => { + *rx_opt = None; + break; + } + Poll::Pending => {} + } + } + } else { + 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)); +} + +/// Stress test that receiver properly receives all the messages +/// after sender dropped. +#[test] +fn stress_drop_sender() { + fn list() -> impl Stream { + 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) { + for i in 1..=3 { + tx.send(i).await.unwrap(); + } +} + +/// Stress test that after receiver dropped, +/// no messages are lost. +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 tx.unbounded_send(i).is_err() { + unwritten_tx.send(i).expect("unwritten_tx"); + return; + } + } + }); + + // Read one message to make sure thread effectively started + 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, count: u32) { + for i in (1..=count).rev() { + sender.send(i).await.unwrap(); + } +} + +/// Tests that after `poll_ready` indicates capacity a channel can always send without waiting. +#[allow(clippy::same_item_push)] +#[test] +fn stress_poll_ready() { + const AMT: u32 = 1000; + const NTHREADS: u32 = 8; + + /// Run a stress test using the specified channel capacity. + 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(); + }); + + let _ = 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(); + + // This should fail + 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(); // should be full + rx.try_next().unwrap(); + rx.try_next().unwrap(); + rx.try_next().unwrap_err(); // should be empty + tx.try_send("hello").unwrap(); + rx.try_next().unwrap(); + rx.try_next().unwrap_err(); // should be empty +} + +#[test] +fn same_receiver() { + let (mut txa1, _) = mpsc::channel::(1); + let txa2 = txa1.clone(); + + let (mut txb1, _) = mpsc::channel::(1); + let txb2 = txb1.clone(); + + assert!(txa1.same_receiver(&txa2)); + assert!(txb1.same_receiver(&txb2)); + assert!(!txa1.same_receiver(&txb1)); + + txa1.disconnect(); + txb1.close_channel(); + + assert!(!txa1.same_receiver(&txa2)); + assert!(txb1.same_receiver(&txb2)); +} + +#[test] +fn is_connected_to() { + let (txa, rxa) = mpsc::channel::(1); + let (txb, rxb) = mpsc::channel::(1); + + assert!(txa.is_connected_to(&rxa)); + assert!(txb.is_connected_to(&rxb)); + assert!(!txa.is_connected_to(&rxb)); + assert!(!txb.is_connected_to(&rxa)); +} + +#[test] +fn hash_receiver() { + use std::collections::hash_map::DefaultHasher; + use std::hash::Hasher; + + let mut hasher_a1 = DefaultHasher::new(); + let mut hasher_a2 = DefaultHasher::new(); + let mut hasher_b1 = DefaultHasher::new(); + let mut hasher_b2 = DefaultHasher::new(); + let (mut txa1, _) = mpsc::channel::(1); + let txa2 = txa1.clone(); + + let (mut txb1, _) = mpsc::channel::(1); + let txb2 = txb1.clone(); + + txa1.hash_receiver(&mut hasher_a1); + let hash_a1 = hasher_a1.finish(); + txa2.hash_receiver(&mut hasher_a2); + let hash_a2 = hasher_a2.finish(); + txb1.hash_receiver(&mut hasher_b1); + let hash_b1 = hasher_b1.finish(); + txb2.hash_receiver(&mut hasher_b2); + let hash_b2 = hasher_b2.finish(); + + assert_eq!(hash_a1, hash_a2); + assert_eq!(hash_b1, hash_b2); + assert!(hash_a1 != hash_b1); + + txa1.disconnect(); + txb1.close_channel(); + + let mut hasher_a1 = DefaultHasher::new(); + let mut hasher_a2 = DefaultHasher::new(); + let mut hasher_b1 = DefaultHasher::new(); + let mut hasher_b2 = DefaultHasher::new(); + + txa1.hash_receiver(&mut hasher_a1); + let hash_a1 = hasher_a1.finish(); + txa2.hash_receiver(&mut hasher_a2); + let hash_a2 = hasher_a2.finish(); + txb1.hash_receiver(&mut hasher_b1); + let hash_b1 = hasher_b1.finish(); + txb2.hash_receiver(&mut hasher_b2); + let hash_b2 = hasher_b2.finish(); + + assert!(hash_a1 != hash_a2); + assert_eq!(hash_b1, hash_b2); +} + +#[test] +fn send_backpressure() { + let (waker, counter) = new_count_waker(); + let mut cx = Context::from_waker(&waker); + + let (mut tx, mut rx) = mpsc::channel(1); + block_on(tx.send(1)).unwrap(); + + let mut task = tx.send(2); + assert_eq!(task.poll_unpin(&mut cx), Poll::Pending); + assert_eq!(counter, 0); + + let item = block_on(rx.next()).unwrap(); + assert_eq!(item, 1); + assert_eq!(counter, 1); + assert_eq!(task.poll_unpin(&mut cx), Poll::Ready(Ok(()))); + + let item = block_on(rx.next()).unwrap(); + assert_eq!(item, 2); +} + +#[test] +fn send_backpressure_multi_senders() { + let (waker, counter) = new_count_waker(); + let mut cx = Context::from_waker(&waker); + + let (mut tx1, mut rx) = mpsc::channel(1); + let mut tx2 = tx1.clone(); + block_on(tx1.send(1)).unwrap(); + + let mut task = tx2.send(2); + assert_eq!(task.poll_unpin(&mut cx), Poll::Pending); + assert_eq!(counter, 0); + + let item = block_on(rx.next()).unwrap(); + assert_eq!(item, 1); + assert_eq!(counter, 1); + assert_eq!(task.poll_unpin(&mut cx), Poll::Ready(Ok(()))); + + let item = block_on(rx.next()).unwrap(); + assert_eq!(item, 2); +} diff --git a/vendor/futures-channel/tests/oneshot.rs b/vendor/futures-channel/tests/oneshot.rs new file mode 100644 index 000000000..979cd8a15 --- /dev/null +++ b/vendor/futures-channel/tests/oneshot.rs @@ -0,0 +1,252 @@ +use futures::channel::oneshot::{self, Sender}; +use futures::executor::block_on; +use futures::future::{poll_fn, FutureExt}; +use futures::task::{Context, Poll}; +use futures_test::task::panic_waker_ref; +use std::sync::mpsc; +use std::thread; + +#[test] +fn smoke_poll() { + let (mut tx, rx) = oneshot::channel::(); + let mut rx = Some(rx); + let f = poll_fn(|cx| { + assert!(tx.poll_canceled(cx).is_pending()); + assert!(tx.poll_canceled(cx).is_pending()); + drop(rx.take()); + assert!(tx.poll_canceled(cx).is_ready()); + assert!(tx.poll_canceled(cx).is_ready()); + Poll::Ready(()) + }); + + block_on(f); +} + +#[test] +fn cancel_notifies() { + let (mut tx, rx) = oneshot::channel::(); + + let t = thread::spawn(move || { + block_on(tx.cancellation()); + }); + drop(rx); + t.join().unwrap(); +} + +#[test] +fn cancel_lots() { + let (tx, rx) = mpsc::channel::<(Sender<_>, mpsc::Sender<_>)>(); + let t = thread::spawn(move || { + for (mut tx, tx2) in rx { + block_on(tx.cancellation()); + tx2.send(()).unwrap(); + } + }); + + for _ in 0..20000 { + let (otx, orx) = oneshot::channel::(); + let (tx2, rx2) = mpsc::channel(); + tx.send((otx, tx2)).unwrap(); + drop(orx); + rx2.recv().unwrap(); + } + drop(tx); + + t.join().unwrap(); +} + +#[test] +fn cancel_after_sender_drop_doesnt_notify() { + let (mut tx, rx) = oneshot::channel::(); + let mut cx = Context::from_waker(panic_waker_ref()); + assert_eq!(tx.poll_canceled(&mut cx), Poll::Pending); + drop(tx); + drop(rx); +} + +#[test] +fn close() { + let (mut tx, mut rx) = oneshot::channel::(); + rx.close(); + block_on(poll_fn(|cx| { + match rx.poll_unpin(cx) { + Poll::Ready(Err(_)) => {} + _ => panic!(), + }; + assert!(tx.poll_canceled(cx).is_ready()); + Poll::Ready(()) + })); +} + +#[test] +fn close_wakes() { + let (mut tx, mut rx) = oneshot::channel::(); + let (tx2, rx2) = mpsc::channel(); + let t = thread::spawn(move || { + rx.close(); + rx2.recv().unwrap(); + }); + block_on(tx.cancellation()); + tx2.send(()).unwrap(); + t.join().unwrap(); +} + +#[test] +fn is_canceled() { + let (tx, rx) = oneshot::channel::(); + assert!(!tx.is_canceled()); + drop(rx); + assert!(tx.is_canceled()); +} + +#[test] +fn cancel_sends() { + let (tx, rx) = mpsc::channel::>(); + let t = thread::spawn(move || { + for otx in rx { + let _ = otx.send(42); + } + }); + + for _ in 0..20000 { + let (otx, mut orx) = oneshot::channel::(); + tx.send(otx).unwrap(); + + orx.close(); + let _ = block_on(orx); + } + + drop(tx); + t.join().unwrap(); +} + +// #[test] +// fn spawn_sends_items() { +// let core = local_executor::Core::new(); +// let future = ok::<_, ()>(1); +// let rx = spawn(future, &core); +// assert_eq!(core.run(rx).unwrap(), 1); +// } +// +// #[test] +// fn spawn_kill_dead_stream() { +// use std::thread; +// use std::time::Duration; +// use futures::future::Either; +// use futures::sync::oneshot; +// +// // a future which never returns anything (forever accepting incoming +// // connections), but dropping it leads to observable side effects +// // (like closing listening sockets, releasing limited resources, +// // ...) +// #[derive(Debug)] +// struct Dead { +// // when dropped you should get Err(oneshot::Canceled) on the +// // receiving end +// done: oneshot::Sender<()>, +// } +// impl Future for Dead { +// type Item = (); +// type Error = (); +// +// fn poll(&mut self) -> Poll { +// Ok(Poll::Pending) +// } +// } +// +// // need to implement a timeout for the test, as it would hang +// // forever right now +// let (timeout_tx, timeout_rx) = oneshot::channel(); +// thread::spawn(move || { +// thread::sleep(Duration::from_millis(1000)); +// let _ = timeout_tx.send(()); +// }); +// +// let core = local_executor::Core::new(); +// let (done_tx, done_rx) = oneshot::channel(); +// let future = Dead{done: done_tx}; +// let rx = spawn(future, &core); +// let res = core.run( +// Ok::<_, ()>(()) +// .into_future() +// .then(move |_| { +// // now drop the spawned future: maybe some timeout exceeded, +// // or some connection on this end was closed by the remote +// // end. +// drop(rx); +// // and wait for the spawned future to release its resources +// done_rx +// }) +// .select2(timeout_rx) +// ); +// match res { +// Err(Either::A((oneshot::Canceled, _))) => (), +// Ok(Either::B(((), _))) => { +// panic!("dead future wasn't canceled (timeout)"); +// }, +// _ => { +// panic!("dead future wasn't canceled (unexpected result)"); +// }, +// } +// } +// +// #[test] +// fn spawn_dont_kill_forgot_dead_stream() { +// use std::thread; +// use std::time::Duration; +// use futures::future::Either; +// use futures::sync::oneshot; +// +// // a future which never returns anything (forever accepting incoming +// // connections), but dropping it leads to observable side effects +// // (like closing listening sockets, releasing limited resources, +// // ...) +// #[derive(Debug)] +// struct Dead { +// // when dropped you should get Err(oneshot::Canceled) on the +// // receiving end +// done: oneshot::Sender<()>, +// } +// impl Future for Dead { +// type Item = (); +// type Error = (); +// +// fn poll(&mut self) -> Poll { +// Ok(Poll::Pending) +// } +// } +// +// // need to implement a timeout for the test, as it would hang +// // forever right now +// let (timeout_tx, timeout_rx) = oneshot::channel(); +// thread::spawn(move || { +// thread::sleep(Duration::from_millis(1000)); +// let _ = timeout_tx.send(()); +// }); +// +// let core = local_executor::Core::new(); +// let (done_tx, done_rx) = oneshot::channel(); +// let future = Dead{done: done_tx}; +// let rx = spawn(future, &core); +// let res = core.run( +// Ok::<_, ()>(()) +// .into_future() +// .then(move |_| { +// // forget the spawned future: should keep running, i.e. hit +// // the timeout below. +// rx.forget(); +// // and wait for the spawned future to release its resources +// done_rx +// }) +// .select2(timeout_rx) +// ); +// match res { +// Err(Either::A((oneshot::Canceled, _))) => { +// panic!("forgotten dead future was canceled"); +// }, +// Ok(Either::B(((), _))) => (), // reached timeout +// _ => { +// panic!("forgotten dead future was canceled (unexpected result)"); +// }, +// } +// } -- cgit v1.2.3