| extern crate futures; |
| |
| use std::sync::{Arc, Weak}; |
| use std::thread; |
| use std::time::{Duration, Instant}; |
| |
| use futures::prelude::*; |
| use futures::sync::mpsc::*; |
| use futures::task; |
| |
| #[test] |
| fn smoke() { |
| let (mut sender, receiver) = channel(1); |
| |
| let t = thread::spawn(move ||{ |
| while let Ok(s) = sender.send(42).wait() { |
| sender = s; |
| } |
| }); |
| |
| receiver.take(3).for_each(|_| Ok(())).wait().unwrap(); |
| |
| t.join().unwrap() |
| } |
| |
| // 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: Receiver<Arc<()>>, |
| // The number of times to query `rx` before dropping it. |
| poll_count: usize |
| } |
| struct TestTask { |
| command_rx: Receiver<TestRx>, |
| test_rx: Option<Receiver<Arc<()>>>, |
| countdown: usize, |
| } |
| impl TestTask { |
| /// Create a new TestTask |
| fn new() -> (TestTask, Sender<TestRx>) { |
| let (command_tx, command_rx) = channel::<TestRx>(0); |
| ( |
| TestTask { |
| command_rx: command_rx, |
| test_rx: None, |
| countdown: 0, // 0 means no countdown is in progress. |
| }, |
| command_tx, |
| ) |
| } |
| } |
| impl Future for TestTask { |
| type Item = (); |
| type Error = (); |
| fn poll(&mut self) -> Poll<(), ()> { |
| // Poll the test channel, if one is present. |
| if let Some(ref mut rx) = self.test_rx { |
| if let Ok(Async::Ready(v)) = rx.poll() { |
| let _ = v.expect("test finished unexpectedly!"); |
| } |
| self.countdown -= 1; |
| // Busy-poll until the countdown is finished. |
| task::current().notify(); |
| } |
| // Accept any newly submitted MPSC channels for testing. |
| match self.command_rx.poll()? { |
| Async::Ready(Some(TestRx { rx, poll_count })) => { |
| self.test_rx = Some(rx); |
| self.countdown = poll_count; |
| task::current().notify(); |
| }, |
| Async::Ready(None) => return Ok(Async::Ready(())), |
| _ => {}, |
| } |
| if self.countdown == 0 { |
| // Countdown complete -- drop the Receiver. |
| self.test_rx = None; |
| } |
| Ok(Async::NotReady) |
| } |
| } |
| let (f, mut cmd_tx) = TestTask::new(); |
| let bg = thread::spawn(move || f.wait()); |
| for i in 0..AMT { |
| let (mut test_tx, rx) = channel(0); |
| let poll_count = i % MAX_COUNTDOWN; |
| cmd_tx.try_send(TestRx { rx: rx, poll_count: poll_count }).unwrap(); |
| let mut prev_weak: Option<Weak<()>> = 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") |
| .expect("background thread result"); |
| } |