Port to crossbeam-deque 0.6

Author: cuviper

Cargo.toml

@@ -20,7 +20,7 @@ exclude = ["ci"]
 
 [dependencies]
 rayon-core = { version = "1.5.0", path = "rayon-core" }
-crossbeam-deque = "0.2.0"
+crossbeam-deque = "0.6.3"
 
 # This is a public dependency!
 [dependencies.either]

rayon-core/Cargo.toml

@@ -17,11 +17,7 @@ categories = ["concurrency"]
 num_cpus = "1.2"
 libc = "0.2.16"
 lazy_static = "1"
-
-# This is deliberately not the latest version, because we want
-# to support older rustc than crossbeam-deque 0.3+ does.
-[dependencies.crossbeam-deque]
-version = "0.2.0"
+crossbeam-deque = "0.6.3"
 
 # Also held back for rustc compatibility
 [dependencies.crossbeam]

rayon-core/src/registry.rs

@@ -1,5 +1,5 @@
 use crossbeam::sync::SegQueue;
-use crossbeam_deque::{Deque, Steal, Stealer};
+use crossbeam_deque::{self as deque, Pop, Steal, Stealer, Worker};
 #[cfg(rayon_unstable)]
 use internal::task::Task;
 #[cfg(rayon_unstable)]
@@ -102,8 +102,15 @@ impl Registry {
         let n_threads = builder.get_num_threads();
         let breadth_first = builder.get_breadth_first();
 
-        let workers: Vec<_> = (0..n_threads).map(|_| Deque::new()).collect();
-        let stealers: Vec<_> = workers.iter().map(|d| d.stealer()).collect();
+        let (workers, stealers): (Vec<_>, Vec<_>) = (0..n_threads)
+            .map(|_| {
+                if breadth_first {
+                    deque::fifo()
+                } else {
+                    deque::lifo()
+                }
+            })
+            .unzip();
 
         let registry = Arc::new(Registry {
             thread_infos: stealers.into_iter().map(|s| ThreadInfo::new(s)).collect(),
@@ -127,9 +134,7 @@ impl Registry {
             if let Some(stack_size) = builder.get_stack_size() {
                 b = b.stack_size(stack_size);
             }
-            if let Err(e) =
-                b.spawn(move || unsafe { main_loop(worker, registry, index, breadth_first) })
-            {
+            if let Err(e) = b.spawn(move || unsafe { main_loop(worker, registry, index) }) {
                 return Err(ThreadPoolBuildError::new(ErrorKind::IOError(e)));
             }
         }
@@ -467,16 +472,13 @@ impl ThreadInfo {
 
 pub struct WorkerThread {
     /// the "worker" half of our local deque
-    worker: Deque<JobRef>,
+    worker: Worker<JobRef>,
 
     /// local queue used for `spawn_fifo` indirection
     fifo: JobFifo,
 
     index: usize,
 
-    /// are these workers configured to steal breadth-first or not?
-    breadth_first: bool,
-
     /// A weak random number generator.
     rng: XorShift64Star,
 
@@ -535,7 +537,7 @@ impl WorkerThread {
 
     #[inline]
     pub fn local_deque_is_empty(&self) -> bool {
-        self.worker.len() == 0
+        self.worker.is_empty()
     }
 
     /// Attempts to obtain a "local" job -- typically this means
@@ -544,15 +546,11 @@ impl WorkerThread {
     /// bottom.
     #[inline]
     pub unsafe fn take_local_job(&self) -> Option<JobRef> {
-        if !self.breadth_first {
-            self.worker.pop()
-        } else {
-            loop {
-                match self.worker.steal() {
-                    Steal::Empty => return None,
-                    Steal::Data(d) => return Some(d),
-                    Steal::Retry => {}
-                }
+        loop {
+            match self.worker.pop() {
+                Pop::Empty => return None,
+                Pop::Data(d) => return Some(d),
+                Pop::Retry => {}
             }
         }
     }
@@ -620,7 +618,7 @@ impl WorkerThread {
     /// local work to do.
     unsafe fn steal(&self) -> Option<JobRef> {
         // we only steal when we don't have any work to do locally
-        debug_assert!(self.worker.pop().is_none());
+        debug_assert!(self.local_deque_is_empty());
 
         // otherwise, try to steal
         let num_threads = self.registry.thread_infos.len();
@@ -654,16 +652,10 @@ impl WorkerThread {
 
 /// ////////////////////////////////////////////////////////////////////////
 
-unsafe fn main_loop(
-    worker: Deque<JobRef>,
-    registry: Arc<Registry>,
-    index: usize,
-    breadth_first: bool,
-) {
+unsafe fn main_loop(worker: Worker<JobRef>, registry: Arc<Registry>, index: usize) {
     let worker_thread = WorkerThread {
         worker: worker,
         fifo: JobFifo::new(),
-        breadth_first: breadth_first,
         index: index,
         rng: XorShift64Star::new(),
         registry: registry.clone(),

src/iter/par_bridge.rs

@@ -1,4 +1,4 @@
-use crossbeam_deque::{Deque, Steal, Stealer};
+use crossbeam_deque::{self as deque, Steal, Stealer, Worker};
 
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Mutex, TryLockError};
@@ -79,10 +79,9 @@ where
         C: UnindexedConsumer<Self::Item>,
     {
         let split_count = AtomicUsize::new(current_num_threads());
-        let deque = Deque::new();
-        let stealer = deque.stealer();
+        let (worker, stealer) = deque::fifo();
         let done = AtomicBool::new(false);
-        let iter = Mutex::new((self.iter, deque));
+        let iter = Mutex::new((self.iter, worker));
 
         bridge_unindexed(
             IterParallelProducer {
@@ -99,7 +98,7 @@ where
 struct IterParallelProducer<'a, Iter: Iterator + 'a> {
     split_count: &'a AtomicUsize,
     done: &'a AtomicBool,
-    iter: &'a Mutex<(Iter, Deque<Iter::Item>)>,
+    iter: &'a Mutex<(Iter, Worker<Iter::Item>)>,
     items: Stealer<Iter::Item>,
 }
 
@@ -167,11 +166,15 @@ where
                                 let count = current_num_threads();
                                 let count = (count * count) * 2;
 
-                                let (ref mut iter, ref deque) = *guard;
+                                let (ref mut iter, ref worker) = *guard;
 
-                                while deque.len() < count {
+                                // while worker.len() < count {
+                                // FIXME the new deque doesn't let us count items.  We can just
+                                // push a number of items, but that doesn't consider active
+                                // stealers elsewhere.
+                                for _ in 0..count {
                                     if let Some(it) = iter.next() {
-                                        deque.push(it);
+                                        worker.push(it);
                                     } else {
                                         self.done.store(true, Ordering::SeqCst);
                                         break;