diff --git a/Cargo.toml b/Cargo.toml index 7d93dcaf9..4b14d0772 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,7 +14,7 @@ members = [ # "tokio-signal", "tokio-sync", "tokio-test", - # "tokio-threadpool", + "tokio-threadpool", "tokio-timer", "tokio-tcp", # "tokio-tls", diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 5fb9aa432..43954510b 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -24,7 +24,9 @@ publish = false [dependencies] tokio-executor = { version = "0.2.0", path = "../tokio-executor" } -futures = "0.1.19" +tokio-sync = { version = "0.2.0", path = "../tokio-sync" } + +arc-waker = { git = "https://github.com/tokio-rs/async" } crossbeam-deque = "0.7.0" crossbeam-queue = "0.1.0" crossbeam-utils = "0.6.4" @@ -35,7 +37,6 @@ log = "0.4" [dev-dependencies] env_logger = "0.5" - -# For comparison benchmarks -futures-cpupool = "0.1.7" -threadpool = "1.7.1" +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" } diff --git a/tokio-threadpool/README.md b/tokio-threadpool/README.md index 103b13ea4..f122403a7 100644 --- a/tokio-threadpool/README.md +++ b/tokio-threadpool/README.md @@ -5,43 +5,6 @@ threads. [Documentation](https://docs.rs/tokio-threadpool/0.1.14/tokio_threadpool) -### Why not Rayon? - -Rayon is designed to handle parallelizing single computations by breaking them -into smaller chunks. The scheduling for each individual chunk doesn't matter as -long as the root computation completes in a timely fashion. In other words, -Rayon does not provide any guarantees of fairness with regards to how each task -gets scheduled. - -On the other hand, `tokio-threadpool` is a general purpose scheduler and -attempts to schedule each task fairly. This is the ideal behavior when -scheduling a set of unrelated tasks. - -### Why not futures-cpupool? - -It's 10x slower. - -## Examples - -```rust -use tokio_threadpool::ThreadPool; -use futures::{Future, lazy}; -use futures::sync::oneshot; - -pub fn main() { - let pool = ThreadPool::new(); - let (tx, rx) = oneshot::channel(); - - pool.spawn(lazy(|| { - println!("Running on the pool"); - tx.send("complete").map_err(|e| println!("send error, {}", e)) - })); - - println!("Result: {:?}", rx.wait()); - pool.shutdown().wait().unwrap(); -} -``` - ## License This project is licensed under the [MIT license](LICENSE). diff --git a/tokio-threadpool/src/blocking.rs b/tokio-threadpool/src/blocking.rs index 094612049..ef1667c2a 100644 --- a/tokio-threadpool/src/blocking.rs +++ b/tokio-threadpool/src/blocking.rs @@ -1,7 +1,8 @@ use crate::worker::Worker; -use futures::{try_ready, Poll}; + use std::error::Error; use std::fmt; +use std::task::Poll; /// Error raised by `blocking`. pub struct BlockingError { @@ -116,7 +117,7 @@ pub struct BlockingError { /// pool.shutdown_on_idle().wait().unwrap(); /// } /// ``` -pub fn blocking(f: F) -> Poll +pub fn blocking(f: F) -> Poll> where F: FnOnce() -> T, { @@ -124,7 +125,7 @@ where let worker = match worker { Some(worker) => worker, None => { - return Err(BlockingError { _p: () }); + return Poll::Ready(Err(BlockingError { _p: () })); } }; @@ -135,7 +136,7 @@ where }); // If the transition cannot happen, exit early - try_ready!(res); + ready!(res)?; // Currently in blocking mode, so call the inner closure let ret = f(); @@ -148,7 +149,7 @@ where }); // Return the result - Ok(ret.into()) + Poll::Ready(Ok(ret)) } impl fmt::Display for BlockingError { diff --git a/tokio-threadpool/src/lib.rs b/tokio-threadpool/src/lib.rs index 2aff652be..dcf95933e 100644 --- a/tokio-threadpool/src/lib.rs +++ b/tokio-threadpool/src/lib.rs @@ -131,21 +131,30 @@ pub mod park; +macro_rules! ready { + ($e:expr) => { + match $e { + ::std::task::Poll::Ready(t) => t, + ::std::task::Poll::Pending => return ::std::task::Poll::Pending, + } + }; +} + mod blocking; mod builder; mod callback; mod config; -mod notifier; mod pool; mod sender; mod shutdown; mod task; mod thread_pool; +mod waker; mod worker; pub use crate::blocking::{blocking, BlockingError}; pub use crate::builder::Builder; pub use crate::sender::Sender; pub use crate::shutdown::Shutdown; -pub use crate::thread_pool::{SpawnHandle, ThreadPool}; +pub use crate::thread_pool::ThreadPool; pub use crate::worker::{Worker, WorkerId}; diff --git a/tokio-threadpool/src/notifier.rs b/tokio-threadpool/src/notifier.rs deleted file mode 100644 index ed1429744..000000000 --- a/tokio-threadpool/src/notifier.rs +++ /dev/null @@ -1,91 +0,0 @@ -use crate::pool::Pool; -use crate::task::Task; -use futures::executor::Notify; -use log::trace; -use std::mem; -use std::ops; -use std::sync::Arc; - -/// Implements the future `Notify` API. -/// -/// This is how external events are able to signal the task, informing it to try -/// to poll the future again. -#[derive(Debug)] -pub(crate) struct Notifier { - pub pool: Arc, -} - -/// A guard that ensures that the inner value gets forgotten. -#[derive(Debug)] -struct Forget(Option); - -impl Notify for Notifier { - fn notify(&self, id: usize) { - trace!("Notifier::notify; id=0x{:x}", id); - - unsafe { - let ptr = id as *const Task; - - // We did not actually take ownership of the `Arc` in this function - // so we must ensure that the Arc is forgotten. - let task = Forget::new(Arc::from_raw(ptr)); - - // TODO: Unify this with Task::notify - if task.schedule() { - // TODO: Check if the pool is still running - // - // Bump the ref count - let task = task.clone(); - - let _ = self.pool.submit(task, &self.pool); - } - } - } - - fn clone_id(&self, id: usize) -> usize { - let ptr = id as *const Task; - - // This function doesn't actually get a strong ref to the task here. - // However, the only method we have to convert a raw pointer -> &Arc - // is to call `Arc::from_raw` which returns a strong ref. So, to - // maintain the invariants, `t1` has to be forgotten. This prevents the - // ref count from being decremented. - let t1 = Forget::new(unsafe { Arc::from_raw(ptr) }); - - // The clone is forgotten so that the fn exits without decrementing the ref - // count. The caller of `clone_id` ensures that `drop_id` is called when - // the ref count needs to be decremented. - let _ = Forget::new(t1.clone()); - - id - } - - fn drop_id(&self, id: usize) { - unsafe { - let ptr = id as *const Task; - let _ = Arc::from_raw(ptr); - } - } -} - -// ===== impl Forget ===== - -impl Forget { - fn new(t: T) -> Self { - Forget(Some(t)) - } -} - -impl ops::Deref for Forget { - type Target = T; - - fn deref(&self) -> &T { - self.0.as_ref().unwrap() - } -} - -impl Drop for Forget { - fn drop(&mut self) { - mem::forget(self.0.take()); - } -} diff --git a/tokio-threadpool/src/pool/mod.rs b/tokio-threadpool/src/pool/mod.rs index 793b5b5e3..0b14fcf12 100644 --- a/tokio-threadpool/src/pool/mod.rs +++ b/tokio-threadpool/src/pool/mod.rs @@ -15,7 +15,7 @@ use crate::task::{Blocking, Task}; use crate::worker::{self, Worker, WorkerId}; use crossbeam_deque::Injector; use crossbeam_utils::CachePadded; -use futures::Poll; + use log::{debug, error, trace}; use rand; use std::cell::Cell; @@ -23,6 +23,7 @@ use std::num::Wrapping; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{AcqRel, Acquire}; use std::sync::{Arc, Weak}; +use std::task::Poll; use std::thread; #[derive(Debug)] @@ -219,7 +220,10 @@ impl Pool { } } - pub fn poll_blocking_capacity(&self, task: &Arc) -> Poll<(), crate::BlockingError> { + pub fn poll_blocking_capacity( + &self, + task: &Arc, + ) -> Poll> { self.blocking.poll_blocking_capacity(task) } diff --git a/tokio-threadpool/src/sender.rs b/tokio-threadpool/src/sender.rs index ef79bed99..b2803806d 100644 --- a/tokio-threadpool/src/sender.rs +++ b/tokio-threadpool/src/sender.rs @@ -1,10 +1,13 @@ use crate::pool::{self, Lifecycle, Pool, MAX_FUTURES}; use crate::task::Task; -use futures::{future, Future}; + +use tokio_executor::{self, SpawnError}; + use log::trace; +use std::future::Future; +use std::pin::Pin; use std::sync::atomic::Ordering::{AcqRel, Acquire}; use std::sync::Arc; -use tokio_executor::{self, SpawnError}; /// Submit futures to the associated thread pool for execution. /// @@ -75,10 +78,10 @@ impl Sender { /// ``` pub fn spawn(&self, future: F) -> Result<(), SpawnError> where - F: Future + Send + 'static, + F: Future + Send + 'static, { let mut s = self; - tokio_executor::Executor::spawn(&mut s, Box::new(future)) + tokio_executor::Executor::spawn(&mut s, Box::pin(future)) } /// Logic to prepare for spawning @@ -128,7 +131,7 @@ impl tokio_executor::Executor for Sender { fn spawn( &mut self, - future: Box + Send>, + future: Pin + Send>>, ) -> Result<(), SpawnError> { let mut s = &*self; tokio_executor::Executor::spawn(&mut s, future) @@ -154,7 +157,7 @@ impl<'a> tokio_executor::Executor for &'a Sender { fn spawn( &mut self, - future: Box + Send>, + future: Pin + Send>>, ) -> Result<(), SpawnError> { self.prepare_for_spawn()?; @@ -175,34 +178,14 @@ impl<'a> tokio_executor::Executor for &'a Sender { impl tokio_executor::TypedExecutor for Sender where - T: Future + Send + 'static, + T: Future + Send + 'static, { fn status(&self) -> Result<(), tokio_executor::SpawnError> { tokio_executor::Executor::status(self) } fn spawn(&mut self, future: T) -> Result<(), SpawnError> { - tokio_executor::Executor::spawn(self, Box::new(future)) - } -} - -impl future::Executor for Sender -where - T: Future + Send + 'static, -{ - fn execute(&self, future: T) -> Result<(), future::ExecuteError> { - if let Err(e) = tokio_executor::Executor::status(self) { - let kind = if e.is_at_capacity() { - future::ExecuteErrorKind::NoCapacity - } else { - future::ExecuteErrorKind::Shutdown - }; - - return Err(future::ExecuteError::new(kind, future)); - } - - let _ = self.spawn(future); - Ok(()) + tokio_executor::Executor::spawn(self, Box::pin(future)) } } diff --git a/tokio-threadpool/src/shutdown.rs b/tokio-threadpool/src/shutdown.rs index 925f7e44a..ea15c4e2d 100644 --- a/tokio-threadpool/src/shutdown.rs +++ b/tokio-threadpool/src/shutdown.rs @@ -1,9 +1,13 @@ use crate::task::Task; use crate::worker; + +use tokio_sync::task::AtomicWaker; + use crossbeam_deque::Injector; -use futures::task::AtomicTask; -use futures::{Async, Future, Poll}; +use std::future::Future; +use std::pin::Pin; use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; /// Future that resolves when the thread pool is shutdown. /// @@ -27,7 +31,7 @@ pub struct Shutdown { #[derive(Debug)] struct Inner { /// The task to notify when the threadpool completes the shutdown process. - task: AtomicTask, + task: AtomicWaker, /// `true` if the threadpool has been shut down. completed: bool, } @@ -38,20 +42,25 @@ impl Shutdown { inner: trigger.inner.clone(), } } + + /// Wait for the shutdown to complete + pub fn wait(self) { + let mut enter = tokio_executor::enter().unwrap(); + enter.block_on(self); + } } impl Future for Shutdown { - type Item = (); - type Error = (); + type Output = (); - fn poll(&mut self) -> Poll<(), ()> { + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { let inner = self.inner.lock().unwrap(); if !inner.completed { - inner.task.register(); - Ok(Async::NotReady) + inner.task.register_by_ref(cx.waker()); + Poll::Pending } else { - Ok(().into()) + Poll::Ready(()) } } } @@ -74,7 +83,7 @@ impl ShutdownTrigger { ) -> ShutdownTrigger { ShutdownTrigger { inner: Arc::new(Mutex::new(Inner { - task: AtomicTask::new(), + task: AtomicWaker::new(), completed: false, })), workers, @@ -96,6 +105,6 @@ impl Drop for ShutdownTrigger { // Notify the task interested in shutdown. let mut inner = self.inner.lock().unwrap(); inner.completed = true; - inner.task.notify(); + inner.task.wake(); } } diff --git a/tokio-threadpool/src/task/blocking.rs b/tokio-threadpool/src/task/blocking.rs index 77a175b2a..75d41cd63 100644 --- a/tokio-threadpool/src/task/blocking.rs +++ b/tokio-threadpool/src/task/blocking.rs @@ -1,12 +1,13 @@ use crate::pool::Pool; use crate::task::{BlockingState, Task}; -use futures::{Async, Poll}; + use std::cell::UnsafeCell; use std::fmt; use std::ptr; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release}; use std::sync::Arc; +use std::task::Poll; use std::thread; /// Manages the state around entering a blocking section and tasks that are @@ -109,7 +110,10 @@ impl Blocking { /// /// The caller must ensure that `task` has not previously been queued to be /// notified when capacity becomes available. - pub fn poll_blocking_capacity(&self, task: &Arc) -> Poll<(), crate::BlockingError> { + pub fn poll_blocking_capacity( + &self, + task: &Arc, + ) -> Poll> { // This requires atomically claiming blocking capacity and if none is // available, queuing &task. @@ -193,7 +197,7 @@ impl Blocking { // The node was queued to be notified once capacity is made // available. - Ok(Async::NotReady) + Poll::Pending } None => { debug_assert!(curr.remaining_capacity() > 0); @@ -208,7 +212,7 @@ impl Blocking { } // Capacity has been obtained - Ok(().into()) + Poll::Ready(Ok(())) } } } diff --git a/tokio-threadpool/src/task/mod.rs b/tokio-threadpool/src/task/mod.rs index 7af070dd4..86348f42c 100644 --- a/tokio-threadpool/src/task/mod.rs +++ b/tokio-threadpool/src/task/mod.rs @@ -5,15 +5,17 @@ mod state; pub(crate) use self::blocking::{Blocking, CanBlock}; use self::blocking_state::BlockingState; use self::state::State; -use crate::notifier::Notifier; use crate::pool::Pool; -use futures::executor::{self, Spawn}; -use futures::{self, Async, Future}; +use crate::waker::Waker; + use log::trace; use std::cell::{Cell, UnsafeCell}; +use std::future::Future; +use std::pin::Pin; use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release}; use std::sync::atomic::{AtomicPtr, AtomicUsize}; use std::sync::Arc; +use std::task::{Context, Poll}; use std::{fmt, panic, ptr}; /// Harness around a future. @@ -48,7 +50,7 @@ pub(crate) struct Task { /// Store the future at the head of the struct /// /// The future is dropped immediately when it transitions to Complete - future: UnsafeCell>>, + future: UnsafeCell>, } #[derive(Debug)] @@ -58,31 +60,27 @@ pub(crate) enum Run { Complete, } -type BoxFuture = Box + Send + 'static>; +type BoxFuture = Pin + Send + 'static>>; // ===== impl Task ===== impl Task { /// Create a new `Task` as a harness for `future`. pub fn new(future: BoxFuture) -> Task { - // Wrap the future with an execution context. - let task_fut = executor::spawn(future); - Task { state: AtomicUsize::new(State::new().into()), blocking: AtomicUsize::new(BlockingState::new().into()), next_blocking: AtomicPtr::new(ptr::null_mut()), reg_worker: Cell::new(None), reg_index: Cell::new(0), - future: UnsafeCell::new(Some(task_fut)), + future: UnsafeCell::new(Some(future)), } } /// Create a fake `Task` to be used as part of the intrusive mpsc channel /// algorithm. fn stub() -> Task { - let future = Box::new(futures::empty()) as BoxFuture; - let task_fut = executor::spawn(future); + let future = Box::pin(Empty) as BoxFuture; Task { state: AtomicUsize::new(State::stub().into()), @@ -90,18 +88,18 @@ impl Task { next_blocking: AtomicPtr::new(ptr::null_mut()), reg_worker: Cell::new(None), reg_index: Cell::new(0), - future: UnsafeCell::new(Some(task_fut)), + future: UnsafeCell::new(Some(future)), } } /// Execute the task returning `Run::Schedule` if the task needs to be /// scheduled again. - pub fn run(&self, unpark: &Arc) -> Run { + pub fn run(me: &Arc, pool: &Arc) -> Run { use self::State::*; // Transition task to running state. At this point, the task must be // scheduled. - let actual: State = self + let actual: State = me .state .compare_and_swap(Scheduled.into(), Running.into(), AcqRel) .into(); @@ -111,14 +109,11 @@ impl Task { _ => panic!("unexpected task state; {:?}", actual), } - trace!( - "Task::run; state={:?}", - State::from(self.state.load(Relaxed)) - ); + trace!("Task::run; state={:?}", State::from(me.state.load(Relaxed))); // The transition to `Running` done above ensures that a lock on the // future has been obtained. - let fut = unsafe { &mut (*self.future.get()) }; + let fut = unsafe { &mut (*me.future.get()) }; // This block deals with the future panicking while being polled. // @@ -126,7 +121,7 @@ impl Task { // `thread::panicking() -> true`. To do this, the future is dropped from // within the catch_unwind block. let res = panic::catch_unwind(panic::AssertUnwindSafe(|| { - struct Guard<'a>(&'a mut Option>, bool); + struct Guard<'a>(&'a mut Option, bool); impl<'a> Drop for Guard<'a> { fn drop(&mut self) { @@ -139,10 +134,14 @@ impl Task { let mut g = Guard(fut, true); - let ret = - g.0.as_mut() - .unwrap() - .poll_future_notify(unpark, self as *const _ as usize); + let mut waker = arc_waker::waker(Arc::new(Waker { + task: me.clone(), + pool: pool.clone(), + })); + + let mut cx = Context::from_waker(&mut waker); + + let ret = g.0.as_mut().unwrap().as_mut().poll(&mut cx); g.1 = false; @@ -150,7 +149,7 @@ impl Task { })); match res { - Ok(Ok(Async::Ready(_))) | Ok(Err(_)) | Err(_) => { + Ok(Poll::Ready(_)) | Err(_) => { trace!(" -> task complete"); // The future has completed. Drop it immediately to free @@ -158,20 +157,20 @@ impl Task { // // The `Task` harness will stay around longer if it is contained // by any of the various queues. - self.drop_future(); + me.drop_future(); // Transition to the completed state - self.state.store(State::Complete.into(), Release); + me.state.store(State::Complete.into(), Release); if let Err(panic_err) = res { - if let Some(ref f) = unpark.pool.config.panic_handler { + if let Some(ref f) = pool.config.panic_handler { f(panic_err); } } Run::Complete } - Ok(Ok(Async::NotReady)) => { + Ok(Poll::Pending) => { trace!(" -> not ready"); // Attempt to transition from Running -> Idle, if successful, @@ -179,7 +178,7 @@ impl Task { // fails, then the task has been unparked concurrent to running, // in which case it transitions immediately back to scheduled // and we return `true`. - let prev: State = self + let prev: State = me .state .compare_and_swap(Running.into(), Idle.into(), AcqRel) .into(); @@ -187,7 +186,7 @@ impl Task { match prev { Running => Run::Idle, Notified => { - self.state.store(Scheduled.into(), Release); + me.state.store(Scheduled.into(), Release); Run::Schedule } _ => unreachable!(), @@ -231,23 +230,23 @@ impl Task { } } - /// Notify the task - pub fn notify(me: Arc, pool: &Arc) { - if me.schedule() { - let _ = pool.submit(me, pool); - } - } - /// Notify the task it has been allocated blocking capacity pub fn notify_blocking(me: Arc, pool: &Arc) { BlockingState::notify_blocking(&me.blocking, AcqRel); - Task::notify(me, pool); + Task::schedule(&me, pool); + } + + pub fn schedule(me: &Arc, pool: &Arc) { + if me.schedule2() { + let task = me.clone(); + let _ = pool.submit(task, &pool); + } } /// Transition the task state to scheduled. /// /// Returns `true` if the caller is permitted to schedule the task. - pub fn schedule(&self) -> bool { + fn schedule2(&self) -> bool { use self::State::*; loop { @@ -300,7 +299,18 @@ impl fmt::Debug for Task { fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { fmt.debug_struct("Task") .field("state", &self.state) - .field("future", &"Spawn") + .field("future", &"BoxFuture") .finish() } } + +struct Empty; + +impl Future for Empty { + type Output = (); + + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + // Never used + unreachable!(); + } +} diff --git a/tokio-threadpool/src/thread_pool.rs b/tokio-threadpool/src/thread_pool.rs index 62abd1e73..afb9a049e 100644 --- a/tokio-threadpool/src/thread_pool.rs +++ b/tokio-threadpool/src/thread_pool.rs @@ -2,8 +2,8 @@ use crate::builder::Builder; use crate::pool::Pool; use crate::sender::Sender; use crate::shutdown::{Shutdown, ShutdownTrigger}; -use futures::sync::oneshot; -use futures::{Future, Poll}; + +use std::future::Future; use std::sync::Arc; /// Work-stealing based thread pool for executing futures. @@ -72,11 +72,14 @@ impl ThreadPool { /// version that returns a `Result` instead of panicking. pub fn spawn(&self, future: F) where - F: Future + Send + 'static, + F: Future + Send + 'static, { self.sender().spawn(future).unwrap(); } + /* + * TODO: Bring back + /// Spawn a future on to the thread pool, return a future representing /// the produced value. /// @@ -114,6 +117,8 @@ impl ThreadPool { SpawnHandle(oneshot::spawn(future, self.sender())) } + */ + /// Return a reference to the sender handle /// /// The handle is used to spawn futures onto the thread pool. It also @@ -183,11 +188,15 @@ impl Drop for ThreadPool { drop(inner); // Wait until all worker threads terminate and the threadpool's resources clean up. - let _ = shutdown.wait(); + let mut enter = tokio_executor::enter().unwrap(); + enter.block_on(shutdown); } } } +/* + * TODO: Bring back + /// Handle returned from ThreadPool::spawn_handle. /// /// This handle is a future representing the completion of a different future @@ -205,3 +214,5 @@ impl Future for SpawnHandle { self.0.poll() } } + +*/ diff --git a/tokio-threadpool/src/waker.rs b/tokio-threadpool/src/waker.rs new file mode 100644 index 000000000..9358bcd25 --- /dev/null +++ b/tokio-threadpool/src/waker.rs @@ -0,0 +1,24 @@ +use crate::pool::Pool; +use crate::task::Task; + +use arc_waker::Wake; +use std::sync::Arc; + +/// Implements the future `Waker` API. +/// +/// This is how external events are able to signal the task, informing it to try +/// to poll the future again. +#[derive(Debug)] +pub(crate) struct Waker { + pub pool: Arc, + pub task: Arc, +} + +unsafe impl Send for Waker {} +unsafe impl Sync for Waker {} + +impl Wake for Waker { + fn wake_by_ref(me: &Arc) { + Task::schedule(&me.task, &me.pool); + } +} diff --git a/tokio-threadpool/src/worker/mod.rs b/tokio-threadpool/src/worker/mod.rs index 9165f3945..ef6636d7d 100644 --- a/tokio-threadpool/src/worker/mod.rs +++ b/tokio-threadpool/src/worker/mod.rs @@ -6,21 +6,22 @@ pub(crate) use self::entry::WorkerEntry as Entry; pub(crate) use self::stack::Stack; pub(crate) use self::state::{Lifecycle, State}; -use crate::notifier::Notifier; use crate::pool::{self, BackupId, Pool}; use crate::sender::Sender; use crate::shutdown::ShutdownTrigger; use crate::task::{self, CanBlock, Task}; -use futures::{Async, Poll}; + +use tokio_executor; + use log::trace; use std::cell::Cell; use std::marker::PhantomData; use std::rc::Rc; use std::sync::atomic::Ordering::{AcqRel, Acquire}; use std::sync::Arc; +use std::task::Poll; use std::thread; use std::time::Duration; -use tokio_executor; /// Thread worker /// @@ -148,7 +149,7 @@ impl Worker { } /// Transition the current worker to a blocking worker - pub(crate) fn transition_to_blocking(&self) -> Poll<(), crate::BlockingError> { + pub(crate) fn transition_to_blocking(&self) -> Poll> { use self::CanBlock::*; // If we get this far, then `current_task` has been set. @@ -161,7 +162,7 @@ impl Worker { // The task has already requested capacity to block, but there is // none yet available. - NoCapacity => return Ok(Async::NotReady), + NoCapacity => return Poll::Pending, // The task has yet to ask for capacity CanRequest => { @@ -169,12 +170,12 @@ impl Worker { // is available, register the task to be notified once capacity // becomes available. match self.pool.poll_blocking_capacity(task_ref)? { - Async::Ready(()) => { + Poll::Ready(()) => { self.current_task.set_can_block(Allocated); } - Async::NotReady => { + Poll::Pending => { self.current_task.set_can_block(NoCapacity); - return Ok(Async::NotReady); + return Poll::Pending; } } } @@ -187,7 +188,7 @@ impl Worker { if self.is_blocking.get() { // The thread is already in blocking mode, so there is nothing else // to do. Return `Ready` and allow the caller to block the thread. - return Ok(().into()); + return Poll::Ready(Ok(())); } trace!("transition to blocking state"); @@ -200,7 +201,7 @@ impl Worker { // Track that the thread has now fully entered the blocking state. self.is_blocking.set(true); - Ok(().into()) + Poll::Ready(Ok(())) } /// Transition from blocking @@ -223,11 +224,6 @@ impl Worker { const MAX_SPINS: usize = 3; const LIGHT_SLEEP_INTERVAL: usize = 32; - // Get the notifier. - let notify = Arc::new(Notifier { - pool: self.pool.clone(), - }); - let mut first = true; let mut spin_cnt = 0; let mut tick = 0; @@ -236,7 +232,7 @@ impl Worker { first = false; // Run the next available task - if self.try_run_task(¬ify) { + if self.try_run_task(&self.pool) { if self.is_blocking.get() { // Exit out of the run state return; @@ -291,12 +287,12 @@ impl Worker { /// /// Returns `true` if work was found. #[inline] - fn try_run_task(&self, notify: &Arc) -> bool { - if self.try_run_owned_task(notify) { + fn try_run_task(&self, pool: &Arc) -> bool { + if self.try_run_owned_task(pool) { return true; } - self.try_steal_task(notify) + self.try_steal_task(pool) } /// Checks the worker's current state, updating it as needed. @@ -381,11 +377,11 @@ impl Worker { /// Runs the next task on this worker's queue. /// /// Returns `true` if work was found. - fn try_run_owned_task(&self, notify: &Arc) -> bool { + fn try_run_owned_task(&self, pool: &Arc) -> bool { // Poll the internal queue for a task to run match self.entry().pop_task() { Some(task) => { - self.run_task(task, notify); + self.run_task(task, pool); true } None => false, @@ -395,7 +391,7 @@ impl Worker { /// Tries to steal a task from another worker. /// /// Returns `true` if work was found - fn try_steal_task(&self, notify: &Arc) -> bool { + fn try_steal_task(&self, pool: &Arc) -> bool { use crossbeam_deque::Steal; debug_assert!(!self.is_blocking.get()); @@ -411,7 +407,7 @@ impl Worker { Steal::Success(task) => { trace!("stole task from another worker"); - self.run_task(task, notify); + self.run_task(task, pool); trace!( "try_steal_task -- signal_work; self={}; from={}", @@ -444,7 +440,7 @@ impl Worker { found_work } - fn run_task(&self, task: Arc, notify: &Arc) { + fn run_task(&self, task: Arc, pool: &Arc) { use crate::task::Run::*; // If this is the first time this task is being polled, register it so that we can keep @@ -454,7 +450,7 @@ impl Worker { self.entry().register_task(&task); } - let run = self.run_task2(&task, notify); + let run = self.run_task2(&task, pool); // TODO: Try to claim back the worker state in case the backup thread // did not start up fast enough. This is a performance optimization. @@ -528,7 +524,7 @@ impl Worker { /// /// Great care is needed to ensure that `current_task` is unset in this /// function. - fn run_task2(&self, task: &Arc, notify: &Arc) -> task::Run { + fn run_task2(&self, task: &Arc, pool: &Arc) -> task::Run { struct Guard<'a> { worker: &'a Worker, } @@ -562,7 +558,7 @@ impl Worker { // function returns, even if the return is caused by a panic. let _g = Guard { worker: self }; - task.run(notify) + Task::run(task, pool) } /// Put the worker to sleep diff --git a/tokio-threadpool/tests/blocking.rs b/tokio-threadpool/tests/blocking.rs index e3c2576fb..e5ef81a64 100644 --- a/tokio-threadpool/tests/blocking.rs +++ b/tokio-threadpool/tests/blocking.rs @@ -1,14 +1,26 @@ -#![deny(warnings, rust_2018_idioms)] +#![deny(/* warnings, */ rust_2018_idioms)] +#![feature(async_await)] -use futures::future::{lazy, poll_fn}; -use futures::*; +use tokio_test::*; +use tokio_threadpool::*; + +use async_util::future::poll_fn; use rand::*; use std::sync::atomic::Ordering::*; use std::sync::atomic::*; use std::sync::*; +use std::task::{Poll, Waker}; use std::thread; use std::time::Duration; -use tokio_threadpool::*; + +macro_rules! ready { + ($e:expr) => { + match $e { + ::std::task::Poll::Ready(t) => t, + ::std::task::Poll::Pending => return ::std::task::Poll::Pending, + } + }; +} #[test] fn basic() { @@ -19,21 +31,18 @@ fn basic() { let (tx1, rx1) = mpsc::channel(); let (tx2, rx2) = mpsc::channel(); - pool.spawn(lazy(move || { + pool.spawn(async move { let res = blocking(|| { let v = rx1.recv().unwrap(); tx2.send(v).unwrap(); - }) - .unwrap(); + }); - assert!(res.is_ready()); - Ok(().into()) - })); + assert_ready!(res).unwrap(); + }); - pool.spawn(lazy(move || { + pool.spawn(async move { tx1.send(()).unwrap(); - Ok(().into()) - })); + }); rx2.recv().unwrap(); } @@ -51,11 +60,11 @@ fn notify_task_on_capacity() { let rem = rem.clone(); let tx = tx.clone(); - pool.spawn(lazy(move || { - poll_fn(move || { + pool.spawn(async move { + poll_fn(move |_| { blocking(|| { thread::sleep(Duration::from_millis(100)); - let prev = rem.fetch_sub(1, Relaxed); + let prev = rem.fetch_sub(1, SeqCst); if prev == 1 { tx.send(()).unwrap(); @@ -63,20 +72,19 @@ fn notify_task_on_capacity() { }) .map_err(|e| panic!("blocking err {:?}", e)) }) - })); + .await + .unwrap() + }); } rx.recv().unwrap(); - assert_eq!(0, rem.load(Relaxed)); + assert_eq!(0, rem.load(SeqCst)); } #[test] fn capacity_is_use_it_or_lose_it() { - use futures::sync::oneshot; - use futures::task::Task; - use futures::Async::*; - use futures::*; + use tokio_sync::oneshot; // TODO: Run w/ bigger pool size @@ -88,74 +96,77 @@ fn capacity_is_use_it_or_lose_it() { let (tx4, rx4) = mpsc::channel(); // First, fill the blocking capacity - pool.spawn(lazy(move || { - poll_fn(move || { + pool.spawn(async move { + poll_fn(move |_| { blocking(|| { rx1.recv().unwrap(); }) - .map_err(|_| panic!()) }) - })); + .await + .unwrap() + }); - pool.spawn(lazy(move || { - rx2.map_err(|_| panic!()).and_then(|task: Task| { - poll_fn(move || { - blocking(|| { - // Notify the other task - task.notify(); + pool.spawn(async move { + let task: Waker = rx2.await.unwrap(); - // Block until woken - rx3.recv().unwrap(); - }) - .map_err(|_| panic!()) + poll_fn(move |_| { + blocking(|| { + // Notify the other task + task.wake_by_ref(); + + // Block until woken + rx3.recv().unwrap(); }) }) - })); + .await + .unwrap(); + }); // Spawn a future that will try to block, get notified, then not actually // use the blocking let mut i = 0; let mut tx2 = Some(tx2); - pool.spawn(lazy(move || { - poll_fn(move || { + pool.spawn(async move { + poll_fn(move |cx| { match i { 0 => { i = 1; let res = blocking(|| unreachable!()).map_err(|_| panic!()); - assert!(res.unwrap().is_not_ready()); + assert_pending!(res); // Unblock the first blocker tx1.send(()).unwrap(); - return Ok(NotReady); + return Poll::Pending; } 1 => { i = 2; // Skip blocking, and notify the second task that it should // start blocking - let me = task::current(); + let me = cx.waker().clone(); tx2.take().unwrap().send(me).unwrap(); - return Ok(NotReady); + return Poll::Pending; } 2 => { let res = blocking(|| unreachable!()).map_err(|_| panic!()); - assert!(res.unwrap().is_not_ready()); + assert_pending!(res); // Unblock the first blocker tx3.send(()).unwrap(); tx4.send(()).unwrap(); - Ok(().into()) + Poll::Ready(()) } _ => unreachable!(), } }) - })); + .await + }); rx4.recv().unwrap(); } @@ -173,33 +184,35 @@ fn blocking_thread_does_not_take_over_shutdown_worker_thread() { { let exited = exited.clone(); - pool.spawn(lazy(move || { - poll_fn(move || { + pool.spawn(async move { + poll_fn(move |_| { blocking(|| { enter_tx.send(()).unwrap(); exit_rx.recv().unwrap(); - exited.store(true, Relaxed); + exited.store(true, SeqCst); }) - .map_err(|_| panic!()) }) - })); + .await + .unwrap() + }); } // Wait for the task to block let _ = enter_rx.recv().unwrap(); // Spawn another task that attempts to block - pool.spawn(lazy(move || { - poll_fn(move || { - let res = blocking(|| {}).unwrap(); + pool.spawn(async move { + poll_fn(move |_| { + let res = blocking(|| {}); - assert_eq!(res.is_ready(), exited.load(Relaxed)); + assert_eq!(res.is_ready(), exited.load(SeqCst)); try_tx.send(res.is_ready()).unwrap(); - Ok(res) + res.map(|_| ()) }) - })); + .await + }); // Wait for the second task to try to block (and not be ready). let res = try_rx.recv().unwrap(); @@ -230,35 +243,35 @@ fn blocking_one_time_gets_capacity_for_multiple_blocks() { let rem = rem.clone(); let tx = tx.clone(); - pool.spawn(lazy(move || { - poll_fn(move || { + pool.spawn(async move { + poll_fn(move |_| { // First block let res = blocking(|| { thread::sleep(Duration::from_millis(100)); - }) - .map_err(|e| panic!("blocking err {:?}", e)); + }); - try_ready!(res); + ready!(res).unwrap(); let res = blocking(|| { thread::sleep(Duration::from_millis(100)); - let prev = rem.fetch_sub(1, Relaxed); + let prev = rem.fetch_sub(1, SeqCst); if prev == 1 { tx.send(()).unwrap(); } }); - assert!(res.unwrap().is_ready()); + assert!(res.is_ready()); - Ok(().into()) + Poll::Ready(()) }) - })); + .await + }); } rx.recv().unwrap(); - assert_eq!(0, rem.load(Relaxed)); + assert_eq!(0, rem.load(SeqCst)); } } @@ -280,10 +293,10 @@ fn shutdown() { .pool_size(1) .max_blocking(BLOCKING) .after_start(move || { - num_inc.fetch_add(1, Relaxed); + num_inc.fetch_add(1, SeqCst); }) .before_stop(move || { - num_dec.fetch_add(1, Relaxed); + num_dec.fetch_add(1, SeqCst); }) .build() }; @@ -294,18 +307,16 @@ fn shutdown() { let barrier = barrier.clone(); let tx = tx.clone(); - pool.spawn(lazy(move || { + pool.spawn(async move { let res = blocking(|| { barrier.wait(); Ok::<_, ()>(()) - }) - .unwrap(); + }); tx.send(()).unwrap(); assert!(res.is_ready()); - Ok(().into()) - })); + }); } for _ in 0..BLOCKING { @@ -315,8 +326,8 @@ fn shutdown() { // Shutdown drop(pool); - assert_eq!(11, num_inc.load(Relaxed)); - assert_eq!(11, num_dec.load(Relaxed)); + assert_eq!(11, num_inc.load(SeqCst)); + assert_eq!(11, num_dec.load(SeqCst)); } } @@ -356,10 +367,10 @@ fn hammer() { let cnt_task = cnt_task.clone(); let cnt_block = cnt_block.clone(); - pool.spawn(lazy(move || { - cnt_task.fetch_add(1, Relaxed); + pool.spawn(async move { + cnt_task.fetch_add(1, SeqCst); - poll_fn(move || { + poll_fn(move |_| { blocking(|| { match sleep { Skip => {} @@ -375,18 +386,20 @@ fn hammer() { } } - cnt_block.fetch_add(1, Relaxed); + cnt_block.fetch_add(1, SeqCst); }) .map_err(|_| panic!()) }) - })); + .await + .unwrap() + }); } // Wait for the work to complete - pool.shutdown_on_idle().wait().unwrap(); + pool.shutdown_on_idle().wait(); - assert_eq!(n, cnt_task.load(Relaxed)); - assert_eq!(n, cnt_block.load(Relaxed)); + assert_eq!(n, cnt_task.load(SeqCst)); + assert_eq!(n, cnt_block.load(SeqCst)); } } } diff --git a/tokio-threadpool/tests/hammer.rs b/tokio-threadpool/tests/hammer.rs index 71db21921..d8533bd8c 100644 --- a/tokio-threadpool/tests/hammer.rs +++ b/tokio-threadpool/tests/hammer.rs @@ -1,16 +1,18 @@ #![deny(warnings, rust_2018_idioms)] +#![feature(async_await)] -use futures::{Future, Poll, Sink, Stream}; +use tokio_sync::{mpsc, oneshot}; +use tokio_threadpool::*; + +use std::future::Future; +use std::pin::Pin; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::*; use std::sync::Arc; -use tokio_threadpool::*; +use std::task::{Context, Poll}; #[test] fn hammer() { - use futures::future; - use futures::sync::{mpsc, oneshot}; - const N: usize = 1000; const ITER: usize = 20; @@ -20,11 +22,13 @@ fn hammer() { } impl Future for Counted { - type Item = T::Item; - type Error = T::Error; + type Output = T::Output; - fn poll(&mut self) -> Poll { - self.inner.poll() + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + unsafe { + let inner = &mut self.get_unchecked_mut().inner; + Pin::new_unchecked(inner).poll(cx) + } } } @@ -41,13 +45,30 @@ fn hammer() { let cnt = Arc::new(AtomicUsize::new(0)); - let (listen_tx, listen_rx) = mpsc::unbounded::>>(); - let mut listen_tx = listen_tx.wait(); + let (mut listen_tx, mut listen_rx) = + mpsc::unbounded_channel::>>(); pool.spawn({ let c1 = cnt.clone(); let c2 = cnt.clone(); let pool = pool.sender().clone(); + let task = async move { + while let Some(tx) = listen_rx.recv().await { + let task = async { + let (tx2, rx2) = oneshot::channel(); + tx.send(tx2).unwrap(); + rx2.await.unwrap() + }; + + pool.spawn(Counted { + inner: task, + cnt: c1.clone(), + }) + .unwrap(); + } + }; + + /* let task = listen_rx .map_err(|e| panic!("accept error = {:?}", e)) .for_each(move |tx| { @@ -68,6 +89,7 @@ fn hammer() { Ok(()) }); + */ Counted { inner: task, @@ -78,21 +100,28 @@ fn hammer() { for _ in 0..N { let cnt = cnt.clone(); let (tx, rx) = oneshot::channel(); - listen_tx.send(tx).unwrap(); + listen_tx.try_send(tx).unwrap(); - pool.spawn({ + pool.spawn(async { + let task = async { + let tx = rx.await.unwrap(); + tx.send(()).unwrap(); + }; + + /* let task = rx.map_err(|e| panic!("rx err={:?}", e)).and_then(|tx| { tx.send(()).unwrap(); Ok(()) }); + */ - Counted { inner: task, cnt } + Counted { inner: task, cnt }.await }); } drop(listen_tx); - pool.shutdown_on_idle().wait().unwrap(); + pool.shutdown_on_idle().wait(); assert_eq!(N * 2 + 1, cnt.load(Relaxed)); } } diff --git a/tokio-threadpool/tests/threadpool.rs b/tokio-threadpool/tests/threadpool.rs index 9b8b8c0f8..398364c0a 100644 --- a/tokio-threadpool/tests/threadpool.rs +++ b/tokio-threadpool/tests/threadpool.rs @@ -1,23 +1,21 @@ #![deny(warnings, rust_2018_idioms)] +#![feature(async_await)] -use futures::future::lazy; -use futures::{Async, Future, Poll, Sink, Stream}; -use std::cell::Cell; -use std::sync::atomic::Ordering::Relaxed; -use std::sync::atomic::*; -use std::sync::{mpsc, Arc}; -use std::time::Duration; use tokio_executor::park::{Park, Unpark}; +use tokio_test::assert_pending; use tokio_threadpool::park::{DefaultPark, DefaultUnpark}; use tokio_threadpool::*; -thread_local!(static FOO: Cell = Cell::new(0)); +use std::cell::Cell; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::Ordering::Relaxed; +use std::sync::atomic::*; +use std::sync::{mpsc, Arc}; +use std::task::{Context, Poll, Waker}; +use std::time::Duration; -fn ignore_results( - f: F, -) -> Box + Send> { - Box::new(f.map(|_| ()).map_err(|_| ())) -} +thread_local!(static FOO: Cell = Cell::new(0)); #[test] fn natural_shutdown_simple_futures() { @@ -47,26 +45,24 @@ fn natural_shutdown_simple_futures() { let a = { let (t, rx) = mpsc::channel(); - tx.spawn(lazy(move || { + tx.spawn(async move { // Makes sure this runs on a worker thread FOO.with(|f| assert_eq!(f.get(), 0)); t.send("one").unwrap(); - Ok(()) - })) + }) .unwrap(); rx }; let b = { let (t, rx) = mpsc::channel(); - tx.spawn(lazy(move || { + tx.spawn(async move { // Makes sure this runs on a worker thread FOO.with(|f| assert_eq!(f.get(), 0)); t.send("two").unwrap(); - Ok(()) - })) + }) .unwrap(); rx }; @@ -77,7 +73,7 @@ fn natural_shutdown_simple_futures() { assert_eq!("two", b.recv().unwrap()); // Wait for the pool to shutdown - pool.shutdown().wait().unwrap(); + pool.shutdown().wait(); // Assert that at least one thread started let num_inc = num_inc.load(Relaxed); @@ -102,11 +98,10 @@ fn force_shutdown_drops_futures() { struct Never(Arc); impl Future for Never { - type Item = (); - type Error = (); + type Output = (); - fn poll(&mut self) -> Poll<(), ()> { - Ok(Async::NotReady) + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + Poll::Pending } } @@ -131,7 +126,7 @@ fn force_shutdown_drops_futures() { tx.spawn(Never(num_drop.clone())).unwrap(); // Wait for the pool to shutdown - pool.shutdown_now().wait().unwrap(); + pool.shutdown_now().wait(); // Assert that only a single thread was spawned. let a = num_inc.load(Relaxed); @@ -159,11 +154,10 @@ fn drop_threadpool_drops_futures() { struct Never(Arc); impl Future for Never { - type Item = (); - type Error = (); + type Output = (); - fn poll(&mut self) -> Poll<(), ()> { - Ok(Async::NotReady) + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + Poll::Pending } } @@ -219,15 +213,14 @@ fn many_oneshot_futures() { for _ in 0..NUM { let cnt = cnt.clone(); - tx.spawn(lazy(move || { + tx.spawn(async move { cnt.fetch_add(1, Relaxed); - Ok(()) - })) + }) .unwrap(); } // Wait for the pool to shutdown - pool.shutdown().wait().unwrap(); + pool.shutdown().wait(); let num = cnt.load(Relaxed); assert_eq!(num, NUM); @@ -236,7 +229,7 @@ fn many_oneshot_futures() { #[test] fn many_multishot_futures() { - use futures::sync::mpsc; + use tokio::sync::mpsc; const CHAIN: usize = 200; const CYCLES: usize = 5; @@ -255,57 +248,61 @@ fn many_multishot_futures() { let (start_tx, mut chain_rx) = mpsc::channel(10); for _ in 0..CHAIN { - let (next_tx, next_rx) = mpsc::channel(10); - - let rx = chain_rx.map_err(|e| panic!("{:?}", e)); + let (mut next_tx, next_rx) = mpsc::channel(10); // Forward all the messages pool_tx - .spawn( - next_tx - .send_all(rx) - .map(|_| ()) - .map_err(|e| panic!("{:?}", e)), - ) + .spawn(async move { + while let Some(v) = chain_rx.recv().await { + next_tx.send(v).await.unwrap(); + } + }) .unwrap(); chain_rx = next_rx; } // This final task cycles if needed - let (final_tx, final_rx) = mpsc::channel(10); - let cycle_tx = start_tx.clone(); + let (mut final_tx, final_rx) = mpsc::channel(10); + let mut cycle_tx = start_tx.clone(); let mut rem = CYCLES; - let task = chain_rx.take(CYCLES as u64).for_each(move |msg| { - rem -= 1; - let send = if rem == 0 { - final_tx.clone().send(msg) - } else { - cycle_tx.clone().send(msg) - }; + pool_tx + .spawn(async move { + for _ in 0..CYCLES { + let msg = chain_rx.recv().await.unwrap(); - send.then(|res| { - res.unwrap(); - Ok(()) + rem -= 1; + + if rem == 0 { + final_tx.send(msg).await.unwrap(); + } else { + cycle_tx.send(msg).await.unwrap(); + } + } }) - }); - pool_tx.spawn(ignore_results(task)).unwrap(); + .unwrap(); start_txs.push(start_tx); final_rxs.push(final_rx); } - for start_tx in start_txs { - start_tx.send("ping").wait().unwrap(); - } + { + let mut e = tokio_executor::enter().unwrap(); - for final_rx in final_rxs { - final_rx.wait().next().unwrap().unwrap(); + e.block_on(async move { + for mut start_tx in start_txs { + start_tx.send("ping").await.unwrap(); + } + + for mut final_rx in final_rxs { + final_rx.recv().await.unwrap(); + } + }); } // Shutdown the pool - pool.shutdown().wait().unwrap(); + pool.shutdown().wait(); } } @@ -316,30 +313,27 @@ fn global_executor_is_configured() { let (signal_tx, signal_rx) = mpsc::channel(); - tx.spawn(lazy(move || { - tokio_executor::spawn(lazy(move || { + tx.spawn(async move { + tokio_executor::spawn(async move { signal_tx.send(()).unwrap(); - Ok(()) - })); - - Ok(()) - })) + }); + }) .unwrap(); signal_rx.recv().unwrap(); - pool.shutdown().wait().unwrap(); + pool.shutdown().wait(); } #[test] fn new_threadpool_is_idle() { let pool = ThreadPool::new(); - pool.shutdown_on_idle().wait().unwrap(); + pool.shutdown_on_idle().wait(); } #[test] fn busy_threadpool_is_not_idle() { - use futures::sync::oneshot; + use tokio_sync::oneshot; // let pool = ThreadPool::new(); let pool = Builder::new().pool_size(4).max_blocking(2).build(); @@ -347,26 +341,31 @@ fn busy_threadpool_is_not_idle() { let (term_tx, term_rx) = oneshot::channel(); - tx.spawn(term_rx.then(|_| Ok(()))).unwrap(); + tx.spawn(async move { + term_rx.await.unwrap(); + }) + .unwrap(); let mut idle = pool.shutdown_on_idle(); struct IdleFut<'a>(&'a mut Shutdown); impl<'a> Future for IdleFut<'a> { - type Item = (); - type Error = (); - fn poll(&mut self) -> Poll<(), ()> { - assert!(self.0.poll().unwrap().is_not_ready()); - Ok(Async::Ready(())) + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + assert_pending!(Pin::new(&mut self.as_mut().0).poll(cx)); + Poll::Ready(()) } } - IdleFut(&mut idle).wait().unwrap(); + let idle_fut = IdleFut(&mut idle); + tokio_executor::enter().unwrap().block_on(idle_fut); term_tx.send(()).unwrap(); - idle.wait().unwrap(); + let idle_fut = IdleFut(&mut idle); + tokio_executor::enter().unwrap().block_on(idle_fut); } #[test] @@ -377,10 +376,9 @@ fn panic_in_task() { struct Boom; impl Future for Boom { - type Item = (); - type Error = (); + type Output = (); - fn poll(&mut self) -> Poll<(), ()> { + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { panic!(); } } @@ -393,7 +391,7 @@ fn panic_in_task() { tx.spawn(Boom).unwrap(); - pool.shutdown_on_idle().wait().unwrap(); + pool.shutdown_on_idle().wait(); } #[test] @@ -407,15 +405,15 @@ fn count_panics() { }) .build(); // Spawn a future that will panic. - pool.spawn(lazy(|| -> Result<(), ()> { panic!() })); - pool.shutdown_on_idle().wait().unwrap(); + pool.spawn(async { panic!() }); + pool.shutdown_on_idle().wait(); let counter = counter.load(Relaxed); assert_eq!(counter, 1); } #[test] fn multi_threadpool() { - use futures::sync::oneshot; + use tokio_sync::oneshot; let pool1 = ThreadPool::new(); let pool2 = ThreadPool::new(); @@ -423,36 +421,22 @@ fn multi_threadpool() { let (tx, rx) = oneshot::channel(); let (done_tx, done_rx) = mpsc::channel(); - pool2.spawn({ - rx.and_then(move |_| { - done_tx.send(()).unwrap(); - Ok(()) - }) - .map_err(|e| panic!("err={:?}", e)) + pool2.spawn(async move { + rx.await.unwrap(); + done_tx.send(()).unwrap(); }); - pool1.spawn(lazy(move || { + pool1.spawn(async move { tx.send(()).unwrap(); - Ok(()) - })); + }); done_rx.recv().unwrap(); } #[test] fn eagerly_drops_futures() { - use futures::future::{empty, lazy, Future}; - use futures::task; use std::sync::mpsc; - struct NotifyOnDrop(mpsc::Sender<()>); - - impl Drop for NotifyOnDrop { - fn drop(&mut self) { - self.0.send(()).unwrap(); - } - } - struct MyPark { inner: DefaultPark, #[allow(dead_code)] @@ -497,9 +481,6 @@ fn eagerly_drops_futures() { let (park_tx, park_rx) = mpsc::sync_channel(0); let (unpark_tx, unpark_rx) = mpsc::sync_channel(0); - // Get the signal that the handler dropped. - let notify_on_drop = NotifyOnDrop(drop_tx); - let pool = tokio_threadpool::Builder::new() .custom_park(move |_| MyPark { inner: DefaultPark::new(), @@ -508,29 +489,33 @@ fn eagerly_drops_futures() { }) .build(); - pool.spawn(lazy(move || { - // Get a handle to the current task. - let task = task::current(); + struct MyTask { + task_tx: Option>, + drop_tx: mpsc::Sender<()>, + } - // Send it to the main thread to hold on to. - task_tx.send(task).unwrap(); + impl Future for MyTask { + type Output = (); - // This future will never resolve, it is only used to hold on to thee - // `notify_on_drop` handle. - empty::<(), ()>().then(move |_| { - // This code path should never be reached. - if true { - panic!() + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + if let Some(tx) = self.get_mut().task_tx.take() { + tx.send(cx.waker().clone()).unwrap(); } - // Explicitly drop `notify_on_drop` here, this is mostly to ensure - // that the `notify_on_drop` handle gets moved into the task. It - // will actually get dropped when the runtime is dropped. - drop(notify_on_drop); + Poll::Pending + } + } - Ok(()) - }) - })); + impl Drop for MyTask { + fn drop(&mut self) { + self.drop_tx.send(()).unwrap(); + } + } + + pool.spawn(MyTask { + task_tx: Some(task_tx), + drop_tx, + }); // Wait until we get the task handle. let task = task_rx.recv().unwrap();