chore: use poll_fn from std (#6810)

tests-integration/src/bin/test-mem.rs

@@ -1,4 +1,4 @@
-use futures::future::poll_fn;
+use std::future::poll_fn;
 
 fn main() {
     let rt = tokio::runtime::Builder::new_multi_thread()

tests-integration/tests/process_stdio.rs

@@ -206,7 +206,7 @@ async fn vectored_writes() {
         let mut input = Bytes::from_static(b"hello\n").chain(Bytes::from_static(b"world!\n"));
         let mut writes_completed = 0;
 
-        futures::future::poll_fn(|cx| loop {
+        std::future::poll_fn(|cx| loop {
             let mut slices = [IoSlice::new(&[]); 2];
             let vectored = input.chunks_vectored(&mut slices);
             if vectored == 0 {

tokio-stream/src/lib.rs

@@ -74,9 +74,6 @@
 #[macro_use]
 mod macros;
 
-mod poll_fn;
-pub(crate) use poll_fn::poll_fn;
-
 pub mod wrappers;
 
 mod stream_ext;

tokio-stream/src/poll_fn.rs

@@ -1,35 +0,0 @@
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-pub(crate) struct PollFn<F> {
-    f: F,
-}
-
-pub(crate) fn poll_fn<T, F>(f: F) -> PollFn<F>
-where
-    F: FnMut(&mut Context<'_>) -> Poll<T>,
-{
-    PollFn { f }
-}
-
-impl<T, F> Future for PollFn<F>
-where
-    F: FnMut(&mut Context<'_>) -> Poll<T>,
-{
-    type Output = T;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
-        // Safety: We never construct a `Pin<&mut F>` anywhere, so accessing `f`
-        // mutably in an unpinned way is sound.
-        //
-        // This use of unsafe cannot be replaced with the pin-project macro
-        // because:
-        //  * If we put `#[pin]` on the field, then it gives us a `Pin<&mut F>`,
-        //    which we can't use to call the closure.
-        //  * If we don't put `#[pin]` on the field, then it makes `PollFn` be
-        //    unconditionally `Unpin`, which we also don't want.
-        let me = unsafe { Pin::into_inner_unchecked(self) };
-        (me.f)(cx)
-    }
-}

tokio-stream/src/stream_map.rs

@@ -1,6 +1,7 @@
-use crate::{poll_fn, Stream};
+use crate::Stream;
 
 use std::borrow::Borrow;
+use std::future::poll_fn;
 use std::hash::Hash;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};

tokio-util/src/util/poll_buf.rs

@@ -17,7 +17,7 @@ use std::task::{ready, Context, Poll};
 /// use tokio_stream as stream;
 /// use tokio::io::Result;
 /// use tokio_util::io::{StreamReader, poll_read_buf};
-/// use futures::future::poll_fn;
+/// use std::future::poll_fn;
 /// use std::pin::Pin;
 /// # #[tokio::main]
 /// # async fn main() -> std::io::Result<()> {
@@ -95,9 +95,9 @@ pub fn poll_read_buf<T: AsyncRead + ?Sized, B: BufMut>(
 /// use tokio::fs::File;
 ///
 /// use bytes::Buf;
+/// use std::future::poll_fn;
 /// use std::io::Cursor;
 /// use std::pin::Pin;
-/// use futures::future::poll_fn;
 ///
 /// #[tokio::main]
 /// async fn main() -> io::Result<()> {

tokio-util/tests/io_inspect.rs

@@ -1,5 +1,5 @@
-use futures::future::poll_fn;
 use std::{
+    future::poll_fn,
     io::IoSlice,
     pin::Pin,
     task::{Context, Poll},

tokio-util/tests/mpsc.rs

@@ -1,5 +1,5 @@
-use futures::future::poll_fn;
 use futures::sink::SinkExt;
+use std::future::poll_fn;
 use tokio::sync::mpsc::channel;
 use tokio_test::task::spawn;
 use tokio_test::{

tokio-util/tests/poll_semaphore.rs

@@ -9,7 +9,7 @@ type SemRet = Option<OwnedSemaphorePermit>;
 fn semaphore_poll(
     sem: &mut PollSemaphore,
 ) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + '_> {
-    let fut = futures::future::poll_fn(move |cx| sem.poll_acquire(cx));
+    let fut = std::future::poll_fn(move |cx| sem.poll_acquire(cx));
     tokio_test::task::spawn(fut)
 }
 
@@ -17,7 +17,7 @@ fn semaphore_poll_many(
     sem: &mut PollSemaphore,
     permits: u32,
 ) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + '_> {
-    let fut = futures::future::poll_fn(move |cx| sem.poll_acquire_many(cx, permits));
+    let fut = std::future::poll_fn(move |cx| sem.poll_acquire_many(cx, permits));
     tokio_test::task::spawn(fut)
 }
 

tokio/src/fs/file.rs

@@ -936,7 +936,7 @@ cfg_windows! {
 
 impl Inner {
     async fn complete_inflight(&mut self) {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         poll_fn(|cx| self.poll_complete_inflight(cx)).await;
     }

tokio/src/fs/read_dir.rs

@@ -76,7 +76,7 @@ impl ReadDir {
     ///
     /// This method is cancellation safe.
     pub async fn next_entry(&mut self) -> io::Result<Option<DirEntry>> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
         poll_fn(|cx| self.poll_next_entry(cx)).await
     }
 

tokio/src/future/mod.rs

@@ -5,10 +5,6 @@
 #[cfg(any(feature = "macros", feature = "process"))]
 pub(crate) mod maybe_done;
 
-mod poll_fn;
-#[allow(unused_imports)]
-pub use poll_fn::poll_fn;
-
 cfg_process! {
     mod try_join;
     pub(crate) use try_join::try_join3;

tokio/src/future/poll_fn.rs

@@ -1,60 +0,0 @@
-#![allow(dead_code)]
-
-//! Definition of the `PollFn` adapter combinator.
-
-use std::fmt;
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-// This struct is intentionally `!Unpin` when `F` is `!Unpin`. This is to
-// mitigate the issue where rust puts noalias on mutable references to the
-// `PollFn` type if it is `Unpin`. If the closure has ownership of a future,
-// then this "leaks" and the future is affected by noalias too, which we don't
-// want.
-//
-// See this thread for more information:
-// <https://internals.rust-lang.org/t/surprising-soundness-trouble-around-pollfn/17484>
-//
-// The fact that `PollFn` is not `Unpin` when it shouldn't be is tested in
-// `tests/async_send_sync.rs`.
-
-/// Future for the [`poll_fn`] function.
-pub struct PollFn<F> {
-    f: F,
-}
-
-/// Creates a new future wrapping around a function returning [`Poll`].
-pub fn poll_fn<T, F>(f: F) -> PollFn<F>
-where
-    F: FnMut(&mut Context<'_>) -> Poll<T>,
-{
-    PollFn { f }
-}
-
-impl<F> fmt::Debug for PollFn<F> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("PollFn").finish()
-    }
-}
-
-impl<T, F> Future for PollFn<F>
-where
-    F: FnMut(&mut Context<'_>) -> Poll<T>,
-{
-    type Output = T;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
-        // Safety: We never construct a `Pin<&mut F>` anywhere, so accessing `f`
-        // mutably in an unpinned way is sound.
-        //
-        // This use of unsafe cannot be replaced with the pin-project macro
-        // because:
-        //  * If we put `#[pin]` on the field, then it gives us a `Pin<&mut F>`,
-        //    which we can't use to call the closure.
-        //  * If we don't put `#[pin]` on the field, then it makes `PollFn` be
-        //    unconditionally `Unpin`, which we also don't want.
-        let me = unsafe { Pin::into_inner_unchecked(self) };
-        (me.f)(cx)
-    }
-}

tokio/src/io/util/copy_bidirectional.rs

@@ -1,8 +1,8 @@
 use super::copy::CopyBuffer;
 
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite};
 
+use std::future::poll_fn;
 use std::io;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};

tokio/src/io/util/lines.rs

@@ -67,7 +67,7 @@ where
     /// # }
     /// ```
     pub async fn next_line(&mut self) -> io::Result<Option<String>> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         poll_fn(|cx| Pin::new(&mut *self).poll_next_line(cx)).await
     }

tokio/src/io/util/split.rs

@@ -59,7 +59,7 @@ where
     /// # }
     /// ```
     pub async fn next_segment(&mut self) -> io::Result<Option<Vec<u8>>> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         poll_fn(|cx| Pin::new(&mut *self).poll_next_segment(cx)).await
     }

tokio/src/macros/support.rs

@@ -1,7 +1,8 @@
 cfg_macros! {
-    pub use crate::future::poll_fn;
     pub use crate::future::maybe_done::maybe_done;
 
+    pub use std::future::poll_fn;
+
     #[doc(hidden)]
     pub fn thread_rng_n(n: u32) -> u32 {
         crate::runtime::context::thread_rng_n(n)

tokio/src/net/tcp/split.rs

@@ -8,10 +8,10 @@
 //! split has no associated overhead and enforces all invariants at the type
 //! level.
 
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite, Interest, ReadBuf, Ready};
 use crate::net::TcpStream;
 
+use std::future::poll_fn;
 use std::io;
 use std::net::{Shutdown, SocketAddr};
 use std::pin::Pin;
@@ -69,7 +69,7 @@ impl ReadHalf<'_> {
     /// use tokio::io::{self, ReadBuf};
     /// use tokio::net::TcpStream;
     ///
-    /// use futures::future::poll_fn;
+    /// use std::future::poll_fn;
     ///
     /// #[tokio::main]
     /// async fn main() -> io::Result<()> {

tokio/src/net/tcp/split_owned.rs

@@ -8,11 +8,11 @@
 //! split has no associated overhead and enforces all invariants at the type
 //! level.
 
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite, Interest, ReadBuf, Ready};
 use crate::net::TcpStream;
 
 use std::error::Error;
+use std::future::poll_fn;
 use std::net::{Shutdown, SocketAddr};
 use std::pin::Pin;
 use std::sync::Arc;
@@ -124,7 +124,7 @@ impl OwnedReadHalf {
     /// use tokio::io::{self, ReadBuf};
     /// use tokio::net::TcpStream;
     ///
-    /// use futures::future::poll_fn;
+    /// use std::future::poll_fn;
     ///
     /// #[tokio::main]
     /// async fn main() -> io::Result<()> {

tokio/src/net/tcp/stream.rs

@@ -1,6 +1,6 @@
 cfg_not_wasi! {
-    use crate::future::poll_fn;
     use crate::net::{to_socket_addrs, ToSocketAddrs};
+    use std::future::poll_fn;
     use std::time::Duration;
 }
 
@@ -340,7 +340,7 @@ impl TcpStream {
     /// use tokio::io::{self, ReadBuf};
     /// use tokio::net::TcpStream;
     ///
-    /// use futures::future::poll_fn;
+    /// use std::future::poll_fn;
     ///
     /// #[tokio::main]
     /// async fn main() -> io::Result<()> {

tokio/src/net/unix/stream.rs

@@ -1,4 +1,3 @@
-use crate::future::poll_fn;
 use crate::io::{AsyncRead, AsyncWrite, Interest, PollEvented, ReadBuf, Ready};
 use crate::net::unix::split::{split, ReadHalf, WriteHalf};
 use crate::net::unix::split_owned::{split_owned, OwnedReadHalf, OwnedWriteHalf};
@@ -6,6 +5,7 @@ use crate::net::unix::ucred::{self, UCred};
 use crate::net::unix::SocketAddr;
 
 use std::fmt;
+use std::future::poll_fn;
 use std::io::{self, Read, Write};
 use std::net::Shutdown;
 #[cfg(target_os = "android")]

tokio/src/runtime/coop.rs

@@ -255,7 +255,7 @@ mod test {
 
     #[test]
     fn budgeting() {
-        use futures::future::poll_fn;
+        use std::future::poll_fn;
         use tokio_test::*;
 
         assert!(get().0.is_none());

tokio/src/runtime/io/registration.rs

@@ -219,7 +219,7 @@ impl Registration {
         loop {
             let event = self.readiness(interest).await?;
 
-            let coop = crate::future::poll_fn(crate::runtime::coop::poll_proceed).await;
+            let coop = std::future::poll_fn(crate::runtime::coop::poll_proceed).await;
 
             match f() {
                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {

tokio/src/runtime/scheduler/current_thread/mod.rs

@@ -1,4 +1,3 @@
-use crate::future::poll_fn;
 use crate::loom::sync::atomic::AtomicBool;
 use crate::loom::sync::Arc;
 use crate::runtime::driver::{self, Driver};
@@ -15,7 +14,7 @@ use crate::util::{waker_ref, RngSeedGenerator, Wake, WakerRef};
 
 use std::cell::RefCell;
 use std::collections::VecDeque;
-use std::future::Future;
+use std::future::{poll_fn, Future};
 use std::sync::atomic::Ordering::{AcqRel, Release};
 use std::task::Poll::{Pending, Ready};
 use std::task::Waker;

tokio/src/runtime/tests/loom_local.rs

@@ -23,7 +23,7 @@ fn wake_during_shutdown() {
         ls.spawn_local(async move {
             let mut send = Some(send);
 
-            let () = futures::future::poll_fn(|cx| {
+            let () = std::future::poll_fn(|cx| {
                 if let Some(send) = send.take() {
                     send.send(cx.waker().clone());
                 }

tokio/src/runtime/tests/loom_multi_thread.rs

@@ -8,7 +8,6 @@ mod yield_now;
 /// Use `LOOM_MAX_PREEMPTIONS=1` to do a "quick" run as a smoke test.
 ///
 /// In order to speed up the C
-use crate::future::poll_fn;
 use crate::runtime::tests::loom_oneshot as oneshot;
 use crate::runtime::{self, Runtime};
 use crate::{spawn, task};
@@ -18,7 +17,7 @@ use loom::sync::atomic::{AtomicBool, AtomicUsize};
 use loom::sync::Arc;
 
 use pin_project_lite::pin_project;
-use std::future::Future;
+use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::atomic::Ordering::{Relaxed, SeqCst};
 use std::task::{ready, Context, Poll};

tokio/src/runtime/tests/loom_multi_thread_alt.rs

@@ -10,7 +10,6 @@ mod yield_now;
 /// Use `LOOM_MAX_PREEMPTIONS=1` to do a "quick" run as a smoke test.
 ///
 /// In order to speed up the C
-use crate::future::poll_fn;
 use crate::runtime::tests::loom_oneshot as oneshot;
 use crate::runtime::{self, Runtime};
 use crate::{spawn, task};
@@ -20,7 +19,7 @@ use loom::sync::atomic::{AtomicBool, AtomicUsize};
 use loom::sync::Arc;
 
 use pin_project_lite::pin_project;
-use std::future::Future;
+use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::atomic::Ordering::{Relaxed, SeqCst};
 use std::task::{ready, Context, Poll};

tokio/src/runtime/tests/task.rs

@@ -228,7 +228,7 @@ fn shutdown_immediately() {
 // Test for https://github.com/tokio-rs/tokio/issues/6729
 #[test]
 fn spawn_niche_in_task() {
-    use crate::future::poll_fn;
+    use std::future::poll_fn;
     use std::task::{Context, Poll, Waker};
 
     with(|rt| {

tokio/src/runtime/time/tests/mod.rs

@@ -54,10 +54,7 @@ fn single_timer() {
             );
             pin!(entry);
 
-            block_on(futures::future::poll_fn(|cx| {
-                entry.as_mut().poll_elapsed(cx)
-            }))
-            .unwrap();
+            block_on(std::future::poll_fn(|cx| entry.as_mut().poll_elapsed(cx))).unwrap();
         });
 
         thread::yield_now();
@@ -126,10 +123,7 @@ fn change_waker() {
                 .as_mut()
                 .poll_elapsed(&mut Context::from_waker(futures::task::noop_waker_ref()));
 
-            block_on(futures::future::poll_fn(|cx| {
-                entry.as_mut().poll_elapsed(cx)
-            }))
-            .unwrap();
+            block_on(std::future::poll_fn(|cx| entry.as_mut().poll_elapsed(cx))).unwrap();
         });
 
         thread::yield_now();
@@ -167,10 +161,7 @@ fn reset_future() {
             entry.as_mut().reset(start + Duration::from_secs(2), true);
 
             // shouldn't complete before 2s
-            block_on(futures::future::poll_fn(|cx| {
-                entry.as_mut().poll_elapsed(cx)
-            }))
-            .unwrap();
+            block_on(std::future::poll_fn(|cx| entry.as_mut().poll_elapsed(cx))).unwrap();
 
             finished_early_.store(true, Ordering::Relaxed);
         });

tokio/src/signal/mod.rs

@@ -84,7 +84,7 @@ impl RxFuture {
     }
 
     async fn recv(&mut self) -> Option<()> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
         poll_fn(|cx| self.poll_recv(cx)).await
     }
 

tokio/src/sync/mpsc/bounded.rs

@@ -238,7 +238,7 @@ impl<T> Receiver<T> {
     /// }
     /// ```
     pub async fn recv(&mut self) -> Option<T> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
         poll_fn(|cx| self.chan.recv(cx)).await
     }
 
@@ -314,7 +314,7 @@ impl<T> Receiver<T> {
     /// }
     /// ```
     pub async fn recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
         poll_fn(|cx| self.chan.recv_many(cx, buffer, limit)).await
     }
 

tokio/src/sync/mpsc/unbounded.rs

@@ -165,7 +165,7 @@ impl<T> UnboundedReceiver<T> {
     /// }
     /// ```
     pub async fn recv(&mut self) -> Option<T> {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         poll_fn(|cx| self.poll_recv(cx)).await
     }
@@ -239,7 +239,7 @@ impl<T> UnboundedReceiver<T> {
     /// }
     /// ```
     pub async fn recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
         poll_fn(|cx| self.chan.recv_many(cx, buffer, limit)).await
     }
 

tokio/src/sync/oneshot.rs

@@ -698,7 +698,7 @@ impl<T> Sender<T> {
     /// }
     /// ```
     pub async fn closed(&mut self) {
-        use crate::future::poll_fn;
+        use std::future::poll_fn;
 
         #[cfg(all(tokio_unstable, feature = "tracing"))]
         let resource_span = self.resource_span.clone();
@@ -775,7 +775,7 @@ impl<T> Sender<T> {
     /// ```
     /// use tokio::sync::oneshot;
     ///
-    /// use futures::future::poll_fn;
+    /// use std::future::poll_fn;
     ///
     /// #[tokio::main]
     /// async fn main() {

tokio/src/sync/tests/loom_atomic_waker.rs

@@ -1,9 +1,9 @@
 use crate::sync::task::AtomicWaker;
 
-use futures::future::poll_fn;
 use loom::future::block_on;
 use loom::sync::atomic::AtomicUsize;
 use loom::thread;
+use std::future::poll_fn;
 use std::sync::atomic::Ordering::Relaxed;
 use std::sync::Arc;
 use std::task::Poll::{Pending, Ready};

tokio/src/sync/tests/loom_mpsc.rs

@@ -1,9 +1,9 @@
 use crate::sync::mpsc;
 
-use futures::future::poll_fn;
 use loom::future::block_on;
 use loom::sync::Arc;
 use loom::thread;
+use std::future::poll_fn;
 use tokio_test::assert_ok;
 
 #[test]

tokio/src/sync/tests/loom_notify.rs

@@ -108,8 +108,7 @@ fn notify_multi() {
 
 #[test]
 fn notify_drop() {
-    use crate::future::poll_fn;
-    use std::future::Future;
+    use std::future::{poll_fn, Future};
     use std::task::Poll;
 
     loom::model(|| {

tokio/src/sync/tests/loom_oneshot.rs

@@ -1,8 +1,8 @@
 use crate::sync::oneshot;
 
-use futures::future::poll_fn;
 use loom::future::block_on;
 use loom::thread;
+use std::future::poll_fn;
 use std::task::Poll::{Pending, Ready};
 
 #[test]

tokio/src/sync/tests/loom_semaphore_batch.rs

@@ -1,10 +1,9 @@
 use crate::sync::batch_semaphore::*;
 
-use futures::future::poll_fn;
 use loom::future::block_on;
 use loom::sync::atomic::AtomicUsize;
 use loom::thread;
-use std::future::Future;
+use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::atomic::Ordering::SeqCst;
 use std::sync::Arc;

tokio/src/task/consume_budget.rs

@@ -27,7 +27,7 @@ use std::task::{ready, Poll};
 pub async fn consume_budget() {
     let mut status = Poll::Pending;
 
-    crate::future::poll_fn(move |cx| {
+    std::future::poll_fn(move |cx| {
         ready!(crate::trace::trace_leaf(cx));
         if status.is_ready() {
             return status;

tokio/src/task/join_set.rs

@@ -281,7 +281,7 @@ impl<T: 'static> JoinSet<T> {
     /// statement and some other branch completes first, it is guaranteed that no tasks were
     /// removed from this `JoinSet`.
     pub async fn join_next(&mut self) -> Option<Result<T, JoinError>> {
-        crate::future::poll_fn(|cx| self.poll_join_next(cx)).await
+        std::future::poll_fn(|cx| self.poll_join_next(cx)).await
     }
 
     /// Waits until one of the tasks in the set completes and returns its
@@ -303,7 +303,7 @@ impl<T: 'static> JoinSet<T> {
     #[cfg(tokio_unstable)]
     #[cfg_attr(docsrs, doc(cfg(tokio_unstable)))]
     pub async fn join_next_with_id(&mut self) -> Option<Result<(Id, T), JoinError>> {
-        crate::future::poll_fn(|cx| self.poll_join_next_with_id(cx)).await
+        std::future::poll_fn(|cx| self.poll_join_next_with_id(cx)).await
     }
 
     /// Tries to join one of the tasks in the set that has completed and return its output.

tokio/src/task/local.rs

@@ -1238,7 +1238,7 @@ mod tests {
             }));
 
             // poll the run until future once
-            crate::future::poll_fn(|cx| {
+            std::future::poll_fn(|cx| {
                 let _ = run_until.as_mut().poll(cx);
                 Poll::Ready(())
             })

tokio/src/time/interval.rs

@@ -1,8 +1,7 @@
-use crate::future::poll_fn;
 use crate::time::{sleep_until, Duration, Instant, Sleep};
 use crate::util::trace;
 
-use std::future::Future;
+use std::future::{poll_fn, Future};
 use std::panic::Location;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};

tokio/tests/io_async_fd.rs

@@ -400,12 +400,12 @@ async fn poll_fns() {
 
     let read_fut = tokio::spawn(async move {
         // Move waker onto this task first
-        assert_pending!(poll!(futures::future::poll_fn(|cx| afd_a_2
+        assert_pending!(poll!(std::future::poll_fn(|cx| afd_a_2
             .as_ref()
             .poll_read_ready(cx))));
         barrier_clone.wait().await;
 
-        let _ = futures::future::poll_fn(|cx| afd_a_2.as_ref().poll_read_ready(cx)).await;
+        let _ = std::future::poll_fn(|cx| afd_a_2.as_ref().poll_read_ready(cx)).await;
     });
 
     let afd_a_2 = afd_a.clone();
@@ -414,12 +414,12 @@ async fn poll_fns() {
 
     let mut write_fut = tokio::spawn(async move {
         // Move waker onto this task first
-        assert_pending!(poll!(futures::future::poll_fn(|cx| afd_a_2
+        assert_pending!(poll!(std::future::poll_fn(|cx| afd_a_2
             .as_ref()
             .poll_write_ready(cx))));
         barrier_clone.wait().await;
 
-        let _ = futures::future::poll_fn(|cx| afd_a_2.as_ref().poll_write_ready(cx)).await;
+        let _ = std::future::poll_fn(|cx| afd_a_2.as_ref().poll_write_ready(cx)).await;
     });
 
     r_barrier.wait().await;
@@ -530,11 +530,11 @@ fn driver_shutdown_wakes_pending_race() {
 }
 
 async fn poll_readable<T: AsRawFd>(fd: &AsyncFd<T>) -> std::io::Result<AsyncFdReadyGuard<'_, T>> {
-    futures::future::poll_fn(|cx| fd.poll_read_ready(cx)).await
+    std::future::poll_fn(|cx| fd.poll_read_ready(cx)).await
 }
 
 async fn poll_writable<T: AsRawFd>(fd: &AsyncFd<T>) -> std::io::Result<AsyncFdReadyGuard<'_, T>> {
-    futures::future::poll_fn(|cx| fd.poll_write_ready(cx)).await
+    std::future::poll_fn(|cx| fd.poll_write_ready(cx)).await
 }
 
 #[test]

tokio/tests/io_take.rs

@@ -33,7 +33,7 @@ async fn issue_4435() {
     let mut read_buf = ReadBuf::new(&mut buf);
     read_buf.put_slice(b"AB");
 
-    futures::future::poll_fn(|cx| rd.as_mut().poll_read(cx, &mut read_buf))
+    std::future::poll_fn(|cx| rd.as_mut().poll_read(cx, &mut read_buf))
         .await
         .unwrap();
     assert_eq!(&buf, &b"ABhell\0\0"[..]);

tokio/tests/macros_select.rs

@@ -11,7 +11,7 @@ use tokio::test as maybe_tokio_test;
 use tokio::sync::oneshot;
 use tokio_test::{assert_ok, assert_pending, assert_ready};
 
-use futures::future::poll_fn;
+use std::future::poll_fn;
 use std::task::Poll::Ready;
 
 #[maybe_tokio_test]

tokio/tests/rt_common.rs

@@ -112,8 +112,7 @@ rt_test! {
     use tokio_test::assert_err;
     use tokio_test::assert_ok;
 
-    use futures::future::poll_fn;
-    use std::future::Future;
+    use std::future::{poll_fn, Future};
     use std::pin::Pin;
 
     #[cfg(not(target_os="wasi"))]
@@ -696,7 +695,7 @@ rt_test! {
                 loop {
                     // Don't use Tokio's `yield_now()` to avoid special defer
                     // logic.
-                    futures::future::poll_fn::<(), _>(|cx| {
+                    std::future::poll_fn::<(), _>(|cx| {
                         cx.waker().wake_by_ref();
                         std::task::Poll::Pending
                     }).await;

tokio/tests/rt_threaded.rs

@@ -8,8 +8,7 @@ use tokio::runtime;
 use tokio::sync::oneshot;
 use tokio_test::{assert_err, assert_ok};
 
-use futures::future::poll_fn;
-use std::future::Future;
+use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::atomic::Ordering::Relaxed;
 use std::sync::atomic::{AtomicUsize, Ordering};

tokio/tests/rt_threaded_alt.rs

@@ -9,8 +9,7 @@ use tokio::runtime;
 use tokio::sync::oneshot;
 use tokio_test::{assert_err, assert_ok};
 
-use futures::future::poll_fn;
-use std::future::Future;
+use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering::Relaxed;

tokio/tests/tcp_accept.rs

@@ -96,8 +96,7 @@ async fn no_extra_poll() {
 
 #[tokio::test]
 async fn accept_many() {
-    use futures::future::poll_fn;
-    use std::future::Future;
+    use std::future::{poll_fn, Future};
     use std::sync::atomic::AtomicBool;
 
     const N: usize = 50;

tokio/tests/tcp_stream.rs

@@ -7,12 +7,11 @@ use tokio::try_join;
 use tokio_test::task;
 use tokio_test::{assert_ok, assert_pending, assert_ready_ok};
 
+use std::future::poll_fn;
 use std::io;
 use std::task::Poll;
 use std::time::Duration;
 
-use futures::future::poll_fn;
-
 #[tokio::test]
 async fn set_linger() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

tokio/tests/udp.rs

@@ -1,7 +1,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support bind or UDP
 
-use futures::future::poll_fn;
+use std::future::poll_fn;
 use std::io;
 use std::sync::Arc;
 use tokio::{io::ReadBuf, net::UdpSocket};

tokio/tests/uds_datagram.rs

@@ -2,11 +2,11 @@
 #![cfg(feature = "full")]
 #![cfg(unix)]
 
-use futures::future::poll_fn;
 use tokio::io::ReadBuf;
 use tokio::net::UnixDatagram;
 use tokio::try_join;
 
+use std::future::poll_fn;
 use std::io;
 use std::sync::Arc;