Release 0.3.26

CHANGELOG.md

@@ -1,3 +1,13 @@
+# 0.3.26 - 2023-01-30
+
+* Add `Either::as_pin_mut` and `Either::as_pin_ref` (#2691)
+* Add `Shared::ptr_eq` and `Shared::ptr_hash` (#2691)
+* Implement `FusedStream` for `Buffered` (#2676)
+* Implement `FusedStream` for all streams in `ReadyChunks` (#2693)
+* Fix bug in `FuturesOrdered::push_front` (#2664)
+* Remove `Fut::Output: Clone` bounds from some `Shared` methods (#2662)
+* Remove `T: Debug` bounds from `Debug` implementations of `mpsc` and `oneshot` types (#2666, #2667)
+
 # 0.3.25 - 2022-10-20
 
 * Fix soundness issue in `join!` and `try_join!` macros (#2649)

README.md

@@ -8,16 +8,16 @@
 
 <p align="center">
  <a href="https:/rust-lang/futures-rs/actions?query=branch%3Amaster">
- <img alt="Build Status" src="https://img.shields.io/github/workflow/status/rust-lang/futures-rs/CI/master">
+ <img alt="Build Status" src="https://img.shields.io/github/actions/workflow/status/rust-lang/futures-rs/ci.yml?branch=master">
  </a>
 
  <a href="https://crates.io/crates/futures">
- <img alt="Crates.io" src="https://img.shields.io/crates/v/futures.svg">
+ <img alt="crates.io" src="https://img.shields.io/crates/v/futures.svg">
  </a>
 </p>
 
 <p align="center">
- <a href="https://docs.rs/futures/">
+ <a href="https://docs.rs/futures">
  Documentation
  </a> | <a href="https://rust-lang.github.io/futures-rs/">
  Website

examples/functional/src/main.rs

@@ -42,5 +42,5 @@ fn main() {
  // to drive all futures. Eventually fut_values will be driven to completion.
  let values: Vec<i32> = executor::block_on(fut_values);
 
- println!("Values={:?}", values);
+ println!("Values={values:?}");
 }

examples/imperative/src/main.rs

@@ -44,5 +44,5 @@ fn main() {
  // to drive all futures. Eventually fut_values will be driven to completion.
  let values: Vec<i32> = executor::block_on(fut_values);
 
- println!("Values={:?}", values);
+ println!("Values={values:?}");
 }

futures-channel/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-channel"
-version = "0.3.25"
+version = "0.3.26"
 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.25", default-features = false }
-futures-sink = { path = "../futures-sink", version = "0.3.25", default-features = false, optional = true }
+futures-core = { path = "../futures-core", version = "0.3.26", default-features = false }
+futures-sink = { path = "../futures-sink", version = "0.3.26", default-features = false, optional = true }
 
 [dev-dependencies]
 futures = { path = "../futures", default-features = true }

futures-channel/src/mpsc/mod.rs

@@ -94,13 +94,11 @@ mod queue;
 #[cfg(feature = "sink")]
 mod sink_impl;
 
-#[derive(Debug)]
 struct UnboundedSenderInner<T> {
  // Channel state shared between the sender and receiver.
  inner: Arc<UnboundedInner<T>>,
 }
 
-#[derive(Debug)]
 struct BoundedSenderInner<T> {
  // Channel state shared between the sender and receiver.
  inner: Arc<BoundedInner<T>>,
@@ -122,13 +120,11 @@ impl<T> Unpin for BoundedSenderInner<T> {}
 /// The transmission end of a bounded mpsc channel.
 ///
 /// This value is created by the [`channel`](channel) function.
-#[derive(Debug)]
 pub struct Sender<T>(Option<BoundedSenderInner<T>>);
 
 /// The transmission end of an unbounded mpsc channel.
 ///
 /// This value is created by the [`unbounded`](unbounded) function.
-#[derive(Debug)]
 pub struct UnboundedSender<T>(Option<UnboundedSenderInner<T>>);
 
 trait AssertKinds: Send + Sync + Clone {}
@@ -137,15 +133,13 @@ impl AssertKinds for UnboundedSender<u32> {}
 /// The receiving end of a bounded mpsc channel.
 ///
 /// This value is created by the [`channel`](channel) function.
-#[derive(Debug)]
 pub struct Receiver<T> {
  inner: Option<Arc<BoundedInner<T>>>,
 }
 
 /// The receiving end of an unbounded mpsc channel.
 ///
 /// This value is created by the [`unbounded`](unbounded) function.
-#[derive(Debug)]
 pub struct UnboundedReceiver<T> {
  inner: Option<Arc<UnboundedInner<T>>>,
 }
@@ -261,7 +255,6 @@ impl fmt::Display for TryRecvError {
 
 impl std::error::Error for TryRecvError {}
 
-#[derive(Debug)]
 struct UnboundedInner<T> {
  // Internal channel state. Consists of the number of messages stored in the
  // channel as well as a flag signalling that the channel is closed.
@@ -277,7 +270,6 @@ struct UnboundedInner<T> {
  recv_task: AtomicWaker,
 }
 
-#[derive(Debug)]
 struct BoundedInner<T> {
  // Max buffer size of the channel. If `None` then the channel is unbounded.
  buffer: usize,
@@ -300,7 +292,7 @@ struct BoundedInner<T> {
 }
 
 // Struct representation of `Inner::state`.
-#[derive(Debug, Clone, Copy)]
+#[derive(Clone, Copy)]
 struct State {
  // `true` when the channel is open
  is_open: bool,
@@ -324,7 +316,6 @@ const MAX_CAPACITY: usize = !(OPEN_MASK);
 const MAX_BUFFER: usize = MAX_CAPACITY >> 1;
 
 // Sent to the consumer to wake up blocked producers
-#[derive(Debug)]
 struct SenderTask {
  task: Option<Waker>,
  is_parked: bool,
@@ -947,6 +938,18 @@ impl<T> Drop for BoundedSenderInner<T> {
  }
 }
 
+impl<T> fmt::Debug for Sender<T> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("Sender").field("closed", &self.is_closed()).finish()
+ }
+}
+
+impl<T> fmt::Debug for UnboundedSender<T> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("UnboundedSender").field("closed", &self.is_closed()).finish()
+ }
+}
+
 /*
  *
  * ===== impl Receiver =====
@@ -1075,6 +1078,14 @@ impl<T> Stream for Receiver<T> {
  }
  }
  }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ if let Some(inner) = &self.inner {
+ decode_state(inner.state.load(SeqCst)).size_hint()
+ } else {
+ (0, Some(0))
+ }
+ }
 }
 
 impl<T> Drop for Receiver<T> {
@@ -1107,6 +1118,18 @@ impl<T> Drop for Receiver<T> {
  }
 }
 
+impl<T> fmt::Debug for Receiver<T> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let closed = if let Some(ref inner) = self.inner {
+ decode_state(inner.state.load(SeqCst)).is_closed()
+ } else {
+ false
+ };
+
+ f.debug_struct("Receiver").field("closed", &closed).finish()
+ }
+}
+
 impl<T> UnboundedReceiver<T> {
  /// Closes the receiving half of a channel, without dropping it.
  ///
@@ -1207,6 +1230,14 @@ impl<T> Stream for UnboundedReceiver<T> {
  }
  }
  }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ if let Some(inner) = &self.inner {
+ decode_state(inner.state.load(SeqCst)).size_hint()
+ } else {
+ (0, Some(0))
+ }
+ }
 }
 
 impl<T> Drop for UnboundedReceiver<T> {
@@ -1239,6 +1270,18 @@ impl<T> Drop for UnboundedReceiver<T> {
  }
 }
 
+impl<T> fmt::Debug for UnboundedReceiver<T> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let closed = if let Some(ref inner) = self.inner {
+ decode_state(inner.state.load(SeqCst)).is_closed()
+ } else {
+ false
+ };
+
+ f.debug_struct("Receiver").field("closed", &closed).finish()
+ }
+}
+
 /*
  *
  * ===== impl Inner =====
@@ -1285,6 +1328,14 @@ impl State {
  fn is_closed(&self) -> bool {
  !self.is_open && self.num_messages == 0
  }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ if self.is_open {
+ (self.num_messages, None)
+ } else {
+ (self.num_messages, Some(self.num_messages))
+ }
+ }
 }
 
 /*

futures-channel/src/mpsc/queue.rs

@@ -61,7 +61,6 @@ pub(super) enum PopResult<T> {
  Inconsistent,
 }
 
-#[derive(Debug)]
 struct Node<T> {
  next: AtomicPtr<Self>,
  value: Option<T>,
@@ -70,7 +69,6 @@ struct Node<T> {
 /// The multi-producer single-consumer structure. This is not cloneable, but it
 /// may be safely shared so long as it is guaranteed that there is only one
 /// popper at a time (many pushers are allowed).
-#[derive(Debug)]
 pub(super) struct Queue<T> {
  head: AtomicPtr<Node<T>>,
  tail: UnsafeCell<*mut Node<T>>,

futures-channel/src/oneshot.rs

@@ -390,7 +390,7 @@ impl<T> Drop for Sender<T> {
  }
 }
 
-impl<T: fmt::Debug> fmt::Debug for Sender<T> {
+impl<T> fmt::Debug for Sender<T> {
  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
  f.debug_struct("Sender").field("complete", &self.inner.complete).finish()
  }
@@ -481,7 +481,7 @@ impl<T> Drop for Receiver<T> {
  }
 }
 
-impl<T: fmt::Debug> fmt::Debug for Receiver<T> {
+impl<T> fmt::Debug for Receiver<T> {
  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
  f.debug_struct("Receiver").field("complete", &self.inner.complete).finish()
  }

futures-channel/tests/mpsc-size_hint.rs

@@ -0,0 +1,40 @@
+use futures::channel::mpsc;
+use futures::stream::Stream;
+
+#[test]
+fn unbounded_size_hint() {
+ let (tx, mut rx) = mpsc::unbounded::<u32>();
+ assert_eq!((0, None), rx.size_hint());
+ tx.unbounded_send(1).unwrap();
+ assert_eq!((1, None), rx.size_hint());
+ rx.try_next().unwrap().unwrap();
+ assert_eq!((0, None), rx.size_hint());
+ tx.unbounded_send(2).unwrap();
+ tx.unbounded_send(3).unwrap();
+ assert_eq!((2, None), rx.size_hint());
+ drop(tx);
+ assert_eq!((2, Some(2)), rx.size_hint());
+ rx.try_next().unwrap().unwrap();
+ assert_eq!((1, Some(1)), rx.size_hint());
+ rx.try_next().unwrap().unwrap();
+ assert_eq!((0, Some(0)), rx.size_hint());
+}
+
+#[test]
+fn channel_size_hint() {
+ let (mut tx, mut rx) = mpsc::channel::<u32>(10);
+ assert_eq!((0, None), rx.size_hint());
+ tx.try_send(1).unwrap();
+ assert_eq!((1, None), rx.size_hint());
+ rx.try_next().unwrap().unwrap();
+ assert_eq!((0, None), rx.size_hint());
+ tx.try_send(2).unwrap();
+ tx.try_send(3).unwrap();
+ assert_eq!((2, None), rx.size_hint());
+ drop(tx);
+ assert_eq!((2, Some(2)), rx.size_hint());
+ rx.try_next().unwrap().unwrap();
+ assert_eq!((1, Some(1)), rx.size_hint());
+ rx.try_next().unwrap().unwrap();
+ assert_eq!((0, Some(0)), rx.size_hint());
+}

futures-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-core"
-version = "0.3.25"
+version = "0.3.26"
 edition = "2018"
 rust-version = "1.36"
 license = "MIT OR Apache-2.0"
@@ -21,6 +21,7 @@ unstable = []
 cfg-target-has-atomic = []
 
 [dependencies]
+portable-atomic = { version = "1", default-features = false, optional = true }
 
 [dev-dependencies]
 futures = { path = "../futures" }

futures-core/src/task/__internal/atomic_waker.rs

@@ -1,9 +1,16 @@
 use core::cell::UnsafeCell;
 use core::fmt;
-use core::sync::atomic::AtomicUsize;
-use core::sync::atomic::Ordering::{AcqRel, Acquire, Release};
 use core::task::Waker;
 
+use atomic::AtomicUsize;
+use atomic::Ordering::{AcqRel, Acquire, Release};
+
+#[cfg(feature = "portable-atomic")]
+use portable_atomic as atomic;
+
+#[cfg(not(feature = "portable-atomic"))]
+use core::sync::atomic;
+
 /// A synchronization primitive for task wakeup.
 ///
 /// Sometimes the task interested in a given event will change over time.

futures-executor/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-executor"
-version = "0.3.25"
+version = "0.3.26"
 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.25", default-features = false }
-futures-task = { path = "../futures-task", version = "0.3.25", default-features = false }
-futures-util = { path = "../futures-util", version = "0.3.25", default-features = false }
+futures-core = { path = "../futures-core", version = "0.3.26", default-features = false }
+futures-task = { path = "../futures-task", version = "0.3.26", default-features = false }
+futures-util = { path = "../futures-util", version = "0.3.26", 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.25"
+version = "0.3.26"
 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.25"
+version = "0.3.26"
 edition = "2018"
 rust-version = "1.45"
 license = "MIT OR Apache-2.0"