diff --git a/.cargo_vcs_info.json b/.cargo_vcs_info.json
index ffd4f55..e483977 100644
--- a/.cargo_vcs_info.json
+++ b/.cargo_vcs_info.json
@@ -1,5 +1,6 @@
 {
   "git": {
-    "sha1": "7caefa51304e78fd5018cd5d2a03f3b9089cc010"
-  }
-}
+    "sha1": "fc1e3250219170e31cddb8857a276cba7dd08d44"
+  },
+  "path_in_vcs": "futures"
+}
\ No newline at end of file
diff --git a/Android.bp b/Android.bp
index bc471eb..a594b53 100644
--- a/Android.bp
+++ b/Android.bp
@@ -42,7 +42,7 @@
     host_supported: true,
     crate_name: "futures",
     cargo_env_compat: true,
-    cargo_pkg_version: "0.3.17",
+    cargo_pkg_version: "0.3.21",
     srcs: ["src/lib.rs"],
     edition: "2018",
     features: [
diff --git a/Cargo.toml b/Cargo.toml
index 17c4d5f..f740f96 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -11,53 +11,72 @@
 
 [package]
 edition = "2018"
+rust-version = "1.45"
 name = "futures"
-version = "0.3.17"
-authors = ["Alex Crichton <alex@alexcrichton.com>"]
-description = "An implementation of futures and streams featuring zero allocations,\ncomposability, and iterator-like interfaces.\n"
+version = "0.3.21"
+description = """
+An implementation of futures and streams featuring zero allocations,
+composability, and iterator-like interfaces.
+"""
 homepage = "https://rust-lang.github.io/futures-rs"
-documentation = "https://docs.rs/futures/0.3"
 readme = "../README.md"
-keywords = ["futures", "async", "future"]
+keywords = [
+    "futures",
+    "async",
+    "future",
+]
 categories = ["asynchronous"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang/futures-rs"
+
 [package.metadata.docs.rs]
 all-features = true
-rustdoc-args = ["--cfg", "docsrs"]
+rustdoc-args = [
+    "--cfg",
+    "docsrs",
+]
 
 [package.metadata.playground]
-features = ["std", "async-await", "compat", "io-compat", "executor", "thread-pool"]
+features = [
+    "std",
+    "async-await",
+    "compat",
+    "io-compat",
+    "executor",
+    "thread-pool",
+]
+
 [dependencies.futures-channel]
-version = "0.3.17"
+version = "0.3.21"
 features = ["sink"]
 default-features = false
 
 [dependencies.futures-core]
-version = "0.3.17"
+version = "0.3.21"
 default-features = false
 
 [dependencies.futures-executor]
-version = "0.3.17"
+version = "0.3.21"
 optional = true
 default-features = false
 
 [dependencies.futures-io]
-version = "0.3.17"
+version = "0.3.21"
 default-features = false
 
 [dependencies.futures-sink]
-version = "0.3.17"
+version = "0.3.21"
 default-features = false
 
 [dependencies.futures-task]
-version = "0.3.17"
+version = "0.3.21"
 default-features = false
 
 [dependencies.futures-util]
-version = "0.3.17"
+version = "0.3.21"
 features = ["sink"]
 default-features = false
+
 [dev-dependencies.assert_matches]
 version = "1.3.0"
 
@@ -74,16 +93,55 @@
 version = "0.1.11"
 
 [features]
-alloc = ["futures-core/alloc", "futures-task/alloc", "futures-sink/alloc", "futures-channel/alloc", "futures-util/alloc"]
-async-await = ["futures-util/async-await", "futures-util/async-await-macro"]
+alloc = [
+    "futures-core/alloc",
+    "futures-task/alloc",
+    "futures-sink/alloc",
+    "futures-channel/alloc",
+    "futures-util/alloc",
+]
+async-await = [
+    "futures-util/async-await",
+    "futures-util/async-await-macro",
+]
 bilock = ["futures-util/bilock"]
 cfg-target-has-atomic = []
-compat = ["std", "futures-util/compat"]
-default = ["std", "async-await", "executor"]
-executor = ["std", "futures-executor/std"]
-io-compat = ["compat", "futures-util/io-compat"]
-read-initializer = ["futures-io/read-initializer", "futures-util/read-initializer"]
-std = ["alloc", "futures-core/std", "futures-task/std", "futures-io/std", "futures-sink/std", "futures-util/std", "futures-util/io", "futures-util/channel"]
-thread-pool = ["executor", "futures-executor/thread-pool"]
-unstable = ["futures-core/unstable", "futures-task/unstable", "futures-channel/unstable", "futures-io/unstable", "futures-util/unstable"]
+compat = [
+    "std",
+    "futures-util/compat",
+]
+default = [
+    "std",
+    "async-await",
+    "executor",
+]
+executor = [
+    "std",
+    "futures-executor/std",
+]
+io-compat = [
+    "compat",
+    "futures-util/io-compat",
+]
+std = [
+    "alloc",
+    "futures-core/std",
+    "futures-task/std",
+    "futures-io/std",
+    "futures-sink/std",
+    "futures-util/std",
+    "futures-util/io",
+    "futures-util/channel",
+]
+thread-pool = [
+    "executor",
+    "futures-executor/thread-pool",
+]
+unstable = [
+    "futures-core/unstable",
+    "futures-task/unstable",
+    "futures-channel/unstable",
+    "futures-io/unstable",
+    "futures-util/unstable",
+]
 write-all-vectored = ["futures-util/write-all-vectored"]
diff --git a/Cargo.toml.orig b/Cargo.toml.orig
index b01b12e..6871f47 100644
--- a/Cargo.toml.orig
+++ b/Cargo.toml.orig
@@ -1,14 +1,13 @@
 [package]
 name = "futures"
+version = "0.3.21"
 edition = "2018"
-version = "0.3.17"
-authors = ["Alex Crichton <alex@alexcrichton.com>"]
+rust-version = "1.45"
 license = "MIT OR Apache-2.0"
 readme = "../README.md"
 keywords = ["futures", "async", "future"]
 repository = "https://github.com/rust-lang/futures-rs"
 homepage = "https://rust-lang.github.io/futures-rs"
-documentation = "https://docs.rs/futures/0.3"
 description = """
 An implementation of futures and streams featuring zero allocations,
 composability, and iterator-like interfaces.
@@ -16,13 +15,13 @@
 categories = ["asynchronous"]
 
 [dependencies]
-futures-core = { path = "../futures-core", version = "0.3.17", default-features = false }
-futures-task = { path = "../futures-task", version = "0.3.17", default-features = false }
-futures-channel = { path = "../futures-channel", version = "0.3.17", default-features = false, features = ["sink"] }
-futures-executor = { path = "../futures-executor", version = "0.3.17", default-features = false, optional = true }
-futures-io = { path = "../futures-io", version = "0.3.17", default-features = false }
-futures-sink = { path = "../futures-sink", version = "0.3.17", default-features = false }
-futures-util = { path = "../futures-util", version = "0.3.17", default-features = false, features = ["sink"] }
+futures-core = { path = "../futures-core", version = "0.3.21", default-features = false }
+futures-task = { path = "../futures-task", version = "0.3.21", default-features = false }
+futures-channel = { path = "../futures-channel", version = "0.3.21", default-features = false, features = ["sink"] }
+futures-executor = { path = "../futures-executor", version = "0.3.21", default-features = false, optional = true }
+futures-io = { path = "../futures-io", version = "0.3.21", default-features = false }
+futures-sink = { path = "../futures-sink", version = "0.3.21", default-features = false }
+futures-util = { path = "../futures-util", version = "0.3.21", default-features = false, features = ["sink"] }
 
 [dev-dependencies]
 futures-executor = { path = "../futures-executor", features = ["thread-pool"] }
@@ -48,7 +47,6 @@
 # `unstable` feature as an explicit opt-in to unstable API.
 unstable = ["futures-core/unstable", "futures-task/unstable", "futures-channel/unstable", "futures-io/unstable", "futures-util/unstable"]
 bilock = ["futures-util/bilock"]
-read-initializer = ["futures-io/read-initializer", "futures-util/read-initializer"]
 write-all-vectored = ["futures-util/write-all-vectored"]
 
 # These features are no longer used.
diff --git a/METADATA b/METADATA
index ba70d3e..554f4df 100644
--- a/METADATA
+++ b/METADATA
@@ -7,13 +7,13 @@
   }
   url {
     type: ARCHIVE
-    value: "https://static.crates.io/crates/futures/futures-0.3.17.crate"
+    value: "https://static.crates.io/crates/futures/futures-0.3.21.crate"
   }
-  version: "0.3.17"
+  version: "0.3.21"
   license_type: NOTICE
   last_upgrade_date {
-    year: 2021
-    month: 9
-    day: 22
+    year: 2022
+    month: 3
+    day: 1
   }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 362aa3c..b8ebc61 100644
--- a/src/lib.rs
+++ b/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;
@@ -78,7 +79,6 @@
 //! The majority of examples and code snippets in this crate assume that they are
 //! inside an async block as written above.
 
-#![cfg_attr(feature = "read-initializer", feature(read_initializer))]
 #![cfg_attr(not(feature = "std"), no_std)]
 #![warn(
     missing_debug_implementations,
@@ -99,9 +99,6 @@
 #[cfg(all(feature = "bilock", not(feature = "unstable")))]
 compile_error!("The `bilock` feature requires the `unstable` feature as an explicit opt-in to unstable features");
 
-#[cfg(all(feature = "read-initializer", not(feature = "unstable")))]
-compile_error!("The `read-initializer` feature requires the `unstable` feature as an explicit opt-in to unstable features");
-
 #[doc(no_inline)]
 pub use futures_core::future::{Future, TryFuture};
 #[doc(no_inline)]
@@ -154,13 +151,73 @@
 
 #[cfg(feature = "executor")]
 #[cfg_attr(docsrs, doc(cfg(feature = "executor")))]
-#[doc(inline)]
-pub use futures_executor as executor;
+pub mod executor {
+    //! Built-in executors and related tools.
+    //!
+    //! All asynchronous computation occurs within an executor, which is
+    //! capable of spawning futures as tasks. This module provides several
+    //! built-in executors, as well as tools for building your own.
+    //!
+    //!
+    //! This module is only available when the `executor` feature of this
+    //! library is activated.
+    //!
+    //! # Using a thread pool (M:N task scheduling)
+    //!
+    //! Most of the time tasks should be executed on a [thread pool](ThreadPool).
+    //! A small set of worker threads can handle a very large set of spawned tasks
+    //! (which are much lighter weight than threads). Tasks spawned onto the pool
+    //! with the [`spawn_ok`](ThreadPool::spawn_ok) function will run ambiently on
+    //! the created threads.
+    //!
+    //! # Spawning additional tasks
+    //!
+    //! Tasks can be spawned onto a spawner by calling its [`spawn_obj`] method
+    //! directly. In the case of `!Send` futures, [`spawn_local_obj`] can be used
+    //! instead.
+    //!
+    //! # Single-threaded execution
+    //!
+    //! In addition to thread pools, it's possible to run a task (and the tasks
+    //! it spawns) entirely within a single thread via the [`LocalPool`] executor.
+    //! Aside from cutting down on synchronization costs, this executor also makes
+    //! it possible to spawn non-`Send` tasks, via [`spawn_local_obj`]. The
+    //! [`LocalPool`] is best suited for running I/O-bound tasks that do relatively
+    //! little work between I/O operations.
+    //!
+    //! There is also a convenience function [`block_on`] for simply running a
+    //! future to completion on the current thread.
+    //!
+    //! [`spawn_obj`]: https://docs.rs/futures/0.3/futures/task/trait.Spawn.html#tymethod.spawn_obj
+    //! [`spawn_local_obj`]: https://docs.rs/futures/0.3/futures/task/trait.LocalSpawn.html#tymethod.spawn_local_obj
+
+    pub use futures_executor::{
+        block_on, block_on_stream, enter, BlockingStream, Enter, EnterError, LocalPool,
+        LocalSpawner,
+    };
+
+    #[cfg(feature = "thread-pool")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "thread-pool")))]
+    pub use futures_executor::{ThreadPool, ThreadPoolBuilder};
+}
 
 #[cfg(feature = "compat")]
 #[cfg_attr(docsrs, doc(cfg(feature = "compat")))]
-#[doc(inline)]
-pub use futures_util::compat;
+pub mod compat {
+    //! Interop between `futures` 0.1 and 0.3.
+    //!
+    //! This module is only available when the `compat` feature of this
+    //! library is activated.
+
+    pub use futures_util::compat::{
+        Compat, Compat01As03, Compat01As03Sink, CompatSink, Executor01As03, Executor01CompatExt,
+        Executor01Future, Future01CompatExt, Sink01CompatExt, Stream01CompatExt,
+    };
+
+    #[cfg(feature = "io-compat")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
+    pub use futures_util::compat::{AsyncRead01CompatExt, AsyncWrite01CompatExt};
+}
 
 pub mod prelude {
     //! A "prelude" for crates using the `futures` crate.
@@ -181,10 +238,12 @@
     pub use crate::stream::{self, Stream, TryStream};
 
     #[doc(no_inline)]
+    #[allow(unreachable_pub)]
     pub use crate::future::{FutureExt as _, TryFutureExt as _};
     #[doc(no_inline)]
     pub use crate::sink::SinkExt as _;
     #[doc(no_inline)]
+    #[allow(unreachable_pub)]
     pub use crate::stream::{StreamExt as _, TryStreamExt as _};
 
     #[cfg(feature = "std")]
@@ -192,6 +251,7 @@
 
     #[cfg(feature = "std")]
     #[doc(no_inline)]
+    #[allow(unreachable_pub)]
     pub use crate::io::{
         AsyncBufReadExt as _, AsyncReadExt as _, AsyncSeekExt as _, AsyncWriteExt as _,
     };
diff --git a/tests/compat.rs b/tests/compat.rs
index c4125d8..ac04a95 100644
--- a/tests/compat.rs
+++ b/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::*;
diff --git a/tests/eventual.rs b/tests/eventual.rs
index bff000d..3461380 100644
--- a/tests/eventual.rs
+++ b/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};
diff --git a/tests/future_join_all.rs b/tests/future_join_all.rs
index ae05a21..44486e1 100644
--- a/tests/future_join_all.rs
+++ b/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 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])
 }
diff --git a/tests/future_shared.rs b/tests/future_shared.rs
index 718d6c4..3ceaebb 100644
--- a/tests/future_shared.rs
+++ b/tests/future_shared.rs
@@ -96,6 +96,7 @@
     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();
diff --git a/tests/future_try_join_all.rs b/tests/future_try_join_all.rs
index a4b3bb7..9a82487 100644
--- a/tests/future_try_join_all.rs
+++ b/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 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]),
     )
 }
diff --git a/tests/io_line_writer.rs b/tests/io_line_writer.rs
new file mode 100644
index 0000000..b483e0f
--- /dev/null
+++ b/tests/io_line_writer.rs
@@ -0,0 +1,73 @@
+use futures::executor::block_on;
+use futures::io::{AsyncWriteExt, LineWriter};
+use std::io;
+
+#[test]
+fn line_writer() {
+    let mut writer = LineWriter::new(Vec::new());
+
+    block_on(writer.write(&[0])).unwrap();
+    assert_eq!(*writer.get_ref(), []);
+
+    block_on(writer.write(&[1])).unwrap();
+    assert_eq!(*writer.get_ref(), []);
+
+    block_on(writer.flush()).unwrap();
+    assert_eq!(*writer.get_ref(), [0, 1]);
+
+    block_on(writer.write(&[0, b'\n', 1, b'\n', 2])).unwrap();
+    assert_eq!(*writer.get_ref(), [0, 1, 0, b'\n', 1, b'\n']);
+
+    block_on(writer.flush()).unwrap();
+    assert_eq!(*writer.get_ref(), [0, 1, 0, b'\n', 1, b'\n', 2]);
+
+    block_on(writer.write(&[3, b'\n'])).unwrap();
+    assert_eq!(*writer.get_ref(), [0, 1, 0, b'\n', 1, b'\n', 2, 3, b'\n']);
+}
+
+#[test]
+fn line_vectored() {
+    let mut line_writer = LineWriter::new(Vec::new());
+    assert_eq!(
+        block_on(line_writer.write_vectored(&[
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(b"\n"),
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(b"a"),
+        ]))
+        .unwrap(),
+        2
+    );
+    assert_eq!(line_writer.get_ref(), b"\n");
+
+    assert_eq!(
+        block_on(line_writer.write_vectored(&[
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(b"b"),
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(b"a"),
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(b"c"),
+        ]))
+        .unwrap(),
+        3
+    );
+    assert_eq!(line_writer.get_ref(), b"\n");
+    block_on(line_writer.flush()).unwrap();
+    assert_eq!(line_writer.get_ref(), b"\nabac");
+    assert_eq!(block_on(line_writer.write_vectored(&[])).unwrap(), 0);
+
+    assert_eq!(
+        block_on(line_writer.write_vectored(&[
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(&[]),
+            io::IoSlice::new(&[]),
+        ]))
+        .unwrap(),
+        0
+    );
+
+    assert_eq!(block_on(line_writer.write_vectored(&[io::IoSlice::new(b"a\nb")])).unwrap(), 3);
+    assert_eq!(line_writer.get_ref(), b"\nabaca\nb");
+}
diff --git a/tests/lock_mutex.rs b/tests/lock_mutex.rs
index 7c33864..c92ef50 100644
--- a/tests/lock_mutex.rs
+++ b/tests/lock_mutex.rs
@@ -34,6 +34,7 @@
     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();
diff --git a/tests/macro_comma_support.rs b/tests/macro_comma_support.rs
index 85871e9..3b082d2 100644
--- a/tests/macro_comma_support.rs
+++ b/tests/macro_comma_support.rs
@@ -14,6 +14,7 @@
     }))
 }
 
+#[cfg_attr(miri, ignore)] // https://github.com/rust-lang/miri/issues/1038
 #[test]
 fn poll() {
     use futures::poll;
diff --git a/tests/ready_queue.rs b/tests/ready_queue.rs
index 8290132..afba8f2 100644
--- a/tests/ready_queue.rs
+++ b/tests/ready_queue.rs
@@ -93,6 +93,9 @@
 
 #[test]
 fn stress() {
+    #[cfg(miri)]
+    const ITER: usize = 30;
+    #[cfg(not(miri))]
     const ITER: usize = 300;
 
     for i in 0..ITER {
diff --git a/tests/recurse.rs b/tests/recurse.rs
index d81753c..f06524f 100644
--- a/tests/recurse.rs
+++ b/tests/recurse.rs
@@ -3,6 +3,7 @@
 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))]
diff --git a/tests/sink.rs b/tests/sink.rs
index f3cf11b..dc826bd 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -288,6 +288,7 @@
 
 // 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();
diff --git a/tests/stream.rs b/tests/stream.rs
index 0d453d1..71ec654 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -1,10 +1,14 @@
+use std::iter;
+use std::sync::Arc;
+
 use futures::channel::mpsc;
 use futures::executor::block_on;
 use futures::future::{self, Future};
+use futures::lock::Mutex;
 use futures::sink::SinkExt;
 use futures::stream::{self, StreamExt};
 use futures::task::Poll;
-use futures::FutureExt;
+use futures::{ready, FutureExt};
 use futures_test::task::noop_context;
 
 #[test]
@@ -50,6 +54,272 @@
 }
 
 #[test]
+fn flatten_unordered() {
+    use futures::executor::block_on;
+    use futures::stream::*;
+    use futures::task::*;
+    use std::convert::identity;
+    use std::pin::Pin;
+    use std::thread;
+    use std::time::Duration;
+
+    struct DataStream {
+        data: Vec<u8>,
+        polled: bool,
+        wake_immediately: bool,
+    }
+
+    impl Stream for DataStream {
+        type Item = u8;
+
+        fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
+            if !self.polled {
+                if !self.wake_immediately {
+                    let waker = ctx.waker().clone();
+                    let sleep_time =
+                        Duration::from_millis(*self.data.first().unwrap_or(&0) as u64 / 10);
+                    thread::spawn(move || {
+                        thread::sleep(sleep_time);
+                        waker.wake_by_ref();
+                    });
+                } else {
+                    ctx.waker().wake_by_ref();
+                }
+                self.polled = true;
+                Poll::Pending
+            } else {
+                self.polled = false;
+                Poll::Ready(self.data.pop())
+            }
+        }
+    }
+
+    struct Interchanger {
+        polled: bool,
+        base: u8,
+        wake_immediately: bool,
+    }
+
+    impl Stream for Interchanger {
+        type Item = DataStream;
+
+        fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
+            if !self.polled {
+                self.polled = true;
+                if !self.wake_immediately {
+                    let waker = ctx.waker().clone();
+                    let sleep_time = Duration::from_millis(self.base as u64);
+                    thread::spawn(move || {
+                        thread::sleep(sleep_time);
+                        waker.wake_by_ref();
+                    });
+                } else {
+                    ctx.waker().wake_by_ref();
+                }
+                Poll::Pending
+            } else {
+                let data: Vec<_> = (0..6).rev().map(|v| v + self.base * 6).collect();
+                self.base += 1;
+                self.polled = false;
+                Poll::Ready(Some(DataStream {
+                    polled: false,
+                    data,
+                    wake_immediately: self.wake_immediately && self.base % 2 == 0,
+                }))
+            }
+        }
+    }
+
+    // basic behaviour
+    {
+        block_on(async {
+            let st = stream::iter(vec![
+                stream::iter(0..=4u8),
+                stream::iter(6..=10),
+                stream::iter(10..=12),
+            ]);
+
+            let fl_unordered = st.flatten_unordered(3).collect::<Vec<_>>().await;
+
+            assert_eq!(fl_unordered, vec![0, 6, 10, 1, 7, 11, 2, 8, 12, 3, 9, 4, 10]);
+        });
+
+        block_on(async {
+            let st = stream::iter(vec![
+                stream::iter(0..=4u8),
+                stream::iter(6..=10),
+                stream::iter(0..=2),
+            ]);
+
+            let mut fm_unordered = st
+                .flat_map_unordered(1, |s| s.filter(|v| futures::future::ready(v % 2 == 0)))
+                .collect::<Vec<_>>()
+                .await;
+
+            fm_unordered.sort_unstable();
+
+            assert_eq!(fm_unordered, vec![0, 0, 2, 2, 4, 6, 8, 10]);
+        });
+    }
+
+    // wake up immediately
+    {
+        block_on(async {
+            let mut fl_unordered = Interchanger { polled: false, base: 0, wake_immediately: true }
+                .take(10)
+                .map(|s| s.map(identity))
+                .flatten_unordered(10)
+                .collect::<Vec<_>>()
+                .await;
+
+            fl_unordered.sort_unstable();
+
+            assert_eq!(fl_unordered, (0..60).collect::<Vec<u8>>());
+        });
+
+        block_on(async {
+            let mut fm_unordered = Interchanger { polled: false, base: 0, wake_immediately: true }
+                .take(10)
+                .flat_map_unordered(10, |s| s.map(identity))
+                .collect::<Vec<_>>()
+                .await;
+
+            fm_unordered.sort_unstable();
+
+            assert_eq!(fm_unordered, (0..60).collect::<Vec<u8>>());
+        });
+    }
+
+    // wake up after delay
+    {
+        block_on(async {
+            let mut fl_unordered = Interchanger { polled: false, base: 0, wake_immediately: false }
+                .take(10)
+                .map(|s| s.map(identity))
+                .flatten_unordered(10)
+                .collect::<Vec<_>>()
+                .await;
+
+            fl_unordered.sort_unstable();
+
+            assert_eq!(fl_unordered, (0..60).collect::<Vec<u8>>());
+        });
+
+        block_on(async {
+            let mut fm_unordered = Interchanger { polled: false, base: 0, wake_immediately: false }
+                .take(10)
+                .flat_map_unordered(10, |s| s.map(identity))
+                .collect::<Vec<_>>()
+                .await;
+
+            fm_unordered.sort_unstable();
+
+            assert_eq!(fm_unordered, (0..60).collect::<Vec<u8>>());
+        });
+
+        block_on(async {
+            let (mut fm_unordered, mut fl_unordered) = futures_util::join!(
+                Interchanger { polled: false, base: 0, wake_immediately: false }
+                    .take(10)
+                    .flat_map_unordered(10, |s| s.map(identity))
+                    .collect::<Vec<_>>(),
+                Interchanger { polled: false, base: 0, wake_immediately: false }
+                    .take(10)
+                    .map(|s| s.map(identity))
+                    .flatten_unordered(10)
+                    .collect::<Vec<_>>()
+            );
+
+            fm_unordered.sort_unstable();
+            fl_unordered.sort_unstable();
+
+            assert_eq!(fm_unordered, fl_unordered);
+            assert_eq!(fm_unordered, (0..60).collect::<Vec<u8>>());
+        });
+    }
+
+    // waker panics
+    {
+        let stream = Arc::new(Mutex::new(
+            Interchanger { polled: false, base: 0, wake_immediately: true }
+                .take(10)
+                .flat_map_unordered(10, |s| s.map(identity)),
+        ));
+
+        struct PanicWaker;
+
+        impl ArcWake for PanicWaker {
+            fn wake_by_ref(_arc_self: &Arc<Self>) {
+                panic!("WAKE UP");
+            }
+        }
+
+        std::thread::spawn({
+            let stream = stream.clone();
+            move || {
+                let mut st = poll_fn(|cx| {
+                    let mut lock = ready!(stream.lock().poll_unpin(cx));
+
+                    let panic_waker = waker(Arc::new(PanicWaker));
+                    let mut panic_cx = Context::from_waker(&panic_waker);
+                    let _ = ready!(lock.poll_next_unpin(&mut panic_cx));
+
+                    Poll::Ready(Some(()))
+                });
+
+                block_on(st.next())
+            }
+        })
+        .join()
+        .unwrap_err();
+
+        block_on(async move {
+            let mut values: Vec<_> = stream.lock().await.by_ref().collect().await;
+            values.sort_unstable();
+
+            assert_eq!(values, (0..60).collect::<Vec<u8>>());
+        });
+    }
+
+    // stream panics
+    {
+        let st = stream::iter(iter::once(
+            once(Box::pin(async { panic!("Polled") })).left_stream::<DataStream>(),
+        ))
+        .chain(
+            Interchanger { polled: false, base: 0, wake_immediately: true }
+                .map(|stream| stream.right_stream())
+                .take(10),
+        );
+
+        let stream = Arc::new(Mutex::new(st.flatten_unordered(10)));
+
+        std::thread::spawn({
+            let stream = stream.clone();
+            move || {
+                let mut st = poll_fn(|cx| {
+                    let mut lock = ready!(stream.lock().poll_unpin(cx));
+                    let data = ready!(lock.poll_next_unpin(cx));
+
+                    Poll::Ready(data)
+                });
+
+                block_on(st.next())
+            }
+        })
+        .join()
+        .unwrap_err();
+
+        block_on(async move {
+            let mut values: Vec<_> = stream.lock().await.by_ref().collect().await;
+            values.sort_unstable();
+
+            assert_eq!(values, (0..60).collect::<Vec<u8>>());
+        });
+    }
+}
+
+#[test]
 fn take_until() {
     fn make_stop_fut(stop_on: u32) -> impl Future<Output = ()> {
         let mut i = 0;
diff --git a/tests/stream_futures_ordered.rs b/tests/stream_futures_ordered.rs
index 7506c65..84e0bcc 100644
--- a/tests/stream_futures_ordered.rs
+++ b/tests/stream_futures_ordered.rs
@@ -26,6 +26,7 @@
     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 @@
     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>>();
diff --git a/tests/stream_futures_unordered.rs b/tests/stream_futures_unordered.rs
index 4b9afcc..f62f733 100644
--- a/tests/stream_futures_unordered.rs
+++ b/tests/stream_futures_unordered.rs
@@ -56,6 +56,7 @@
     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 @@
     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>();
diff --git a/tests/stream_try_stream.rs b/tests/stream_try_stream.rs
index 194e74d..d83fc54 100644
--- a/tests/stream_try_stream.rs
+++ b/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,
diff --git a/tests/task_atomic_waker.rs b/tests/task_atomic_waker.rs
index cec3db2..2d1612a 100644
--- a/tests/task_atomic_waker.rs
+++ b/tests/task_atomic_waker.rs
@@ -6,6 +6,7 @@
 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());