Implement in_place_scope: a scope that runs its driver closure on the…

… same thread as the creator of the scope

rayon-core/src/latch.rs

@@ -298,11 +298,9 @@ impl CountLatch {
 
     /// Decrements the latch counter by one. If this is the final
     /// count, then the latch is **set**, and calls to `probe()` will
-    /// return true. Returns whether the latch was set. This is an
-    /// internal operation, as it does not tickle, and to fail to
-    /// tickle would lead to deadlock.
+    /// return true. Returns whether the latch was set.
     #[inline]
-    fn set(&self) -> bool {
+    pub(super) fn set(&self) -> bool {
         if self.counter.fetch_sub(1, Ordering::SeqCst) == 1 {
             self.core_latch.set();
             true
@@ -329,6 +327,41 @@ impl AsCoreLatch for CountLatch {
     }
 }
 
+#[derive(Debug)]
+pub(super) struct CountLockLatch {
+    lock_latch: LockLatch,
+    counter: AtomicUsize,
+}
+
+impl CountLockLatch {
+    #[inline]
+    pub(super) fn new() -> CountLockLatch {
+        CountLockLatch {
+            lock_latch: LockLatch::new(),
+            counter: AtomicUsize::new(1),
+        }
+    }
+
+    #[inline]
+    pub(super) fn increment(&self) {
+        let old_counter = self.counter.fetch_add(1, Ordering::Relaxed);
+        debug_assert!(old_counter != 0);
+    }
+
+    pub(super) fn wait(&self) {
+        self.lock_latch.wait();
+    }
+}
+
+impl Latch for CountLockLatch {
+    #[inline]
+    fn set(&self) {
+        if self.counter.fetch_sub(1, Ordering::SeqCst) == 1 {
+            self.lock_latch.set();
+        }
+    }
+}
+
 impl<'a, L> Latch for &'a L
 where
     L: Latch,

rayon-core/src/lib.rs

@@ -78,7 +78,7 @@ mod test;
 
 pub use self::join::{join, join_context};
 pub use self::registry::ThreadBuilder;
-pub use self::scope::{scope, Scope};
+pub use self::scope::{in_place_scope, scope, Scope};
 pub use self::scope::{scope_fifo, ScopeFifo};
 pub use self::spawn::{spawn, spawn_fifo};
 pub use self::thread_pool::current_thread_has_pending_tasks;

rayon-core/src/registry.rs

@@ -164,7 +164,7 @@ static THE_REGISTRY_SET: Once = Once::new();
 /// Starts the worker threads (if that has not already happened). If
 /// initialization has not already occurred, use the default
 /// configuration.
-fn global_registry() -> &'static Arc<Registry> {
+pub(super) fn global_registry() -> &'static Arc<Registry> {
     set_global_registry(|| Registry::new(ThreadPoolBuilder::new()))
         .or_else(|err| unsafe { THE_REGISTRY.as_ref().ok_or(err) })
         .expect("The global thread pool has not been initialized.")