removed lots of inline hints because they probably SUCK + benchmark

distaff/Cargo.toml

@@ -3,6 +3,12 @@ name = "distaff"
 version = "0.1.0"
 edition = "2024"
 
+[profile.bench]
+debug = true
+
+[profile.release]
+debug = true
+
 [features]
 default = ["metrics"]
 tracing = ["dep:tracing"]
@@ -23,3 +29,11 @@ async-task = "4.7.1"
 tracing-test = {version = "0.2"}
 tracing-tracy = {version = "0.11"}
 futures = "0.3"
+
+divan = "0.1.14"
+rayon = "1.10.0"
+chili = {path = "../../chili"}
+
+[[bench]]
+name = "overhead"
+harness = false

distaff/benches/overhead.rs

@@ -0,0 +1,119 @@
+use distaff::{Scope, ThreadPool};
+use divan::Bencher;
+
+struct Node {
+    val: u64,
+    left: Option<Box<Node>>,
+    right: Option<Box<Node>>,
+}
+
+impl Node {
+    pub fn tree(layers: usize) -> Self {
+        Self {
+            val: 1,
+            left: (layers != 1).then(|| Box::new(Self::tree(layers - 1))),
+            right: (layers != 1).then(|| Box::new(Self::tree(layers - 1))),
+        }
+    }
+}
+
+const LAYERS: &[usize] = &[10, 24];
+fn nodes() -> impl Iterator<Item = (usize, usize)> {
+    LAYERS.iter().map(|&l| (l, (1 << l) - 1))
+}
+
+#[divan::bench(args = nodes())]
+fn no_overhead(bencher: Bencher, nodes: (usize, usize)) {
+    fn join_no_overhead<A, B, RA, RB>(scope: &Scope<'_, '_>, a: A, b: B) -> (RA, RB)
+    where
+        A: FnOnce(&Scope<'_, '_>) -> RA + Send,
+        B: FnOnce(&Scope<'_, '_>) -> RB + Send,
+        RA: Send,
+        RB: Send,
+    {
+        (a(scope), b(scope))
+    }
+
+    #[inline]
+    fn sum(node: &Node, scope: &Scope<'_, '_>) -> u64 {
+        let (left, right) = join_no_overhead(
+            scope,
+            |s| node.left.as_deref().map(|n| sum(n, s)).unwrap_or_default(),
+            |s| node.right.as_deref().map(|n| sum(n, s)).unwrap_or_default(),
+        );
+
+        node.val + left + right
+    }
+
+    let tree = Node::tree(nodes.0);
+    let pool = ThreadPool::global();
+
+    bencher.bench_local(move || {
+        pool.scope(|scope| {
+            assert_eq!(sum(&tree, scope), nodes.1 as u64);
+        });
+    });
+}
+
+#[divan::bench(args = nodes())]
+fn distaff_overhead(bencher: Bencher, nodes: (usize, usize)) {
+    fn sum<'scope, 'env>(node: &Node, scope: &'scope Scope<'scope, 'env>) -> u64 {
+        let (left, right) = scope.join(
+            |s| node.left.as_deref().map(|n| sum(n, s)).unwrap_or_default(),
+            |s| node.right.as_deref().map(|n| sum(n, s)).unwrap_or_default(),
+        );
+
+        node.val + left + right
+    }
+
+    let tree = Node::tree(nodes.0);
+    let pool = ThreadPool::global();
+
+    bencher.bench_local(move || {
+        pool.scope(|scope| {
+            assert_eq!(sum(&tree, scope), nodes.1 as u64);
+        });
+    });
+}
+
+#[divan::bench(args = nodes())]
+fn rayon_overhead(bencher: Bencher, nodes: (usize, usize)) {
+    fn sum(node: &Node) -> u64 {
+        let (left, right) = rayon::join(
+            || node.left.as_deref().map(sum).unwrap_or_default(),
+            || node.right.as_deref().map(sum).unwrap_or_default(),
+        );
+
+        node.val + left + right
+    }
+
+    let tree = Node::tree(nodes.0);
+
+    bencher.bench_local(move || {
+        assert_eq!(sum(&tree), nodes.1 as u64);
+    });
+}
+
+#[divan::bench(args = nodes())]
+fn chili_overhead(bencher: Bencher, nodes: (usize, usize)) {
+    use chili::Scope;
+    fn sum(node: &Node, scope: &mut Scope<'_>) -> u64 {
+        let (left, right) = scope.join(
+            |s| node.left.as_deref().map(|n| sum(n, s)).unwrap_or_default(),
+            |s| node.right.as_deref().map(|n| sum(n, s)).unwrap_or_default(),
+        );
+
+        node.val + left + right
+    }
+
+    let tree = Node::tree(nodes.0);
+    let mut scope = Scope::global();
+
+    bencher.bench_local(move || {
+        assert_eq!(sum(&tree, &mut scope), nodes.1 as u64);
+    });
+}
+
+fn main() {
+    divan::main();
+}

distaff/src/context.rs

