3 files changed, 715 insertions, 0 deletions
diff --git a/third_party/rust/mio-extras/test/mod.rs b/third_party/rust/mio-extras/test/mod.rs
new file mode 100644
index 0000000000..217069466a
--- /dev/null
+++ b/third_party/rust/mio-extras/test/mod.rs
@@ -0,0 +1,45 @@
+extern crate mio;
+extern crate mio_extras;
+
+use mio::event::Event;
+use mio::{Events, Poll};
+use std::time::Duration;
+
+mod test_poll_channel;
+mod test_timer;
+
+pub fn expect_events(
+    poll: &Poll,
+    event_buffer: &mut Events,
+    poll_try_count: usize,
+    mut expected: Vec<Event>,
+) {
+    const MS: u64 = 1_000;
+
+    for _ in 0..poll_try_count {
+        poll.poll(event_buffer, Some(Duration::from_millis(MS)))
+            .unwrap();
+        for event in event_buffer.iter() {
+            let pos_opt = match expected.iter().position(|exp_event| {
+                (event.token() == exp_event.token())
+                    && event.readiness().contains(exp_event.readiness())
+            }) {
+                Some(x) => Some(x),
+                None => None,
+            };
+            if let Some(pos) = pos_opt {
+                expected.remove(pos);
+            }
+        }
+
+        if expected.is_empty() {
+            break;
+        }
+    }
+
+    assert!(
+        expected.is_empty(),
+        "The following expected events were not found: {:?}",
+        expected
+    );
+}
diff --git a/third_party/rust/mio-extras/test/test_poll_channel.rs b/third_party/rust/mio-extras/test/test_poll_channel.rs
new file mode 100644
index 0000000000..7314f26661
--- /dev/null
+++ b/third_party/rust/mio-extras/test/test_poll_channel.rs
@@ -0,0 +1,362 @@
+use crate::expect_events;
+use mio::event::Event;
+use mio::{Events, Poll, PollOpt, Ready, Token};
+use mio_extras::channel;
+use std::sync::mpsc::TryRecvError;
+use std::thread;
+use std::time::Duration;
+
+#[test]
+pub fn test_poll_channel_edge() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let (tx, rx) = channel::channel();
+
+    poll.register(&rx, Token(123), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    // Wait, but nothing should happen
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Push the value
+    tx.send("hello").unwrap();
+
+    // Polling will contain the event
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(1, num);
+
+    let event = events.iter().next().unwrap();
+    assert_eq!(event.token(), Token(123));
+    assert_eq!(event.readiness(), Ready::readable());
+
+    // Poll again and there should be no events
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Read the value
+    assert_eq!("hello", rx.try_recv().unwrap());
+
+    // Poll again, nothing
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Push a value
+    tx.send("goodbye").unwrap();
+
+    // Have an event
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(1, num);
+
+    let event = events.iter().next().unwrap();
+    assert_eq!(event.token(), Token(123));
+    assert_eq!(event.readiness(), Ready::readable());
+
+    // Read the value
+    rx.try_recv().unwrap();
+
+    // Drop the sender half
+    drop(tx);
+
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(1, num);
+
+    let event = events.iter().next().unwrap();
+    assert_eq!(event.token(), Token(123));
+    assert_eq!(event.readiness(), Ready::readable());
+
+    match rx.try_recv() {
+        Err(TryRecvError::Disconnected) => {}
+        no => panic!("unexpected value {:?}", no),
+    }
+}
+
+#[test]
+pub fn test_poll_channel_oneshot() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let (tx, rx) = channel::channel();
+
+    poll.register(
+        &rx,
+        Token(123),
+        Ready::readable(),
+        PollOpt::edge() | PollOpt::oneshot(),
+    )
+    .unwrap();
+
+    // Wait, but nothing should happen
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Push the value
+    tx.send("hello").unwrap();
+
+    // Polling will contain the event
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(1, num);
+
+    let event = events.iter().next().unwrap();
+    assert_eq!(event.token(), Token(123));
+    assert_eq!(event.readiness(), Ready::readable());
+
+    // Poll again and there should be no events
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Read the value
+    assert_eq!("hello", rx.try_recv().unwrap());
+
+    // Poll again, nothing
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Push a value
+    tx.send("goodbye").unwrap();
+
+    // Poll again, nothing
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Reregistering will re-trigger the notification
+    for _ in 0..3 {
+        poll.reregister(
+            &rx,
+            Token(123),
+            Ready::readable(),
+            PollOpt::edge() | PollOpt::oneshot(),
+        )
+        .unwrap();
+
+        // Have an event
+        let num = poll
+            .poll(&mut events, Some(Duration::from_millis(300)))
+            .unwrap();
+        assert_eq!(1, num);
+
+        let event = events.iter().next().unwrap();
+        assert_eq!(event.token(), Token(123));
+        assert_eq!(event.readiness(), Ready::readable());
+    }
+
+    // Get the value
+    assert_eq!("goodbye", rx.try_recv().unwrap());
+
+    poll.reregister(
+        &rx,
+        Token(123),
+        Ready::readable(),
+        PollOpt::edge() | PollOpt::oneshot(),
+    )
+    .unwrap();
+
+    // Have an event
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    poll.reregister(
+        &rx,
+        Token(123),
+        Ready::readable(),
+        PollOpt::edge() | PollOpt::oneshot(),
+    )
+    .unwrap();
+
+    // Have an event
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+}
+
+#[test]
+pub fn test_poll_channel_level() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let (tx, rx) = channel::channel();
+
+    poll.register(&rx, Token(123), Ready::readable(), PollOpt::level())
+        .unwrap();
+
+    // Wait, but nothing should happen
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Push the value
+    tx.send("hello").unwrap();
+
+    // Polling will contain the event
+    for i in 0..5 {
+        let num = poll
+            .poll(&mut events, Some(Duration::from_millis(300)))
+            .unwrap();
+        assert!(1 == num, "actually got {} on iteration {}", num, i);
+
+        let event = events.iter().next().unwrap();
+        assert_eq!(event.token(), Token(123));
+        assert_eq!(event.readiness(), Ready::readable());
+    }
+
+    // Read the value
+    assert_eq!("hello", rx.try_recv().unwrap());
+
+    // Wait, but nothing should happen
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+}
+
+#[test]
+pub fn test_poll_channel_writable() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let (tx, rx) = channel::channel();
+
+    poll.register(&rx, Token(123), Ready::writable(), PollOpt::edge())
+        .unwrap();
+
+    // Wait, but nothing should happen
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+
+    // Push the value
+    tx.send("hello").unwrap();
+
+    // Wait, but nothing should happen
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+}
+
+#[test]
+pub fn test_dropping_receive_before_poll() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let (tx, rx) = channel::channel();
+
+    poll.register(&rx, Token(123), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    // Push the value
+    tx.send("hello").unwrap();
+
+    // Drop the receive end
+    drop(rx);
+
+    // Wait, but nothing should happen
+    let num = poll
+        .poll(&mut events, Some(Duration::from_millis(300)))
+        .unwrap();
+    assert_eq!(0, num);
+}
+
+#[test]
+pub fn test_mixing_channel_with_socket() {
+    use mio::net::{TcpListener, TcpStream};
+
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let (tx, rx) = channel::channel();
+
+    // Create the listener
+    let l = TcpListener::bind(&"127.0.0.1:0".parse().unwrap()).unwrap();
+
+    // Register the listener with `Poll`
+    poll.register(&l, Token(0), Ready::readable(), PollOpt::edge())
+        .unwrap();
+    poll.register(&rx, Token(1), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    // Push a value onto the channel
+    tx.send("hello").unwrap();
+
+    // Connect a TCP socket
+    let s1 = TcpStream::connect(&l.local_addr().unwrap()).unwrap();
+
+    // Register the socket
+    poll.register(&s1, Token(2), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    // Sleep a bit to ensure it arrives at dest
+    thread::sleep(Duration::from_millis(250));
+
+    expect_events(
+        &poll,
+        &mut events,
+        2,
+        vec![
+            Event::new(Ready::empty(), Token(0)),
+            Event::new(Ready::empty(), Token(1)),
+        ],
+    );
+}
+
+#[test]
+pub fn test_sending_from_other_thread_while_polling() {
+    const ITERATIONS: usize = 20;
+    const THREADS: usize = 5;
+
+    // Make sure to run multiple times
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+
+    for _ in 0..ITERATIONS {
+        let (tx, rx) = channel::channel();
+        poll.register(&rx, Token(0), Ready::readable(), PollOpt::edge())
+            .unwrap();
+
+        for _ in 0..THREADS {
+            let tx = tx.clone();
+
+            thread::spawn(move || {
+                thread::sleep(Duration::from_millis(50));
+                tx.send("ping").unwrap();
+            });
+        }
+
+        let mut recv = 0;
+
+        while recv < THREADS {
+            let num = poll.poll(&mut events, None).unwrap();
+
+            if num != 0 {
+                assert_eq!(1, num);
+                assert_eq!(events.iter().next().unwrap().token(), Token(0));
+
+                while let Ok(_) = rx.try_recv() {
+                    recv += 1;
+                }
+            }
+        }
+    }
+}
diff --git a/third_party/rust/mio-extras/test/test_timer.rs b/third_party/rust/mio-extras/test/test_timer.rs
new file mode 100644
index 0000000000..ac49833523
--- /dev/null
+++ b/third_party/rust/mio-extras/test/test_timer.rs
@@ -0,0 +1,308 @@
+use mio::{Events, Poll, PollOpt, Ready, Token};
+use mio_extras::timer::{self, Timer};
+
+use std::thread;
+use std::time::Duration;
+
+#[test]
+fn test_basic_timer_without_poll() {
+    let mut timer = Timer::default();
+
+    // Set the timeout
+    timer.set_timeout(Duration::from_millis(200), "hello");
+
+    // Nothing when polled immediately
+    assert!(timer.poll().is_none());
+
+    // Wait for the timeout
+    thread::sleep(Duration::from_millis(250));
+
+    assert_eq!(Some("hello"), timer.poll());
+    assert!(timer.poll().is_none());
+}
+
+#[test]
+fn test_basic_timer_with_poll_edge_set_timeout_after_register() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let mut timer = Timer::default();
+
+    poll.register(&timer, Token(0), Ready::readable(), PollOpt::edge())
+        .unwrap();
+    timer.set_timeout(Duration::from_millis(200), "hello");
+
+    let elapsed = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+
+        assert_eq!(num, 1);
+        let event = events.iter().next().unwrap();
+        assert_eq!(Token(0), event.token());
+        assert_eq!(Ready::readable(), event.readiness());
+    });
+
+    assert!(is_about(200, elapsed), "actual={:?}", elapsed);
+    assert_eq!("hello", timer.poll().unwrap());
+    assert_eq!(None, timer.poll());
+}
+
+#[test]
+fn test_basic_timer_with_poll_edge_set_timeout_before_register() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let mut timer = Timer::default();
+
+    timer.set_timeout(Duration::from_millis(200), "hello");
+    poll.register(&timer, Token(0), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    let elapsed = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+
+        assert_eq!(num, 1);
+        let event = events.iter().next().unwrap();
+        assert_eq!(Token(0), event.token());
+        assert_eq!(Ready::readable(), event.readiness());
+    });
+
+    assert!(is_about(200, elapsed), "actual={:?}", elapsed);
+    assert_eq!("hello", timer.poll().unwrap());
+    assert_eq!(None, timer.poll());
+}
+
+#[test]
+fn test_setting_later_timeout_then_earlier_one() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let mut timer = Timer::default();
+
+    poll.register(&timer, Token(0), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    timer.set_timeout(Duration::from_millis(600), "hello");
+    timer.set_timeout(Duration::from_millis(200), "world");
+
+    let elapsed = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+
+        assert_eq!(num, 1);
+        let event = events.iter().next().unwrap();
+        assert_eq!(Token(0), event.token());
+        assert_eq!(Ready::readable(), event.readiness());
+    });
+
+    assert!(is_about(200, elapsed), "actual={:?}", elapsed);
+    assert_eq!("world", timer.poll().unwrap());
+    assert_eq!(None, timer.poll());
+
+    let elapsed = self::elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+
+        assert_eq!(num, 1);
+        let event = events.iter().next().unwrap();
+        assert_eq!(Token(0), event.token());
+        assert_eq!(Ready::readable(), event.readiness());
+    });
+
+    assert!(is_about(400, elapsed), "actual={:?}", elapsed);
+    assert_eq!("hello", timer.poll().unwrap());
+    assert_eq!(None, timer.poll());
+}
+
+#[test]
+fn test_timer_with_looping_wheel() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let mut timer = timer::Builder::default().num_slots(2).build();
+
+    poll.register(&timer, Token(0), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    const TOKENS: &[&str] = &["hello", "world", "some", "thing"];
+
+    for (i, msg) in TOKENS.iter().enumerate() {
+        timer.set_timeout(Duration::from_millis(500 * (i as u64 + 1)), msg);
+    }
+
+    for msg in TOKENS {
+        let elapsed = elapsed(|| {
+            let num = poll.poll(&mut events, None).unwrap();
+
+            assert_eq!(num, 1);
+            let event = events.iter().next().unwrap();
+            assert_eq!(Token(0), event.token());
+            assert_eq!(Ready::readable(), event.readiness());
+        });
+
+        assert!(
+            is_about(500, elapsed),
+            "actual={:?}; msg={:?}",
+            elapsed,
+            msg
+        );
+        assert_eq!(Some(msg), timer.poll());
+        assert_eq!(None, timer.poll());
+    }
+}
+
+#[test]
+fn test_edge_without_polling() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let mut timer = Timer::default();
+
+    poll.register(&timer, Token(0), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    timer.set_timeout(Duration::from_millis(400), "hello");
+
+    let ms = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+        assert_eq!(num, 1);
+        let event = events.iter().next().unwrap();
+        assert_eq!(Token(0), event.token());
+        assert_eq!(Ready::readable(), event.readiness());
+    });
+
+    assert!(is_about(400, ms), "actual={:?}", ms);
+
+    let ms = elapsed(|| {
+        let num = poll
+            .poll(&mut events, Some(Duration::from_millis(300)))
+            .unwrap();
+        assert_eq!(num, 0);
+    });
+
+    assert!(is_about(300, ms), "actual={:?}", ms);
+}
+
+#[test]
+fn test_level_triggered() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let mut timer = Timer::default();
+
+    poll.register(&timer, Token(0), Ready::readable(), PollOpt::level())
+        .unwrap();
+
+    timer.set_timeout(Duration::from_millis(400), "hello");
+
+    let ms = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+        assert_eq!(num, 1);
+        let event = events.iter().next().unwrap();
+        assert_eq!(Token(0), event.token());
+        assert_eq!(Ready::readable(), event.readiness());
+    });
+
+    assert!(is_about(400, ms), "actual={:?}", ms);
+
+    let ms = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+        assert_eq!(num, 1);
+        let event = events.iter().next().unwrap();
+        assert_eq!(Token(0), event.token());
+        assert_eq!(Ready::readable(), event.readiness());
+    });
+
+    assert!(is_about(0, ms), "actual={:?}", ms);
+}
+
+#[test]
+fn test_edge_oneshot_triggered() {
+    let poll = Poll::new().unwrap();
+    let mut events = Events::with_capacity(1024);
+    let mut timer = Timer::default();
+
+    poll.register(
+        &timer,
+        Token(0),
+        Ready::readable(),
+        PollOpt::edge() | PollOpt::oneshot(),
+    )
+    .unwrap();
+
+    timer.set_timeout(Duration::from_millis(200), "hello");
+
+    let ms = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+        assert_eq!(num, 1);
+    });
+
+    assert!(is_about(200, ms), "actual={:?}", ms);
+
+    let ms = elapsed(|| {
+        let num = poll
+            .poll(&mut events, Some(Duration::from_millis(300)))
+            .unwrap();
+        assert_eq!(num, 0);
+    });
+
+    assert!(is_about(300, ms), "actual={:?}", ms);
+
+    poll.reregister(
+        &timer,
+        Token(0),
+        Ready::readable(),
+        PollOpt::edge() | PollOpt::oneshot(),
+    )
+    .unwrap();
+
+    let ms = elapsed(|| {
+        let num = poll.poll(&mut events, None).unwrap();
+        assert_eq!(num, 1);
+    });
+
+    assert!(is_about(0, ms));
+}
+
+#[test]
+fn test_cancel_timeout() {
+    use std::time::Instant;
+
+    let mut timer: Timer<u32> = Default::default();
+    let timeout = timer.set_timeout(Duration::from_millis(200), 1);
+    timer.cancel_timeout(&timeout);
+
+    let poll = Poll::new().unwrap();
+    poll.register(&timer, Token(0), Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    let mut events = Events::with_capacity(16);
+
+    let now = Instant::now();
+    let dur = Duration::from_millis(500);
+    let mut i = 0;
+
+    while Instant::now() - now < dur {
+        if i > 10 {
+            panic!("iterated too many times");
+        }
+
+        i += 1;
+
+        let elapsed = Instant::now() - now;
+
+        poll.poll(&mut events, Some(dur - elapsed)).unwrap();
+
+        while let Some(_) = timer.poll() {
+            panic!("did not expect to receive timeout");
+        }
+    }
+}
+
+fn elapsed<F: FnMut()>(mut f: F) -> u64 {
+    use std::time::Instant;
+
+    let now = Instant::now();
+
+    f();
+
+    let elapsed = now.elapsed();
+    elapsed.as_secs() * 1000 + u64::from(elapsed.subsec_millis())
+}
+
+fn is_about(expect: u64, val: u64) -> bool {
+    const WINDOW: i64 = 200;
+
+    ((expect as i64) - (val as i64)).abs() <= WINDOW
+}