4c8f274db9
## Motivation When the thread pool shuts down, futures that have been polled at least once but not completed yet are simply leaked. We should drop them instead. ## Solution Multiple changes are introduced: * Tasks are assigned a home worker the first time they are polled. * Each worker contains a set of tasks (`Arc<Task>`) it is home to. When a task is assigned a home worker, it is registered in that worker's set of tasks. When the task is completed, it is unregistered from the set. * When the thread pool shuts down and after all worker threads stop, the remaining tasks in workers' sets are aborted, i.e. they are switched to the `Aborted` state and their `Future`s are dropped. * The thread pool shutdown process is refactored to make it more robust. We don't track the number of active threads manually anymore. Instead, there's `Arc<ShutdownTrigger>` that aborts remaining tasks and completes the `Shutdown` future once it gets destroyed (when all `Worker`s and `ThreadPool` get dropped because they're the only ones to contain strong references to the `ShutdownTrigger`). Closes #424 Closes #428
Tokio Thread Pool
A library for scheduling execution of futures concurrently across a pool of threads.
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
extern crate tokio_threadpool;
extern crate futures;
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.
Contribution
Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in Tokio by you, shall be licensed as MIT, without any additional terms or conditions.