Run more tests with Miri (#2584)

Author: taiki-e

.github/workflows/ci.yml

@@ -273,10 +273,10 @@ jobs:
       - 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
+      - run: cargo miri test --workspace --all-features
         env:
           MIRIFLAGS: -Zmiri-check-number-validity -Zmiri-symbolic-alignment-check -Zmiri-tag-raw-pointers -Zmiri-disable-isolation
+          RUSTFLAGS: ${{ env.RUSTFLAGS }} -Z randomize-layout
 
   san:
     name: cargo test -Z sanitizer=${{ matrix.sanitizer }}

futures-executor/Cargo.toml

@@ -22,7 +22,7 @@ futures-util = { path = "../futures-util", version = "0.3.21", default-features
 num_cpus = { version = "1.8.0", optional = true }
 
 [dev-dependencies]
-futures = { path = "../futures" }
+futures = { path = "../futures", features = ["thread-pool"] }
 
 [package.metadata.docs.rs]
 all-features = true

futures-executor/src/thread_pool.rs

@@ -108,12 +108,15 @@ impl ThreadPool {
     /// completion.
     ///
     /// ```
+    /// # {
     /// use futures::executor::ThreadPool;
     ///
     /// let pool = ThreadPool::new().unwrap();
     ///
     /// let future = async { /* ... */ };
     /// pool.spawn_ok(future);
+    /// # }
+    /// # std::thread::sleep(std::time::Duration::from_secs(1)); // wait for background threads closed
     /// ```
     ///
     /// > **Note**: This method is similar to `SpawnExt::spawn`, except that
@@ -359,16 +362,19 @@ mod tests {
 
     #[test]
     fn test_drop_after_start() {
-        let (tx, rx) = mpsc::sync_channel(2);
-        let _cpu_pool = ThreadPoolBuilder::new()
-            .pool_size(2)
-            .after_start(move |_| tx.send(1).unwrap())
-            .create()
-            .unwrap();
-
-        // After ThreadPoolBuilder is deconstructed, the tx should be dropped
-        // so that we can use rx as an iterator.
-        let count = rx.into_iter().count();
-        assert_eq!(count, 2);
+        {
+            let (tx, rx) = mpsc::sync_channel(2);
+            let _cpu_pool = ThreadPoolBuilder::new()
+                .pool_size(2)
+                .after_start(move |_| tx.send(1).unwrap())
+                .create()
+                .unwrap();
+
+            // After ThreadPoolBuilder is deconstructed, the tx should be dropped
+            // so that we can use rx as an iterator.
+            let count = rx.into_iter().count();
+            assert_eq!(count, 2);
+        }
+        std::thread::sleep(std::time::Duration::from_secs(1)); // wait for background threads closed
     }
 }

futures-executor/tests/local_pool.rs

@@ -288,6 +288,9 @@ fn run_until_stalled_runs_spawned_sub_futures() {
 
 #[test]
 fn run_until_stalled_executes_all_ready() {
+    #[cfg(miri)]
+    const ITER: usize = 50;
+    #[cfg(not(miri))]
     const ITER: usize = 200;
     const PER_ITER: usize = 3;
 

futures-util/src/task/spawn.rs

@@ -34,14 +34,16 @@ 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;
     ///
     /// let executor = ThreadPool::new().unwrap();
     ///
     /// let future = async { /* ... */ };
     /// executor.spawn(future).unwrap();
+    /// # }
+    /// # std::thread::sleep(std::time::Duration::from_secs(1)); // wait for background threads closed
     /// ```
     #[cfg(feature = "alloc")]
     fn spawn<Fut>(&self, future: Fut) -> Result<(), SpawnError>
@@ -59,7 +61,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;
@@ -69,6 +71,8 @@ pub trait SpawnExt: Spawn {
     /// let future = future::ready(1);
     /// let join_handle_fut = executor.spawn_with_handle(future).unwrap();
     /// assert_eq!(block_on(join_handle_fut), 1);
+    /// # }
+    /// # std::thread::sleep(std::time::Duration::from_secs(1)); // wait for background threads closed
     /// ```
     #[cfg(feature = "channel")]
     #[cfg_attr(docsrs, doc(cfg(feature = "channel")))]
@@ -138,7 +142,6 @@ 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,13 +25,13 @@
 //! 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;
 //! # use futures::StreamExt;
 //! #
 //! fn main() {
+//!     # {
 //!     let pool = ThreadPool::new().expect("Failed to build pool");
 //!     let (tx, rx) = mpsc::unbounded::<i32>();
 //!
@@ -73,6 +73,8 @@
 //!     let values: Vec<i32> = executor::block_on(fut_values);
 //!
 //!     println!("Values={:?}", values);
+//!     # }
+//!     # std::thread::sleep(std::time::Duration::from_secs(1)); // wait for background threads closed
 //! }
 //! ```
 //!

futures/tests/eventual.rs

@@ -1,5 +1,3 @@
-#![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};
@@ -136,6 +134,11 @@ fn select3() {
 
 #[test]
 fn select4() {
+    #[cfg(miri)]
+    const N: usize = 100;
+    #[cfg(not(miri))]
+    const N: usize = 10000;
+
     let (tx, rx) = mpsc::channel::<oneshot::Sender<i32>>();
 
     let t = thread::spawn(move || {
@@ -145,7 +148,7 @@ fn select4() {
     });
 
     let (tx2, rx2) = mpsc::channel();
-    for _ in 0..10000 {
+    for _ in 0..N {
         let (c1, p1) = oneshot::channel::<i32>();
         let (c2, p2) = oneshot::channel::<i32>();
 

futures/tests/future_shared.rs

@@ -97,7 +97,6 @@ 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/lock_mutex.rs

@@ -34,34 +34,36 @@ 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();
-    let pool = ThreadPool::builder().pool_size(16).create().unwrap();
+    {
+        let (tx, mut rx) = mpsc::unbounded();
+        let pool = ThreadPool::builder().pool_size(16).create().unwrap();
 
-    let tx = Arc::new(tx);
-    let mutex = Arc::new(Mutex::new(0));
+        let tx = Arc::new(tx);
+        let mutex = Arc::new(Mutex::new(0));
 
-    let num_tasks = 1000;
-    for _ in 0..num_tasks {
-        let tx = tx.clone();
-        let mutex = mutex.clone();
-        pool.spawn(async move {
-            let mut lock = mutex.lock().await;
-            ready(()).pending_once().await;
-            *lock += 1;
-            tx.unbounded_send(()).unwrap();
-            drop(lock);
-        })
-        .unwrap();
-    }
-
-    block_on(async {
+        let num_tasks = 1000;
         for _ in 0..num_tasks {
-            rx.next().await.unwrap();
+            let tx = tx.clone();
+            let mutex = mutex.clone();
+            pool.spawn(async move {
+                let mut lock = mutex.lock().await;
+                ready(()).pending_once().await;
+                *lock += 1;
+                tx.unbounded_send(()).unwrap();
+                drop(lock);
+            })
+            .unwrap();
         }
-        let lock = mutex.lock().await;
-        assert_eq!(num_tasks, *lock);
-    })
+
+        block_on(async {
+            for _ in 0..num_tasks {
+                rx.next().await.unwrap();
+            }
+            let lock = mutex.lock().await;
+            assert_eq!(num_tasks, *lock);
+        });
+    }
+    std::thread::sleep(std::time::Duration::from_secs(1)); // wait for background threads closed
 }

futures/tests/macro_comma_support.rs

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

futures/tests/recurse.rs

@@ -3,7 +3,6 @@ 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,7 +288,6 @@ 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,7 +26,6 @@ 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>();
@@ -55,7 +54,6 @@ 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,7 +56,6 @@ 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>();
@@ -86,7 +85,6 @@ 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,5 +1,3 @@
-#![cfg(not(miri))] // https://github.com/rust-lang/miri/issues/1038
-
 use futures::{
     stream::{self, StreamExt, TryStreamExt},
     task::Poll,