Add scope_fifo

Author: cuviper

rayon-core/Cargo.toml

@@ -23,6 +23,10 @@ lazy_static = "1"
 [dependencies.crossbeam-deque]
 version = "0.2.0"
 
+# Also held back for rustc compatibility
+[dependencies.crossbeam]
+version = "0.3.0"
+
 [dev-dependencies]
 rand = "0.5"
 

rayon-core/src/job.rs

@@ -1,3 +1,4 @@
+use crossbeam::sync::SegQueue;
 use latch::Latch;
 use std::any::Any;
 use std::cell::UnsafeCell;
@@ -179,3 +180,35 @@ impl<T> JobResult<T> {
         }
     }
 }
+
+/// Indirect queue to provide FIFO job priority.
+pub struct JobFifo {
+    inner: SegQueue<JobRef>,
+}
+
+impl JobFifo {
+    pub fn new() -> Self {
+        JobFifo {
+            inner: SegQueue::new(),
+        }
+    }
+
+    pub unsafe fn push(&self, job_ref: JobRef) -> JobRef {
+        // A little indirection ensures that spawns are always prioritized in FIFO order.  The
+        // jobs in a thread's deque may be popped from the back (LIFO) or stolen from the front
+        // (FIFO), but either way they will end up popping from the front of this queue.
+        self.inner.push(job_ref);
+        JobRef::new(self)
+    }
+}
+
+impl Job for JobFifo {
+    unsafe fn execute(this: *const Self) {
+        // We "execute" a queue by executing its first job, FIFO.
+        (*this)
+            .inner
+            .try_pop()
+            .expect("job in fifo queue")
+            .execute()
+    }
+}

rayon-core/src/lib.rs

@@ -31,6 +31,7 @@ use std::io;
 use std::marker::PhantomData;
 use std::str::FromStr;
 
+extern crate crossbeam;
 extern crate crossbeam_deque;
 #[macro_use]
 extern crate lazy_static;
@@ -61,6 +62,7 @@ mod test;
 pub mod internal;
 pub use join::{join, join_context};
 pub use scope::{scope, Scope};
+pub use scope::{scope_fifo, ScopeFifo};
 pub use spawn::spawn;
 pub use thread_pool::current_thread_has_pending_tasks;
 pub use thread_pool::current_thread_index;

rayon-core/src/registry.rs

@@ -1,3 +1,4 @@
+use crossbeam::sync::SegQueue;
 use crossbeam_deque::{Deque, Steal, Stealer};
 #[cfg(rayon_unstable)]
 use internal::task::Task;
@@ -13,7 +14,7 @@ use std::collections::hash_map::DefaultHasher;
 use std::hash::Hasher;
 use std::mem;
 use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
-use std::sync::{Arc, Mutex, Once, ONCE_INIT};
+use std::sync::{Arc, Once, ONCE_INIT};
 use std::thread;
 use std::usize;
 use unwind;
@@ -22,9 +23,8 @@ use {ErrorKind, ExitHandler, PanicHandler, StartHandler, ThreadPoolBuildError, T
 
 pub struct Registry {
     thread_infos: Vec<ThreadInfo>,
-    state: Mutex<RegistryState>,
     sleep: Sleep,
-    job_uninjector: Stealer<JobRef>,
+    injected_jobs: SegQueue<JobRef>,
     panic_handler: Option<Box<PanicHandler>>,
     start_handler: Option<Box<StartHandler>>,
     exit_handler: Option<Box<ExitHandler>>,
@@ -45,10 +45,6 @@ pub struct Registry {
     terminate_latch: CountLatch,
 }
 
-struct RegistryState {
-    job_injector: Deque<JobRef>,
-}
-
 /// ////////////////////////////////////////////////////////////////////////
 /// Initialization
 
@@ -104,16 +100,13 @@ impl Registry {
         let n_threads = builder.get_num_threads();
         let breadth_first = builder.get_breadth_first();
 
-        let inj_worker = Deque::new();
-        let inj_stealer = inj_worker.stealer();
         let workers: Vec<_> = (0..n_threads).map(|_| Deque::new()).collect();
         let stealers: Vec<_> = workers.iter().map(|d| d.stealer()).collect();
 
         let registry = Arc::new(Registry {
             thread_infos: stealers.into_iter().map(|s| ThreadInfo::new(s)).collect(),
-            state: Mutex::new(RegistryState::new(inj_worker)),
             sleep: Sleep::new(),
-            job_uninjector: inj_stealer,
+            injected_jobs: SegQueue::new(),
             terminate_latch: CountLatch::new(),
             panic_handler: builder.take_panic_handler(),
             start_handler: builder.take_start_handler(),
@@ -175,6 +168,18 @@ impl Registry {
         }
     }
 
+    /// Returns the current `WorkerThread` if it's part of this `Registry`.
+    pub fn current_thread(&self) -> Option<&WorkerThread> {
+        unsafe {
+            if let Some(worker) = WorkerThread::current().as_ref() {
+                if worker.registry().id() == self.id() {
+                    return Some(worker);
+                }
+            }
+            None
+        }
+    }
+
     /// Returns an opaque identifier for this registry.
     pub fn id(&self) -> RegistryId {
         // We can rely on `self` not to change since we only ever create
@@ -297,39 +302,31 @@ impl Registry {
         log!(InjectJobs {
             count: injected_jobs.len()
         });
-        {
-            let state = self.state.lock().unwrap();
-
-            // It should not be possible for `state.terminate` to be true
-            // here. It is only set to true when the user creates (and
-            // drops) a `ThreadPool`; and, in that case, they cannot be
-            // calling `inject()` later, since they dropped their
-            // `ThreadPool`.
-            assert!(
-                !self.terminate_latch.probe(),
-                "inject() sees state.terminate as true"
-            );
-
-            for &job_ref in injected_jobs {
-                state.job_injector.push(job_ref);
-            }
+
+        // It should not be possible for `state.terminate` to be true
+        // here. It is only set to true when the user creates (and
+        // drops) a `ThreadPool`; and, in that case, they cannot be
+        // calling `inject()` later, since they dropped their
+        // `ThreadPool`.
+        assert!(
+            !self.terminate_latch.probe(),
+            "inject() sees state.terminate as true"
+        );
+
+        for &job_ref in injected_jobs {
+            self.injected_jobs.push(job_ref);
         }
         self.sleep.tickle(usize::MAX);
     }
 
     fn pop_injected_job(&self, worker_index: usize) -> Option<JobRef> {
-        loop {
-            match self.job_uninjector.steal() {
-                Steal::Empty => return None,
-                Steal::Data(d) => {
-                    log!(UninjectedWork {
-                        worker: worker_index
-                    });
-                    return Some(d);
-                }
-                Steal::Retry => {}
-            }
+        let job = self.injected_jobs.try_pop();
+        if job.is_some() {
+            log!(UninjectedWork {
+                worker: worker_index
+            });
         }
+        job
     }
 
     /// If already in a worker-thread of this registry, just execute `op`.
@@ -439,14 +436,6 @@ pub struct RegistryId {
     addr: usize,
 }
 
-impl RegistryState {
-    pub fn new(job_injector: Deque<JobRef>) -> RegistryState {
-        RegistryState {
-            job_injector: job_injector,
-        }
-    }
-}
-
 struct ThreadInfo {
     /// Latch set once thread has started and we are entering into the
     /// main loop. Used to wait for worker threads to become primed,