diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 04721cfa4..d0bbfe33c 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -22,10 +22,9 @@ categories = ["concurrency", "asynchronous"] [dependencies] tokio-executor = { version = "0.1.2", path = "../tokio-executor" } futures = "0.1.19" -crossbeam = "0.6.0" -crossbeam-channel = "0.3.3" -crossbeam-deque = "0.6.1" -crossbeam-utils = "0.6.2" +crossbeam-deque = "0.7.0" +crossbeam-queue = "0.1.0" +crossbeam-utils = "0.6.4" num_cpus = "1.2" rand = "0.6" slab = "0.4.1" diff --git a/tokio-threadpool/src/builder.rs b/tokio-threadpool/src/builder.rs index 9c49135e0..82c7cac65 100644 --- a/tokio-threadpool/src/builder.rs +++ b/tokio-threadpool/src/builder.rs @@ -3,7 +3,6 @@ use config::{Config, MAX_WORKERS}; use park::{BoxPark, BoxedPark, DefaultPark}; use shutdown::ShutdownTrigger; use pool::{Pool, MAX_BACKUP}; -use task::Queue; use thread_pool::ThreadPool; use worker::{self, Worker, WorkerId}; @@ -13,6 +12,7 @@ use std::sync::Arc; use std::time::Duration; use std::cmp::max; +use crossbeam_deque::Injector; use num_cpus; use tokio_executor::Enter; use tokio_executor::park::Park; @@ -414,7 +414,7 @@ impl Builder { workers.into() }; - let queue = Arc::new(Queue::new()); + let queue = Arc::new(Injector::new()); // Create a trigger that will clean up resources on shutdown. // diff --git a/tokio-threadpool/src/lib.rs b/tokio-threadpool/src/lib.rs index 3340cca9f..45ee6626a 100644 --- a/tokio-threadpool/src/lib.rs +++ b/tokio-threadpool/src/lib.rs @@ -79,9 +79,8 @@ extern crate tokio_executor; -extern crate crossbeam; -extern crate crossbeam_channel; -extern crate crossbeam_deque as deque; +extern crate crossbeam_deque; +extern crate crossbeam_queue; extern crate crossbeam_utils; #[macro_use] extern crate futures; diff --git a/tokio-threadpool/src/pool/mod.rs b/tokio-threadpool/src/pool/mod.rs index 3d7a93773..2326fca04 100644 --- a/tokio-threadpool/src/pool/mod.rs +++ b/tokio-threadpool/src/pool/mod.rs @@ -15,7 +15,7 @@ use self::backup_stack::BackupStack; use config::Config; use shutdown::ShutdownTrigger; -use task::{Blocking, Queue, Task}; +use task::{Blocking, Task}; use worker::{self, Worker, WorkerId}; use futures::Poll; @@ -27,6 +27,7 @@ use std::sync::atomic::AtomicUsize; use std::sync::{Arc, Weak}; use std::thread; +use crossbeam_deque::Injector; use crossbeam_utils::CachePadded; use rand; @@ -57,7 +58,7 @@ pub(crate) struct Pool { // // Spawned tasks are pushed into this queue. Although worker threads have their own dedicated // task queues, they periodically steal tasks from this global queue, too. - pub queue: Arc, + pub queue: Arc>>, // Completes the shutdown process when the `ThreadPool` and all `Worker`s get dropped. // @@ -90,7 +91,7 @@ impl Pool { trigger: Weak, max_blocking: usize, config: Config, - queue: Arc, + queue: Arc>>, ) -> Pool { let pool_size = workers.len(); let total_size = max_blocking + pool_size; diff --git a/tokio-threadpool/src/shutdown.rs b/tokio-threadpool/src/shutdown.rs index 880fc8220..290cb182c 100644 --- a/tokio-threadpool/src/shutdown.rs +++ b/tokio-threadpool/src/shutdown.rs @@ -1,6 +1,7 @@ -use task::Queue; +use task::Task; use worker; +use crossbeam_deque::Injector; use futures::{Future, Poll, Async}; use futures::task::AtomicTask; @@ -62,14 +63,17 @@ impl Future for Shutdown { pub(crate) struct ShutdownTrigger { inner: Arc>, workers: Arc<[worker::Entry]>, - queue: Arc, + queue: Arc>>, } unsafe impl Send for ShutdownTrigger {} unsafe impl Sync for ShutdownTrigger {} impl ShutdownTrigger { - pub(crate) fn new(workers: Arc<[worker::Entry]>, queue: Arc) -> ShutdownTrigger { + pub(crate) fn new( + workers: Arc<[worker::Entry]>, + queue: Arc>>, + ) -> ShutdownTrigger { ShutdownTrigger { inner: Arc::new(Mutex::new(Inner { task: AtomicTask::new(), @@ -84,7 +88,7 @@ impl ShutdownTrigger { impl Drop for ShutdownTrigger { fn drop(&mut self) { // Drain the global task queue. - while self.queue.pop().is_some() {} + while !self.queue.steal().is_empty() {} // Drop the remaining incomplete tasks and parkers assosicated with workers. for worker in self.workers.iter() { diff --git a/tokio-threadpool/src/task/mod.rs b/tokio-threadpool/src/task/mod.rs index 90592f203..fe535b30d 100644 --- a/tokio-threadpool/src/task/mod.rs +++ b/tokio-threadpool/src/task/mod.rs @@ -1,10 +1,8 @@ mod blocking; mod blocking_state; -mod queue; mod state; pub(crate) use self::blocking::{Blocking, CanBlock}; -pub(crate) use self::queue::Queue; use self::blocking_state::BlockingState; use self::state::State; diff --git a/tokio-threadpool/src/task/queue.rs b/tokio-threadpool/src/task/queue.rs deleted file mode 100644 index 166438e78..000000000 --- a/tokio-threadpool/src/task/queue.rs +++ /dev/null @@ -1,34 +0,0 @@ -use task::Task; - -use std::sync::Arc; - -use crossbeam_channel::{unbounded, Receiver, Sender}; - -#[derive(Debug)] -pub(crate) struct Queue { - // TODO(stjepang): Use a custom, faster MPMC queue implementation that supports `steal_many()`. - chan: (Sender>, Receiver>), -} - -// ===== impl Queue ===== - -impl Queue { - /// Create a new, empty, `Queue`. - pub fn new() -> Queue { - Queue { - chan: unbounded(), - } - } - - /// Push a task onto the queue. - #[inline] - pub fn push(&self, task: Arc) { - self.chan.0.send(task).unwrap(); - } - - /// Pop a task from the queue. - #[inline] - pub fn pop(&self) -> Option> { - self.chan.1.try_recv().ok() - } -} diff --git a/tokio-threadpool/src/worker/entry.rs b/tokio-threadpool/src/worker/entry.rs index 07bee5bd2..e3a363276 100644 --- a/tokio-threadpool/src/worker/entry.rs +++ b/tokio-threadpool/src/worker/entry.rs @@ -9,9 +9,9 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::atomic::Ordering::{Acquire, AcqRel, Relaxed, Release}; use std::time::Duration; -use crossbeam::queue::SegQueue; +use crossbeam_deque::{Steal, Stealer, Worker}; +use crossbeam_queue::SegQueue; use crossbeam_utils::CachePadded; -use deque; use slab::Slab; // TODO: None of the fields should be public @@ -29,10 +29,10 @@ pub(crate) struct WorkerEntry { next_sleeper: UnsafeCell, // Worker half of deque - worker: deque::Worker>, + pub worker: Worker>, // Stealer half of deque - stealer: deque::Stealer>, + stealer: Stealer>, // Thread parker park: UnsafeCell>, @@ -53,7 +53,8 @@ pub(crate) struct WorkerEntry { impl WorkerEntry { pub fn new(park: BoxPark, unpark: BoxUnpark) -> Self { - let (w, s) = deque::fifo(); + let w = Worker::new_fifo(); + let s = w.stealer(); WorkerEntry { state: CachePadded::new(AtomicUsize::new(State::default().into())), @@ -187,7 +188,7 @@ impl WorkerEntry { /// This **must** only be called by the thread that owns the worker entry. /// This function is not `Sync`. #[inline] - pub fn pop_task(&self) -> deque::Pop> { + pub fn pop_task(&self) -> Option> { self.worker.pop() } @@ -199,23 +200,15 @@ impl WorkerEntry { /// At the same time, this method steals some additional tasks and moves /// them into `dest` in order to balance the work distribution among /// workers. - pub fn steal_tasks(&self, dest: &Self) -> deque::Steal> { - self.stealer.steal_many(&dest.worker) + pub fn steal_tasks(&self, dest: &Self) -> Steal> { + self.stealer.steal_batch_and_pop(&dest.worker) } /// Drain (and drop) all tasks that are queued for work. /// /// This is called when the pool is shutting down. pub fn drain_tasks(&self) { - use deque::Pop::*; - - loop { - match self.worker.pop() { - Data(_) => {} - Empty => break, - Retry => {} - } - } + while self.worker.pop().is_some() {} } /// Parks the worker thread. @@ -284,7 +277,6 @@ impl WorkerEntry { } running_tasks.clear(); - // Drop the parker. unsafe { *self.park.get() = None; *self.unpark.get() = None; @@ -297,7 +289,7 @@ impl WorkerEntry { if self.needs_drain.compare_and_swap(true, false, Acquire) { let running_tasks = unsafe { &mut *self.running_tasks.get() }; - while let Some(task) = self.remotely_completed_tasks.try_pop() { + while let Ok(task) = self.remotely_completed_tasks.pop() { running_tasks.remove(task.reg_index.get()); } } diff --git a/tokio-threadpool/src/worker/mod.rs b/tokio-threadpool/src/worker/mod.rs index e270c10d9..939d0063f 100644 --- a/tokio-threadpool/src/worker/mod.rs +++ b/tokio-threadpool/src/worker/mod.rs @@ -386,16 +386,13 @@ impl Worker { /// /// Returns `true` if work was found. fn try_run_owned_task(&self, notify: &Arc) -> bool { - use deque::Pop; - // Poll the internal queue for a task to run match self.entry().pop_task() { - Pop::Data(task) => { + Some(task) => { self.run_task(task, notify); true } - Pop::Empty => false, - Pop::Retry => true, + None => false, } } @@ -403,7 +400,7 @@ impl Worker { /// /// Returns `true` if work was found fn try_steal_task(&self, notify: &Arc) -> bool { - use deque::Steal; + use crossbeam_deque::Steal; debug_assert!(!self.is_blocking.get()); @@ -415,7 +412,7 @@ impl Worker { loop { if idx < len { match self.pool.workers[idx].steal_tasks(self.entry()) { - Steal::Data(task) => { + Steal::Success(task) => { trace!("stole task from another worker"); self.run_task(task, notify); @@ -701,15 +698,17 @@ impl Worker { /// /// Returns `true` if this worker has tasks in its queue. fn sleep_light(&self) { - const STEAL_COUNT: usize = 32; - self.entry().park_timeout(Duration::from_millis(0)); - for _ in 0..STEAL_COUNT { - if let Some(task) = self.pool.queue.pop() { - self.pool.submit(task, &self.pool); - } else { - break; + use crossbeam_deque::Steal; + loop { + match self.pool.queue.steal_batch(&self.entry().worker) { + Steal::Success(()) => { + self.pool.signal_work(&self.pool); + break; + } + Steal::Empty => break, + Steal::Retry => {} } } }