Release 0.3.24

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.3.24 - 2022-08-29
+
+* Fix incorrect termination of `select_with_strategy` streams (#2635)
+
 # 0.3.23 - 2022-08-14
 
 * Work around MSRV increase due to a cargo bug.

futures-channel/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-channel"
-version = "0.3.23"
+version = "0.3.24"
 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.23", default-features = false }
-futures-sink = { path = "../futures-sink", version = "0.3.23", default-features = false, optional = true }
+futures-core = { path = "../futures-core", version = "0.3.24", default-features = false }
+futures-sink = { path = "../futures-sink", version = "0.3.24", 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.23"
+version = "0.3.24"
 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.23"
+version = "0.3.24"
 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.23", default-features = false }
-futures-task = { path = "../futures-task", version = "0.3.23", default-features = false }
-futures-util = { path = "../futures-util", version = "0.3.23", default-features = false }
+futures-core = { path = "../futures-core", version = "0.3.24", default-features = false }
+futures-task = { path = "../futures-task", version = "0.3.24", default-features = false }
+futures-util = { path = "../futures-util", version = "0.3.24", 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.23"
+version = "0.3.24"
 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.23"
+version = "0.3.24"
 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.23"
+version = "0.3.24"
 edition = "2018"
 rust-version = "1.36"
 license = "MIT OR Apache-2.0"

futures-task/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-task"
-version = "0.3.23"
+version = "0.3.24"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"

futures-test/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-test"
-version = "0.3.23"
+version = "0.3.24"
 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.23", path = "../futures-core", default-features = false }
-futures-task = { version = "0.3.23", path = "../futures-task", default-features = false }
-futures-io = { version = "0.3.23", path = "../futures-io", default-features = false }
-futures-util = { version = "0.3.23", path = "../futures-util", default-features = false }
-futures-executor = { version = "0.3.23", path = "../futures-executor", default-features = false }
-futures-sink = { version = "0.3.23", path = "../futures-sink", default-features = false }
-futures-macro = { version = "=0.3.23", path = "../futures-macro", default-features = false }
+futures-core = { version = "0.3.24", path = "../futures-core", default-features = false }
+futures-task = { version = "0.3.24", path = "../futures-task", default-features = false }
+futures-io = { version = "0.3.24", path = "../futures-io", default-features = false }
+futures-util = { version = "0.3.24", path = "../futures-util", default-features = false }
+futures-executor = { version = "0.3.24", path = "../futures-executor", default-features = false }
+futures-sink = { version = "0.3.24", path = "../futures-sink", default-features = false }
+futures-macro = { version = "=0.3.24", path = "../futures-macro", default-features = false }
 pin-utils = { version = "0.1.0", default-features = false }
 pin-project = "1.0.11"
 

futures-util/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-util"
-version = "0.3.23"
+version = "0.3.24"
 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.23", default-features = false }
-futures-task = { path = "../futures-task", version = "0.3.23", default-features = false }
-futures-channel = { path = "../futures-channel", version = "0.3.23", default-features = false, features = ["std"], optional = true }
-futures-io = { path = "../futures-io", version = "0.3.23", default-features = false, features = ["std"], optional = true }
-futures-sink = { path = "../futures-sink", version = "0.3.23", default-features = false, optional = true }
-futures-macro = { path = "../futures-macro", version = "=0.3.23", default-features = false, optional = true }
+futures-core = { path = "../futures-core", version = "0.3.24", default-features = false }
+futures-task = { path = "../futures-task", version = "0.3.24", default-features = false }
+futures-channel = { path = "../futures-channel", version = "0.3.24", default-features = false, features = ["std"], optional = true }
+futures-io = { path = "../futures-io", version = "0.3.24", default-features = false, features = ["std"], optional = true }
+futures-sink = { path = "../futures-sink", version = "0.3.24", default-features = false, optional = true }
+futures-macro = { path = "../futures-macro", version = "=0.3.24", 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/src/stream/select_with_strategy.rs

@@ -231,18 +231,23 @@ where
     St1: Stream,
     St2: Stream<Item = St1::Item>,
 {
-    match poll_side(select, side, cx) {
+    let first_done = match poll_side(select, side, cx) {
         Poll::Ready(Some(item)) => return Poll::Ready(Some(item)),
         Poll::Ready(None) => {
             select.internal_state.finish(side);
+            true
         }
-        Poll::Pending => (),
+        Poll::Pending => false,
     };
     let other = side.other();
     match poll_side(select, other, cx) {
         Poll::Ready(None) => {
             select.internal_state.finish(other);
-            Poll::Ready(None)
+            if first_done {
+                Poll::Ready(None)
+            } else {
+                Poll::Pending
+            }
         }
         a => a,
     }

futures/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures"
-version = "0.3.23"
+version = "0.3.24"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"
@@ -15,13 +15,13 @@ composability, and iterator-like interfaces.
 categories = ["asynchronous"]
 
 [dependencies]
-futures-core = { path = "../futures-core", version = "0.3.23", default-features = false }
-futures-task = { path = "../futures-task", version = "0.3.23", default-features = false }
-futures-channel = { path = "../futures-channel", version = "0.3.23", default-features = false, features = ["sink"] }
-futures-executor = { path = "../futures-executor", version = "0.3.23", default-features = false, optional = true }
-futures-io = { path = "../futures-io", version = "0.3.23", default-features = false }
-futures-sink = { path = "../futures-sink", version = "0.3.23", default-features = false }
-futures-util = { path = "../futures-util", version = "0.3.23", default-features = false, features = ["sink"] }
+futures-core = { path = "../futures-core", version = "0.3.24", default-features = false }
+futures-task = { path = "../futures-task", version = "0.3.24", default-features = false }
+futures-channel = { path = "../futures-channel", version = "0.3.24", default-features = false, features = ["sink"] }
+futures-executor = { path = "../futures-executor", version = "0.3.24", default-features = false, optional = true }
+futures-io = { path = "../futures-io", version = "0.3.24", default-features = false }
+futures-sink = { path = "../futures-sink", version = "0.3.24", default-features = false }
+futures-util = { path = "../futures-util", version = "0.3.24", default-features = false, features = ["sink"] }
 
 [dev-dependencies]
 futures-executor = { path = "../futures-executor", features = ["thread-pool"] }

futures/tests/stream.rs

@@ -1,5 +1,9 @@
+use std::cell::Cell;
 use std::iter;
+use std::pin::Pin;
+use std::rc::Rc;
 use std::sync::Arc;
+use std::task::Context;
 
 use futures::channel::mpsc;
 use futures::executor::block_on;
@@ -9,6 +13,7 @@ use futures::sink::SinkExt;
 use futures::stream::{self, StreamExt};
 use futures::task::Poll;
 use futures::{ready, FutureExt};
+use futures_core::Stream;
 use futures_test::task::noop_context;
 
 #[test]
@@ -419,3 +424,40 @@ fn ready_chunks() {
         assert_eq!(s.next().await.unwrap(), vec![4]);
     });
 }
+
+struct SlowStream {
+    times_should_poll: usize,
+    times_polled: Rc<Cell<usize>>,
+}
+impl Stream for SlowStream {
+    type Item = usize;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.times_polled.set(self.times_polled.get() + 1);
+        if self.times_polled.get() % 2 == 0 {
+            cx.waker().wake_by_ref();
+            return Poll::Pending;
+        }
+        if self.times_polled.get() >= self.times_should_poll {
+            return Poll::Ready(None);
+        }
+        Poll::Ready(Some(self.times_polled.get()))
+    }
+}
+
+#[test]
+fn select_with_strategy_doesnt_terminate_early() {
+    for side in [stream::PollNext::Left, stream::PollNext::Right] {
+        let times_should_poll = 10;
+        let count = Rc::new(Cell::new(0));
+        let b = stream::iter([10, 20]);
+
+        let mut selected = stream::select_with_strategy(
+            SlowStream { times_should_poll, times_polled: count.clone() },
+            b,
+            |_: &mut ()| side,
+        );
+        block_on(async move { while selected.next().await.is_some() {} });
+        assert_eq!(count.get(), times_should_poll + 1);
+    }
+}

no_atomic_cas.rs

@@ -2,6 +2,7 @@
 // It is not intended for manual editing.
 
 const NO_ATOMIC_CAS: &[&str] = &[
+    "armv4t-none-eabi",
     "avr-unknown-gnu-atmega328",
     "bpfeb-unknown-none",
     "bpfel-unknown-none",