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");
}