Run Miri on CI

.github/workflows/ci.yml

@@ -1,5 +1,8 @@
 name: CI
 
+permissions:
+  contents: read
+
 on:
   pull_request:
   push:
@@ -10,8 +13,12 @@ on:
     - cron: '0 1 * * *'
 
 env:
-  RUSTFLAGS: -D warnings
+  CARGO_INCREMENTAL: 0
+  CARGO_NET_RETRY: 10
+  CARGO_TERM_COLOR: always
   RUST_BACKTRACE: 1
+  RUSTFLAGS: -D warnings
+  RUSTUP_MAX_RETRIES: 10
 
 defaults:
   run:
@@ -229,6 +236,16 @@ jobs:
       - run: ci/no_atomic_cas.sh
       - run: git diff --exit-code
 
+  miri:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install Rust
+        run: rustup toolchain install nightly --component miri && rustup default nightly
+      - run: cargo miri test --workspace --no-fail-fast
+        # env:
+        #   MIRIFLAGS: -Zmiri-tag-raw-pointers
+
   san:
     name: cargo test -Z sanitizer=${{ matrix.sanitizer }}
     strategy:
@@ -261,6 +278,7 @@ jobs:
       - run: cargo clippy --workspace --all-features --all-targets
 
   fmt:
+    name: cargo fmt
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2

futures-channel/tests/mpsc.rs

