tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-07 00:00:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

threadpool: update to std::future (#1219)

Browse Source 
An initial pass at updating `tokio-threadpool` to `std::future`. The
codebase and tests both now run using `std::future` but the wake
mechanism is not ideal. Follow up work will be required to improve on
this.

Refs: #1200
This commit is contained in:
Carl Lerche
2019-06-27 22:30:56 -07:00
committed by GitHub
parent e4415d986a
commit e7488d983e
17 changed files with 436 additions and 485 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -14,7 +14,7 @@ members = [
# "tokio-signal",
"tokio-sync",
"tokio-test",
# "tokio-threadpool",
"tokio-threadpool",
"tokio-timer",
"tokio-tcp",
# "tokio-tls",
tokio-threadpool/Cargo.toml
+6 -5
View File
@@ -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" }
tokio-threadpool/README.md
-37
View File
@@ -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).
tokio-threadpool/src/blocking.rs
+6 -5
View File
@@ -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, T>(f: F) -> Poll<T, BlockingError>
pub fn blocking<F, T>(f: F) -> Poll<Result<T, BlockingError>>
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 {
tokio-threadpool/src/lib.rs
+11 -2
View File
@@ -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};
tokio-threadpool/src/notifier.rs
-91
View File
@@ -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<Pool>,
}
/// A guard that ensures that the inner value gets forgotten.
#[derive(Debug)]
struct Forget<T>(Option<T>);
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<T>
// 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<T> Forget<T> {
fn new(t: T) -> Self {
Forget(Some(t))
}
}
impl<T> ops::Deref for Forget<T> {
type Target = T;
fn deref(&self) -> &T {
self.0.as_ref().unwrap()
}
}
impl<T> Drop for Forget<T> {
fn drop(&mut self) {
mem::forget(self.0.take());
}
}
tokio-threadpool/src/pool/mod.rs
+6 -2
View File
@@ -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<Task>) -> Poll<(), crate::BlockingError> {
pub fn poll_blocking_capacity(
&self,
task: &Arc<Task>,
) -> Poll<Result<(), crate::BlockingError>> {
self.blocking.poll_blocking_capacity(task)
}
tokio-threadpool/src/sender.rs
+11 -28
View File
@@ -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<F>(&self, future: F) -> Result<(), SpawnError>
where
F: Future<Item = (), Error = ()> + Send + 'static,
F: Future<Output = ()> + 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<dyn Future<Item = (), Error = ()> + Send>,
future: Pin<Box<dyn Future<Output = ()> + 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<dyn Future<Item = (), Error = ()> + Send>,
future: Pin<Box<dyn Future<Output = ()> + Send>>,
) -> Result<(), SpawnError> {
self.prepare_for_spawn()?;
@@ -175,34 +178,14 @@ impl<'a> tokio_executor::Executor for &'a Sender {
impl<T> tokio_executor::TypedExecutor<T> for Sender
where
T: Future<Item = (), Error = ()> + Send + 'static,
T: Future<Output = ()> + 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<T> future::Executor<T> for Sender
where
T: Future<Item = (), Error = ()> + Send + 'static,
{
fn execute(&self, future: T) -> Result<(), future::ExecuteError<T>> {
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))
}
}
tokio-threadpool/src/shutdown.rs
+20 -11
View File
@@ -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();
}
}
tokio-threadpool/src/task/blocking.rs
+8 -4
View File
@@ -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<Task>) -> Poll<(), crate::BlockingError> {
pub fn poll_blocking_capacity(
&self,
task: &Arc<Task>,
) -> Poll<Result<(), crate::BlockingError>> {
// 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(()))
}
}
}
tokio-threadpool/src/task/mod.rs
+51 -41
View File
@@ -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<Option<Spawn<BoxFuture>>>,
future: UnsafeCell<Option<BoxFuture>>,
}
#[derive(Debug)]
@@ -58,31 +60,27 @@ pub(crate) enum Run {
Complete,
}
type BoxFuture = Box<dyn Future<Item = (), Error = ()> + Send + 'static>;
type BoxFuture = Pin<Box<dyn Future<Output = ()> + 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<Notifier>) -> Run {
pub fn run(me: &Arc<Task>, pool: &Arc<Pool>) -> 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<Spawn<BoxFuture>>, bool);
struct Guard<'a>(&'a mut Option<BoxFuture>, 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<Task>, pool: &Arc<Pool>) {
if me.schedule() {
let _ = pool.submit(me, pool);
}
}
/// Notify the task it has been allocated blocking capacity
pub fn notify_blocking(me: Arc<Task>, pool: &Arc<Pool>) {
BlockingState::notify_blocking(&me.blocking, AcqRel);
Task::notify(me, pool);
Task::schedule(&me, pool);
}
pub fn schedule(me: &Arc<Self>, pool: &Arc<Pool>) {
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<BoxFuture>")
.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!();
}
}
tokio-threadpool/src/thread_pool.rs
+15 -4
View File
@@ -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<F>(&self, future: F)
where
F: Future<Item = (), Error = ()> + Send + 'static,
F: Future<Output = ()> + 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<T, E> Future for SpawnHandle<T, E> {
self.0.poll()
}
}
*/
tokio-threadpool/src/waker.rs
+24
View File
@@ -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<Pool>,
pub task: Arc<Task>,
}
unsafe impl Send for Waker {}
unsafe impl Sync for Waker {}
impl Wake for Waker {
fn wake_by_ref(me: &Arc<Self>) {
Task::schedule(&me.task, &me.pool);
}
}
tokio-threadpool/src/worker/mod.rs
+23 -27
View File
@@ -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<Result<(), crate::BlockingError>> {
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(&notify) {
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<Notifier>) -> bool {
if self.try_run_owned_task(notify) {
fn try_run_task(&self, pool: &Arc<Pool>) -> 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<Notifier>) -> bool {
fn try_run_owned_task(&self, pool: &Arc<Pool>) -> 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<Notifier>) -> bool {
fn try_steal_task(&self, pool: &Arc<Pool>) -> 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<Task>, notify: &Arc<Notifier>) {
fn run_task(&self, task: Arc<Task>, pool: &Arc<Pool>) {
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<Task>, notify: &Arc<Notifier>) -> task::Run {
fn run_task2(&self, task: &Arc<Task>, pool: &Arc<Pool>) -> 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
tokio-threadpool/tests/blocking.rs
+97 -84
View File
@@ -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));
}
}
}
tokio-threadpool/tests/hammer.rs
+44 -15
View File
@@ -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<T: Future> Future for Counted<T> {
type Item = T::Item;
type Error = T::Error;
type Output = T::Output;
fn poll(&mut self) -> Poll<T::Item, T::Error> {
self.inner.poll()
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T::Output> {
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::<oneshot::Sender<oneshot::Sender<()>>>();
let mut listen_tx = listen_tx.wait();
let (mut listen_tx, mut listen_rx) =
mpsc::unbounded_channel::<oneshot::Sender<oneshot::Sender<()>>>();
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));
}
}
tokio-threadpool/tests/threadpool.rs
+113 -128
View File
@@ -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<u32> = 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: Future + Send + 'static>(
f: F,
) -> Box<dyn Future<Item = (), Error = ()> + Send> {
Box::new(f.map(|_| ()).map_err(|_| ()))
}
thread_local!(static FOO: Cell<u32> = 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<AtomicUsize>);
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<AtomicUsize>);
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<mpsc::Sender<Waker>>,
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();