@@ -53,7 +53,6 @@ impl Shared {
 }
 
 impl Context {
-    #[inline]
     pub(crate) fn shared(&self) -> parking_lot::MutexGuard<'_, Shared> {
         self.shared.lock()
     }

distaff/src/join.rs

@@ -11,7 +11,6 @@ use crate::{
 };
 
 impl WorkerThread {
-    #[inline]
     #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
     fn join_seq<A, B, RA, RB>(&self, a: A, b: B) -> (RA, RB)
     where
@@ -24,7 +23,6 @@ impl WorkerThread {
         (ra, rb)
     }
 
-    #[inline]
     pub(crate) fn join_heartbeat_every<A, B, RA, RB>(&self, a: A, b: B) -> (RA, RB)
     where
         A: FnOnce() -> RA + Send,
@@ -61,7 +59,6 @@ impl WorkerThread {
     }
 
     /// This function must be called from a worker thread.
-    #[inline]
     #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
     fn join_heartbeat<A, B, RA, RB>(&self, a: A, b: B) -> (RA, RB)
     where
@@ -131,7 +128,6 @@ impl WorkerThread {
 }
 
 impl Context {
-    #[inline]
     pub fn join<A, B, RA, RB>(self: &Arc<Self>, a: A, b: B) -> (RA, RB)
     where
         A: FnOnce() -> RA + Send,

distaff/src/scope.rs

@@ -257,7 +257,6 @@ impl<'scope, 'env> Scope<'scope, 'env> {
         task
     }
 
-    #[inline]
     pub fn join<A, B, RA, RB>(&'scope self, a: A, b: B) -> (RA, RB)
     where
         RA: Send,

distaff/src/threadpool.rs

@@ -8,8 +8,8 @@ pub struct ThreadPool {
 
 impl Drop for ThreadPool {
     fn drop(&mut self) {
-        // Ensure that the context is properly cleaned up when the thread pool is dropped.
+        // TODO: Ensure that the context is properly cleaned up when the thread pool is dropped.
-        self.context.set_should_exit();
+        // self.context.set_should_exit();
     }
 }
 
@@ -25,6 +25,11 @@ impl ThreadPool {
         Self { context }
     }
 
+    pub fn global() -> Self {
+        let context = Context::global_context().clone();
+        Self { context }
+    }
+
     pub fn scope<'env, F, R>(&self, f: F) -> R
     where
         F: for<'scope> FnOnce(&'scope Scope<'scope, 'env>) -> R + Send,

distaff/src/util.rs

@@ -182,7 +182,6 @@ impl<T, const BITS: u8> TaggedAtomicPtr<T, BITS> {
     }
 
     /// returns tag
-    #[inline]
     #[allow(dead_code)]
     pub fn compare_exchange_tag(
         &self,
@@ -201,7 +200,6 @@ impl<T, const BITS: u8> TaggedAtomicPtr<T, BITS> {
     }
 
     /// returns tag
-    #[inline]
     pub fn compare_exchange_weak_tag(
         &self,
         old: usize,

distaff/src/workerthread.rs

@@ -147,7 +147,6 @@ impl WorkerThread {
         None
     }
 
-    #[inline]
     fn find_work(&self) -> Option<SharedJob> {
         let mut guard = self.context.shared();
 
@@ -162,7 +161,6 @@ impl WorkerThread {
         None
     }
 
-    #[inline(always)]
     pub(crate) fn tick(&self) {
         if self.heartbeat.take() {
             #[cfg(feature = "metrics")]
@@ -176,7 +174,6 @@ impl WorkerThread {
         }
     }
 
-    #[inline]
     #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip(self)))]
     fn execute(&self, job: SharedJob) {
         unsafe { SharedJob::execute(job, self) };
@@ -209,28 +206,23 @@ impl WorkerThread {
 }
 
 impl WorkerThread {
-    #[inline]
     pub fn pop_back(&self) -> Option<NonNull<Job>> {
         unsafe { self.queue.as_mut_unchecked().pop_back() }
     }
 
-    #[inline]
     pub fn push_back<T>(&self, job: *const Job<T>) {
         unsafe { self.queue.as_mut_unchecked().push_back(job.cast()) }
     }
-    #[inline]
     pub fn push_front<T>(&self, job: *const Job<T>) {
         unsafe { self.queue.as_mut_unchecked().push_front(job.cast()) }
     }
 
-    #[inline]
     pub fn pop_front(&self) -> Option<NonNull<Job>> {
         unsafe { self.queue.as_mut_unchecked().pop_front() }
     }
 }
 
 impl WorkerThread {
-    #[inline]
     pub fn current_ref<'a>() -> Option<&'a Self> {
         unsafe { (*WORKER.with(UnsafeCell::get)).map(|ptr| ptr.as_ref()) }
     }
@@ -352,9 +344,10 @@ impl WorkerThread {
                 unsafe {
                     SharedJob::execute(job, self);
                 }
-            } else {
+                continue;
-                break;
             }
+
+            recv.wait();
         }
 
         Some(recv.recv())