@@ -200,6 +200,9 @@ fn tx_close_gets_none() {
 
 #[test]
 fn stress_shared_unbounded() {
+    #[cfg(miri)]
+    const AMT: u32 = 100;
+    #[cfg(not(miri))]
     const AMT: u32 = 10000;
     const NTHREADS: u32 = 8;
     let (tx, rx) = mpsc::unbounded::<i32>();
@@ -229,6 +232,9 @@ fn stress_shared_unbounded() {
 
 #[test]
 fn stress_shared_bounded_hard() {
+    #[cfg(miri)]
+    const AMT: u32 = 100;
+    #[cfg(not(miri))]
     const AMT: u32 = 10000;
     const NTHREADS: u32 = 8;
     let (tx, rx) = mpsc::channel::<i32>(0);
@@ -259,6 +265,9 @@ fn stress_shared_bounded_hard() {
 #[allow(clippy::same_item_push)]
 #[test]
 fn stress_receiver_multi_task_bounded_hard() {
+    #[cfg(miri)]
+    const AMT: usize = 100;
+    #[cfg(not(miri))]
     const AMT: usize = 10_000;
     const NTHREADS: u32 = 2;
 
@@ -327,6 +336,11 @@ fn stress_receiver_multi_task_bounded_hard() {
 /// after sender dropped.
 #[test]
 fn stress_drop_sender() {
+    #[cfg(miri)]
+    const ITER: usize = 100;
+    #[cfg(not(miri))]
+    const ITER: usize = 10000;
+
     fn list() -> impl Stream<Item = i32> {
         let (tx, rx) = mpsc::channel(1);
         thread::spawn(move || {
@@ -335,7 +349,7 @@ fn stress_drop_sender() {
         rx
     }
 
-    for _ in 0..10000 {
+    for _ in 0..ITER {
         let v: Vec<_> = block_on(list().collect());
         assert_eq!(v, vec![1, 2, 3]);
     }
@@ -380,9 +394,12 @@ fn stress_close_receiver_iter() {
     }
 }
 
+#[cfg_attr(miri, ignore)] // Miri is too slow
 #[test]
 fn stress_close_receiver() {
-    for _ in 0..10000 {
+    const ITER: usize = 10000;
+
+    for _ in 0..ITER {
         stress_close_receiver_iter();
     }
 }
@@ -397,6 +414,9 @@ async fn stress_poll_ready_sender(mut sender: mpsc::Sender<u32>, count: u32) {
 #[allow(clippy::same_item_push)]
 #[test]
 fn stress_poll_ready() {
+    #[cfg(miri)]
+    const AMT: u32 = 100;
+    #[cfg(not(miri))]
     const AMT: u32 = 1000;
     const NTHREADS: u32 = 8;
 
@@ -424,6 +444,7 @@ fn stress_poll_ready() {
     stress(16);
 }
 
+#[cfg_attr(miri, ignore)] // Miri is too slow
 #[test]
 fn try_send_1() {
     const N: usize = 3000;

futures/tests/compat.rs

@@ -1,4 +1,5 @@
 #![cfg(feature = "compat")]
+#![cfg(not(miri))] // unsupported operation: can't call foreign function: kqueue
 
 use futures::compat::Future01CompatExt;
 use futures::prelude::*;

futures/tests/eventual.rs

@@ -1,3 +1,5 @@
+#![cfg(not(miri))] // https://github.com/rust-lang/miri/issues/1038
+
 use futures::channel::oneshot;
 use futures::executor::ThreadPool;
 use futures::future::{self, ok, Future, FutureExt, TryFutureExt};

futures/tests/future_join_all.rs

@@ -1,22 +1,24 @@
+use futures::pin_mut;
 use futures::executor::block_on;
 use futures::future::{join_all, ready, Future, JoinAll};
 use std::fmt::Debug;
 
-fn assert_done<T, F>(actual_fut: F, expected: T)
+#[track_caller]
+fn assert_done<T>(actual_fut: impl Future<Output = T>, expected: T)
 where
     T: PartialEq + Debug,
-    F: FnOnce() -> Box<dyn Future<Output = T> + Unpin>,
 {
-    let output = block_on(actual_fut());
+    pin_mut!(actual_fut);
+    let output = block_on(actual_fut);
     assert_eq!(output, expected);
 }
 
 #[test]
 fn collect_collects() {
-    assert_done(|| Box::new(join_all(vec![ready(1), ready(2)])), vec![1, 2]);
-    assert_done(|| Box::new(join_all(vec![ready(1)])), vec![1]);
+    assert_done(join_all(vec![ready(1), ready(2)]), vec![1, 2]);
+    assert_done(join_all(vec![ready(1)]), vec![1]);
     // REVIEW: should this be implemented?
-    // assert_done(|| Box::new(join_all(Vec::<i32>::new())), vec![]);
+    // assert_done(join_all(Vec::<i32>::new()), vec![]);
 
     // TODO: needs more tests
 }
@@ -25,18 +27,18 @@ fn collect_collects() {
 fn join_all_iter_lifetime() {
     // In futures-rs version 0.1, this function would fail to typecheck due to an overly
     // conservative type parameterization of `JoinAll`.
-    fn sizes(bufs: Vec<&[u8]>) -> Box<dyn Future<Output = Vec<usize>> + Unpin> {
+    fn sizes(bufs: Vec<&[u8]>) -> impl Future<Output = Vec<usize>>  {
         let iter = bufs.into_iter().map(|b| ready::<usize>(b.len()));
-        Box::new(join_all(iter))
+        join_all(iter)
     }
 
-    assert_done(|| sizes(vec![&[1, 2, 3], &[], &[0]]), vec![3_usize, 0, 1]);
+    assert_done(sizes(vec![&[1, 2, 3], &[], &[0]]), vec![3_usize, 0, 1]);
 }
 
 #[test]
 fn join_all_from_iter() {
     assert_done(
-        || Box::new(vec![ready(1), ready(2)].into_iter().collect::<JoinAll<_>>()),
+        vec![ready(1), ready(2)].into_iter().collect::<JoinAll<_>>(),
         vec![1, 2],
     )
 }

futures/tests/future_shared.rs

@@ -96,6 +96,7 @@ fn drop_in_poll() {
     assert_eq!(block_on(future1), 1);
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn peek() {
     let mut local_pool = LocalPool::new();

futures/tests/future_try_join_all.rs

@@ -1,24 +1,26 @@
 use futures::executor::block_on;
+use futures::pin_mut;
 use futures_util::future::{err, ok, try_join_all, TryJoinAll};
 use std::fmt::Debug;
 use std::future::Future;
 
-fn assert_done<T, F>(actual_fut: F, expected: T)
+#[track_caller]
+fn assert_done<T>(actual_fut: impl Future<Output = T>, expected: T)
 where
     T: PartialEq + Debug,
-    F: FnOnce() -> Box<dyn Future<Output = T> + Unpin>,
 {
-    let output = block_on(actual_fut());
+    pin_mut!(actual_fut);
+    let output = block_on(actual_fut);
     assert_eq!(output, expected);
 }
 
 #[test]
 fn collect_collects() {
-    assert_done(|| Box::new(try_join_all(vec![ok(1), ok(2)])), Ok::<_, usize>(vec![1, 2]));
-    assert_done(|| Box::new(try_join_all(vec![ok(1), err(2)])), Err(2));
-    assert_done(|| Box::new(try_join_all(vec![ok(1)])), Ok::<_, usize>(vec![1]));
+    assert_done(try_join_all(vec![ok(1), ok(2)]), Ok::<_, usize>(vec![1, 2]));
+    assert_done(try_join_all(vec![ok(1), err(2)]), Err(2));
+    assert_done(try_join_all(vec![ok(1)]), Ok::<_, usize>(vec![1]));
     // REVIEW: should this be implemented?
-    // assert_done(|| Box::new(try_join_all(Vec::<i32>::new())), Ok(vec![]));
+    // assert_done(try_join_all(Vec::<i32>::new()), Ok(vec![]));
 
     // TODO: needs more tests
 }
@@ -27,18 +29,18 @@ fn collect_collects() {
 fn try_join_all_iter_lifetime() {
     // In futures-rs version 0.1, this function would fail to typecheck due to an overly
     // conservative type parameterization of `TryJoinAll`.
-    fn sizes(bufs: Vec<&[u8]>) -> Box<dyn Future<Output = Result<Vec<usize>, ()>> + Unpin> {
+    fn sizes(bufs: Vec<&[u8]>) -> impl Future<Output = Result<Vec<usize>, ()>>  {
         let iter = bufs.into_iter().map(|b| ok::<usize, ()>(b.len()));
-        Box::new(try_join_all(iter))
+        try_join_all(iter)
     }
 
-    assert_done(|| sizes(vec![&[1, 2, 3], &[], &[0]]), Ok(vec![3_usize, 0, 1]));
+    assert_done(sizes(vec![&[1, 2, 3], &[], &[0]]), Ok(vec![3_usize, 0, 1]));
 }
 
 #[test]
 fn try_join_all_from_iter() {
     assert_done(
-        || Box::new(vec![ok(1), ok(2)].into_iter().collect::<TryJoinAll<_>>()),
+        vec![ok(1), ok(2)].into_iter().collect::<TryJoinAll<_>>(),
         Ok::<_, usize>(vec![1, 2]),
     )
 }

futures/tests/lock_mutex.rs

@@ -34,6 +34,7 @@ fn mutex_wakes_waiters() {
     assert!(waiter.poll_unpin(&mut panic_context()).is_ready());
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn mutex_contested() {
     let (tx, mut rx) = mpsc::unbounded();

futures/tests/macro_comma_support.rs

@@ -14,6 +14,7 @@ fn ready() {
     }))
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn poll() {
     use futures::poll;

futures/tests/ready_queue.rs

@@ -93,6 +93,9 @@ fn dropping_ready_queue() {
 
 #[test]
 fn stress() {
+    #[cfg(miri)]
+    const ITER: usize = 30;
+    #[cfg(not(miri))]
     const ITER: usize = 300;
 
     for i in 0..ITER {

futures/tests/recurse.rs

@@ -3,6 +3,7 @@ use futures::future::{self, BoxFuture, FutureExt};
 use std::sync::mpsc;
 use std::thread;
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn lots() {
     #[cfg(not(futures_sanitizer))]

futures/tests/sink.rs

@@ -288,6 +288,7 @@ fn mpsc_blocking_start_send() {
 
 // test `flush` by using `with` to make the first insertion into a sink block
 // until a oneshot is completed
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn with_flush() {
     let (tx, rx) = oneshot::channel();

futures/tests/stream_futures_ordered.rs

@@ -26,6 +26,7 @@ fn works_1() {
     assert_eq!(None, iter.next());
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn works_2() {
     let (a_tx, a_rx) = oneshot::channel::<i32>();
@@ -54,6 +55,7 @@ fn from_iterator() {
     assert_eq!(block_on(stream.collect::<Vec<_>>()), vec![1, 2, 3]);
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn queue_never_unblocked() {
     let (_a_tx, a_rx) = oneshot::channel::<Box<dyn Any + Send>>();

futures/tests/stream_futures_unordered.rs

@@ -56,6 +56,7 @@ fn works_1() {
     assert_eq!(None, iter.next());
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn works_2() {
     let (a_tx, a_rx) = oneshot::channel::<i32>();
@@ -85,6 +86,7 @@ fn from_iterator() {
     assert_eq!(block_on(stream.collect::<Vec<_>>()), vec![1, 2, 3]);
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn finished_future() {
     let (_a_tx, a_rx) = oneshot::channel::<i32>();

futures/tests/stream_try_stream.rs

@@ -1,3 +1,5 @@
+#![cfg(not(miri))] // https://github.com/rust-lang/miri/issues/1038
+
 use futures::{
     stream::{self, StreamExt, TryStreamExt},
     task::Poll,

futures/tests/task_atomic_waker.rs

@@ -6,6 +6,7 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::thread;
 
+#[cfg_attr(miri, ignore)] // Miri is too slow
 #[test]
 fn basic() {
     let atomic_waker = Arc::new(AtomicWaker::new());