Release 0.3.20

CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.3.20 - 2022-02-06
+
+* Fix stacked borrows violations when `-Zmiri-tag-raw-pointers` is enabled. This raises MSRV of `futures-task` to 1.45. (#2548, #2550)
+* Change `FuturesUnordered` to respect yielding from future (#2551)
+* Add `StreamExt::{flatten_unordered, flat_map_unordered}` (#2083)
+
 # 0.3.19 - 2021-12-18
 
 * Remove unstable `read-initializer` feature (#2534)

futures-channel/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-channel"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"
@@ -22,8 +22,8 @@ unstable = []
 cfg-target-has-atomic = []
 
 [dependencies]
-futures-core = { path = "../futures-core", version = "0.3.19", default-features = false }
-futures-sink = { path = "../futures-sink", version = "0.3.19", default-features = false, optional = true }
+futures-core = { path = "../futures-core", version = "0.3.20", default-features = false }
+futures-sink = { path = "../futures-sink", version = "0.3.20", default-features = false, optional = true }
 
 [dev-dependencies]
 futures = { path = "../futures", default-features = true }

futures-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-core"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.36"
 license = "MIT OR Apache-2.0"

futures-executor/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-executor"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"
@@ -16,9 +16,9 @@ std = ["futures-core/std", "futures-task/std", "futures-util/std"]
 thread-pool = ["std", "num_cpus"]
 
 [dependencies]
-futures-core = { path = "../futures-core", version = "0.3.19", default-features = false }
-futures-task = { path = "../futures-task", version = "0.3.19", default-features = false }
-futures-util = { path = "../futures-util", version = "0.3.19", default-features = false }
+futures-core = { path = "../futures-core", version = "0.3.20", default-features = false }
+futures-task = { path = "../futures-task", version = "0.3.20", default-features = false }
+futures-util = { path = "../futures-util", version = "0.3.20", default-features = false }
 num_cpus = { version = "1.8.0", optional = true }
 
 [dev-dependencies]

futures-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-io"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.36"
 license = "MIT OR Apache-2.0"

futures-macro/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-macro"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"

futures-sink/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-sink"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.36"
 license = "MIT OR Apache-2.0"

futures-task/Cargo.toml

@@ -1,8 +1,8 @@
 [package]
 name = "futures-task"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
-rust-version = "1.36"
+rust-version = "1.45"
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang/futures-rs"
 homepage = "https://rust-lang.github.io/futures-rs"

futures-test/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-test"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"
@@ -11,13 +11,13 @@ Common utilities for testing components built off futures-rs.
 """
 
 [dependencies]
-futures-core = { version = "0.3.19", path = "../futures-core", default-features = false }
-futures-task = { version = "0.3.19", path = "../futures-task", default-features = false }
-futures-io = { version = "0.3.19", path = "../futures-io", default-features = false }
-futures-util = { version = "0.3.19", path = "../futures-util", default-features = false }
-futures-executor = { version = "0.3.19", path = "../futures-executor", default-features = false }
-futures-sink = { version = "0.3.19", path = "../futures-sink", default-features = false }
-futures-macro = { version = "=0.3.19", path = "../futures-macro", default-features = false }
+futures-core = { version = "0.3.20", path = "../futures-core", default-features = false }
+futures-task = { version = "0.3.20", path = "../futures-task", default-features = false }
+futures-io = { version = "0.3.20", path = "../futures-io", default-features = false }
+futures-util = { version = "0.3.20", path = "../futures-util", default-features = false }
+futures-executor = { version = "0.3.20", path = "../futures-executor", default-features = false }
+futures-sink = { version = "0.3.20", path = "../futures-sink", default-features = false }
+futures-macro = { version = "=0.3.20", path = "../futures-macro", default-features = false }
 pin-utils = { version = "0.1.0", default-features = false }
 pin-project = "1.0.1"
 

futures-util/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-util"
-version = "0.3.19"
+version = "0.3.20"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"
@@ -34,12 +34,12 @@ write-all-vectored = ["io"]
 cfg-target-has-atomic = []
 
 [dependencies]
-futures-core = { path = "../futures-core", version = "0.3.19", default-features = false }
-futures-task = { path = "../futures-task", version = "0.3.19", default-features = false }
-futures-channel = { path = "../futures-channel", version = "0.3.19", default-features = false, features = ["std"], optional = true }
-futures-io = { path = "../futures-io", version = "0.3.19", default-features = false, features = ["std"], optional = true }
-futures-sink = { path = "../futures-sink", version = "0.3.19", default-features = false, optional = true }
-futures-macro = { path = "../futures-macro", version = "=0.3.19", default-features = false, optional = true }
+futures-core = { path = "../futures-core", version = "0.3.20", default-features = false }
+futures-task = { path = "../futures-task", version = "0.3.20", default-features = false }
+futures-channel = { path = "../futures-channel", version = "0.3.20", default-features = false, features = ["std"], optional = true }
+futures-io = { path = "../futures-io", version = "0.3.20", default-features = false, features = ["std"], optional = true }
+futures-sink = { path = "../futures-sink", version = "0.3.20", default-features = false, optional = true }
+futures-macro = { path = "../futures-macro", version = "=0.3.20", default-features = false, optional = true }
 slab = { version = "0.4.2", optional = true }
 memchr = { version = "2.2", optional = true }
 futures_01 = { version = "0.1.25", optional = true, package = "futures" }

futures-util/benches/flatten_unordered.rs

@@ -0,0 +1,66 @@
+#![feature(test)]
+
+extern crate test;
+use crate::test::Bencher;
+
+use futures::channel::oneshot;
+use futures::executor::block_on;
+use futures::future::{self, FutureExt};
+use futures::stream::{self, StreamExt};
+use futures::task::Poll;
+use std::collections::VecDeque;
+use std::thread;
+
+#[bench]
+fn oneshot_streams(b: &mut Bencher) {
+    const STREAM_COUNT: usize = 10_000;
+    const STREAM_ITEM_COUNT: usize = 1;
+
+    b.iter(|| {
+        let mut txs = VecDeque::with_capacity(STREAM_COUNT);
+        let mut rxs = Vec::new();
+
+        for _ in 0..STREAM_COUNT {
+            let (tx, rx) = oneshot::channel();
+            txs.push_back(tx);
+            rxs.push(rx);
+        }
+
+        thread::spawn(move || {
+            let mut last = 1;
+            while let Some(tx) = txs.pop_front() {
+                let _ = tx.send(stream::iter(last..last + STREAM_ITEM_COUNT));
+                last += STREAM_ITEM_COUNT;
+            }
+        });
+
+        let mut flatten = stream::unfold(rxs.into_iter(), |mut vals| {
+            async {
+                if let Some(next) = vals.next() {
+                    let val = next.await.unwrap();
+                    Some((val, vals))
+                } else {
+                    None
+                }
+            }
+            .boxed()
+        })
+        .flatten_unordered(None);
+
+        block_on(future::poll_fn(move |cx| {
+            let mut count = 0;
+            loop {
+                match flatten.poll_next_unpin(cx) {
+                    Poll::Ready(None) => break,
+                    Poll::Ready(Some(_)) => {
+                        count += 1;
+                    }
+                    _ => {}
+                }
+            }
+            assert_eq!(count, STREAM_COUNT * STREAM_ITEM_COUNT);
+
+            Poll::Ready(())
+        }))
+    });
+}