Use crossbeam_deque::Injector instead of crossbeam_queue::SegQueue

rayon-core/Cargo.toml

@@ -20,7 +20,6 @@ num_cpus = "1.2"
 lazy_static = "1"
 crossbeam-channel = "0.4.2"
 crossbeam-deque = "0.7.2"
-crossbeam-queue = "0.2"
 crossbeam-utils = "0.7"
 
 [dev-dependencies]

rayon-core/src/job.rs

@@ -1,6 +1,6 @@
 use crate::latch::Latch;
 use crate::unwind;
-use crossbeam_queue::SegQueue;
+use crossbeam_deque::{Injector, Steal};
 use std::any::Any;
 use std::cell::UnsafeCell;
 use std::mem;
@@ -184,13 +184,13 @@ impl<T> JobResult<T> {
 
 /// Indirect queue to provide FIFO job priority.
 pub(super) struct JobFifo {
-    inner: SegQueue<JobRef>,
+    inner: Injector<JobRef>,
 }
 
 impl JobFifo {
     pub(super) fn new() -> Self {
         JobFifo {
-            inner: SegQueue::new(),
+            inner: Injector::new(),
         }
     }
 
@@ -206,6 +206,12 @@ impl JobFifo {
 impl Job for JobFifo {
     unsafe fn execute(this: *const Self) {
         // We "execute" a queue by executing its first job, FIFO.
-        (*this).inner.pop().expect("job in fifo queue").execute()
+        loop {
+            match (*this).inner.steal() {
+                Steal::Success(job_ref) => break job_ref.execute(),
+                Steal::Empty => panic!("FIFO is empty"),
+                Steal::Retry => {}
+            }
+        }
     }
 }

rayon-core/src/registry.rs

@@ -8,8 +8,7 @@ use crate::util::leak;
 use crate::{
     ErrorKind, ExitHandler, PanicHandler, StartHandler, ThreadPoolBuildError, ThreadPoolBuilder,
 };
-use crossbeam_deque::{Steal, Stealer, Worker};
-use crossbeam_queue::SegQueue;
+use crossbeam_deque::{Injector, Steal, Stealer, Worker};
 use std::any::Any;
 use std::cell::Cell;
 use std::collections::hash_map::DefaultHasher;
@@ -136,7 +135,7 @@ pub(super) struct Registry {
     logger: Logger,
     thread_infos: Vec<ThreadInfo>,
     sleep: Sleep,
-    injected_jobs: SegQueue<JobRef>,
+    injected_jobs: Injector<JobRef>,
     panic_handler: Option<Box<PanicHandler>>,
     start_handler: Option<Box<StartHandler>>,
     exit_handler: Option<Box<ExitHandler>>,
@@ -240,7 +239,7 @@ impl Registry {
             logger: logger.clone(),
             thread_infos: stealers.into_iter().map(ThreadInfo::new).collect(),
             sleep: Sleep::new(logger, n_threads),
-            injected_jobs: SegQueue::new(),
+            injected_jobs: Injector::new(),
             terminate_count: AtomicUsize::new(1),
             panic_handler: builder.take_panic_handler(),
             start_handler: builder.take_start_handler(),
@@ -413,13 +412,18 @@ impl Registry {
     }
 
     fn pop_injected_job(&self, worker_index: usize) -> Option<JobRef> {
-        let job = self.injected_jobs.pop().ok();
-        if job.is_some() {
-            self.log(|| JobUninjected {
-                worker: worker_index,
-            });
+        loop {
+            match self.injected_jobs.steal() {
+                Steal::Success(job) => {
+                    self.log(|| JobUninjected {
+                        worker: worker_index,
+                    });
+                    return Some(job);
+                }
+                Steal::Empty => return None,
+                Steal::Retry => {}
+            }
         }
-        job
     }
 
     /// If already in a worker-thread of this registry, just execute `op`.