V4 optimize recursive computations

Cargo.toml

@@ -21,7 +21,8 @@ orx-pinned-concurrent-col = { version = "2.18.0", default-features = false }
 orx-iterable = { version = "1.3.0", default-features = false }
 orx-priority-queue = { version = "1.7.0", default-features = false }
 orx-pseudo-default = { version = "2.1.0", default-features = false }
-orx-concurrent-recursive-iter = { version = "2.0.0", default-features = false }
+# orx-concurrent-recursive-iter = { version = "2.0.0", default-features = false }
+orx-concurrent-recursive-iter = { path = "../orx-concurrent-recursive-iter", default-features = false }
 
 # optional: generic iterator
 rayon = { version = "1.11.0", optional = true, default-features = false }

benches/rec/file_system.rs

@@ -128,6 +128,7 @@ fn orx_sum(fs: &FileSystem, work: usize) -> u64 {
         .iter()
         .copied()
         .into_par_recursive(extend)
+        .runner(runner::recursive_runner())
         .map(|idx| fs.nodes[idx].compute_score(work))
         .reduce(|a, b| a + b)
         .unwrap_or(0)
@@ -237,6 +238,12 @@ fn run(c: &mut Criterion) {
             max_children: 8,
             work: 250,
         },
+        Input {
+            num_nodes: 160_000,
+            num_roots: 100,
+            max_children: 8,
+            work: 250,
+        },
     ];
 
     let variants: Vec<_> = all::<Method>().collect();

examples/recursive_tune.rs

