diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 50557c4bbd..aafcfee053 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -47,8 +47,9 @@ jobs: - os: ubuntu-latest target: i686-unknown-linux-gnu runs-on: ${{ matrix.os }} + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust # --no-self-update is necessary because the windows environment cannot self-update rustup.exe. run: rustup update nightly --no-self-update && rustup default nightly @@ -70,8 +71,9 @@ jobs: # When updating this, the reminder to update the minimum required version in README.md and Cargo.toml. - '1.36' runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update ${{ matrix.rust }} && rustup default ${{ matrix.rust }} # cargo does not support for --features/--no-default-features with workspace, so use cargo-hack instead. @@ -104,8 +106,9 @@ jobs: # When updating this, the reminder to update the minimum required version in README.md and Cargo.toml. - '1.56' runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update ${{ matrix.rust }} && rustup default ${{ matrix.rust }} - name: Install cargo-hack @@ -136,8 +139,9 @@ jobs: - beta - nightly runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update ${{ matrix.rust }} && rustup default ${{ matrix.rust }} - name: Install cargo-hack @@ -148,8 +152,9 @@ jobs: minimal-versions: name: cargo minimal-versions build runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update nightly && rustup default nightly - name: Install cargo-hack @@ -169,8 +174,9 @@ jobs: - thumbv7m-none-eabi - thumbv6m-none-eabi runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update nightly && rustup default nightly - run: rustup target add ${{ matrix.target }} @@ -201,8 +207,9 @@ jobs: bench: name: cargo bench runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update nightly && rustup default nightly - run: cargo bench --workspace @@ -211,8 +218,9 @@ jobs: features: name: cargo hack check --feature-powerset runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update nightly && rustup default nightly - name: Install cargo-hack @@ -233,14 +241,21 @@ jobs: miri: name: cargo miri test runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup toolchain install nightly --component miri && rustup default nightly - - run: cargo miri test --workspace --all-features + - run: cargo miri test --workspace --all-features -- --skip panic_on_drop_fut env: MIRIFLAGS: -Zmiri-strict-provenance -Zmiri-symbolic-alignment-check -Zmiri-disable-isolation RUSTFLAGS: ${{ env.RUSTFLAGS }} -Z randomize-layout + # This test is expected to leak. + - run: cargo miri test --workspace --all-features --test stream_futures_unordered -- panic_on_drop_fut + env: + MIRIFLAGS: -Zmiri-strict-provenance -Zmiri-symbolic-alignment-check -Zmiri-disable-isolation -Zmiri-ignore-leaks + RUSTDOCFLAGS: ${{ env.RUSTDOCFLAGS }} -Z randomize-layout + RUSTFLAGS: ${{ env.RUSTFLAGS }} -Z randomize-layout san: name: cargo test -Z sanitizer=${{ matrix.sanitizer }} @@ -252,11 +267,14 @@ jobs: - memory - thread runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup toolchain install nightly --component rust-src && rustup default nightly - - run: cargo -Z build-std test --workspace --all-features --target x86_64-unknown-linux-gnu --lib --tests + # https://github.com/google/sanitizers/issues/1716 / https://github.com/actions/runner-images/issues/9491 + - run: sudo sysctl vm.mmap_rnd_bits=28 + - run: cargo -Z build-std test --workspace --all-features --target x86_64-unknown-linux-gnu --lib --tests -- --skip panic_on_drop_fut env: # TODO: Once `cfg(sanitize = "..")` is stable, replace # `cfg(futures_sanitizer)` with `cfg(sanitize = "..")` and remove @@ -268,8 +286,9 @@ jobs: # clippy: # name: cargo clippy # runs-on: ubuntu-latest + # timeout-minutes: 60 # steps: - # - uses: actions/checkout@v4 + # - uses: taiki-e/checkout-action@v1 # - name: Install Rust # run: rustup toolchain install nightly --component clippy && rustup default nightly # - run: cargo clippy --workspace --all-features --all-targets @@ -277,8 +296,9 @@ jobs: fmt: name: cargo fmt runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update stable - run: cargo fmt --all -- --check @@ -286,8 +306,9 @@ jobs: docs: name: cargo doc runs-on: ubuntu-latest + timeout-minutes: 60 steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - name: Install Rust run: rustup update nightly && rustup default nightly - run: RUSTDOCFLAGS="-D warnings --cfg docsrs" cargo doc --workspace --no-deps --all-features diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5fec08e5bd..d56fcdddc3 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -10,7 +10,7 @@ jobs: if: github.repository_owner == 'rust-lang' runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: taiki-e/checkout-action@v1 - uses: taiki-e/create-gh-release-action@v1 with: changelog: CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md index a1b633b2bc..992a457a1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,16 @@ +# 0.3.31 - 2024-10-05 + +* Fix use after free of task in `FuturesUnordered` when dropped future panics (#2886) +* Fix soundness bug in `task::waker_ref` (#2830) + This is a breaking change but allowed because it is soundness bug fix. +* Fix bugs in `AsyncBufRead::read_line` and `AsyncBufReadExt::lines` (#2884) +* Fix parsing issue in `select!`/`select_biased!` (#2832) + This is technically a breaking change as it will now reject a very odd undocumented syntax that was previously accidentally accepted. +* Work around issue due to upstream `Waker::will_wake` change (#2865) +* Add `stream::Iter::{get_ref,get_mut,into_inner}` (#2875) +* Add `future::AlwaysReady` (#2825) +* Relax trait bound on non-constructor methods of `io::{BufReader,BufWriter}` (#2848) + # 0.3.30 - 2023-12-24 * Add `{BiLock,SplitStream,SplitSink,ReadHalf,WriteHalf}::is_pair_of` (#2797) diff --git a/Cargo.toml b/Cargo.toml index d27a9f2885..1a90362f31 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,3 +18,10 @@ members = [ "examples/functional", "examples/imperative", ] + +[workspace.lints.rust] +missing_debug_implementations = "warn" +rust_2018_idioms = "warn" +single_use_lifetimes = "warn" +unreachable_pub = "warn" +unexpected_cfgs = { level = "warn", check-cfg = ['cfg(futures_sanitizer)'] } diff --git a/examples/functional/Cargo.toml b/examples/functional/Cargo.toml index 7b8b494d98..bb95a6f6a3 100644 --- a/examples/functional/Cargo.toml +++ b/examples/functional/Cargo.toml @@ -1,8 +1,11 @@ [package] name = "futures-example-functional" +version = "0.0.0" edition = "2018" -version = "0.1.0" publish = false [dependencies] futures = { path = "../../futures", features = ["thread-pool"] } + +[lints] +workspace = true diff --git a/examples/imperative/Cargo.toml b/examples/imperative/Cargo.toml index 3405451f00..afd7a20428 100644 --- a/examples/imperative/Cargo.toml +++ b/examples/imperative/Cargo.toml @@ -1,8 +1,11 @@ [package] name = "futures-example-imperative" +version = "0.0.0" edition = "2018" -version = "0.1.0" publish = false [dependencies] futures = { path = "../../futures", features = ["thread-pool"] } + +[lints] +workspace = true diff --git a/futures-channel/Cargo.toml b/futures-channel/Cargo.toml index 7e0e9dde2c..4d430b78d7 100644 --- a/futures-channel/Cargo.toml +++ b/futures-channel/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-channel" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.56" license = "MIT OR Apache-2.0" @@ -22,8 +22,8 @@ unstable = [] cfg-target-has-atomic = [] [dependencies] -futures-core = { path = "../futures-core", version = "0.3.30", default-features = false } -futures-sink = { path = "../futures-sink", version = "0.3.30", default-features = false, optional = true } +futures-core = { path = "../futures-core", version = "0.3.31", default-features = false } +futures-sink = { path = "../futures-sink", version = "0.3.31", default-features = false, optional = true } [dev-dependencies] futures = { path = "../futures", default-features = true } @@ -32,3 +32,6 @@ futures-test = { path = "../futures-test", default-features = true } [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] + +[lints] +workspace = true diff --git a/futures-channel/src/lib.rs b/futures-channel/src/lib.rs index f611e6b9f5..09d6a1e2de 100644 --- a/futures-channel/src/lib.rs +++ b/futures-channel/src/lib.rs @@ -11,14 +11,7 @@ //! All items are only available when the `std` or `alloc` feature of this //! library is activated, and it is activated by default. -#![cfg_attr(not(feature = "std"), no_std)] -#![warn( - missing_debug_implementations, - missing_docs, - rust_2018_idioms, - single_use_lifetimes, - unreachable_pub -)] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -26,10 +19,13 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, unsafe_op_in_unsafe_fn)] #[cfg_attr(target_os = "none", cfg(target_has_atomic = "ptr"))] #[cfg(feature = "alloc")] extern crate alloc; +#[cfg(feature = "std")] +extern crate std; #[cfg_attr(target_os = "none", cfg(target_has_atomic = "ptr"))] #[cfg(feature = "alloc")] diff --git a/futures-channel/src/mpsc/mod.rs b/futures-channel/src/mpsc/mod.rs index 64f7526fa4..03a8a53898 100644 --- a/futures-channel/src/mpsc/mod.rs +++ b/futures-channel/src/mpsc/mod.rs @@ -127,6 +127,7 @@ pub struct Sender(Option>); /// This value is created by the [`unbounded`] function. pub struct UnboundedSender(Option>); +#[allow(dead_code)] trait AssertKinds: Send + Sync + Clone {} impl AssertKinds for UnboundedSender {} @@ -302,13 +303,13 @@ struct State { } // The `is_open` flag is stored in the left-most bit of `Inner::state` -const OPEN_MASK: usize = usize::max_value() - (usize::max_value() >> 1); +const OPEN_MASK: usize = usize::MAX - (usize::MAX >> 1); // When a new channel is created, it is created in the open state with no // pending messages. const INIT_STATE: usize = OPEN_MASK; -// The maximum number of messages that a channel can track is `usize::max_value() >> 1` +// The maximum number of messages that a channel can track is `usize::MAX >> 1` const MAX_CAPACITY: usize = !(OPEN_MASK); // The maximum requested buffer size must be less than the maximum capacity of diff --git a/futures-channel/src/mpsc/queue.rs b/futures-channel/src/mpsc/queue.rs index 02ec633fe0..78cbdce6ec 100644 --- a/futures-channel/src/mpsc/queue.rs +++ b/futures-channel/src/mpsc/queue.rs @@ -43,6 +43,7 @@ pub(super) use self::PopResult::*; +use std::boxed::Box; use std::cell::UnsafeCell; use std::ptr; use std::sync::atomic::{AtomicPtr, Ordering}; @@ -113,22 +114,24 @@ impl Queue { /// /// This function is unsafe because only one thread can call it at a time. pub(super) unsafe fn pop(&self) -> PopResult { - let tail = *self.tail.get(); - let next = (*tail).next.load(Ordering::Acquire); - - if !next.is_null() { - *self.tail.get() = next; - assert!((*tail).value.is_none()); - assert!((*next).value.is_some()); - let ret = (*next).value.take().unwrap(); - drop(Box::from_raw(tail)); - return Data(ret); - } + unsafe { + let tail = *self.tail.get(); + let next = (*tail).next.load(Ordering::Acquire); + + if !next.is_null() { + *self.tail.get() = next; + assert!((*tail).value.is_none()); + assert!((*next).value.is_some()); + let ret = (*next).value.take().unwrap(); + drop(Box::from_raw(tail)); + return Data(ret); + } - if self.head.load(Ordering::Acquire) == tail { - Empty - } else { - Inconsistent + if self.head.load(Ordering::Acquire) == tail { + Empty + } else { + Inconsistent + } } } @@ -138,7 +141,7 @@ impl Queue { /// This function is unsafe because only one thread can call it at a time. pub(super) unsafe fn pop_spin(&self) -> Option { loop { - match self.pop() { + match unsafe { self.pop() } { Empty => return None, Data(t) => return Some(t), // Inconsistent means that there will be a message to pop diff --git a/futures-channel/tests/mpsc-close.rs b/futures-channel/tests/mpsc-close.rs index 1a14067eca..2d49936fc1 100644 --- a/futures-channel/tests/mpsc-close.rs +++ b/futures-channel/tests/mpsc-close.rs @@ -174,10 +174,10 @@ fn stress_try_send_as_receiver_closes() { } impl TestTask { /// Create a new TestTask - fn new() -> (TestTask, mpsc::Sender) { + fn new() -> (Self, mpsc::Sender) { let (command_tx, command_rx) = mpsc::channel::(0); ( - TestTask { + Self { command_rx, test_rx: None, countdown: 0, // 0 means no countdown is in progress. diff --git a/futures-channel/tests/mpsc.rs b/futures-channel/tests/mpsc.rs index f5d7198d22..28d476b05a 100644 --- a/futures-channel/tests/mpsc.rs +++ b/futures-channel/tests/mpsc.rs @@ -10,6 +10,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::thread; +#[allow(dead_code)] trait AssertSend: Send {} impl AssertSend for mpsc::Sender {} impl AssertSend for mpsc::Receiver {} diff --git a/futures-core/Cargo.toml b/futures-core/Cargo.toml index 0e27db0a66..a7d710e6a5 100644 --- a/futures-core/Cargo.toml +++ b/futures-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-core" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.36" license = "MIT OR Apache-2.0" @@ -29,3 +29,6 @@ futures = { path = "../futures" } [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] + +[lints] +workspace = true diff --git a/futures-core/src/future.rs b/futures-core/src/future.rs index 7540cd027e..30c0323a6b 100644 --- a/futures-core/src/future.rs +++ b/futures-core/src/future.rs @@ -9,10 +9,20 @@ pub use core::future::Future; /// An owned dynamically typed [`Future`] for use in cases where you can't /// statically type your result or need to add some indirection. +/// +/// This type is often created by the [`boxed`] method on [`FutureExt`]. See its documentation for more. +/// +/// [`boxed`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html#method.boxed +/// [`FutureExt`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html #[cfg(feature = "alloc")] pub type BoxFuture<'a, T> = Pin + Send + 'a>>; /// `BoxFuture`, but without the `Send` requirement. +/// +/// This type is often created by the [`boxed_local`] method on [`FutureExt`]. See its documentation for more. +/// +/// [`boxed_local`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html#method.boxed_local +/// [`FutureExt`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html #[cfg(feature = "alloc")] pub type LocalBoxFuture<'a, T> = Pin + 'a>>; diff --git a/futures-core/src/lib.rs b/futures-core/src/lib.rs index 9c31d8d90b..6ff6b974b1 100644 --- a/futures-core/src/lib.rs +++ b/futures-core/src/lib.rs @@ -1,9 +1,6 @@ //! Core traits and types for asynchronous operations in Rust. -#![cfg_attr(not(feature = "std"), no_std)] -#![warn(missing_debug_implementations, missing_docs, rust_2018_idioms, unreachable_pub)] -// It cannot be included in the published code because this lints have false positives in the minimum required version. -#![cfg_attr(test, warn(single_use_lifetimes))] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -11,9 +8,12 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, /* unsafe_op_in_unsafe_fn */)] // unsafe_op_in_unsafe_fn requires Rust 1.52 #[cfg(feature = "alloc")] extern crate alloc; +#[cfg(feature = "std")] +extern crate std; pub mod future; #[doc(no_inline)] diff --git a/futures-core/src/stream.rs b/futures-core/src/stream.rs index ad5350b795..dd07d5aa01 100644 --- a/futures-core/src/stream.rs +++ b/futures-core/src/stream.rs @@ -6,10 +6,20 @@ use core::task::{Context, Poll}; /// An owned dynamically typed [`Stream`] for use in cases where you can't /// statically type your result or need to add some indirection. +/// +/// This type is often created by the [`boxed`] method on [`StreamExt`]. See its documentation for more. +/// +/// [`boxed`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html#method.boxed +/// [`StreamExt`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html #[cfg(feature = "alloc")] pub type BoxStream<'a, T> = Pin + Send + 'a>>; /// `BoxStream`, but without the `Send` requirement. +/// +/// This type is often created by the [`boxed_local`] method on [`StreamExt`]. See its documentation for more. +/// +/// [`boxed_local`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html#method.boxed_local +/// [`StreamExt`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html #[cfg(feature = "alloc")] pub type LocalBoxStream<'a, T> = Pin + 'a>>; @@ -38,15 +48,15 @@ pub trait Stream { /// stream state: /// /// - `Poll::Pending` means that this stream's next value is not ready - /// yet. Implementations will ensure that the current task will be notified - /// when the next value may be ready. + /// yet. Implementations will ensure that the current task will be notified + /// when the next value may be ready. /// /// - `Poll::Ready(Some(val))` means that the stream has successfully - /// produced a value, `val`, and may produce further values on subsequent - /// `poll_next` calls. + /// produced a value, `val`, and may produce further values on subsequent + /// `poll_next` calls. /// /// - `Poll::Ready(None)` means that the stream has terminated, and - /// `poll_next` should not be invoked again. + /// `poll_next` should not be invoked again. /// /// # Panics /// diff --git a/futures-core/src/task/__internal/atomic_waker.rs b/futures-core/src/task/__internal/atomic_waker.rs index 2fc594b8a9..3b82fb7cc8 100644 --- a/futures-core/src/task/__internal/atomic_waker.rs +++ b/futures-core/src/task/__internal/atomic_waker.rs @@ -206,6 +206,7 @@ impl AtomicWaker { /// Create an `AtomicWaker`. pub const fn new() -> Self { // Make sure that task is Sync + #[allow(dead_code)] trait AssertSync: Sync {} impl AssertSync for Waker {} diff --git a/futures-executor/Cargo.toml b/futures-executor/Cargo.toml index 391a5adc4d..978f8ae018 100644 --- a/futures-executor/Cargo.toml +++ b/futures-executor/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-executor" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.56" 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.30", default-features = false } -futures-task = { path = "../futures-task", version = "0.3.30", default-features = false } -futures-util = { path = "../futures-util", version = "0.3.30", default-features = false } +futures-core = { path = "../futures-core", version = "0.3.31", default-features = false } +futures-task = { path = "../futures-task", version = "0.3.31", default-features = false } +futures-util = { path = "../futures-util", version = "0.3.31", default-features = false } num_cpus = { version = "1.8.0", optional = true } [dev-dependencies] @@ -27,3 +27,6 @@ futures = { path = "../futures", features = ["thread-pool"] } [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] + +[lints] +workspace = true diff --git a/futures-executor/src/enter.rs b/futures-executor/src/enter.rs index cb58c30bb7..b0e43e0593 100644 --- a/futures-executor/src/enter.rs +++ b/futures-executor/src/enter.rs @@ -1,7 +1,7 @@ use std::cell::Cell; use std::fmt; -thread_local!(static ENTERED: Cell = Cell::new(false)); +std::thread_local!(static ENTERED: Cell = Cell::new(false)); /// Represents an executor context. /// diff --git a/futures-executor/src/lib.rs b/futures-executor/src/lib.rs index b1af87545f..a0dc49d7ba 100644 --- a/futures-executor/src/lib.rs +++ b/futures-executor/src/lib.rs @@ -36,14 +36,7 @@ //! [`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 -#![cfg_attr(not(feature = "std"), no_std)] -#![warn( - missing_debug_implementations, - missing_docs, - rust_2018_idioms, - single_use_lifetimes, - unreachable_pub -)] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -51,8 +44,12 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, unsafe_op_in_unsafe_fn)] #![cfg_attr(docsrs, feature(doc_cfg))] +#[cfg(feature = "std")] +extern crate std; + #[cfg(feature = "std")] mod local_pool; #[cfg(feature = "std")] diff --git a/futures-executor/src/local_pool.rs b/futures-executor/src/local_pool.rs index 8a9bc2fc90..90c2a41520 100644 --- a/futures-executor/src/local_pool.rs +++ b/futures-executor/src/local_pool.rs @@ -15,6 +15,7 @@ use std::sync::{ Arc, }; use std::thread::{self, Thread}; +use std::vec::Vec; /// A single-threaded task pool for polling futures to completion. /// @@ -33,8 +34,7 @@ pub struct LocalPool { incoming: Rc, } -/// A handle to a [`LocalPool`](LocalPool) that implements -/// [`Spawn`](futures_task::Spawn). +/// A handle to a [`LocalPool`] that implements [`Spawn`](futures_task::Spawn). #[derive(Clone, Debug)] pub struct LocalSpawner { incoming: Weak, @@ -53,7 +53,7 @@ pub(crate) struct ThreadNotify { unparked: AtomicBool, } -thread_local! { +std::thread_local! { static CURRENT_THREAD_NOTIFY: Arc = Arc::new(ThreadNotify { thread: thread::current(), unparked: AtomicBool::new(false), @@ -310,8 +310,7 @@ impl Default for LocalPool { /// /// This function will block the caller until the given future has completed. /// -/// Use a [`LocalPool`](LocalPool) if you need finer-grained control over -/// spawned tasks. +/// Use a [`LocalPool`] if you need finer-grained control over spawned tasks. pub fn block_on(f: F) -> F::Output { pin_mut!(f); run_executor(|cx| f.as_mut().poll(cx)) diff --git a/futures-executor/src/thread_pool.rs b/futures-executor/src/thread_pool.rs index 5371008953..c4442e4eeb 100644 --- a/futures-executor/src/thread_pool.rs +++ b/futures-executor/src/thread_pool.rs @@ -5,9 +5,12 @@ use futures_core::task::{Context, Poll}; use futures_task::{waker_ref, ArcWake}; use futures_task::{FutureObj, Spawn, SpawnError}; use futures_util::future::FutureExt; +use std::boxed::Box; use std::cmp; use std::fmt; +use std::format; use std::io; +use std::string::String; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc; use std::sync::{Arc, Mutex}; @@ -42,6 +45,7 @@ pub struct ThreadPoolBuilder { before_stop: Option>, } +#[allow(dead_code)] trait AssertSendSync: Send + Sync {} impl AssertSendSync for ThreadPool {} @@ -358,7 +362,6 @@ impl ArcWake for WakeHandle { #[cfg(test)] mod tests { use super::*; - use std::sync::mpsc; #[test] fn test_drop_after_start() { diff --git a/futures-executor/src/unpark_mutex.rs b/futures-executor/src/unpark_mutex.rs index ac5112cfa2..f45164be9d 100644 --- a/futures-executor/src/unpark_mutex.rs +++ b/futures-executor/src/unpark_mutex.rs @@ -108,7 +108,7 @@ impl UnparkMutex { /// Callable only from the `POLLING`/`REPOLL` states, i.e. between /// successful calls to `notify` and `wait`/`complete`. pub(crate) unsafe fn wait(&self, data: D) -> Result<(), D> { - *self.inner.get() = Some(data); + unsafe { *self.inner.get() = Some(data) } match self.status.compare_exchange(POLLING, WAITING, SeqCst, SeqCst) { // no unparks came in while we were running @@ -119,7 +119,7 @@ impl UnparkMutex { Err(status) => { assert_eq!(status, REPOLL); self.status.store(POLLING, SeqCst); - Err((*self.inner.get()).take().unwrap()) + Err(unsafe { (*self.inner.get()).take().unwrap() }) } } } diff --git a/futures-executor/tests/local_pool.rs b/futures-executor/tests/local_pool.rs index 72ce74b744..d9b2e9d797 100644 --- a/futures-executor/tests/local_pool.rs +++ b/futures-executor/tests/local_pool.rs @@ -3,6 +3,7 @@ use futures::executor::LocalPool; use futures::future::{self, lazy, poll_fn, Future}; use futures::task::{Context, LocalSpawn, LocalSpawnExt, Poll, Spawn, SpawnExt, Waker}; use std::cell::{Cell, RefCell}; +use std::marker::PhantomData; use std::pin::Pin; use std::rc::Rc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -10,7 +11,7 @@ use std::sync::Arc; use std::thread; use std::time::Duration; -struct Pending(Rc<()>); +struct Pending(PhantomData>); impl Future for Pending { type Output = (); @@ -21,7 +22,7 @@ impl Future for Pending { } fn pending() -> Pending { - Pending(Rc::new(())) + Pending(PhantomData) } #[test] diff --git a/futures-io/Cargo.toml b/futures-io/Cargo.toml index 9911316ac4..e5320d4281 100644 --- a/futures-io/Cargo.toml +++ b/futures-io/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-io" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.36" license = "MIT OR Apache-2.0" @@ -24,3 +24,6 @@ unstable = [] [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] + +[lints] +workspace = true diff --git a/futures-io/src/lib.rs b/futures-io/src/lib.rs index e91eb78492..9c2fa056ad 100644 --- a/futures-io/src/lib.rs +++ b/futures-io/src/lib.rs @@ -8,10 +8,7 @@ //! All items of this library are only available when the `std` feature of this //! library is activated, and it is activated by default. -#![cfg_attr(not(feature = "std"), no_std)] -#![warn(missing_debug_implementations, missing_docs, rust_2018_idioms, unreachable_pub)] -// It cannot be included in the published code because this lints have false positives in the minimum required version. -#![cfg_attr(test, warn(single_use_lifetimes))] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -19,14 +16,20 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, /* unsafe_op_in_unsafe_fn */)] // unsafe_op_in_unsafe_fn requires Rust 1.52 #![cfg_attr(docsrs, feature(doc_cfg))] +#[cfg(feature = "std")] +extern crate std; + #[cfg(feature = "std")] mod if_std { + use std::boxed::Box; use std::io; use std::ops::DerefMut; use std::pin::Pin; use std::task::{Context, Poll}; + use std::vec::Vec; // Re-export some types from `std::io` so that users don't have to deal // with conflicts when `use`ing `futures::io` and `std::io`. diff --git a/futures-macro/Cargo.toml b/futures-macro/Cargo.toml index 6bd4baff12..fdd7be3dd8 100644 --- a/futures-macro/Cargo.toml +++ b/futures-macro/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-macro" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.56" license = "MIT OR Apache-2.0" @@ -18,4 +18,7 @@ proc-macro = true [dependencies] proc-macro2 = "1.0.60" quote = "1.0" -syn = { version = "2.0.8", features = ["full"] } +syn = { version = "2.0.52", features = ["full"] } + +[lints] +workspace = true diff --git a/futures-macro/src/join.rs b/futures-macro/src/join.rs index 94e356f729..0e891b3aad 100644 --- a/futures-macro/src/join.rs +++ b/futures-macro/src/join.rs @@ -1,4 +1,4 @@ -//! The futures-rs `join! macro implementation. +//! The futures-rs `join!` macro implementation. use proc_macro::TokenStream; use proc_macro2::{Span, TokenStream as TokenStream2}; diff --git a/futures-macro/src/lib.rs b/futures-macro/src/lib.rs index 0afe34b83b..c1b94785e8 100644 --- a/futures-macro/src/lib.rs +++ b/futures-macro/src/lib.rs @@ -1,6 +1,5 @@ //! The futures-rs procedural macro implementations. -#![warn(rust_2018_idioms, single_use_lifetimes, unreachable_pub)] #![doc(test( no_crate_inject, attr( @@ -9,11 +8,6 @@ ) ))] -// Since https://github.com/rust-lang/cargo/pull/7700 `proc_macro` is part of the prelude for -// proc-macro crates, but to support older compilers we still need this explicit `extern crate`. -#[allow(unused_extern_crates)] -extern crate proc_macro; - use proc_macro::TokenStream; mod executor; diff --git a/futures-macro/src/select.rs b/futures-macro/src/select.rs index 2789b3e659..ca83a87e4a 100644 --- a/futures-macro/src/select.rs +++ b/futures-macro/src/select.rs @@ -1,4 +1,4 @@ -//! The futures-rs `select! macro implementation. +//! The futures-rs `select!` macro implementation. use proc_macro::TokenStream; use proc_macro2::Span; @@ -59,7 +59,7 @@ impl Parse for Select { // `=> ` input.parse::]>()?; - let expr = input.parse::()?; + let expr = Expr::parse_with_earlier_boundary_rule(input)?; // Commas after the expression are only optional if it's a `Block` // or it is the last branch in the `match`. @@ -232,7 +232,7 @@ fn select_inner(input: TokenStream, random: bool) -> TokenStream { let branches = parsed.normal_fut_handlers.into_iter().zip(variant_names.iter()).map( |((pat, expr), variant_name)| { quote! { - #enum_ident::#variant_name(#pat) => { #expr }, + #enum_ident::#variant_name(#pat) => #expr, } }, ); diff --git a/futures-sink/Cargo.toml b/futures-sink/Cargo.toml index 6994ba6fcd..21f8259155 100644 --- a/futures-sink/Cargo.toml +++ b/futures-sink/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-sink" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.36" license = "MIT OR Apache-2.0" @@ -19,3 +19,6 @@ alloc = [] [package.metadata.docs.rs] all-features = true + +[lints] +workspace = true diff --git a/futures-sink/src/lib.rs b/futures-sink/src/lib.rs index 0328740efd..52394e8c2a 100644 --- a/futures-sink/src/lib.rs +++ b/futures-sink/src/lib.rs @@ -3,10 +3,7 @@ //! This crate contains the `Sink` trait which allows values to be sent //! asynchronously. -#![cfg_attr(not(feature = "std"), no_std)] -#![warn(missing_debug_implementations, missing_docs, rust_2018_idioms, unreachable_pub)] -// It cannot be included in the published code because this lints have false positives in the minimum required version. -#![cfg_attr(test, warn(single_use_lifetimes))] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -14,9 +11,12 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, /* unsafe_op_in_unsafe_fn */)] // unsafe_op_in_unsafe_fn requires Rust 1.52 #[cfg(feature = "alloc")] extern crate alloc; +#[cfg(feature = "std")] +extern crate std; use core::ops::DerefMut; use core::pin::Pin; diff --git a/futures-task/Cargo.toml b/futures-task/Cargo.toml index 4116a3144b..f1848dacbd 100644 --- a/futures-task/Cargo.toml +++ b/futures-task/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-task" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.56" license = "MIT OR Apache-2.0" @@ -27,3 +27,6 @@ futures = { path = "../futures" } [package.metadata.docs.rs] all-features = true + +[lints] +workspace = true diff --git a/futures-task/src/future_obj.rs b/futures-task/src/future_obj.rs index 071392af6c..a611382193 100644 --- a/futures-task/src/future_obj.rs +++ b/futures-task/src/future_obj.rs @@ -29,14 +29,14 @@ impl Unpin for LocalFutureObj<'_, T> {} unsafe fn remove_future_lifetime<'a, T>( ptr: *mut (dyn Future + 'a), ) -> *mut (dyn Future + 'static) { - mem::transmute(ptr) + unsafe { mem::transmute(ptr) } } #[allow(single_use_lifetimes)] unsafe fn remove_drop_lifetime<'a, T>( ptr: unsafe fn(*mut (dyn Future + 'a)), ) -> unsafe fn(*mut (dyn Future + 'static)) { - mem::transmute(ptr) + unsafe { mem::transmute(ptr) } } impl<'a, T> LocalFutureObj<'a, T> { @@ -225,7 +225,7 @@ mod if_alloc { } unsafe fn drop(ptr: *mut (dyn Future + 'a)) { - drop(Box::from_raw(ptr.cast::())) + drop(unsafe { Box::from_raw(ptr.cast::()) }) } } @@ -235,7 +235,7 @@ mod if_alloc { } unsafe fn drop(ptr: *mut (dyn Future + 'a)) { - drop(Box::from_raw(ptr)) + drop(unsafe { Box::from_raw(ptr) }) } } @@ -245,7 +245,7 @@ mod if_alloc { } unsafe fn drop(ptr: *mut (dyn Future + 'a)) { - drop(Box::from_raw(ptr)) + drop(unsafe { Box::from_raw(ptr) }) } } @@ -259,7 +259,7 @@ mod if_alloc { } unsafe fn drop(ptr: *mut (dyn Future + 'a)) { - drop(Pin::from(Box::from_raw(ptr))) + drop(Pin::from(unsafe { Box::from_raw(ptr) })) } } @@ -270,7 +270,7 @@ mod if_alloc { } unsafe fn drop(ptr: *mut (dyn Future + 'a)) { - drop(Pin::from(Box::from_raw(ptr))) + drop(Pin::from(unsafe { Box::from_raw(ptr) })) } } @@ -281,7 +281,7 @@ mod if_alloc { } unsafe fn drop(ptr: *mut (dyn Future + 'a)) { - drop(Pin::from(Box::from_raw(ptr))) + drop(Pin::from(unsafe { Box::from_raw(ptr) })) } } diff --git a/futures-task/src/lib.rs b/futures-task/src/lib.rs index 33896d8b14..c119b6b1e4 100644 --- a/futures-task/src/lib.rs +++ b/futures-task/src/lib.rs @@ -1,9 +1,6 @@ //! Tools for working with tasks. -#![cfg_attr(not(feature = "std"), no_std)] -#![warn(missing_debug_implementations, missing_docs, rust_2018_idioms, unreachable_pub)] -// It cannot be included in the published code because this lints have false positives in the minimum required version. -#![cfg_attr(test, warn(single_use_lifetimes))] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -11,9 +8,12 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, unsafe_op_in_unsafe_fn)] #[cfg(feature = "alloc")] extern crate alloc; +#[cfg(feature = "std")] +extern crate std; mod spawn; pub use crate::spawn::{LocalSpawn, Spawn, SpawnError}; diff --git a/futures-task/src/waker.rs b/futures-task/src/waker.rs index 79112569c5..ce90aa83d9 100644 --- a/futures-task/src/waker.rs +++ b/futures-task/src/waker.rs @@ -3,7 +3,7 @@ use alloc::sync::Arc; use core::mem; use core::task::{RawWaker, RawWakerVTable, Waker}; -pub(super) fn waker_vtable() -> &'static RawWakerVTable { +pub(super) fn waker_vtable() -> &'static RawWakerVTable { &RawWakerVTable::new( clone_arc_raw::, wake_arc_raw::, @@ -29,31 +29,32 @@ where // code here. We should guard against this by aborting. #[allow(clippy::redundant_clone)] // The clone here isn't actually redundant. -unsafe fn increase_refcount(data: *const ()) { +unsafe fn increase_refcount(data: *const ()) { // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop - let arc = mem::ManuallyDrop::new(Arc::::from_raw(data.cast::())); + let arc = mem::ManuallyDrop::new(unsafe { Arc::::from_raw(data.cast::()) }); // Now increase refcount, but don't drop new refcount either let _arc_clone: mem::ManuallyDrop<_> = arc.clone(); } // used by `waker_ref` -unsafe fn clone_arc_raw(data: *const ()) -> RawWaker { - increase_refcount::(data); +#[inline(always)] +unsafe fn clone_arc_raw(data: *const ()) -> RawWaker { + unsafe { increase_refcount::(data) } RawWaker::new(data, waker_vtable::()) } -unsafe fn wake_arc_raw(data: *const ()) { - let arc: Arc = Arc::from_raw(data.cast::()); +unsafe fn wake_arc_raw(data: *const ()) { + let arc: Arc = unsafe { Arc::from_raw(data.cast::()) }; ArcWake::wake(arc); } // used by `waker_ref` -unsafe fn wake_by_ref_arc_raw(data: *const ()) { +unsafe fn wake_by_ref_arc_raw(data: *const ()) { // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop - let arc = mem::ManuallyDrop::new(Arc::::from_raw(data.cast::())); + let arc = mem::ManuallyDrop::new(unsafe { Arc::::from_raw(data.cast::()) }); ArcWake::wake_by_ref(&arc); } -unsafe fn drop_arc_raw(data: *const ()) { - drop(Arc::::from_raw(data.cast::())) +unsafe fn drop_arc_raw(data: *const ()) { + drop(unsafe { Arc::::from_raw(data.cast::()) }) } diff --git a/futures-task/src/waker_ref.rs b/futures-task/src/waker_ref.rs index aac4109577..5957b4d46a 100644 --- a/futures-task/src/waker_ref.rs +++ b/futures-task/src/waker_ref.rs @@ -54,7 +54,7 @@ impl Deref for WakerRef<'_> { #[inline] pub fn waker_ref(wake: &Arc) -> WakerRef<'_> where - W: ArcWake, + W: ArcWake + 'static, { // simply copy the pointer instead of using Arc::into_raw, // as we don't actually keep a refcount by using ManuallyDrop.< diff --git a/futures-test/Cargo.toml b/futures-test/Cargo.toml index 8499e034df..46d933415f 100644 --- a/futures-test/Cargo.toml +++ b/futures-test/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-test" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.56" license = "MIT OR Apache-2.0" @@ -11,14 +11,13 @@ Common utilities for testing components built off futures-rs. """ [dependencies] -futures-core = { version = "0.3.30", path = "../futures-core", default-features = false } -futures-task = { version = "0.3.30", path = "../futures-task", default-features = false } -futures-io = { version = "0.3.30", path = "../futures-io", default-features = false } -futures-util = { version = "0.3.30", path = "../futures-util", default-features = false } -futures-executor = { version = "0.3.30", path = "../futures-executor", default-features = false } -futures-sink = { version = "0.3.30", path = "../futures-sink", default-features = false } -futures-macro = { version = "=0.3.30", path = "../futures-macro", default-features = false } -pin-utils = { version = "0.1.0", default-features = false } +futures-core = { version = "0.3.31", path = "../futures-core", default-features = false } +futures-task = { version = "0.3.31", path = "../futures-task", default-features = false } +futures-io = { version = "0.3.31", path = "../futures-io", default-features = false } +futures-util = { version = "0.3.31", path = "../futures-util", default-features = false } +futures-executor = { version = "0.3.31", path = "../futures-executor", default-features = false } +futures-sink = { version = "0.3.31", path = "../futures-sink", default-features = false } +futures-macro = { version = "=0.3.31", path = "../futures-macro", default-features = false } pin-project = "1.0.11" [dev-dependencies] @@ -30,3 +29,6 @@ std = ["futures-core/std", "futures-task/std", "futures-io/std", "futures-util/s [package.metadata.docs.rs] all-features = true + +[lints] +workspace = true diff --git a/futures-test/src/lib.rs b/futures-test/src/lib.rs index 49f834846e..4e420eac88 100644 --- a/futures-test/src/lib.rs +++ b/futures-test/src/lib.rs @@ -1,12 +1,5 @@ //! Utilities to make testing [`Future`s](futures_core::future::Future) easier -#![warn( - missing_debug_implementations, - missing_docs, - rust_2018_idioms, - single_use_lifetimes, - unreachable_pub -)] #![doc(test( no_crate_inject, attr( @@ -14,6 +7,8 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, unsafe_op_in_unsafe_fn)] +#![allow(clippy::test_attr_in_doctest)] #[cfg(not(feature = "std"))] compile_error!( diff --git a/futures-util/Cargo.toml b/futures-util/Cargo.toml index dcdbce459e..825a204125 100644 --- a/futures-util/Cargo.toml +++ b/futures-util/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures-util" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.56" license = "MIT OR Apache-2.0" @@ -35,12 +35,12 @@ write-all-vectored = ["io"] cfg-target-has-atomic = [] [dependencies] -futures-core = { path = "../futures-core", version = "0.3.30", default-features = false } -futures-task = { path = "../futures-task", version = "0.3.30", default-features = false } -futures-channel = { path = "../futures-channel", version = "0.3.30", default-features = false, features = ["std"], optional = true } -futures-io = { path = "../futures-io", version = "0.3.30", default-features = false, features = ["std"], optional = true } -futures-sink = { path = "../futures-sink", version = "0.3.30", default-features = false, optional = true } -futures-macro = { path = "../futures-macro", version = "=0.3.30", default-features = false, optional = true } +futures-core = { path = "../futures-core", version = "0.3.31", default-features = false } +futures-task = { path = "../futures-task", version = "0.3.31", default-features = false } +futures-channel = { path = "../futures-channel", version = "0.3.31", default-features = false, features = ["std"], optional = true } +futures-io = { path = "../futures-io", version = "0.3.31", default-features = false, features = ["std"], optional = true } +futures-sink = { path = "../futures-sink", version = "0.3.31", default-features = false, optional = true } +futures-macro = { path = "../futures-macro", version = "=0.3.31", 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" } @@ -56,3 +56,6 @@ tokio = "0.1.11" [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] + +[lints] +workspace = true diff --git a/futures-util/src/abortable.rs b/futures-util/src/abortable.rs index 9dbcfc2b52..e1e79e371d 100644 --- a/futures-util/src/abortable.rs +++ b/futures-util/src/abortable.rs @@ -64,7 +64,7 @@ impl Abortable { /// This means that it will return `true` even if: /// * `abort` was called after the task had completed. /// * `abort` was called while the task was being polled - the task may still be running and - /// will not be stopped until `poll` returns. + /// will not be stopped until `poll` returns. pub fn is_aborted(&self) -> bool { self.inner.aborted.load(Ordering::Relaxed) } @@ -200,7 +200,7 @@ impl AbortHandle { /// even if: /// * `abort` was called after the task had completed. /// * `abort` was called while the task was being polled - the task may still be running and - /// will not be stopped until `poll` returns. + /// will not be stopped until `poll` returns. /// /// This operation has a Relaxed ordering. pub fn is_aborted(&self) -> bool { diff --git a/futures-util/src/async_await/random.rs b/futures-util/src/async_await/random.rs index 4f8c7254b4..2ac2f78a87 100644 --- a/futures-util/src/async_await/random.rs +++ b/futures-util/src/async_await/random.rs @@ -25,7 +25,7 @@ fn gen_index(n: usize) -> usize { /// /// [xorshift*]: https://en.wikipedia.org/wiki/Xorshift#xorshift* fn random() -> u64 { - thread_local! { + std::thread_local! { static RNG: Cell> = Cell::new(Wrapping(prng_seed())); } diff --git a/futures-util/src/async_await/select_mod.rs b/futures-util/src/async_await/select_mod.rs index 1d13067d38..80c3c7504a 100644 --- a/futures-util/src/async_await/select_mod.rs +++ b/futures-util/src/async_await/select_mod.rs @@ -3,6 +3,7 @@ macro_rules! document_select_macro { // This branch is required for `futures 0.3.1`, from before select_biased was introduced ($select:item) => { + #[allow(clippy::too_long_first_doc_paragraph)] /// Polls multiple futures and streams simultaneously, executing the branch /// for the future that finishes first. If multiple futures are ready, /// one will be pseudo-randomly selected at runtime. Futures directly @@ -153,6 +154,7 @@ macro_rules! document_select_macro { ($select:item $select_biased:item) => { document_select_macro!($select); + #[allow(clippy::too_long_first_doc_paragraph)] /// Polls multiple futures and streams simultaneously, executing the branch /// for the future that finishes first. Unlike [`select!`], if multiple futures are ready, /// one will be selected in order of declaration. Futures directly diff --git a/futures-util/src/async_await/stream_select_mod.rs b/futures-util/src/async_await/stream_select_mod.rs index 61e3fa1c66..c3e4f0052c 100644 --- a/futures-util/src/async_await/stream_select_mod.rs +++ b/futures-util/src/async_await/stream_select_mod.rs @@ -4,6 +4,7 @@ #[doc(hidden)] pub use futures_macro::stream_select_internal; +#[allow(clippy::too_long_first_doc_paragraph)] /// Combines several streams, all producing the same `Item` type, into one stream. /// This is similar to `select_all` but does not require the streams to all be the same type. /// It also keeps the streams inline, and does not require `Box`s to be allocated. diff --git a/futures-util/src/compat/compat01as03.rs b/futures-util/src/compat/compat01as03.rs index 36de1da98d..0de2d53301 100644 --- a/futures-util/src/compat/compat01as03.rs +++ b/futures-util/src/compat/compat01as03.rs @@ -8,6 +8,7 @@ use futures_01::{AsyncSink as AsyncSink01, Sink as Sink01}; use futures_core::{future::Future as Future03, stream::Stream as Stream03, task as task03}; #[cfg(feature = "sink")] use futures_sink::Sink as Sink03; +use std::boxed::Box; use std::pin::Pin; use std::task::Context; @@ -346,7 +347,7 @@ unsafe impl UnsafeNotify01 for NotifyWaker { unsafe fn drop_raw(&self) { let ptr: *const dyn UnsafeNotify01 = self; - drop(Box::from_raw(ptr as *mut dyn UnsafeNotify01)); + drop(unsafe { Box::from_raw(ptr as *mut dyn UnsafeNotify01) }); } } diff --git a/futures-util/src/compat/compat03as01.rs b/futures-util/src/compat/compat03as01.rs index 5d3a6e920b..3c5bb77b49 100644 --- a/futures-util/src/compat/compat03as01.rs +++ b/futures-util/src/compat/compat03as01.rs @@ -15,6 +15,7 @@ use futures_sink::Sink as Sink03; use std::marker::PhantomData; use std::{mem, pin::Pin, sync::Arc, task::Context}; +#[allow(clippy::too_long_first_doc_paragraph)] // clippy bug, see https://github.com/rust-lang/rust-clippy/issues/13315 /// Converts a futures 0.3 [`TryFuture`](futures_core::future::TryFuture) or /// [`TryStream`](futures_core::stream::TryStream) into a futures 0.1 /// [`Future`](futures_01::future::Future) or @@ -156,7 +157,7 @@ impl Current { fn as_waker(&self) -> WakerRef<'_> { unsafe fn ptr_to_current<'a>(ptr: *const ()) -> &'a Current { - &*(ptr as *const Current) + unsafe { &*(ptr as *const Current) } } fn current_to_ptr(current: &Current) -> *const () { current as *const Current as *const () @@ -166,13 +167,15 @@ impl Current { // Lazily create the `Arc` only when the waker is actually cloned. // FIXME: remove `transmute` when a `Waker` -> `RawWaker` conversion // function is landed in `core`. - mem::transmute::(task03::waker(Arc::new( - ptr_to_current(ptr).clone(), - ))) + unsafe { + mem::transmute::(task03::waker(Arc::new( + ptr_to_current(ptr).clone(), + ))) + } } unsafe fn drop(_: *const ()) {} unsafe fn wake(ptr: *const ()) { - ptr_to_current(ptr).0.notify() + unsafe { ptr_to_current(ptr).0.notify() } } let ptr = current_to_ptr(self); diff --git a/futures-util/src/future/always_ready.rs b/futures-util/src/future/always_ready.rs new file mode 100644 index 0000000000..28625e7272 --- /dev/null +++ b/futures-util/src/future/always_ready.rs @@ -0,0 +1,62 @@ +use super::assert_future; +use core::pin::Pin; +use futures_core::future::{FusedFuture, Future}; +use futures_core::task::{Context, Poll}; + +/// Future for the [`always_ready`](always_ready()) function. +#[must_use = "futures do nothing unless you `.await` or poll them"] +pub struct AlwaysReady T>(F); + +impl T> core::fmt::Debug for AlwaysReady { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.debug_tuple("AlwaysReady").finish() + } +} + +impl T + Clone> Clone for AlwaysReady { + fn clone(&self) -> Self { + Self(self.0.clone()) + } +} + +impl T + Copy> Copy for AlwaysReady {} + +impl T> Unpin for AlwaysReady {} + +impl T> FusedFuture for AlwaysReady { + fn is_terminated(&self) -> bool { + false + } +} + +impl T> Future for AlwaysReady { + type Output = T; + + #[inline] + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll { + Poll::Ready(self.0()) + } +} + +/// Creates a future that is always immediately ready with a value. +/// +/// This is particularly useful in avoiding a heap allocation when an API needs [`Box>`], +/// as [`AlwaysReady`] does not have to store a boolean for `is_finished`. +/// +/// # Examples +/// +/// ``` +/// # futures::executor::block_on(async { +/// use std::mem::size_of_val; +/// +/// use futures::future; +/// +/// let a = future::always_ready(|| 1); +/// assert_eq!(size_of_val(&a), 0); +/// assert_eq!(a.await, 1); +/// assert_eq!(a.await, 1); +/// # }); +/// ``` +pub fn always_ready T>(prod: F) -> AlwaysReady { + assert_future::(AlwaysReady(prod)) +} diff --git a/futures-util/src/future/either.rs b/futures-util/src/future/either.rs index 27e5064dfb..018e51e72f 100644 --- a/futures-util/src/future/either.rs +++ b/futures-util/src/future/either.rs @@ -39,9 +39,9 @@ impl Either { // SAFETY: We can use `new_unchecked` because the `inner` parts are // guaranteed to be pinned, as they come from `self` which is pinned. unsafe { - match *Pin::get_ref(self) { - Either::Left(ref inner) => Either::Left(Pin::new_unchecked(inner)), - Either::Right(ref inner) => Either::Right(Pin::new_unchecked(inner)), + match self.get_ref() { + Self::Left(inner) => Either::Left(Pin::new_unchecked(inner)), + Self::Right(inner) => Either::Right(Pin::new_unchecked(inner)), } } } @@ -55,9 +55,9 @@ impl Either { // offer an unpinned `&mut A` or `&mut B` through `Pin<&mut Self>`. We // also don't have an implementation of `Drop`, nor manual `Unpin`. unsafe { - match *Pin::get_unchecked_mut(self) { - Either::Left(ref mut inner) => Either::Left(Pin::new_unchecked(inner)), - Either::Right(ref mut inner) => Either::Right(Pin::new_unchecked(inner)), + match self.get_unchecked_mut() { + Self::Left(inner) => Either::Left(Pin::new_unchecked(inner)), + Self::Right(inner) => Either::Right(Pin::new_unchecked(inner)), } } } @@ -69,8 +69,8 @@ impl Either<(T, A), (T, B)> { /// Here, the homogeneous type is the first element of the pairs. pub fn factor_first(self) -> (T, Either) { match self { - Either::Left((x, a)) => (x, Either::Left(a)), - Either::Right((x, b)) => (x, Either::Right(b)), + Self::Left((x, a)) => (x, Either::Left(a)), + Self::Right((x, b)) => (x, Either::Right(b)), } } } @@ -81,8 +81,8 @@ impl Either<(A, T), (B, T)> { /// Here, the homogeneous type is the second element of the pairs. pub fn factor_second(self) -> (Either, T) { match self { - Either::Left((a, x)) => (Either::Left(a), x), - Either::Right((b, x)) => (Either::Right(b), x), + Self::Left((a, x)) => (Either::Left(a), x), + Self::Right((b, x)) => (Either::Right(b), x), } } } @@ -91,8 +91,7 @@ impl Either { /// Extract the value of an either over two equivalent types. pub fn into_inner(self) -> T { match self { - Either::Left(x) => x, - Either::Right(x) => x, + Self::Left(x) | Self::Right(x) => x, } } } @@ -119,8 +118,8 @@ where { fn is_terminated(&self) -> bool { match self { - Either::Left(x) => x.is_terminated(), - Either::Right(x) => x.is_terminated(), + Self::Left(x) => x.is_terminated(), + Self::Right(x) => x.is_terminated(), } } } @@ -141,8 +140,8 @@ where fn size_hint(&self) -> (usize, Option) { match self { - Either::Left(x) => x.size_hint(), - Either::Right(x) => x.size_hint(), + Self::Left(x) => x.size_hint(), + Self::Right(x) => x.size_hint(), } } } @@ -154,8 +153,8 @@ where { fn is_terminated(&self) -> bool { match self { - Either::Left(x) => x.is_terminated(), - Either::Right(x) => x.is_terminated(), + Self::Left(x) => x.is_terminated(), + Self::Right(x) => x.is_terminated(), } } } @@ -202,8 +201,6 @@ where mod if_std { use super::*; - use core::pin::Pin; - use core::task::{Context, Poll}; use futures_io::{ AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSlice, IoSliceMut, Result, SeekFrom, }; diff --git a/futures-util/src/future/future/catch_unwind.rs b/futures-util/src/future/future/catch_unwind.rs index 0e09d6eeb0..ed49e314d2 100644 --- a/futures-util/src/future/future/catch_unwind.rs +++ b/futures-util/src/future/future/catch_unwind.rs @@ -1,5 +1,6 @@ use core::any::Any; use core::pin::Pin; +use std::boxed::Box; use std::panic::{catch_unwind, AssertUnwindSafe, UnwindSafe}; use futures_core::future::Future; diff --git a/futures-util/src/future/future/map.rs b/futures-util/src/future/future/map.rs index 7471aba000..6991a35022 100644 --- a/futures-util/src/future/future/map.rs +++ b/futures-util/src/future/future/map.rs @@ -53,7 +53,7 @@ where match self.as_mut().project() { MapProj::Incomplete { future, .. } => { let output = ready!(future.poll(cx)); - match self.project_replace(Map::Complete) { + match self.project_replace(Self::Complete) { MapProjReplace::Incomplete { f, .. } => Poll::Ready(f.call_once(output)), MapProjReplace::Complete => unreachable!(), } diff --git a/futures-util/src/future/future/remote_handle.rs b/futures-util/src/future/future/remote_handle.rs index 1358902cab..6c4b7698ae 100644 --- a/futures-util/src/future/future/remote_handle.rs +++ b/futures-util/src/future/future/remote_handle.rs @@ -9,6 +9,7 @@ use { pin_project_lite::pin_project, std::{ any::Any, + boxed::Box, fmt, panic::{self, AssertUnwindSafe}, pin::Pin, diff --git a/futures-util/src/future/future/shared.rs b/futures-util/src/future/future/shared.rs index 9ab3b4f1d6..158fa70761 100644 --- a/futures-util/src/future/future/shared.rs +++ b/futures-util/src/future/future/shared.rs @@ -82,7 +82,7 @@ const POLLING: usize = 1; const COMPLETE: usize = 2; const POISONED: usize = 3; -const NULL_WAKER_KEY: usize = usize::max_value(); +const NULL_WAKER_KEY: usize = usize::MAX; impl Shared { pub(super) fn new(future: Fut) -> Self { @@ -192,8 +192,8 @@ where /// Safety: callers must first ensure that `self.inner.state` /// is `COMPLETE` unsafe fn output(&self) -> &Fut::Output { - match &*self.future_or_output.get() { - FutureOrOutput::Output(ref item) => item, + match unsafe { &*self.future_or_output.get() } { + FutureOrOutput::Output(item) => item, FutureOrOutput::Future(_) => unreachable!(), } } @@ -235,7 +235,7 @@ where FutureOrOutput::Output(item) => item, FutureOrOutput::Future(_) => unreachable!(), }, - Err(inner) => inner.output().clone(), + Err(inner) => unsafe { inner.output().clone() }, } } } diff --git a/futures-util/src/future/maybe_done.rs b/futures-util/src/future/maybe_done.rs index 26e6c27588..1e21e66f5e 100644 --- a/futures-util/src/future/maybe_done.rs +++ b/futures-util/src/future/maybe_done.rs @@ -53,7 +53,7 @@ impl MaybeDone { pub fn output_mut(self: Pin<&mut Self>) -> Option<&mut Fut::Output> { unsafe { match self.get_unchecked_mut() { - MaybeDone::Done(res) => Some(res), + Self::Done(res) => Some(res), _ => None, } } @@ -69,7 +69,7 @@ impl MaybeDone { } unsafe { match mem::replace(self.get_unchecked_mut(), Self::Gone) { - MaybeDone::Done(output) => Some(output), + Self::Done(output) => Some(output), _ => unreachable!(), } } @@ -91,12 +91,12 @@ impl Future for MaybeDone { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { unsafe { match self.as_mut().get_unchecked_mut() { - MaybeDone::Future(f) => { + Self::Future(f) => { let res = ready!(Pin::new_unchecked(f).poll(cx)); self.set(Self::Done(res)); } - MaybeDone::Done(_) => {} - MaybeDone::Gone => panic!("MaybeDone polled after value taken"), + Self::Done(_) => {} + Self::Gone => panic!("MaybeDone polled after value taken"), } } Poll::Ready(()) diff --git a/futures-util/src/future/mod.rs b/futures-util/src/future/mod.rs index 2d8fa4f654..873d2b2d10 100644 --- a/futures-util/src/future/mod.rs +++ b/futures-util/src/future/mod.rs @@ -74,6 +74,9 @@ pub use self::poll_immediate::{poll_immediate, PollImmediate}; mod ready; pub use self::ready::{err, ok, ready, Ready}; +mod always_ready; +pub use self::always_ready::{always_ready, AlwaysReady}; + mod join; pub use self::join::{join, join3, join4, join5, Join, Join3, Join4, Join5}; diff --git a/futures-util/src/future/try_maybe_done.rs b/futures-util/src/future/try_maybe_done.rs index 24044d2c27..97af6ad437 100644 --- a/futures-util/src/future/try_maybe_done.rs +++ b/futures-util/src/future/try_maybe_done.rs @@ -38,7 +38,7 @@ impl TryMaybeDone { pub fn output_mut(self: Pin<&mut Self>) -> Option<&mut Fut::Ok> { unsafe { match self.get_unchecked_mut() { - TryMaybeDone::Done(res) => Some(res), + Self::Done(res) => Some(res), _ => None, } } @@ -54,7 +54,7 @@ impl TryMaybeDone { } unsafe { match mem::replace(self.get_unchecked_mut(), Self::Gone) { - TryMaybeDone::Done(output) => Some(output), + Self::Done(output) => Some(output), _ => unreachable!(), } } @@ -76,15 +76,15 @@ impl Future for TryMaybeDone { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { unsafe { match self.as_mut().get_unchecked_mut() { - TryMaybeDone::Future(f) => match ready!(Pin::new_unchecked(f).try_poll(cx)) { + Self::Future(f) => match ready!(Pin::new_unchecked(f).try_poll(cx)) { Ok(res) => self.set(Self::Done(res)), Err(e) => { self.set(Self::Gone); return Poll::Ready(Err(e)); } }, - TryMaybeDone::Done(_) => {} - TryMaybeDone::Gone => panic!("TryMaybeDone polled after value taken"), + Self::Done(_) => {} + Self::Gone => panic!("TryMaybeDone polled after value taken"), } } Poll::Ready(Ok(())) diff --git a/futures-util/src/io/allow_std.rs b/futures-util/src/io/allow_std.rs index ec30ee31e5..96133cbc6f 100644 --- a/futures-util/src/io/allow_std.rs +++ b/futures-util/src/io/allow_std.rs @@ -1,6 +1,8 @@ use futures_core::task::{Context, Poll}; use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSlice, IoSliceMut, SeekFrom}; use std::pin::Pin; +use std::string::String; +use std::vec::Vec; use std::{fmt, io}; /// A simple wrapper type which allows types which implement only diff --git a/futures-util/src/io/buf_reader.rs b/futures-util/src/io/buf_reader.rs index 0334a9f081..9a919f7183 100644 --- a/futures-util/src/io/buf_reader.rs +++ b/futures-util/src/io/buf_reader.rs @@ -4,8 +4,10 @@ use futures_core::ready; use futures_core::task::{Context, Poll}; use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSliceMut, SeekFrom}; use pin_project_lite::pin_project; +use std::boxed::Box; use std::io::{self, Read}; use std::pin::Pin; +use std::vec; use std::{cmp, fmt}; pin_project! { @@ -46,14 +48,13 @@ impl BufReader { /// Creates a new `BufReader` with the specified buffer capacity. pub fn with_capacity(capacity: usize, inner: R) -> Self { - unsafe { - let mut buffer = Vec::with_capacity(capacity); - buffer.set_len(capacity); - super::initialize(&inner, &mut buffer); - Self { inner, buffer: buffer.into_boxed_slice(), pos: 0, cap: 0 } - } + // TODO: consider using Box<[u8]>::new_uninit_slice once it stabilized + let buffer = vec![0; capacity]; + Self { inner, buffer: buffer.into_boxed_slice(), pos: 0, cap: 0 } } +} +impl BufReader { delegate_access_inner!(inner, R, ()); /// Returns a reference to the internally buffered data. @@ -212,7 +213,7 @@ impl AsyncSeek for BufReader { // it should be safe to assume that remainder fits within an i64 as the alternative // means we managed to allocate 8 exbibytes and that's absurd. // But it's not out of the realm of possibility for some weird underlying reader to - // support seeking by i64::min_value() so we need to handle underflow when subtracting + // support seeking by i64::MIN so we need to handle underflow when subtracting // remainder. if let Some(offset) = n.checked_sub(remainder) { result = diff --git a/futures-util/src/io/buf_writer.rs b/futures-util/src/io/buf_writer.rs index cb74863ad0..ddcf5e1b4d 100644 --- a/futures-util/src/io/buf_writer.rs +++ b/futures-util/src/io/buf_writer.rs @@ -7,6 +7,7 @@ use std::fmt; use std::io::{self, Write}; use std::pin::Pin; use std::ptr; +use std::vec::Vec; pin_project! { /// Wraps a writer and buffers its output. @@ -78,6 +79,26 @@ impl BufWriter { Poll::Ready(ret) } + /// Write directly using `inner`, bypassing buffering + pub(super) fn inner_poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + self.project().inner.poll_write(cx, buf) + } + + /// Write directly using `inner`, bypassing buffering + pub(super) fn inner_poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[IoSlice<'_>], + ) -> Poll> { + self.project().inner.poll_write_vectored(cx, bufs) + } +} + +impl BufWriter { delegate_access_inner!(inner, W, ()); /// Returns a reference to the internally buffered data. @@ -124,27 +145,11 @@ impl BufWriter { let old_len = this.buf.len(); let buf_len = buf.len(); let src = buf.as_ptr(); - let dst = this.buf.as_mut_ptr().add(old_len); - ptr::copy_nonoverlapping(src, dst, buf_len); - this.buf.set_len(old_len + buf_len); - } - - /// Write directly using `inner`, bypassing buffering - pub(super) fn inner_poll_write( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - self.project().inner.poll_write(cx, buf) - } - - /// Write directly using `inner`, bypassing buffering - pub(super) fn inner_poll_write_vectored( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - bufs: &[IoSlice<'_>], - ) -> Poll> { - self.project().inner.poll_write_vectored(cx, bufs) + unsafe { + let dst = this.buf.as_mut_ptr().add(old_len); + ptr::copy_nonoverlapping(src, dst, buf_len); + this.buf.set_len(old_len + buf_len); + } } } diff --git a/futures-util/src/io/cursor.rs b/futures-util/src/io/cursor.rs index c6e2aeea28..01c8e5f4dc 100644 --- a/futures-util/src/io/cursor.rs +++ b/futures-util/src/io/cursor.rs @@ -1,7 +1,9 @@ use futures_core::task::{Context, Poll}; use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSlice, IoSliceMut, SeekFrom}; +use std::boxed::Box; use std::io; use std::pin::Pin; +use std::vec::Vec; /// A `Cursor` wraps an in-memory buffer and provides it with a /// [`AsyncSeek`] implementation. diff --git a/futures-util/src/io/line_writer.rs b/futures-util/src/io/line_writer.rs index 71cd668325..de1bb53924 100644 --- a/futures-util/src/io/line_writer.rs +++ b/futures-util/src/io/line_writer.rs @@ -25,13 +25,13 @@ pub struct LineWriter { impl LineWriter { /// Create a new `LineWriter` with default buffer capacity. The default is currently 1KB /// which was taken from `std::io::LineWriter` - pub fn new(inner: W) -> LineWriter { - LineWriter::with_capacity(1024, inner) + pub fn new(inner: W) -> Self { + Self::with_capacity(1024, inner) } /// Creates a new `LineWriter` with the specified buffer capacity. - pub fn with_capacity(capacity: usize, inner: W) -> LineWriter { - LineWriter { buf_writer: BufWriter::with_capacity(capacity, inner) } + pub fn with_capacity(capacity: usize, inner: W) -> Self { + Self { buf_writer: BufWriter::with_capacity(capacity, inner) } } /// Flush `buf_writer` if last char is "new line" diff --git a/futures-util/src/io/lines.rs b/futures-util/src/io/lines.rs index b5561bfa7d..0a1abf4bf6 100644 --- a/futures-util/src/io/lines.rs +++ b/futures-util/src/io/lines.rs @@ -7,6 +7,8 @@ use pin_project_lite::pin_project; use std::io; use std::mem; use std::pin::Pin; +use std::string::String; +use std::vec::Vec; pin_project! { /// Stream for the [`lines`](super::AsyncBufReadExt::lines) method. @@ -33,6 +35,7 @@ impl Stream for Lines { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let this = self.project(); let n = ready!(read_line_internal(this.reader, cx, this.buf, this.bytes, this.read))?; + *this.read = 0; if n == 0 && this.buf.is_empty() { return Poll::Ready(None); } diff --git a/futures-util/src/io/mod.rs b/futures-util/src/io/mod.rs index fdad60b1fa..c07fe3fe47 100644 --- a/futures-util/src/io/mod.rs +++ b/futures-util/src/io/mod.rs @@ -21,7 +21,7 @@ use crate::compat::Compat; use crate::future::assert_future; use crate::stream::assert_stream; -use std::{pin::Pin, ptr}; +use std::{pin::Pin, ptr, string::String, vec::Vec}; // Re-export some types from `std::io` so that users don't have to deal // with conflicts when `use`ing `futures::io` and `std::io`. @@ -39,7 +39,7 @@ const DEFAULT_BUF_SIZE: usize = 8 * 1024; /// A buffer is currently always initialized. #[inline] unsafe fn initialize(_reader: &R, buf: &mut [u8]) { - ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len()) + unsafe { ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len()) } } mod allow_std; diff --git a/futures-util/src/io/read_line.rs b/futures-util/src/io/read_line.rs index df782c9570..43942add54 100644 --- a/futures-util/src/io/read_line.rs +++ b/futures-util/src/io/read_line.rs @@ -7,6 +7,8 @@ use std::io; use std::mem; use std::pin::Pin; use std::str; +use std::string::String; +use std::vec::Vec; /// Future for the [`read_line`](super::AsyncBufReadExt::read_line) method. #[derive(Debug)] @@ -16,13 +18,14 @@ pub struct ReadLine<'a, R: ?Sized> { buf: &'a mut String, bytes: Vec, read: usize, + finished: bool, } impl Unpin for ReadLine<'_, R> {} impl<'a, R: AsyncBufRead + ?Sized + Unpin> ReadLine<'a, R> { pub(super) fn new(reader: &'a mut R, buf: &'a mut String) -> Self { - Self { reader, bytes: mem::take(buf).into_bytes(), buf, read: 0 } + Self { reader, bytes: mem::take(buf).into_bytes(), buf, read: 0, finished: false } } } @@ -33,26 +36,42 @@ pub(super) fn read_line_internal( bytes: &mut Vec, read: &mut usize, ) -> Poll> { - let ret = ready!(read_until_internal(reader, cx, b'\n', bytes, read)); - if str::from_utf8(bytes).is_err() { - bytes.clear(); - Poll::Ready(ret.and_then(|_| { - Err(io::Error::new(io::ErrorKind::InvalidData, "stream did not contain valid UTF-8")) - })) - } else { - debug_assert!(buf.is_empty()); - debug_assert_eq!(*read, 0); - // Safety: `bytes` is a valid UTF-8 because `str::from_utf8` returned `Ok`. - mem::swap(unsafe { buf.as_mut_vec() }, bytes); - Poll::Ready(ret) + let mut ret = ready!(read_until_internal(reader, cx, b'\n', bytes, read)); + if str::from_utf8(&bytes[bytes.len() - *read..bytes.len()]).is_err() { + bytes.truncate(bytes.len() - *read); + if ret.is_ok() { + ret = Err(io::Error::new( + io::ErrorKind::InvalidData, + "stream did not contain valid UTF-8", + )); + } } + *read = 0; + // Safety: `bytes` is valid UTF-8 because it was taken from a String + // and the newly read bytes are either valid UTF-8 or have been removed. + mem::swap(unsafe { buf.as_mut_vec() }, bytes); + Poll::Ready(ret) } impl Future for ReadLine<'_, R> { type Output = io::Result; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let Self { reader, buf, bytes, read } = &mut *self; - read_line_internal(Pin::new(reader), cx, buf, bytes, read) + let Self { reader, buf, bytes, read, finished: _ } = &mut *self; + let ret = ready!(read_line_internal(Pin::new(reader), cx, buf, bytes, read)); + self.finished = true; + Poll::Ready(ret) + } +} + +impl Drop for ReadLine<'_, R> { + fn drop(&mut self) { + // restore old string contents + if !self.finished { + self.bytes.truncate(self.bytes.len() - self.read); + // Safety: `bytes` is valid UTF-8 because it was taken from a String + // and the newly read bytes have been removed. + mem::swap(unsafe { self.buf.as_mut_vec() }, &mut self.bytes); + } } } diff --git a/futures-util/src/io/read_to_string.rs b/futures-util/src/io/read_to_string.rs index c175396d81..a075ca2662 100644 --- a/futures-util/src/io/read_to_string.rs +++ b/futures-util/src/io/read_to_string.rs @@ -4,6 +4,7 @@ use futures_core::ready; use futures_core::task::{Context, Poll}; use futures_io::AsyncRead; use std::pin::Pin; +use std::string::String; use std::vec::Vec; use std::{io, mem, str}; diff --git a/futures-util/src/io/read_until.rs b/futures-util/src/io/read_until.rs index 72b59eab13..adc359db51 100644 --- a/futures-util/src/io/read_until.rs +++ b/futures-util/src/io/read_until.rs @@ -3,8 +3,8 @@ use futures_core::ready; use futures_core::task::{Context, Poll}; use futures_io::AsyncBufRead; use std::io; -use std::mem; use std::pin::Pin; +use std::vec::Vec; /// Future for the [`read_until`](super::AsyncBufReadExt::read_until) method. #[derive(Debug)] @@ -45,7 +45,7 @@ pub(super) fn read_until_internal( reader.as_mut().consume(used); *read += used; if done || used == 0 { - return Poll::Ready(Ok(mem::replace(read, 0))); + return Poll::Ready(Ok(*read)); } } } diff --git a/futures-util/src/io/split.rs b/futures-util/src/io/split.rs index 81d1e6dcb5..5b6bc18937 100644 --- a/futures-util/src/io/split.rs +++ b/futures-util/src/io/split.rs @@ -45,7 +45,7 @@ impl ReadHalf { pub fn reunite(self, other: WriteHalf) -> Result> { self.handle .reunite(other.handle) - .map_err(|err| ReuniteError(ReadHalf { handle: err.0 }, WriteHalf { handle: err.1 })) + .map_err(|err| ReuniteError(Self { handle: err.0 }, WriteHalf { handle: err.1 })) } } diff --git a/futures-util/src/io/write_all_vectored.rs b/futures-util/src/io/write_all_vectored.rs index a8fc4c641c..26b3f25d92 100644 --- a/futures-util/src/io/write_all_vectored.rs +++ b/futures-util/src/io/write_all_vectored.rs @@ -49,6 +49,8 @@ mod tests { use std::io; use std::pin::Pin; use std::task::{Context, Poll}; + use std::vec; + use std::vec::Vec; use crate::io::{AsyncWrite, AsyncWriteExt, IoSlice}; use crate::task::noop_waker; diff --git a/futures-util/src/lib.rs b/futures-util/src/lib.rs index 208eb73aa5..2201e4f2c9 100644 --- a/futures-util/src/lib.rs +++ b/futures-util/src/lib.rs @@ -1,15 +1,7 @@ //! Combinators and utilities for working with `Future`s, `Stream`s, `Sink`s, //! and the `AsyncRead` and `AsyncWrite` traits. -#![cfg_attr(feature = "write-all-vectored", feature(io_slice_advance))] -#![cfg_attr(not(feature = "std"), no_std)] -#![warn( - missing_debug_implementations, - missing_docs, - rust_2018_idioms, - single_use_lifetimes, - unreachable_pub -)] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -17,6 +9,7 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, unsafe_op_in_unsafe_fn)] #![cfg_attr(docsrs, feature(doc_cfg))] #[cfg(all(feature = "bilock", not(feature = "unstable")))] @@ -24,6 +17,8 @@ compile_error!("The `bilock` feature requires the `unstable` feature as an expli #[cfg(feature = "alloc")] extern crate alloc; +#[cfg(feature = "std")] +extern crate std; // Macro re-exports pub use futures_core::ready; diff --git a/futures-util/src/lock/mutex.rs b/futures-util/src/lock/mutex.rs index 335ad14273..c77048a786 100644 --- a/futures-util/src/lock/mutex.rs +++ b/futures-util/src/lock/mutex.rs @@ -541,11 +541,17 @@ unsafe impl Sync for OwnedMutexGuard {} unsafe impl Send for MappedMutexGuard<'_, T, U> {} unsafe impl Sync for MappedMutexGuard<'_, T, U> {} -#[test] -fn test_mutex_guard_debug_not_recurse() { - let mutex = Mutex::new(42); - let guard = mutex.try_lock().unwrap(); - let _ = format!("{:?}", guard); - let guard = MutexGuard::map(guard, |n| n); - let _ = format!("{:?}", guard); +#[cfg(test)] +mod tests { + use super::*; + use std::format; + + #[test] + fn test_mutex_guard_debug_not_recurse() { + let mutex = Mutex::new(42); + let guard = mutex.try_lock().unwrap(); + let _ = format!("{:?}", guard); + let guard = MutexGuard::map(guard, |n| n); + let _ = format!("{:?}", guard); + } } diff --git a/futures-util/src/stream/futures_unordered/mod.rs b/futures-util/src/stream/futures_unordered/mod.rs index dedf75dee2..ca62b10b91 100644 --- a/futures-util/src/stream/futures_unordered/mod.rs +++ b/futures-util/src/stream/futures_unordered/mod.rs @@ -256,16 +256,6 @@ impl FuturesUnordered { // `wake` from doing any work in the future let prev = task.queued.swap(true, SeqCst); - // Drop the future, even if it hasn't finished yet. This is safe - // because we're dropping the future on the thread that owns - // `FuturesUnordered`, which correctly tracks `Fut`'s lifetimes and - // such. - unsafe { - // Set to `None` rather than `take()`ing to prevent moving the - // future. - *task.future.get() = None; - } - // If the queued flag was previously set, then it means that this task // is still in our internal ready to run queue. We then transfer // ownership of our reference count to the ready to run queue, and it'll @@ -277,8 +267,25 @@ impl FuturesUnordered { // enqueue the task, so our task will never see the ready to run queue // again. The task itself will be deallocated once all reference counts // have been dropped elsewhere by the various wakers that contain it. - if prev { - mem::forget(task); + // + // Use ManuallyDrop to transfer the reference count ownership before + // dropping the future so unwinding won't release the reference count. + let md_slot; + let task = if prev { + md_slot = mem::ManuallyDrop::new(task); + &*md_slot + } else { + &task + }; + + // Drop the future, even if it hasn't finished yet. This is safe + // because we're dropping the future on the thread that owns + // `FuturesUnordered`, which correctly tracks `Fut`'s lifetimes and + // such. + unsafe { + // Set to `None` rather than `take()`ing to prevent moving the + // future. + *task.future.get() = None; } } @@ -324,35 +331,37 @@ impl FuturesUnordered { /// This method is unsafe because it has be guaranteed that `task` is a /// valid pointer. unsafe fn unlink(&mut self, task: *const Task) -> Arc> { - // Compute the new list length now in case we're removing the head node - // and won't be able to retrieve the correct length later. - let head = *self.head_all.get_mut(); - debug_assert!(!head.is_null()); - let new_len = *(*head).len_all.get() - 1; - - let task = Arc::from_raw(task); - let next = task.next_all.load(Relaxed); - let prev = *task.prev_all.get(); - task.next_all.store(self.pending_next_all(), Relaxed); - *task.prev_all.get() = ptr::null_mut(); - - if !next.is_null() { - *(*next).prev_all.get() = prev; - } + unsafe { + // Compute the new list length now in case we're removing the head node + // and won't be able to retrieve the correct length later. + let head = *self.head_all.get_mut(); + debug_assert!(!head.is_null()); + let new_len = *(*head).len_all.get() - 1; - if !prev.is_null() { - (*prev).next_all.store(next, Relaxed); - } else { - *self.head_all.get_mut() = next; - } + let task = Arc::from_raw(task); + let next = task.next_all.load(Relaxed); + let prev = *task.prev_all.get(); + task.next_all.store(self.pending_next_all(), Relaxed); + *task.prev_all.get() = ptr::null_mut(); - // Store the new list length in the head node. - let head = *self.head_all.get_mut(); - if !head.is_null() { - *(*head).len_all.get() = new_len; - } + if !next.is_null() { + *(*next).prev_all.get() = prev; + } + + if !prev.is_null() { + (*prev).next_all.store(next, Relaxed); + } else { + *self.head_all.get_mut() = next; + } + + // Store the new list length in the head node. + let head = *self.head_all.get_mut(); + if !head.is_null() { + *(*head).len_all.get() = new_len; + } - task + task + } } /// Returns the reserved value for `Task::next_all` to indicate a pending @@ -509,7 +518,8 @@ impl Stream for FuturesUnordered { // We are only interested in whether the future is awoken before it // finishes polling, so reset the flag here. task.woken.store(false, Relaxed); - let waker = Task::waker_ref(task); + // SAFETY: see the comments of Bomb and this block. + let waker = unsafe { Task::waker_ref(task) }; let mut cx = Context::from_waker(&waker); // Safety: We won't move the future ever again @@ -564,15 +574,27 @@ impl FuturesUnordered { impl Drop for FuturesUnordered { fn drop(&mut self) { + // Before the strong reference to the queue is dropped we need all + // futures to be dropped. See note at the bottom of this method. + // + // If there is a panic before this completes, we leak the queue. + struct LeakQueueOnDrop<'a, Fut>(&'a mut FuturesUnordered); + impl Drop for LeakQueueOnDrop<'_, Fut> { + fn drop(&mut self) { + mem::forget(Arc::clone(&self.0.ready_to_run_queue)); + } + } + let guard = LeakQueueOnDrop(self); // When a `FuturesUnordered` is dropped we want to drop all futures // associated with it. At the same time though there may be tons of // wakers flying around which contain `Task` references // inside them. We'll let those naturally get deallocated. - while !self.head_all.get_mut().is_null() { - let head = *self.head_all.get_mut(); - let task = unsafe { self.unlink(head) }; - self.release_task(task); + while !guard.0.head_all.get_mut().is_null() { + let head = *guard.0.head_all.get_mut(); + let task = unsafe { guard.0.unlink(head) }; + guard.0.release_task(task); } + mem::forget(guard); // safe to release strong reference to queue // Note that at this point we could still have a bunch of tasks in the // ready to run queue. None of those tasks, however, have futures diff --git a/futures-util/src/stream/futures_unordered/ready_to_run_queue.rs b/futures-util/src/stream/futures_unordered/ready_to_run_queue.rs index a924935d23..6ffaf554dd 100644 --- a/futures-util/src/stream/futures_unordered/ready_to_run_queue.rs +++ b/futures-util/src/stream/futures_unordered/ready_to_run_queue.rs @@ -27,6 +27,8 @@ pub(super) struct ReadyToRunQueue { /// An MPSC queue into which the tasks containing the futures are inserted /// whenever the future inside is scheduled for polling. impl ReadyToRunQueue { + // FIXME: this takes raw pointer without safety conditions. + /// The enqueue function from the 1024cores intrusive MPSC queue algorithm. pub(super) fn enqueue(&self, task: *const Task) { unsafe { @@ -47,39 +49,41 @@ impl ReadyToRunQueue { /// Note that this is unsafe as it required mutual exclusion (only one /// thread can call this) to be guaranteed elsewhere. pub(super) unsafe fn dequeue(&self) -> Dequeue { - let mut tail = *self.tail.get(); - let mut next = (*tail).next_ready_to_run.load(Acquire); + unsafe { + let mut tail = *self.tail.get(); + let mut next = (*tail).next_ready_to_run.load(Acquire); + + if tail == self.stub() { + if next.is_null() { + return Dequeue::Empty; + } - if tail == self.stub() { - if next.is_null() { - return Dequeue::Empty; + *self.tail.get() = next; + tail = next; + next = (*next).next_ready_to_run.load(Acquire); } - *self.tail.get() = next; - tail = next; - next = (*next).next_ready_to_run.load(Acquire); - } + if !next.is_null() { + *self.tail.get() = next; + debug_assert!(tail != self.stub()); + return Dequeue::Data(tail); + } - if !next.is_null() { - *self.tail.get() = next; - debug_assert!(tail != self.stub()); - return Dequeue::Data(tail); - } + if self.head.load(Acquire) as *const _ != tail { + return Dequeue::Inconsistent; + } - if self.head.load(Acquire) as *const _ != tail { - return Dequeue::Inconsistent; - } + self.enqueue(self.stub()); - self.enqueue(self.stub()); + next = (*tail).next_ready_to_run.load(Acquire); - next = (*tail).next_ready_to_run.load(Acquire); + if !next.is_null() { + *self.tail.get() = next; + return Dequeue::Data(tail); + } - if !next.is_null() { - *self.tail.get() = next; - return Dequeue::Data(tail); + Dequeue::Inconsistent } - - Dequeue::Inconsistent } pub(super) fn stub(&self) -> *const Task { diff --git a/futures-util/src/stream/futures_unordered/task.rs b/futures-util/src/stream/futures_unordered/task.rs index ec2114effa..2ae4cb7d85 100644 --- a/futures-util/src/stream/futures_unordered/task.rs +++ b/futures-util/src/stream/futures_unordered/task.rs @@ -5,7 +5,7 @@ use core::sync::atomic::{AtomicBool, AtomicPtr}; use super::abort::abort; use super::ReadyToRunQueue; -use crate::task::{waker_ref, ArcWake, WakerRef}; +use crate::task::ArcWake; pub(super) struct Task { // The future @@ -77,8 +77,8 @@ impl ArcWake for Task { impl Task { /// Returns a waker reference for this task without cloning the Arc. - pub(super) fn waker_ref(this: &Arc) -> WakerRef<'_> { - waker_ref(this) + pub(super) unsafe fn waker_ref(this: &Arc) -> waker_ref::WakerRef<'_> { + unsafe { waker_ref::waker_ref(this) } } /// Spins until `next_all` is no longer set to `pending_next_all`. @@ -123,3 +123,92 @@ impl Drop for Task { } } } + +mod waker_ref { + use alloc::sync::Arc; + use core::marker::PhantomData; + use core::mem; + use core::mem::ManuallyDrop; + use core::ops::Deref; + use core::task::{RawWaker, RawWakerVTable, Waker}; + use futures_task::ArcWake; + + pub(crate) struct WakerRef<'a> { + waker: ManuallyDrop, + _marker: PhantomData<&'a ()>, + } + + impl WakerRef<'_> { + #[inline] + fn new_unowned(waker: ManuallyDrop) -> Self { + Self { waker, _marker: PhantomData } + } + } + + impl Deref for WakerRef<'_> { + type Target = Waker; + + #[inline] + fn deref(&self) -> &Waker { + &self.waker + } + } + + /// Copy of `future_task::waker_ref` without `W: 'static` bound. + /// + /// # Safety + /// + /// The caller must guarantee that use-after-free will not occur. + #[inline] + pub(crate) unsafe fn waker_ref(wake: &Arc) -> WakerRef<'_> + where + W: ArcWake, + { + // simply copy the pointer instead of using Arc::into_raw, + // as we don't actually keep a refcount by using ManuallyDrop.< + let ptr = Arc::as_ptr(wake).cast::<()>(); + + let waker = + ManuallyDrop::new(unsafe { Waker::from_raw(RawWaker::new(ptr, waker_vtable::())) }); + WakerRef::new_unowned(waker) + } + + fn waker_vtable() -> &'static RawWakerVTable { + &RawWakerVTable::new( + clone_arc_raw::, + wake_arc_raw::, + wake_by_ref_arc_raw::, + drop_arc_raw::, + ) + } + + // FIXME: panics on Arc::clone / refcount changes could wreak havoc on the + // code here. We should guard against this by aborting. + + unsafe fn increase_refcount(data: *const ()) { + // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop + let arc = mem::ManuallyDrop::new(unsafe { Arc::::from_raw(data.cast::()) }); + // Now increase refcount, but don't drop new refcount either + let _arc_clone: mem::ManuallyDrop<_> = arc.clone(); + } + + unsafe fn clone_arc_raw(data: *const ()) -> RawWaker { + unsafe { increase_refcount::(data) } + RawWaker::new(data, waker_vtable::()) + } + + unsafe fn wake_arc_raw(data: *const ()) { + let arc: Arc = unsafe { Arc::from_raw(data.cast::()) }; + ArcWake::wake(arc); + } + + unsafe fn wake_by_ref_arc_raw(data: *const ()) { + // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop + let arc = mem::ManuallyDrop::new(unsafe { Arc::::from_raw(data.cast::()) }); + ArcWake::wake_by_ref(&arc); + } + + unsafe fn drop_arc_raw(data: *const ()) { + drop(unsafe { Arc::::from_raw(data.cast::()) }) + } +} diff --git a/futures-util/src/stream/iter.rs b/futures-util/src/stream/iter.rs index 20471c2ed0..48b6519a39 100644 --- a/futures-util/src/stream/iter.rs +++ b/futures-util/src/stream/iter.rs @@ -10,6 +10,23 @@ pub struct Iter { iter: I, } +impl Iter { + /// Acquires a reference to the underlying iterator that this stream is pulling from. + pub fn get_ref(&self) -> &I { + &self.iter + } + + /// Acquires a mutable reference to the underlying iterator that this stream is pulling from. + pub fn get_mut(&mut self) -> &mut I { + &mut self.iter + } + + /// Consumes this stream, returning the underlying iterator. + pub fn into_inner(self) -> I { + self.iter + } +} + impl Unpin for Iter {} /// Converts an `Iterator` into a `Stream` which is always ready diff --git a/futures-util/src/stream/repeat.rs b/futures-util/src/stream/repeat.rs index 3f9aa87d5c..e09cfa2e41 100644 --- a/futures-util/src/stream/repeat.rs +++ b/futures-util/src/stream/repeat.rs @@ -44,7 +44,7 @@ where } fn size_hint(&self) -> (usize, Option) { - (usize::max_value(), None) + (usize::MAX, None) } } diff --git a/futures-util/src/stream/repeat_with.rs b/futures-util/src/stream/repeat_with.rs index f5a81b4ed4..a482510705 100644 --- a/futures-util/src/stream/repeat_with.rs +++ b/futures-util/src/stream/repeat_with.rs @@ -24,7 +24,7 @@ impl A> Stream for RepeatWith { } fn size_hint(&self) -> (usize, Option) { - (usize::max_value(), None) + (usize::MAX, None) } } diff --git a/futures-util/src/stream/select_with_strategy.rs b/futures-util/src/stream/select_with_strategy.rs index 224d5f821c..4e19873af7 100644 --- a/futures-util/src/stream/select_with_strategy.rs +++ b/futures-util/src/stream/select_with_strategy.rs @@ -21,17 +21,17 @@ impl PollNext { old } - fn other(&self) -> PollNext { + fn other(&self) -> Self { match self { - PollNext::Left => PollNext::Right, - PollNext::Right => PollNext::Left, + Self::Left => Self::Right, + Self::Right => Self::Left, } } } impl Default for PollNext { fn default() -> Self { - PollNext::Left + Self::Left } } @@ -45,15 +45,14 @@ enum InternalState { impl InternalState { fn finish(&mut self, ps: PollNext) { match (&self, ps) { - (InternalState::Start, PollNext::Left) => { - *self = InternalState::LeftFinished; + (Self::Start, PollNext::Left) => { + *self = Self::LeftFinished; } - (InternalState::Start, PollNext::Right) => { - *self = InternalState::RightFinished; + (Self::Start, PollNext::Right) => { + *self = Self::RightFinished; } - (InternalState::LeftFinished, PollNext::Right) - | (InternalState::RightFinished, PollNext::Left) => { - *self = InternalState::BothFinished; + (Self::LeftFinished, PollNext::Right) | (Self::RightFinished, PollNext::Left) => { + *self = Self::BothFinished; } _ => {} } @@ -75,6 +74,7 @@ pin_project! { } } +#[allow(clippy::too_long_first_doc_paragraph)] /// This function will attempt to pull items from both streams. You provide a /// closure to tell [`SelectWithStrategy`] which stream to poll. The closure can /// store state on `SelectWithStrategy` to which it will receive a `&mut` on every diff --git a/futures-util/src/stream/stream/catch_unwind.rs b/futures-util/src/stream/stream/catch_unwind.rs index 09a6dc1b76..dfc388839d 100644 --- a/futures-util/src/stream/stream/catch_unwind.rs +++ b/futures-util/src/stream/stream/catch_unwind.rs @@ -2,6 +2,7 @@ use futures_core::stream::{FusedStream, Stream}; use futures_core::task::{Context, Poll}; use pin_project_lite::pin_project; use std::any::Any; +use std::boxed::Box; use std::panic::{catch_unwind, AssertUnwindSafe, UnwindSafe}; use std::pin::Pin; diff --git a/futures-util/src/stream/stream/cycle.rs b/futures-util/src/stream/stream/cycle.rs index 507431d24f..60fe745399 100644 --- a/futures-util/src/stream/stream/cycle.rs +++ b/futures-util/src/stream/stream/cycle.rs @@ -1,5 +1,4 @@ use core::pin::Pin; -use core::usize; use futures_core::ready; use futures_core::stream::{FusedStream, Stream}; use futures_core::task::{Context, Poll}; @@ -48,7 +47,7 @@ where match self.orig.size_hint() { size @ (0, Some(0)) => size, (0, _) => (0, None), - _ => (usize::max_value(), None), + _ => (usize::MAX, None), } } } diff --git a/futures-util/src/stream/stream/flatten_unordered.rs b/futures-util/src/stream/stream/flatten_unordered.rs index 44c6ace2f7..c79edbd213 100644 --- a/futures-util/src/stream/stream/flatten_unordered.rs +++ b/futures-util/src/stream/stream/flatten_unordered.rs @@ -56,15 +56,13 @@ struct SharedPollState { impl SharedPollState { /// Constructs new `SharedPollState` with the given state. - fn new(value: u8) -> SharedPollState { - SharedPollState { state: Arc::new(AtomicU8::new(value)) } + fn new(value: u8) -> Self { + Self { state: Arc::new(AtomicU8::new(value)) } } /// Attempts to start polling, returning stored state in case of success. /// Returns `None` if either waker is waking at the moment. - fn start_polling( - &self, - ) -> Option<(u8, PollStateBomb<'_, impl FnOnce(&SharedPollState) -> u8>)> { + fn start_polling(&self) -> Option<(u8, PollStateBomb<'_, impl FnOnce(&Self) -> u8>)> { let value = self .state .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |value| { @@ -75,7 +73,7 @@ impl SharedPollState { } }) .ok()?; - let bomb = PollStateBomb::new(self, SharedPollState::reset); + let bomb = PollStateBomb::new(self, Self::reset); Some((value, bomb)) } @@ -87,7 +85,7 @@ impl SharedPollState { fn start_waking( &self, to_poll: u8, - ) -> Option<(u8, PollStateBomb<'_, impl FnOnce(&SharedPollState) -> u8>)> { + ) -> Option<(u8, PollStateBomb<'_, impl FnOnce(&Self) -> u8>)> { let value = self .state .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |value| { @@ -106,7 +104,7 @@ impl SharedPollState { // Only start the waking process if we're not in the polling/waking phase and the stream isn't woken already if value & (WOKEN | POLLING | WAKING) == NONE { - let bomb = PollStateBomb::new(self, SharedPollState::stop_waking); + let bomb = PollStateBomb::new(self, Self::stop_waking); Some((value, bomb)) } else { @@ -121,7 +119,7 @@ impl SharedPollState { /// - `!WAKING` as /// * Wakers called during the `POLLING` phase won't propagate their calls /// * `POLLING` phase can't start if some of the wakers are active - /// So no wrapped waker can touch the inner waker's cell, it's safe to poll again. + /// So no wrapped waker can touch the inner waker's cell, it's safe to poll again. fn stop_polling(&self, to_poll: u8, will_be_woken: bool) -> u8 { self.state .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |mut value| { @@ -210,7 +208,7 @@ impl WrappedWaker { /// This function will modify waker's `inner_waker` via `UnsafeCell`, so /// it should be used only during `POLLING` phase by one thread at the time. unsafe fn replace_waker(self_arc: &mut Arc, cx: &Context<'_>) { - *self_arc.inner_waker.get() = cx.waker().clone().into(); + unsafe { *self_arc.inner_waker.get() = cx.waker().clone().into() } } /// Attempts to start the waking process for the waker with the given value. @@ -261,7 +259,7 @@ impl PollStreamFut { } impl Future for PollStreamFut { - type Output = Option<(St::Item, PollStreamFut)>; + type Output = Option<(St::Item, Self)>; fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let mut stream = self.project().stream; @@ -271,7 +269,7 @@ impl Future for PollStreamFut { } else { None }; - let next_item_fut = PollStreamFut::new(stream.get_mut().take()); + let next_item_fut = Self::new(stream.get_mut().take()); let out = item.map(|item| (item, next_item_fut)); Poll::Ready(out) @@ -320,13 +318,10 @@ where Fc: FlowController::Item>, St::Item: Stream + Unpin, { - pub(crate) fn new( - stream: St, - limit: Option, - ) -> FlattenUnorderedWithFlowController { + pub(crate) fn new(stream: St, limit: Option) -> Self { let poll_state = SharedPollState::new(NEED_TO_POLL_STREAM); - FlattenUnorderedWithFlowController { + Self { inner_streams: FuturesUnordered::new(), stream, is_stream_done: false, diff --git a/futures-util/src/stream/stream/mod.rs b/futures-util/src/stream/stream/mod.rs index 2da7036b24..d365cfe43f 100644 --- a/futures-util/src/stream/stream/mod.rs +++ b/futures-util/src/stream/stream/mod.rs @@ -360,7 +360,7 @@ pub trait StreamExt: Stream { /// # Overflow Behavior /// /// The method does no guarding against overflows, so enumerating more than - /// [`prim@usize::max_value()`] elements either produces the wrong result or panics. If + /// [`usize::MAX`] elements either produces the wrong result or panics. If /// debug assertions are enabled, a panic is guaranteed. /// /// # Panics diff --git a/futures-util/src/stream/stream/split.rs b/futures-util/src/stream/stream/split.rs index 1a7fdcb387..e192f8a92d 100644 --- a/futures-util/src/stream/stream/split.rs +++ b/futures-util/src/stream/stream/split.rs @@ -160,7 +160,7 @@ impl std::error::Error for ReuniteError {} #[cfg(test)] mod tests { use super::*; - use crate::{sink::Sink, stream::StreamExt}; + use crate::stream::StreamExt; use core::marker::PhantomData; struct NopStream { diff --git a/futures-util/src/stream/try_stream/mod.rs b/futures-util/src/stream/try_stream/mod.rs index 7b55444b3e..ee004b51e8 100644 --- a/futures-util/src/stream/try_stream/mod.rs +++ b/futures-util/src/stream/try_stream/mod.rs @@ -660,7 +660,7 @@ pub trait TryStreamExt: TryStream { /// them into a local vector. At most `capacity` items will get buffered /// before they're yielded from the returned stream. If the underlying stream /// returns `Poll::Pending`, and the collected chunk is not empty, it will - /// be immidiatly returned. + /// be immediately returned. /// /// Note that the vectors returned from this iterator may not always have /// `capacity` elements. If the underlying stream ended and only a partial diff --git a/futures-util/src/unfold_state.rs b/futures-util/src/unfold_state.rs index 0edc15e437..b66956bbd8 100644 --- a/futures-util/src/unfold_state.rs +++ b/futures-util/src/unfold_state.rs @@ -29,7 +29,7 @@ impl UnfoldState { pub(crate) fn take_value(self: Pin<&mut Self>) -> Option { match &*self { - UnfoldState::Value { .. } => match self.project_replace(UnfoldState::Empty) { + Self::Value { .. } => match self.project_replace(Self::Empty) { UnfoldStateProjReplace::Value { value } => Some(value), _ => unreachable!(), }, diff --git a/futures/Cargo.toml b/futures/Cargo.toml index 6208f616ea..102f9215cc 100644 --- a/futures/Cargo.toml +++ b/futures/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "futures" -version = "0.3.30" +version = "0.3.31" edition = "2018" rust-version = "1.56" license = "MIT OR Apache-2.0" @@ -15,20 +15,19 @@ composability, and iterator-like interfaces. categories = ["asynchronous"] [dependencies] -futures-core = { path = "../futures-core", version = "0.3.30", default-features = false } -futures-task = { path = "../futures-task", version = "0.3.30", default-features = false } -futures-channel = { path = "../futures-channel", version = "0.3.30", default-features = false, features = ["sink"] } -futures-executor = { path = "../futures-executor", version = "0.3.30", default-features = false, optional = true } -futures-io = { path = "../futures-io", version = "0.3.30", default-features = false } -futures-sink = { path = "../futures-sink", version = "0.3.30", default-features = false } -futures-util = { path = "../futures-util", version = "0.3.30", default-features = false, features = ["sink"] } +futures-core = { path = "../futures-core", version = "0.3.31", default-features = false } +futures-task = { path = "../futures-task", version = "0.3.31", default-features = false } +futures-channel = { path = "../futures-channel", version = "0.3.31", default-features = false, features = ["sink"] } +futures-executor = { path = "../futures-executor", version = "0.3.31", default-features = false, optional = true } +futures-io = { path = "../futures-io", version = "0.3.31", default-features = false } +futures-sink = { path = "../futures-sink", version = "0.3.31", default-features = false } +futures-util = { path = "../futures-util", version = "0.3.31", default-features = false, features = ["sink"] } [dev-dependencies] futures-executor = { path = "../futures-executor", features = ["thread-pool"] } futures-test = { path = "../futures-test" } assert_matches = "1.3.0" pin-project = "1.0.11" -pin-utils = "0.1.0" static_assertions = "1" tokio = "0.1.11" @@ -59,3 +58,6 @@ rustdoc-args = ["--cfg", "docsrs"] [package.metadata.playground] features = ["std", "async-await", "compat", "io-compat", "executor", "thread-pool"] + +[lints] +workspace = true diff --git a/futures/src/lib.rs b/futures/src/lib.rs index b972f51754..654576fcda 100644 --- a/futures/src/lib.rs +++ b/futures/src/lib.rs @@ -81,14 +81,7 @@ //! The majority of examples and code snippets in this crate assume that they are //! inside an async block as written above. -#![cfg_attr(not(feature = "std"), no_std)] -#![warn( - missing_debug_implementations, - missing_docs, - rust_2018_idioms, - single_use_lifetimes, - unreachable_pub -)] +#![no_std] #![doc(test( no_crate_inject, attr( @@ -96,6 +89,7 @@ allow(dead_code, unused_assignments, unused_variables) ) ))] +#![warn(missing_docs, unsafe_op_in_unsafe_fn)] #![cfg_attr(docsrs, feature(doc_cfg))] #[cfg(all(feature = "bilock", not(feature = "unstable")))] diff --git a/futures/tests/async_await_macros.rs b/futures/tests/async_await_macros.rs index 82a617f2c2..0bc79e4ae8 100644 --- a/futures/tests/async_await_macros.rs +++ b/futures/tests/async_await_macros.rs @@ -57,6 +57,18 @@ fn select() { assert!(ran); } +#[test] +fn select_grammar() { + // Parsing after `=>` using Expr::parse would parse `{}() = future::ready(())` + // as one expression. + block_on(async { + select! { + () = future::pending::<()>() => {} + () = future::ready(()) => {} + } + }); +} + #[test] fn select_biased() { let (tx1, rx1) = oneshot::channel::(); diff --git a/futures/tests/auto_traits.rs b/futures/tests/auto_traits.rs index 004fda1e71..3385360591 100644 --- a/futures/tests/auto_traits.rs +++ b/futures/tests/auto_traits.rs @@ -1,4 +1,5 @@ #![cfg(feature = "compat")] +#![allow(dead_code)] //! Assert Send/Sync/Unpin for all public types. @@ -12,49 +13,49 @@ use static_assertions::{assert_impl_all as assert_impl, assert_not_impl_all as a use std::marker::PhantomPinned; use std::{marker::PhantomData, pin::Pin}; -pub type LocalFuture = Pin>>; -pub type LocalTryFuture = LocalFuture>; -pub type SendFuture = Pin + Send>>; -pub type SendTryFuture = SendFuture>; -pub type SyncFuture = Pin + Sync>>; -pub type SyncTryFuture = SyncFuture>; -pub type SendSyncFuture = Pin + Send + Sync>>; -pub type SendSyncTryFuture = SendSyncFuture>; -pub type UnpinFuture = LocalFuture; -pub type UnpinTryFuture = UnpinFuture>; -pub struct PinnedFuture(PhantomPinned, PhantomData); +type LocalFuture = Pin>>; +type LocalTryFuture = LocalFuture>; +type SendFuture = Pin + Send>>; +type SendTryFuture = SendFuture>; +type SyncFuture = Pin + Sync>>; +type SyncTryFuture = SyncFuture>; +type SendSyncFuture = Pin + Send + Sync>>; +type SendSyncTryFuture = SendSyncFuture>; +type UnpinFuture = LocalFuture; +type UnpinTryFuture = UnpinFuture>; +struct PinnedFuture(PhantomPinned, PhantomData); impl Future for PinnedFuture { type Output = T; fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll { unimplemented!() } } -pub type PinnedTryFuture = PinnedFuture>; - -pub type LocalStream = Pin>>; -pub type LocalTryStream = LocalStream>; -pub type SendStream = Pin + Send>>; -pub type SendTryStream = SendStream>; -pub type SyncStream = Pin + Sync>>; -pub type SyncTryStream = SyncStream>; -pub type SendSyncStream = Pin + Send + Sync>>; -pub type SendSyncTryStream = SendSyncStream>; -pub type UnpinStream = LocalStream; -pub type UnpinTryStream = UnpinStream>; -pub struct PinnedStream(PhantomPinned, PhantomData); +type PinnedTryFuture = PinnedFuture>; + +type LocalStream = Pin>>; +type LocalTryStream = LocalStream>; +type SendStream = Pin + Send>>; +type SendTryStream = SendStream>; +type SyncStream = Pin + Sync>>; +type SyncTryStream = SyncStream>; +type SendSyncStream = Pin + Send + Sync>>; +type SendSyncTryStream = SendSyncStream>; +type UnpinStream = LocalStream; +type UnpinTryStream = UnpinStream>; +struct PinnedStream(PhantomPinned, PhantomData); impl Stream for PinnedStream { type Item = T; fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { unimplemented!() } } -pub type PinnedTryStream = PinnedStream>; +type PinnedTryStream = PinnedStream>; -pub type LocalSink = Pin>>; -pub type SendSink = Pin + Send>>; -pub type SyncSink = Pin + Sync>>; -pub type UnpinSink = LocalSink; -pub struct PinnedSink(PhantomPinned, PhantomData<(T, E)>); +type LocalSink = Pin>>; +type SendSink = Pin + Send>>; +type SyncSink = Pin + Sync>>; +type UnpinSink = LocalSink; +struct PinnedSink(PhantomPinned, PhantomData<(T, E)>); impl Sink for PinnedSink { type Error = E; fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { @@ -72,7 +73,7 @@ impl Sink for PinnedSink { } /// Assert Send/Sync/Unpin for all public types in `futures::channel`. -pub mod channel { +mod channel { use super::*; use futures::channel::*; @@ -119,11 +120,11 @@ pub mod channel { assert_impl!(oneshot::Canceled: Sync); assert_impl!(oneshot::Canceled: Unpin); - assert_impl!(oneshot::Cancellation<()>: Send); - assert_not_impl!(oneshot::Cancellation<*const ()>: Send); - assert_impl!(oneshot::Cancellation<()>: Sync); - assert_not_impl!(oneshot::Cancellation<*const ()>: Sync); - assert_impl!(oneshot::Cancellation: Unpin); + assert_impl!(oneshot::Cancellation<'_, ()>: Send); + assert_not_impl!(oneshot::Cancellation<'_, *const ()>: Send); + assert_impl!(oneshot::Cancellation<'_, ()>: Sync); + assert_not_impl!(oneshot::Cancellation<'_, *const ()>: Sync); + assert_impl!(oneshot::Cancellation<'_, PhantomPinned>: Unpin); assert_impl!(oneshot::Receiver<()>: Send); assert_not_impl!(oneshot::Receiver<*const ()>: Send); @@ -139,7 +140,7 @@ pub mod channel { } /// Assert Send/Sync/Unpin for all public types in `futures::compat`. -pub mod compat { +mod compat { use super::*; use futures::compat::*; @@ -181,7 +182,7 @@ pub mod compat { } /// Assert Send/Sync/Unpin for all public types in `futures::executor`. -pub mod executor { +mod executor { use super::*; use futures::executor::*; @@ -219,7 +220,7 @@ pub mod executor { } /// Assert Send/Sync/Unpin for all public types in `futures::future`. -pub mod future { +mod future { use super::*; use futures::future::*; @@ -305,9 +306,9 @@ pub mod future { assert_impl!(Fuse: Unpin); assert_not_impl!(Fuse: Unpin); - assert_impl!(FutureObj<*const ()>: Send); - assert_not_impl!(FutureObj<()>: Sync); - assert_impl!(FutureObj: Unpin); + assert_impl!(FutureObj<'_, *const ()>: Send); + assert_not_impl!(FutureObj<'_, ()>: Sync); + assert_impl!(FutureObj<'_, PhantomPinned>: Unpin); assert_impl!(Inspect: Send); assert_not_impl!(Inspect: Send); @@ -381,9 +382,9 @@ pub mod future { assert_not_impl!(Lazy<*const ()>: Sync); assert_impl!(Lazy: Unpin); - assert_not_impl!(LocalFutureObj<()>: Send); - assert_not_impl!(LocalFutureObj<()>: Sync); - assert_impl!(LocalFutureObj: Unpin); + assert_not_impl!(LocalFutureObj<'_, ()>: Send); + assert_not_impl!(LocalFutureObj<'_, ()>: Sync); + assert_impl!(LocalFutureObj<'_, PhantomPinned>: Unpin); assert_impl!(Map: Send); assert_not_impl!(Map: Send); @@ -652,7 +653,7 @@ pub mod future { } /// Assert Send/Sync/Unpin for all public types in `futures::io`. -pub mod io { +mod io { use super::*; use futures::io::{Sink, *}; @@ -693,23 +694,23 @@ pub mod io { assert_impl!(Close<'_, ()>: Unpin); assert_not_impl!(Close<'_, PhantomPinned>: Unpin); - assert_impl!(Copy<(), ()>: Send); - assert_not_impl!(Copy<(), *const ()>: Send); - assert_not_impl!(Copy<*const (), ()>: Send); - assert_impl!(Copy<(), ()>: Sync); - assert_not_impl!(Copy<(), *const ()>: Sync); - assert_not_impl!(Copy<*const (), ()>: Sync); - assert_impl!(Copy<(), PhantomPinned>: Unpin); - assert_not_impl!(Copy: Unpin); - - assert_impl!(CopyBuf<(), ()>: Send); - assert_not_impl!(CopyBuf<(), *const ()>: Send); - assert_not_impl!(CopyBuf<*const (), ()>: Send); - assert_impl!(CopyBuf<(), ()>: Sync); - assert_not_impl!(CopyBuf<(), *const ()>: Sync); - assert_not_impl!(CopyBuf<*const (), ()>: Sync); - assert_impl!(CopyBuf<(), PhantomPinned>: Unpin); - assert_not_impl!(CopyBuf: Unpin); + assert_impl!(Copy<'_, (), ()>: Send); + assert_not_impl!(Copy<'_, (), *const ()>: Send); + assert_not_impl!(Copy<'_, *const (), ()>: Send); + assert_impl!(Copy<'_, (), ()>: Sync); + assert_not_impl!(Copy<'_, (), *const ()>: Sync); + assert_not_impl!(Copy<'_, *const (), ()>: Sync); + assert_impl!(Copy<'_, (), PhantomPinned>: Unpin); + assert_not_impl!(Copy<'_, PhantomPinned, ()>: Unpin); + + assert_impl!(CopyBuf<'_, (), ()>: Send); + assert_not_impl!(CopyBuf<'_, (), *const ()>: Send); + assert_not_impl!(CopyBuf<'_, *const (), ()>: Send); + assert_impl!(CopyBuf<'_, (), ()>: Sync); + assert_not_impl!(CopyBuf<'_, (), *const ()>: Sync); + assert_not_impl!(CopyBuf<'_, *const (), ()>: Sync); + assert_impl!(CopyBuf<'_, (), PhantomPinned>: Unpin); + assert_not_impl!(CopyBuf<'_, PhantomPinned, ()>: Unpin); assert_impl!(Cursor<()>: Send); assert_not_impl!(Cursor<*const ()>: Send); @@ -890,7 +891,7 @@ pub mod io { } /// Assert Send/Sync/Unpin for all public types in `futures::lock`. -pub mod lock { +mod lock { use super::*; use futures::lock::*; @@ -966,7 +967,7 @@ pub mod lock { } /// Assert Send/Sync/Unpin for all public types in `futures::sink`. -pub mod sink { +mod sink { use super::*; use futures::sink::{self, *}; use std::marker::Send; @@ -1095,7 +1096,7 @@ pub mod sink { } /// Assert Send/Sync/Unpin for all public types in `futures::stream`. -pub mod stream { +mod stream { use super::*; use futures::{io, stream::*}; @@ -1835,33 +1836,33 @@ pub mod stream { assert_not_impl!(Zip: Unpin); assert_not_impl!(Zip: Unpin); - assert_impl!(futures_unordered::Iter<()>: Send); - assert_not_impl!(futures_unordered::Iter<*const ()>: Send); - assert_impl!(futures_unordered::Iter<()>: Sync); - assert_not_impl!(futures_unordered::Iter<*const ()>: Sync); - assert_impl!(futures_unordered::Iter<()>: Unpin); + assert_impl!(futures_unordered::Iter<'_, ()>: Send); + assert_not_impl!(futures_unordered::Iter<'_, *const ()>: Send); + assert_impl!(futures_unordered::Iter<'_, ()>: Sync); + assert_not_impl!(futures_unordered::Iter<'_, *const ()>: Sync); + assert_impl!(futures_unordered::Iter<'_, ()>: Unpin); // The definition of futures_unordered::Iter has `Fut: Unpin` bounds. - // assert_not_impl!(futures_unordered::Iter: Unpin); + // assert_not_impl!(futures_unordered::Iter<'_, PhantomPinned>: Unpin); - assert_impl!(futures_unordered::IterMut<()>: Send); - assert_not_impl!(futures_unordered::IterMut<*const ()>: Send); - assert_impl!(futures_unordered::IterMut<()>: Sync); - assert_not_impl!(futures_unordered::IterMut<*const ()>: Sync); - assert_impl!(futures_unordered::IterMut<()>: Unpin); + assert_impl!(futures_unordered::IterMut<'_, ()>: Send); + assert_not_impl!(futures_unordered::IterMut<'_, *const ()>: Send); + assert_impl!(futures_unordered::IterMut<'_, ()>: Sync); + assert_not_impl!(futures_unordered::IterMut<'_, *const ()>: Sync); + assert_impl!(futures_unordered::IterMut<'_, ()>: Unpin); // The definition of futures_unordered::IterMut has `Fut: Unpin` bounds. - // assert_not_impl!(futures_unordered::IterMut: Unpin); + // assert_not_impl!(futures_unordered::IterMut<'_, PhantomPinned>: Unpin); - assert_impl!(futures_unordered::IterPinMut<()>: Send); - assert_not_impl!(futures_unordered::IterPinMut<*const ()>: Send); - assert_impl!(futures_unordered::IterPinMut<()>: Sync); - assert_not_impl!(futures_unordered::IterPinMut<*const ()>: Sync); - assert_impl!(futures_unordered::IterPinMut: Unpin); + assert_impl!(futures_unordered::IterPinMut<'_, ()>: Send); + assert_not_impl!(futures_unordered::IterPinMut<'_, *const ()>: Send); + assert_impl!(futures_unordered::IterPinMut<'_, ()>: Sync); + assert_not_impl!(futures_unordered::IterPinMut<'_, *const ()>: Sync); + assert_impl!(futures_unordered::IterPinMut<'_, PhantomPinned>: Unpin); - assert_impl!(futures_unordered::IterPinRef<()>: Send); - assert_not_impl!(futures_unordered::IterPinRef<*const ()>: Send); - assert_impl!(futures_unordered::IterPinRef<()>: Sync); - assert_not_impl!(futures_unordered::IterPinRef<*const ()>: Sync); - assert_impl!(futures_unordered::IterPinRef: Unpin); + assert_impl!(futures_unordered::IterPinRef<'_, ()>: Send); + assert_not_impl!(futures_unordered::IterPinRef<'_, *const ()>: Send); + assert_impl!(futures_unordered::IterPinRef<'_, ()>: Sync); + assert_not_impl!(futures_unordered::IterPinRef<'_, *const ()>: Sync); + assert_impl!(futures_unordered::IterPinRef<'_, PhantomPinned>: Unpin); assert_impl!(futures_unordered::IntoIter<()>: Send); assert_not_impl!(futures_unordered::IntoIter<*const ()>: Send); @@ -1872,7 +1873,7 @@ pub mod stream { } /// Assert Send/Sync/Unpin for all public types in `futures::task`. -pub mod task { +mod task { use super::*; use futures::task::*; @@ -1880,13 +1881,13 @@ pub mod task { assert_impl!(AtomicWaker: Sync); assert_impl!(AtomicWaker: Unpin); - assert_impl!(FutureObj<*const ()>: Send); - assert_not_impl!(FutureObj<()>: Sync); - assert_impl!(FutureObj: Unpin); + assert_impl!(FutureObj<'_, *const ()>: Send); + assert_not_impl!(FutureObj<'_, ()>: Sync); + assert_impl!(FutureObj<'_, PhantomPinned>: Unpin); - assert_not_impl!(LocalFutureObj<()>: Send); - assert_not_impl!(LocalFutureObj<()>: Sync); - assert_impl!(LocalFutureObj: Unpin); + assert_not_impl!(LocalFutureObj<'_, ()>: Send); + assert_not_impl!(LocalFutureObj<'_, ()>: Sync); + assert_impl!(LocalFutureObj<'_, PhantomPinned>: Unpin); assert_impl!(SpawnError: Send); assert_impl!(SpawnError: Sync); diff --git a/futures/tests/future_join.rs b/futures/tests/future_join.rs index f5df9d7775..0556a6e62c 100644 --- a/futures/tests/future_join.rs +++ b/futures/tests/future_join.rs @@ -1,5 +1,5 @@ use futures::executor::block_on; -use futures::future::Future; +use futures::future::{self, Future}; use std::task::Poll; /// This tests verifies (through miri) that self-referencing @@ -21,7 +21,7 @@ async fn trouble() { fn yield_now() -> impl Future { let mut yielded = false; - std::future::poll_fn(move |cx| { + future::poll_fn(move |cx| { if core::mem::replace(&mut yielded, true) { Poll::Ready(()) } else { diff --git a/futures/tests/future_try_join_all.rs b/futures/tests/future_try_join_all.rs index 9a824872f7..892e775920 100644 --- a/futures/tests/future_try_join_all.rs +++ b/futures/tests/future_try_join_all.rs @@ -1,8 +1,7 @@ use futures::executor::block_on; +use futures::future::{err, ok, try_join_all, Future, TryJoinAll}; use futures::pin_mut; -use futures_util::future::{err, ok, try_join_all, TryJoinAll}; use std::fmt::Debug; -use std::future::Future; #[track_caller] fn assert_done(actual_fut: impl Future, expected: T) diff --git a/futures/tests/io_lines.rs b/futures/tests/io_lines.rs index 5ce01a6945..62afef326a 100644 --- a/futures/tests/io_lines.rs +++ b/futures/tests/io_lines.rs @@ -1,6 +1,6 @@ use futures::executor::block_on; use futures::future::{Future, FutureExt}; -use futures::io::{AsyncBufReadExt, Cursor}; +use futures::io::{AsyncBufReadExt, AsyncRead, Cursor}; use futures::stream::{self, StreamExt, TryStreamExt}; use futures::task::Poll; use futures_test::io::AsyncReadTestExt; @@ -27,6 +27,24 @@ macro_rules! run_next { }; } +struct IOErrorRead(bool); + +impl AsyncRead for IOErrorRead { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + b: &mut [u8], + ) -> Poll> { + if self.0 { + Poll::Ready(Err(std::io::ErrorKind::InvalidInput.into())) + } else { + self.0 = true; + b[..16].fill(b'x'); + Ok(16).into() + } + } +} + #[test] fn lines() { let buf = Cursor::new(&b"12\r"[..]); @@ -58,3 +76,9 @@ fn maybe_pending() { assert_eq!(run_next!(s), "".to_string()); assert!(run(s.next()).is_none()); } + +#[test] +fn issue2862() { + let mut lines = futures::io::BufReader::new(IOErrorRead(false)).lines(); + assert!(block_on(lines.next()).unwrap().is_err()) +} diff --git a/futures/tests/io_read_line.rs b/futures/tests/io_read_line.rs index 88a877928a..c7559c593b 100644 --- a/futures/tests/io_read_line.rs +++ b/futures/tests/io_read_line.rs @@ -3,6 +3,7 @@ use futures::future::{Future, FutureExt}; use futures::io::{AsyncBufReadExt, Cursor}; use futures::stream::{self, StreamExt, TryStreamExt}; use futures::task::Poll; +use futures::AsyncRead; use futures_test::io::AsyncReadTestExt; use futures_test::task::noop_context; @@ -15,6 +16,24 @@ fn run(mut f: F) -> F::Output { } } +struct IOErrorRead(bool); + +impl AsyncRead for IOErrorRead { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + b: &mut [u8], + ) -> Poll> { + if self.0 { + Poll::Ready(Err(std::io::ErrorKind::InvalidInput.into())) + } else { + self.0 = true; + b[..16].fill(b'x'); + Ok(16).into() + } + } +} + #[test] fn read_line() { let mut buf = Cursor::new(b"12"); @@ -34,6 +53,30 @@ fn read_line() { assert_eq!(v, ""); } +#[test] +fn read_line_drop() { + // string contents should be preserved if the future is dropped + let mut buf = Cursor::new(b"12\n\n"); + let mut v = String::from("abc"); + drop(buf.read_line(&mut v)); + assert_eq!(v, "abc"); +} + +#[test] +fn read_line_io_error() { + let mut r = futures::io::BufReader::new(IOErrorRead(false)); + let _ = block_on(r.read_line(&mut String::new())); +} + +#[test] +fn read_line_utf8_error() { + let mut buf = Cursor::new(b"12\xFF\n\n"); + let mut v = String::from("abc"); + let res = block_on(buf.read_line(&mut v)); + assert_eq!(res.unwrap_err().kind(), std::io::ErrorKind::InvalidData); + assert_eq!(v, "abc"); +} + #[test] fn maybe_pending() { let mut buf = b"12".interleave_pending(); diff --git a/futures/tests/io_read_to_end.rs b/futures/tests/io_read_to_end.rs index 7122511fcb..0441b3bc4f 100644 --- a/futures/tests/io_read_to_end.rs +++ b/futures/tests/io_read_to_end.rs @@ -14,14 +14,14 @@ fn issue2310() { impl MyRead { fn new() -> Self { - MyRead { first: false } + Self { first: false } } } impl AsyncRead for MyRead { fn poll_read( mut self: Pin<&mut Self>, - _cx: &mut Context, + _cx: &mut Context<'_>, _buf: &mut [u8], ) -> Poll> { Poll::Ready(if !self.first { @@ -41,7 +41,7 @@ fn issue2310() { impl VecWrapper { fn new() -> Self { - VecWrapper { inner: Vec::new() } + Self { inner: Vec::new() } } } diff --git a/futures/tests/macro-reexport/Cargo.toml b/futures/tests/macro-reexport/Cargo.toml index a648ee54a5..2ddd1587a0 100644 --- a/futures/tests/macro-reexport/Cargo.toml +++ b/futures/tests/macro-reexport/Cargo.toml @@ -1,8 +1,11 @@ [package] name = "macro-reexport" -version = "0.1.0" +version = "0.0.0" edition = "2018" publish = false [dependencies] futures03 = { path = "../..", package = "futures" } + +[lints] +workspace = true diff --git a/futures/tests/macro-tests/Cargo.toml b/futures/tests/macro-tests/Cargo.toml index 963da731ac..db72bd41b2 100644 --- a/futures/tests/macro-tests/Cargo.toml +++ b/futures/tests/macro-tests/Cargo.toml @@ -1,9 +1,12 @@ [package] name = "macro-tests" -version = "0.1.0" +version = "0.0.0" edition = "2018" publish = false [dependencies] futures03 = { path = "../..", package = "futures" } macro-reexport = { path = "../macro-reexport" } + +[lints] +workspace = true diff --git a/futures/tests/no-std/Cargo.toml b/futures/tests/no-std/Cargo.toml index ed5d0c146a..ec7d4c7871 100644 --- a/futures/tests/no-std/Cargo.toml +++ b/futures/tests/no-std/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "no-std" -version = "0.1.0" +version = "0.0.0" edition = "2018" publish = false @@ -19,3 +19,6 @@ futures-task = { path = "../../../futures-task", optional = true, default-featur futures-channel = { path = "../../../futures-channel", optional = true, default-features = false } futures-util = { path = "../../../futures-util", optional = true, default-features = false } futures = { path = "../..", optional = true, default-features = false } + +[lints] +workspace = true diff --git a/futures/tests/no-std/build.rs b/futures/tests/no-std/build.rs index a96a68274b..ba79f203f2 100644 --- a/futures/tests/no-std/build.rs +++ b/futures/tests/no-std/build.rs @@ -1,6 +1,7 @@ use std::{env, process::Command}; fn main() { + println!("cargo:rustc-check-cfg=cfg(nightly)"); if is_nightly() { println!("cargo:rustc-cfg=nightly"); } diff --git a/futures/tests/stream.rs b/futures/tests/stream.rs index 6cbef7516c..3122b587d0 100644 --- a/futures/tests/stream.rs +++ b/futures/tests/stream.rs @@ -79,7 +79,7 @@ fn flatten_unordered() { impl Stream for DataStream { type Item = u8; - fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll> { + fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll> { if !self.polled { if !self.wake_immediately { let waker = ctx.waker().clone(); @@ -110,7 +110,7 @@ fn flatten_unordered() { impl Stream for Interchanger { type Item = DataStream; - fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll> { + fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll> { if !self.polled { self.polled = true; if !self.wake_immediately { diff --git a/futures/tests/stream_futures_ordered.rs b/futures/tests/stream_futures_ordered.rs index 5a4a3e22ee..f06ce263eb 100644 --- a/futures/tests/stream_futures_ordered.rs +++ b/futures/tests/stream_futures_ordered.rs @@ -72,7 +72,7 @@ fn test_push_front() { stream.push_front(d_rx); d_tx.send(4).unwrap(); - // we pushed `d_rx` to the front and sent 4, so we should recieve 4 next + // we pushed `d_rx` to the front and sent 4, so we should receive 4 next // and then 3 after it assert_eq!(Poll::Ready(Some(Ok(4))), stream.poll_next_unpin(&mut cx)); assert_eq!(Poll::Ready(Some(Ok(3))), stream.poll_next_unpin(&mut cx)); @@ -165,7 +165,7 @@ fn test_push_front_negative() { b_tx.send(2).unwrap(); c_tx.send(3).unwrap(); - // These should all be recieved in reverse order + // These should all be received in reverse order assert_eq!(Poll::Ready(Some(Ok(3))), stream.poll_next_unpin(&mut cx)); assert_eq!(Poll::Ready(Some(Ok(2))), stream.poll_next_unpin(&mut cx)); assert_eq!(Poll::Ready(Some(Ok(1))), stream.poll_next_unpin(&mut cx)); diff --git a/futures/tests/stream_futures_unordered.rs b/futures/tests/stream_futures_unordered.rs index 7bdf5432ca..e25e755cde 100644 --- a/futures/tests/stream_futures_unordered.rs +++ b/futures/tests/stream_futures_unordered.rs @@ -336,7 +336,7 @@ fn polled_only_once_at_most_per_iteration() { impl Future for F { type Output = (); - fn poll(mut self: Pin<&mut Self>, _: &mut Context) -> Poll { + fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll { if self.polled { panic!("polled twice") } else { @@ -406,3 +406,26 @@ fn clear_in_loop() { } }); } + +// https://github.com/rust-lang/futures-rs/issues/2863#issuecomment-2219441515 +#[test] +#[should_panic] +fn panic_on_drop_fut() { + struct BadFuture; + + impl Drop for BadFuture { + fn drop(&mut self) { + panic!() + } + } + + impl Future for BadFuture { + type Output = (); + + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll { + Poll::Pending + } + } + + FuturesUnordered::default().push(BadFuture); +} diff --git a/futures/tests/stream_try_stream.rs b/futures/tests/stream_try_stream.rs index ef38c510b8..57bdd4f3f3 100644 --- a/futures/tests/stream_try_stream.rs +++ b/futures/tests/stream_try_stream.rs @@ -91,7 +91,7 @@ fn try_flatten_unordered() { impl Stream for ErrorStream { type Item = Result>, ()>; - fn poll_next(mut self: Pin<&mut Self>, _: &mut Context) -> Poll> { + fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { if self.polled > self.error_after { panic!("Polled after error"); } else {