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,18 @@ jobs:
       - run: ci/no_atomic_cas.sh
       - run: git diff --exit-code
 
+  miri:
+    name: cargo miri test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install Rust
+        run: rustup toolchain install nightly --component miri && rustup default nightly
+      # futures-executor uses boxed futures so many tests trigger https://github.com/rust-lang/miri/issues/1038
+      - run: cargo miri test --workspace --exclude futures-executor --all-features
+        env:
+          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-tag-raw-pointers
+
   san:
     name: cargo test -Z sanitizer=${{ matrix.sanitizer }}
     strategy:
@@ -261,6 +280,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-close.rs

@@ -147,6 +147,7 @@ fn single_receiver_drop_closes_channel_and_drains() {
 
 // Stress test that `try_send()`s occurring concurrently with receiver
 // close/drops don't appear as successful sends.
+#[cfg_attr(miri, ignore)] // Miri is too slow
 #[test]
 fn stress_try_send_as_receiver_closes() {
     const AMT: usize = 10000;

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-channel/tests/oneshot.rs

@@ -35,6 +35,11 @@ fn cancel_notifies() {
 
 #[test]
 fn cancel_lots() {
+    #[cfg(miri)]
+    const N: usize = 100;
+    #[cfg(not(miri))]
+    const N: usize = 20000;
+
     let (tx, rx) = mpsc::channel::<(Sender<_>, mpsc::Sender<_>)>();
     let t = thread::spawn(move || {
         for (mut tx, tx2) in rx {
@@ -43,7 +48,7 @@ fn cancel_lots() {
         }
     });
 
-    for _ in 0..20000 {
+    for _ in 0..N {
         let (otx, orx) = oneshot::channel::<u32>();
         let (tx2, rx2) = mpsc::channel();
         tx.send((otx, tx2)).unwrap();
@@ -101,6 +106,11 @@ fn is_canceled() {
 
 #[test]
 fn cancel_sends() {
+    #[cfg(miri)]
+    const N: usize = 100;
+    #[cfg(not(miri))]
+    const N: usize = 20000;
+
     let (tx, rx) = mpsc::channel::<Sender<_>>();
     let t = thread::spawn(move || {
         for otx in rx {
@@ -108,7 +118,7 @@ fn cancel_sends() {
         }
     });
 
-    for _ in 0..20000 {
+    for _ in 0..N {
         let (otx, mut orx) = oneshot::channel::<u32>();
         tx.send(otx).unwrap();
 

futures-util/src/compat/compat01as03.rs

@@ -64,6 +64,7 @@ pub trait Future01CompatExt: Future01 {
     /// [`Future<Output = Result<T, E>>`](futures_core::future::Future).
     ///
     /// ```
+    /// # if cfg!(miri) { return; } // https://github.com/rust-lang/futures-rs/issues/2514
     /// # futures::executor::block_on(async {
     /// # // TODO: These should be all using `futures::compat`, but that runs up against Cargo
     /// # // feature issues
@@ -90,6 +91,7 @@ pub trait Stream01CompatExt: Stream01 {
     /// [`Stream<Item = Result<T, E>>`](futures_core::stream::Stream).
     ///
     /// ```
+    /// # if cfg!(miri) { return; } // https://github.com/rust-lang/futures-rs/issues/2514
     /// # futures::executor::block_on(async {
     /// use futures::stream::StreamExt;
     /// use futures_util::compat::Stream01CompatExt;
@@ -119,6 +121,7 @@ pub trait Sink01CompatExt: Sink01 {
     /// [`Sink<T, Error = E>`](futures_sink::Sink).
     ///
     /// ```
+    /// # if cfg!(miri) { return; } // https://github.com/rust-lang/futures-rs/issues/2514
     /// # futures::executor::block_on(async {
     /// use futures::{sink::SinkExt, stream::StreamExt};
     /// use futures_util::compat::{Stream01CompatExt, Sink01CompatExt};
@@ -362,6 +365,7 @@ mod io {
         /// [`AsyncRead`](futures_io::AsyncRead).
         ///
         /// ```
+        /// # if cfg!(miri) { return; } // https://github.com/rust-lang/futures-rs/issues/2514
         /// # futures::executor::block_on(async {
         /// use futures::io::AsyncReadExt;
         /// use futures_util::compat::AsyncRead01CompatExt;
@@ -391,6 +395,7 @@ mod io {
         /// [`AsyncWrite`](futures_io::AsyncWrite).
         ///
         /// ```
+        /// # if cfg!(miri) { return; } // https://github.com/rust-lang/futures-rs/issues/2514
         /// # futures::executor::block_on(async {
         /// use futures::io::AsyncWriteExt;
         /// use futures_util::compat::AsyncWrite01CompatExt;

futures-util/src/compat/executor.rs

@@ -17,6 +17,7 @@ pub trait Executor01CompatExt: Executor01<Executor01Future> + Clone + Send + 'st
     /// futures 0.3 [`Spawn`](futures_task::Spawn).
     ///
     /// ```
+    /// # if cfg!(miri) { return; } // Miri does not support epoll
     /// use futures::task::SpawnExt;
     /// use futures::future::{FutureExt, TryFutureExt};
     /// use futures_util::compat::Executor01CompatExt;

futures-util/src/stream/try_stream/mod.rs

@@ -894,6 +894,7 @@ pub trait TryStreamExt: TryStream {
     /// Wraps a [`TryStream`] into a stream compatible with libraries using
     /// futures 0.1 `Stream`. Requires the `compat` feature to be enabled.
     /// ```
+    /// # if cfg!(miri) { return; } // Miri does not support epoll
     /// use futures::future::{FutureExt, TryFutureExt};
     /// # let (tx, rx) = futures::channel::oneshot::channel();
     ///

futures-util/src/task/spawn.rs

@@ -34,6 +34,7 @@ pub trait SpawnExt: Spawn {
     /// today. Feel free to use this method in the meantime.
     ///
     /// ```
+    /// # if cfg!(miri) { return; } // https://github.com/rust-lang/miri/issues/1038
     /// use futures::executor::ThreadPool;
     /// use futures::task::SpawnExt;
     ///
@@ -58,6 +59,7 @@ pub trait SpawnExt: Spawn {
     /// resolves to the output of the spawned future.
     ///
     /// ```
+    /// # if cfg!(miri) { return; } // https://github.com/rust-lang/miri/issues/1038
     /// use futures::executor::{block_on, ThreadPool};
     /// use futures::future;
     /// use futures::task::SpawnExt;
@@ -136,6 +138,7 @@ pub trait LocalSpawnExt: LocalSpawn {
     /// resolves to the output of the spawned future.
     ///
     /// ```
+    /// # if cfg!(miri) { return; } // https://github.com/rust-lang/miri/issues/1038
     /// use futures::executor::LocalPool;
     /// use futures::task::LocalSpawnExt;
     ///

futures/src/lib.rs

@@ -25,6 +25,7 @@
 //! within macros and keywords such as async and await!.
 //!
 //! ```rust
+//! # if cfg!(miri) { return; } // https://github.com/rust-lang/miri/issues/1038
 //! # use futures::channel::mpsc;
 //! # use futures::executor; ///standard executors to provide a context for futures and streams
 //! # use futures::executor::ThreadPool;

futures/tests/compat.rs

@@ -1,4 +1,5 @@
 #![cfg(feature = "compat")]
+#![cfg(not(miri))] // Miri does not support epoll
 
 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::executor::block_on;
 use futures::future::{join_all, ready, Future, JoinAll};
+use futures::pin_mut;
 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,15 @@ 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![1, 2],
-    )
+    assert_done(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());