@@ -134,6 +134,8 @@ struct Args {
     /// Print ORX workload diagnostics.
     #[arg(long, default_value_t = false)]
     diagnostics: bool,
+    #[arg(long, default_value_t = false)]
+    rec_runner: bool,
 }
 
 fn seq_sum(fs: &FileSystem, work: usize) -> u64 {
@@ -181,22 +183,48 @@ fn orx_sum(fs: &FileSystem, work: usize, args: &Args) -> u64 {
         queue.extend(fs.nodes[*idx].children.iter().copied());
     };
 
-    let input = fs
-        .roots
-        .iter()
-        .copied()
-        .into_par_recursive(extend)
-        .num_threads(args.num_threads)
-        .chunk_size(args.chunk_size)
-        .map(|idx| fs.nodes[idx].compute_score(work));
-
-    if args.diagnostics {
-        input
+    match (args.diagnostics, args.rec_runner) {
+        (false, false) => fs
+            .roots
+            .iter()
+            .copied()
+            .into_par_recursive(extend)
+            .num_threads(args.num_threads)
+            .chunk_size(args.chunk_size)
+            .map(|idx| fs.nodes[idx].compute_score(work))
+            .sum(),
+        (false, true) => fs
+            .roots
+            .iter()
+            .copied()
+            .into_par_recursive(extend)
+            .num_threads(args.num_threads)
+            .chunk_size(args.chunk_size)
+            .runner(runner::recursive_runner())
+            .map(|idx| fs.nodes[idx].compute_score(work))
+            .sum(),
+        (true, false) => fs
+            .roots
+            .iter()
+            .copied()
+            .into_par_recursive(extend)
+            .num_threads(args.num_threads)
+            .chunk_size(args.chunk_size)
+            .runner_with_diagnostics()
+            .map(|idx| fs.nodes[idx].compute_score(work))
+            .sum(),
+        (true, true) => fs
+            .roots
+            .iter()
+            .copied()
+            .into_par_recursive(extend)
+            .num_threads(args.num_threads)
+            .runner(runner::recursive_runner())
             .runner_with_diagnostics()
-            .reduce(|a, b| a + b)
-            .unwrap_or(0)
-    } else {
-        input.reduce(|a, b| a + b).unwrap_or(0)
+            .chunk_size(args.chunk_size)
+            .map(|idx| fs.nodes[idx].compute_score(work))
+            .sum(),
+        _ => todo!(),
     }
 }
 

src/infallible/par_runner.rs

@@ -14,12 +14,14 @@ pub trait ParRunnerInfallible: ParRunner {
         X::O: Send,
     {
         let mut spawned = 0;
-        let (max_nt, state) = self.nt_state(params, iter.try_get_len());
+        let (max_nt, state) = self.nt_state(params, iter.size_hint());
         let results_bag = ConcurrentBag::with_fixed_capacity(max_nt);
 
         let (iter, st, results, x) = (&iter, &state, &results_bag, x);
         self.pool_mut().scoped_computation(move |s| {
-            while let Some(th_idx) = Self::do_spawn_new(spawned, st) {
+            while let Some(th_idx) =
+                Self::do_spawn_new_with_queue_len(spawned, st, iter.size_hint())
+            {
                 spawned += 1;
                 <Self::Pool as ParThreadPool>::run_in_scope(&s, move || {
                     Self::begin_thread(st, th_idx);
@@ -41,12 +43,14 @@ pub trait ParRunnerInfallible: ParRunner {
         X::O: Send,
     {
         let mut spawned = 0;
-        let (max_nt, state) = self.nt_state(params, iter.try_get_len());
+        let (max_nt, state) = self.nt_state(params, iter.size_hint());
         let results_bag = ConcurrentBag::with_fixed_capacity(max_nt);
 
         let (iter, st, results, x) = (&iter, &state, &results_bag, x);
         self.pool_mut().scoped_computation(move |s| {
-            while let Some(th_idx) = Self::do_spawn_new(spawned, st) {
+            while let Some(th_idx) =
+                Self::do_spawn_new_with_queue_len(spawned, st, iter.size_hint())
+            {
                 spawned += 1;
                 <Self::Pool as ParThreadPool>::run_in_scope(&s, move || {
                     Self::begin_thread(st, th_idx);
@@ -69,12 +73,14 @@ pub trait ParRunnerInfallible: ParRunner {
         X::O: Send,
     {
         let mut spawned = 0;
-        let (max_nt, state) = self.nt_state(params, iter.try_get_len());
+        let (max_nt, state) = self.nt_state(params, iter.size_hint());
         let results_bag = ConcurrentBag::with_fixed_capacity(max_nt);
 
         let (iter, st, results, x) = (&iter, &state, &results_bag, x);
         self.pool_mut().scoped_computation(move |s| {
-            while let Some(th_idx) = Self::do_spawn_new(spawned, st) {
+            while let Some(th_idx) =
+                Self::do_spawn_new_with_queue_len(spawned, st, iter.size_hint())
+            {
                 spawned += 1;
                 <Self::Pool as ParThreadPool>::run_in_scope(&s, move || {
                     Self::begin_thread(st, th_idx);
@@ -96,12 +102,14 @@ pub trait ParRunnerInfallible: ParRunner {
         X::O: Send,
     {
         let mut spawned = 0;
-        let (max_nt, state) = self.nt_state(params, iter.try_get_len());
+        let (max_nt, state) = self.nt_state(params, iter.size_hint());
         let results_bag = ConcurrentBag::with_fixed_capacity(max_nt);
 
         let (iter, st, results) = (&iter, &state, &results_bag);
         self.pool_mut().scoped_computation(move |s| {
-            while let Some(th_idx) = Self::do_spawn_new(spawned, st) {
+            while let Some(th_idx) =
+                Self::do_spawn_new_with_queue_len(spawned, st, iter.size_hint())
+            {
                 spawned += 1;
                 <Self::Pool as ParThreadPool>::run_in_scope(&s, move || {
                     Self::begin_thread(st, th_idx);
@@ -123,12 +131,14 @@ pub trait ParRunnerInfallible: ParRunner {
         X::O: Send,
     {
         let mut spawned = 0;
-        let (max_nt, state) = self.nt_state(params, iter.try_get_len());
+        let (max_nt, state) = self.nt_state(params, iter.size_hint());
         let results_bag = ConcurrentBag::with_fixed_capacity(max_nt);
 
         let (iter, st, results) = (&iter, &state, &results_bag);
         self.pool_mut().scoped_computation(move |s| {
-            while let Some(th_idx) = Self::do_spawn_new(spawned, st) {
+            while let Some(th_idx) =
+                Self::do_spawn_new_with_queue_len(spawned, st, iter.size_hint())
+            {
                 spawned += 1;
                 <Self::Pool as ParThreadPool>::run_in_scope(&s, move || {
                     Self::begin_thread(st, th_idx);
@@ -160,11 +170,13 @@ pub trait ParRunnerInfallible: ParRunner {
         };
 
         let mut spawned = 0;
-        let (_, state) = self.nt_state(params, iter.try_get_len());
+        let (_, state) = self.nt_state(params, iter.size_hint());
 
         let (iter, st, bag) = (&iter, &state, &results_bag);
         self.pool_mut().scoped_computation(move |s| {
-            while let Some(th_idx) = Self::do_spawn_new(spawned, st) {
+            while let Some(th_idx) =
+                Self::do_spawn_new_with_queue_len(spawned, st, iter.size_hint())
+            {
                 spawned += 1;
                 <Self::Pool as ParThreadPool>::run_in_scope(&s, move || {
                     Self::begin_thread(st, th_idx);

src/infallible/thread_execution/collect.rs

@@ -17,10 +17,14 @@ where
     loop {
         let chunk_size = Q::next_chunk_size(state, iter.try_get_len());
         let chunk_state = Q::begin_chunk(th_idx, chunk_size);
+        let mut non_empty = false;
 
         match chunk_size {
             0 | 1 => match item_puller.next() {
-                Some((idx, i)) => out.extend(idx, x.xap(i)),
+                Some((idx, i)) => {
+                    non_empty = true;
+                    out.extend(idx, x.xap(i))
+                }
                 None if iter.is_completed_when_none_returned() => break,
                 None => {}
             },
@@ -30,14 +34,20 @@ where
                 }
 
                 match chunk_puller.pull_with_idx() {
-                    Some((idx, chunk)) => out.extend(idx, chunk.flat_map(|i| x.xap(i))),
+                    Some((idx, chunk)) => {
+                        non_empty = true;
+                        out.extend(idx, chunk.flat_map(|i| x.xap(i)))
+                    }
                     None if iter.is_completed_when_none_returned() => break,
                     None => {}
                 }
             }
         }
 
-        Q::complete_chunk(state, chunk_state);
+        match non_empty {
+            true => Q::complete_chunk_non_empty(state, chunk_state),
+            false => Q::complete_chunk_empty(state, chunk_state),
+        }
     }
 
     collected

src/infallible/thread_execution/collect_arb.rs

@@ -18,10 +18,14 @@ where
     loop {
         let chunk_size = Q::next_chunk_size(state, iter.try_get_len());
         let chunk_state = Q::begin_chunk(th_idx, chunk_size);
+        let mut non_empty = false;
 
         match chunk_size {
             0 | 1 => match item_puller.next() {
-                Some(i) => vec.extend(x.xap(i)),
+                Some(i) => {
+                    non_empty = true;
+                    vec.extend(x.xap(i))
+                }
                 None if iter.is_completed_when_none_returned() => break,
                 None => {}
             },
@@ -31,14 +35,20 @@ where
                 }
 
                 match chunk_puller.pull() {
-                    Some(chunk) => vec.extend(chunk.flat_map(|i| x.xap(i))),
+                    Some(chunk) => {
+                        non_empty = true;
+                        vec.extend(chunk.flat_map(|i| x.xap(i)))
+                    }
                     None if iter.is_completed_when_none_returned() => break,
                     None => {}
                 }
             }
         }
 
-        Q::complete_chunk(state, chunk_state);
+        match non_empty {
+            true => Q::complete_chunk_non_empty(state, chunk_state),
+            false => Q::complete_chunk_empty(state, chunk_state),
+        }
     }
 
     collected

src/infallible/thread_execution/collect_arb_over_bag.rs

@@ -23,10 +23,12 @@ pub fn collect_arb_over_bag<Q, I, X, P>(
     loop {
         let chunk_size = Q::next_chunk_size(state, iter.try_get_len());
         let chunk_state = Q::begin_chunk(th_idx, chunk_size);
+        let mut non_empty = false;
 
         match chunk_size {
             0 | 1 => match item_puller.next() {
                 Some(i) => {
+                    non_empty = true;
                     // TODO: possible to try to get len and bag.extend(values_vt.values()) when available, same holds for chunk below
                     for i in x.xap(i).into_iter() {
                         _ = bag.push(i);
@@ -42,6 +44,7 @@ pub fn collect_arb_over_bag<Q, I, X, P>(
 
                 match chunk_puller.pull() {
                     Some(chunk) => {
+                        non_empty = true;
                         for i in chunk.flat_map(|i| x.xap(i)) {
                             _ = bag.push(i);
                         }
@@ -52,6 +55,9 @@ pub fn collect_arb_over_bag<Q, I, X, P>(
             }
         }
 
-        Q::complete_chunk(state, chunk_state);
+        match non_empty {
+            true => Q::complete_chunk_non_empty(state, chunk_state),
+            false => Q::complete_chunk_empty(state, chunk_state),
+        }
     }
 }

src/infallible/thread_execution/next.rs

@@ -15,10 +15,12 @@ where
     loop {
         let chunk_size = Q::next_chunk_size(state, iter.try_get_len());
         let chunk_state = Q::begin_chunk(th_idx, chunk_size);
+        let mut non_empty = false;
 
         match chunk_size {
             0 | 1 => match item_puller.next() {
                 Some((idx, i)) => {
+                    non_empty = true;
                     if let Some(val) = x.xap(i).into_iter().next() {
                         Q::broadcast_stop(iter, state, chunk_state);
                         return Some(ValIdx::new(val, idx));
@@ -34,6 +36,7 @@ where
 
                 match chunk_puller.pull_with_idx() {
                     Some((idx, chunk)) => {
+                        non_empty = true;
                         if let Some(val) = chunk.flat_map(|i| x.xap(i)).next() {
                             Q::broadcast_stop(iter, state, chunk_state);
                             return Some(ValIdx::new(val, idx));
@@ -45,7 +48,10 @@ where
             }
         }
 
-        Q::complete_chunk(state, chunk_state);
+        match non_empty {
+            true => Q::complete_chunk_non_empty(state, chunk_state),
+            false => Q::complete_chunk_empty(state, chunk_state),
+        }
     }
 
     None

src/infallible/thread_execution/next_any.rs

@@ -14,10 +14,12 @@ where
     loop {
         let chunk_size = Q::next_chunk_size(state, iter.try_get_len());
         let chunk_state = Q::begin_chunk(th_idx, chunk_size);
+        let mut non_empty = false;
 
         match chunk_size {
             0 | 1 => match item_puller.next() {
                 Some(i) => {
+                    non_empty = true;
                     if let Some(val) = x.xap(i).into_iter().next() {
                         Q::broadcast_stop(iter, state, chunk_state);
                         return Some(val);
@@ -33,6 +35,7 @@ where
 
                 match chunk_puller.pull() {
                     Some(chunk) => {
+                        non_empty = true;
                         if let Some(val) = chunk.flat_map(|i| x.xap(i)).next() {
                             Q::broadcast_stop(iter, state, chunk_state);
                             return Some(val);
@@ -43,7 +46,10 @@ where
                 }
             }
         }
-        Q::complete_chunk(state, chunk_state);
+        match non_empty {
+            true => Q::complete_chunk_non_empty(state, chunk_state),
+            false => Q::complete_chunk_empty(state, chunk_state),
+        }
     }
 
     None