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,14 +106,19 @@ 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 {
             let _ = otx.send(42);
         }
     });
 
-    for _ in 0..20000 {
+    for _ in 0..N {
         let (otx, mut orx) = oneshot::channel::<u32>();
         tx.send(otx).unwrap();
 

futures-task/src/future_obj.rs

@@ -224,7 +224,7 @@ mod if_alloc {
         }
 
         unsafe fn drop(ptr: *mut (dyn Future<Output = T> + 'a)) {
-            drop(Box::from_raw(ptr as *mut F))
+            drop(Box::from_raw(ptr.cast::<F>()))
         }
     }
 
@@ -252,10 +252,9 @@ mod if_alloc {
     where
         F: Future<Output = T> + 'a,
     {
-        fn into_raw(mut self) -> *mut (dyn Future<Output = T> + 'a) {
-            let ptr = unsafe { self.as_mut().get_unchecked_mut() as *mut _ };
-            mem::forget(self);
-            ptr
+        fn into_raw(self) -> *mut (dyn Future<Output = T> + 'a) {
+            let mut this = mem::ManuallyDrop::new(self);
+            unsafe { this.as_mut().get_unchecked_mut() as *mut _ }
         }
 
         unsafe fn drop(ptr: *mut (dyn Future<Output = T> + 'a)) {
@@ -264,10 +263,9 @@ mod if_alloc {
     }
 
     unsafe impl<'a, T: 'a> UnsafeFutureObj<'a, T> for Pin<Box<dyn Future<Output = T> + 'a>> {
-        fn into_raw(mut self) -> *mut (dyn Future<Output = T> + 'a) {
-            let ptr = unsafe { self.as_mut().get_unchecked_mut() as *mut _ };
-            mem::forget(self);
-            ptr
+        fn into_raw(self) -> *mut (dyn Future<Output = T> + 'a) {
+            let mut this = mem::ManuallyDrop::new(self);
+            unsafe { this.as_mut().get_unchecked_mut() as *mut _ }
         }
 
         unsafe fn drop(ptr: *mut (dyn Future<Output = T> + 'a)) {
@@ -276,10 +274,9 @@ mod if_alloc {
     }
 
     unsafe impl<'a, T: 'a> UnsafeFutureObj<'a, T> for Pin<Box<dyn Future<Output = T> + Send + 'a>> {
-        fn into_raw(mut self) -> *mut (dyn Future<Output = T> + 'a) {
-            let ptr = unsafe { self.as_mut().get_unchecked_mut() as *mut _ };
-            mem::forget(self);
-            ptr
+        fn into_raw(self) -> *mut (dyn Future<Output = T> + 'a) {
+            let mut this = mem::ManuallyDrop::new(self);
+            unsafe { this.as_mut().get_unchecked_mut() as *mut _ }
         }
 
         unsafe fn drop(ptr: *mut (dyn Future<Output = T> + 'a)) {

futures-task/src/waker.rs

@@ -20,7 +20,7 @@ pub fn waker<W>(wake: Arc<W>) -> Waker
 where
     W: ArcWake + 'static,
 {
-    let ptr = Arc::into_raw(wake) as *const ();
+    let ptr = Arc::into_raw(wake).cast::<()>();
 
     unsafe { Waker::from_raw(RawWaker::new(ptr, waker_vtable::<W>())) }
 }
@@ -31,7 +31,7 @@ where
 #[allow(clippy::redundant_clone)] // The clone here isn't actually redundant.
 unsafe fn increase_refcount<T: ArcWake>(data: *const ()) {
     // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop
-    let arc = mem::ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
+    let arc = mem::ManuallyDrop::new(Arc::<T>::from_raw(data.cast::<T>()));
     // Now increase refcount, but don't drop new refcount either
     let _arc_clone: mem::ManuallyDrop<_> = arc.clone();
 }
@@ -43,17 +43,17 @@ unsafe fn clone_arc_raw<T: ArcWake>(data: *const ()) -> RawWaker {
 }
 
 unsafe fn wake_arc_raw<T: ArcWake>(data: *const ()) {
-    let arc: Arc<T> = Arc::from_raw(data as *const T);
+    let arc: Arc<T> = Arc::from_raw(data.cast::<T>());
     ArcWake::wake(arc);
 }
 
 // used by `waker_ref`
 unsafe fn wake_by_ref_arc_raw<T: ArcWake>(data: *const ()) {
     // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop
-    let arc = mem::ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
+    let arc = mem::ManuallyDrop::new(Arc::<T>::from_raw(data.cast::<T>()));
     ArcWake::wake_by_ref(&arc);
 }
 
 unsafe fn drop_arc_raw<T: ArcWake>(data: *const ()) {
-    drop(Arc::<T>::from_raw(data as *const T))
+    drop(Arc::<T>::from_raw(data.cast::<T>()))
 }

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