From 19da6ff59aa9aeec8920f1a3b3c33d58e5222689 Mon Sep 17 00:00:00 2001 From: Stjepan Glavina Date: Wed, 11 Jul 2018 21:24:10 +0200 Subject: [PATCH] New version of crossbeam-deque (#468) --- ci/tsan | 15 +++++------ tokio-threadpool/Cargo.toml | 2 +- tokio-threadpool/src/worker/entry.rs | 27 ++++++++++---------- tokio-threadpool/src/worker/mod.rs | 38 ++++++++++------------------ 4 files changed, 35 insertions(+), 47 deletions(-) diff --git a/ci/tsan b/ci/tsan index 22a358abf..6fda523cc 100644 --- a/ci/tsan +++ b/ci/tsan @@ -12,15 +12,14 @@ race:std*mpsc_queue # Probably more fences in std. race:__call_tls_dtors -# The crossbeam deque uses fences. -race:crossbeam_deque +# The epoch-based GC uses fences. +race:crossbeam_epoch -# This is excluded as this race shows up due to using the stealing features of -# the deque. Unfortunately, the implementation uses a fence, which makes tsan -# unhappy. -# -# TODO: It would be nice to not have to filter this out. -race:try_steal_task +# Push and steal operations in crossbeam-deque may cause data races, but such +# data races are safe. If a data race happens, the value read by `steal` is +# forgotten and the steal operation is then retried. +race:crossbeam_deque*push +race:crossbeam_deque*steal # This filters out expected data race in the treiber stack implementations. # Treiber stacks are inherently racy. The pop operation will attempt to access diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 04c571e96..5b75dfbc0 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -19,7 +19,7 @@ categories = ["concurrency", "asynchronous"] [dependencies] tokio-executor = { version = "0.1.2", path = "../tokio-executor" } futures = "0.1.19" -crossbeam-deque = "0.3" +crossbeam-deque = "0.5.0" num_cpus = "1.2" rand = "0.5" log = "0.4" diff --git a/tokio-threadpool/src/worker/entry.rs b/tokio-threadpool/src/worker/entry.rs index 752ad2955..5c981d45d 100644 --- a/tokio-threadpool/src/worker/entry.rs +++ b/tokio-threadpool/src/worker/entry.rs @@ -25,10 +25,10 @@ pub(crate) struct WorkerEntry { next_sleeper: UnsafeCell, // Worker half of deque - deque: deque::Deque>, + worker: deque::Worker>, // Stealer half of deque - steal: deque::Stealer>, + stealer: deque::Stealer>, // Thread parker pub park: UnsafeCell, @@ -42,14 +42,13 @@ pub(crate) struct WorkerEntry { impl WorkerEntry { pub fn new(park: BoxPark, unpark: BoxUnpark) -> Self { - let w = deque::Deque::new(); - let s = w.stealer(); + let (w, s) = deque::fifo(); WorkerEntry { state: AtomicUsize::new(State::default().into()), next_sleeper: UnsafeCell::new(0), - deque: w, - steal: s, + worker: w, + stealer: s, inbound: Queue::new(), park: UnsafeCell::new(park), unpark, @@ -188,23 +187,23 @@ impl WorkerEntry { /// /// This **must** only be called by the thread that owns the worker entry. /// This function is not `Sync`. - pub fn pop_task(&self) -> deque::Steal> { - self.deque.steal() + pub fn pop_task(&self) -> Option> { + self.worker.pop() } /// Steal a task /// /// This is called by *other* workers to steal a task for processing. This /// function is `Sync`. - pub fn steal_task(&self) -> deque::Steal> { - self.steal.steal() + pub fn steal_task(&self) -> Option> { + self.stealer.steal() } /// Drain (and drop) all tasks that are queued for work. /// /// This is called when the pool is shutting down. pub fn drain_tasks(&self) { - while let Some(_) = self.deque.pop() { + while let Some(_) = self.worker.pop() { } } @@ -215,7 +214,7 @@ impl WorkerEntry { #[inline] pub fn push_internal(&self, task: Arc) { - self.deque.push(task); + self.worker.push(task); } #[inline] @@ -239,8 +238,8 @@ impl fmt::Debug for WorkerEntry { fmt.debug_struct("WorkerEntry") .field("state", &self.state.load(Relaxed)) .field("next_sleeper", &"UnsafeCell") - .field("deque", &self.deque) - .field("steal", &self.steal) + .field("worker", &self.worker) + .field("stealer", &self.stealer) .field("park", &"UnsafeCell") .field("unpark", &"BoxUnpark") .field("inbound", &self.inbound) diff --git a/tokio-threadpool/src/worker/mod.rs b/tokio-threadpool/src/worker/mod.rs index 35feae6b7..0dcebab36 100644 --- a/tokio-threadpool/src/worker/mod.rs +++ b/tokio-threadpool/src/worker/mod.rs @@ -377,16 +377,13 @@ impl Worker { /// /// Returns `true` if work was found. fn try_run_owned_task(&self, notify: &Arc, sender: &mut Sender) -> bool { - use deque::Steal::*; - // Poll the internal queue for a task to run match self.entry().pop_task() { - Data(task) => { + Some(task) => { self.run_task(task, notify, sender); true } - Empty => false, - Retry => true, + None => false, } } @@ -394,36 +391,29 @@ impl Worker { /// /// Returns `true` if work was found fn try_steal_task(&self, notify: &Arc, sender: &mut Sender) -> bool { - use deque::Steal::*; - debug_assert!(!self.is_blocking.get()); let len = self.inner.workers.len(); let mut idx = self.inner.rand_usize() % len; - let mut found_work = false; let start = idx; loop { if idx < len { - match self.inner.workers[idx].steal_task() { - Data(task) => { - trace!("stole task"); + if let Some(task) = self.inner.workers[idx].steal_task() { + trace!("stole task"); - self.run_task(task, notify, sender); + self.run_task(task, notify, sender); - trace!("try_steal_task -- signal_work; self={}; from={}", - self.id.0, idx); + trace!("try_steal_task -- signal_work; self={}; from={}", + self.id.0, idx); - // Signal other workers that work is available - // - // TODO: Should this be called here or before - // `run_task`? - self.inner.signal_work(&self.inner); + // Signal other workers that work is available + // + // TODO: Should this be called here or before + // `run_task`? + self.inner.signal_work(&self.inner); - return true; - } - Empty => {} - Retry => found_work = true, + return true; } idx += 1; @@ -436,7 +426,7 @@ impl Worker { } } - found_work + false } fn run_task(&self, task: Arc, notify: &Arc, sender: &mut Sender) {