From 7268b0bb3a3f661869a539efaacfd8d049a74e9f Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Wed, 7 Aug 2019 16:26:44 -0400 Subject: [PATCH] Migrate threadpool to futures-util (#1403) * Migrate threadpool to futures-util Signed-off-by: Lucio Franco * fmt --- tokio-tcp/Cargo.toml | 2 +- tokio-tcp/src/listener.rs | 2 +- tokio-tcp/src/stream.rs | 2 +- tokio-threadpool/Cargo.toml | 6 +++--- tokio-threadpool/src/task/mod.rs | 3 ++- tokio-threadpool/src/waker.rs | 4 ++-- tokio-threadpool/tests/blocking.rs | 2 +- tokio-timer/Cargo.toml | 2 +- tokio-timer/src/delay_queue.rs | 3 +-- tokio-timer/src/interval.rs | 3 +-- 10 files changed, 14 insertions(+), 15 deletions(-) diff --git a/tokio-tcp/Cargo.toml b/tokio-tcp/Cargo.toml index f643831f3..af9208fd5 100644 --- a/tokio-tcp/Cargo.toml +++ b/tokio-tcp/Cargo.toml @@ -25,7 +25,7 @@ publish = false async-traits = [] [dependencies] -async-util = { git = "https://github.com/tokio-rs/async" } +futures-util-preview = "= 0.3.0-alpha.17" tokio-io = { version = "0.2.0", path = "../tokio-io" } tokio-reactor = { version = "0.2.0", path = "../tokio-reactor" } bytes = "0.4" diff --git a/tokio-tcp/src/listener.rs b/tokio-tcp/src/listener.rs index bc961f566..bd63d0916 100644 --- a/tokio-tcp/src/listener.rs +++ b/tokio-tcp/src/listener.rs @@ -2,6 +2,7 @@ use super::incoming::Incoming; use super::TcpStream; use futures_core::ready; +use futures_util::future::poll_fn; use mio; use std::convert::TryFrom; use std::fmt; @@ -87,7 +88,6 @@ impl TcpListener { /// ``` #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 pub async fn accept(&mut self) -> io::Result<(TcpStream, SocketAddr)> { - use async_util::future::poll_fn; poll_fn(|cx| self.poll_accept(cx)).await } diff --git a/tokio-tcp/src/stream.rs b/tokio-tcp/src/stream.rs index 9eb6d6592..e993ce2f9 100644 --- a/tokio-tcp/src/stream.rs +++ b/tokio-tcp/src/stream.rs @@ -2,9 +2,9 @@ use crate::split::{ split, split_mut, TcpStreamReadHalf, TcpStreamReadHalfMut, TcpStreamWriteHalf, TcpStreamWriteHalfMut, }; -use async_util::future::poll_fn; use bytes::{Buf, BufMut}; use futures_core::ready; +use futures_util::future::poll_fn; use iovec::IoVec; use mio; use std::convert::TryFrom; diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 836a50665..69630c2d0 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -25,9 +25,9 @@ publish = false [dependencies] tokio-executor = { version = "0.2.0", path = "../tokio-executor" } tokio-sync = { version = "0.2.0", path = "../tokio-sync" } -futures-core-preview = "0.3.0-alpha.17" +futures-core-preview = "= 0.3.0-alpha.17" +futures-util-preview = "= 0.3.0-alpha.17" -arc-waker = { git = "https://github.com/tokio-rs/async" } crossbeam-deque = "0.7.0" crossbeam-queue = "0.1.0" crossbeam-utils = "0.6.4" @@ -39,6 +39,6 @@ lazy_static = "1" [dev-dependencies] rand = "0.7" env_logger = "0.5" -async-util = { git = "https://github.com/tokio-rs/async" } tokio = { version = "0.2.0", path = "../tokio" } tokio-test = { version = "0.2.0", path = "../tokio-test" } +futures-util-preview = "= 0.3.0-alpha.17" diff --git a/tokio-threadpool/src/task/mod.rs b/tokio-threadpool/src/task/mod.rs index 0477dd517..81bb7c2d5 100644 --- a/tokio-threadpool/src/task/mod.rs +++ b/tokio-threadpool/src/task/mod.rs @@ -8,6 +8,7 @@ use self::state::State; use crate::pool::Pool; use crate::waker::Waker; +use futures_util::task::ArcWake; use log::trace; use std::cell::{Cell, UnsafeCell}; use std::future::Future; @@ -134,7 +135,7 @@ impl Task { let mut g = Guard(fut, true); - let waker = arc_waker::waker(Arc::new(Waker { + let waker = ArcWake::into_waker(Arc::new(Waker { task: me.clone(), pool: pool.clone(), })); diff --git a/tokio-threadpool/src/waker.rs b/tokio-threadpool/src/waker.rs index 9358bcd25..258a82a21 100644 --- a/tokio-threadpool/src/waker.rs +++ b/tokio-threadpool/src/waker.rs @@ -1,7 +1,7 @@ use crate::pool::Pool; use crate::task::Task; -use arc_waker::Wake; +use futures_util::task::ArcWake; use std::sync::Arc; /// Implements the future `Waker` API. @@ -17,7 +17,7 @@ pub(crate) struct Waker { unsafe impl Send for Waker {} unsafe impl Sync for Waker {} -impl Wake for Waker { +impl ArcWake for Waker { fn wake_by_ref(me: &Arc) { Task::schedule(&me.task, &me.pool); } diff --git a/tokio-threadpool/tests/blocking.rs b/tokio-threadpool/tests/blocking.rs index 012594816..96e945eb3 100644 --- a/tokio-threadpool/tests/blocking.rs +++ b/tokio-threadpool/tests/blocking.rs @@ -4,8 +4,8 @@ use tokio_test::*; use tokio_threadpool::*; -use async_util::future::poll_fn; use futures_core::ready; +use futures_util::future::poll_fn; use rand::*; use std::sync::atomic::Ordering::*; use std::sync::atomic::*; diff --git a/tokio-timer/Cargo.toml b/tokio-timer/Cargo.toml index 5cdbaa49f..3fb3cbd2b 100644 --- a/tokio-timer/Cargo.toml +++ b/tokio-timer/Cargo.toml @@ -28,8 +28,8 @@ async-traits = [] tokio-executor = { version = "0.2.0", path = "../tokio-executor" } tokio-sync = { version = "0.2.0", path = "../tokio-sync" } futures-core-preview = "0.3.0-alpha.17" +futures-util-preview = "0.3.0-alpha.17" -async-util = { git = "https://github.com/tokio-rs/async" } crossbeam-utils = "0.6.0" # Backs `DelayQueue` slab = "0.4.1" diff --git a/tokio-timer/src/delay_queue.rs b/tokio-timer/src/delay_queue.rs index 32ae1c26e..531007738 100644 --- a/tokio-timer/src/delay_queue.rs +++ b/tokio-timer/src/delay_queue.rs @@ -10,6 +10,7 @@ use crate::wheel::{self, Wheel}; use crate::{Delay, Error}; use futures_core::ready; +use futures_util::future::poll_fn; use slab::Slab; use std::cmp; use std::future::Future; @@ -373,8 +374,6 @@ impl DelayQueue { #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 #[allow(clippy::should_implement_trait)] // false positive : https://github.com/rust-lang/rust-clippy/issues/4290 pub async fn next(&mut self) -> Option, Error>> { - use async_util::future::poll_fn; - poll_fn(|cx| self.poll_next(cx)).await } diff --git a/tokio-timer/src/interval.rs b/tokio-timer/src/interval.rs index 2d37647c5..3aa98b1aa 100644 --- a/tokio-timer/src/interval.rs +++ b/tokio-timer/src/interval.rs @@ -2,6 +2,7 @@ use crate::clock; use crate::Delay; use futures_core::ready; +use futures_util::future::poll_fn; use std::future::Future; use std::pin::Pin; use std::task::{self, Poll}; @@ -75,8 +76,6 @@ impl Interval { #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 #[allow(clippy::should_implement_trait)] // false positive : https://github.com/rust-lang/rust-clippy/issues/4290 pub async fn next(&mut self) -> Option { - use async_util::future::poll_fn; - poll_fn(|cx| self.poll_next(cx)).await } }