parallel status check

Author: pascalkuthe

gix-features/src/parallel/in_parallel.rs

@@ -34,6 +34,76 @@ pub fn build_thread() -> std::thread::Builder {
     std::thread::Builder::new()
 }
 
+/// Read items from `input` and `consume` them in multiple threads,
+/// whose output output is collected by a `reducer`. Its task is to
+/// aggregate these outputs into the final result returned by this function with the benefit of not having to be thread-safe.
+///
+/// * if `thread_limit` is `Some`, the given amount of threads will be used. If `None`, all logical cores will be used.
+/// * `new_thread_state(thread_number) -> State` produces thread-local state once per thread to be based to `consume`
+/// * `consume(Item, &mut State) -> Output` produces an output given an input obtained by `input` along with mutable state initially
+///   created by `new_thread_state(…)`.
+/// * For `reducer`, see the [`Reduce`] trait
+pub fn in_parallel_chunks<'a, I, S, O, R>(
+    input: &'a mut [I],
+    chunk_size: usize,
+    thread_limit: Option<usize>,
+    new_thread_state: impl Fn(usize) -> S + Send + Clone,
+    consume: impl Fn(&'a mut I, &mut S) -> Option<O> + Send + Clone,
+    mut reducer: R,
+) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>
+where
+    R: Reduce<Input = O>,
+    I: Send,
+    O: Send,
+{
+    let num_threads = num_threads(thread_limit);
+    std::thread::scope(move |s| {
+        let receive_result = {
+            let (send_input, receive_input) = crossbeam_channel::bounded::<&mut [I]>(num_threads);
+            let (send_result, receive_result) = crossbeam_channel::bounded::<O>(num_threads);
+            for thread_id in 0..num_threads {
+                std::thread::Builder::new()
+                    .name(format!("gitoxide.in_parallel.produce.{thread_id}"))
+                    .spawn_scoped(s, {
+                        let send_result = send_result.clone();
+                        let receive_input = receive_input.clone();
+                        let new_thread_state = new_thread_state.clone();
+                        let consume = consume.clone();
+                        move || {
+                            let mut state = new_thread_state(thread_id);
+                            for chunk in receive_input {
+                                for item in chunk {
+                                    if let Some(output) = consume(item, &mut state) {
+                                        if send_result.send(output).is_err() {
+                                            break;
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    })
+                    .expect("valid name");
+            }
+            std::thread::Builder::new()
+                .name("gitoxide.in_parallel.feed".into())
+                .spawn_scoped(s, move || {
+                    for chunk in input.chunks_mut(chunk_size) {
+                        if send_input.send(chunk).is_err() {
+                            break;
+                        }
+                    }
+                })
+                .expect("valid name");
+            receive_result
+        };
+
+        for item in receive_result {
+            drop(reducer.feed(item)?);
+        }
+        reducer.finalize()
+    })
+}
+
 /// Read items from `input` and `consume` them in multiple threads,
 /// whose output output is collected by a `reducer`. Its task is to
 /// aggregate these outputs into the final result returned by this function with the benefit of not having to be thread-safe.

gix-features/src/parallel/mod.rs

@@ -35,11 +35,11 @@
 #[cfg(feature = "parallel")]
 mod in_parallel;
 #[cfg(feature = "parallel")]
-pub use in_parallel::{build_thread, in_parallel, in_parallel_with_slice, join, threads};
+pub use in_parallel::{build_thread, in_parallel, in_parallel_chunks, in_parallel_with_slice, join, threads};
 
 mod serial;
 #[cfg(not(feature = "parallel"))]
-pub use serial::{build_thread, in_parallel, in_parallel_with_slice, join, threads};
+pub use serial::{build_thread, in_parallel, in_parallel_chunks, in_parallel_with_slice, join, threads};
 
 mod in_order;
 pub use in_order::{InOrderIter, SequenceId};
@@ -128,6 +128,53 @@ pub fn num_threads(thread_limit: Option<usize>) -> usize {
         .unwrap_or(logical_cores)
 }
 
+/// Run [`in_parallel()`] only if the given `condition()` returns true when eagerly evaluated.
+///
+/// For parameters, see the documentation of [`in_parallel()`]
+#[cfg(feature = "parallel")]
+pub fn in_parallel_chunks_if<'a, I, S, O, R>(
+    condition: impl FnOnce() -> bool,
+    input: &'a mut [I],
+    chunk_size: usize,
+    thread_limit: Option<usize>,
+    new_thread_state: impl Fn(usize) -> S + Send + Clone,
+    consume: impl Fn(&'a mut I, &mut S) -> Option<O> + Send + Clone,
+    reducer: R,
+) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>
+where
+    R: Reduce<Input = O>,
+    I: Send,
+    O: Send,
+{
+    if num_threads(thread_limit) > 1 && condition() {
+        in_parallel_chunks(input, chunk_size, thread_limit, new_thread_state, consume, reducer)
+    } else {
+        serial::in_parallel_chunks(input, chunk_size, thread_limit, new_thread_state, consume, reducer)
+    }
+}
+
+/// Run [`in_parallel()`] only if the given `condition()` returns true when eagerly evaluated.
+///
+/// For parameters, see the documentation of [`in_parallel()`]
+///
+/// Note that the non-parallel version is equivalent to [`in_parallel()`].
+#[cfg(not(feature = "parallel"))]
+pub fn in_parallel_chunks_if<'a, I, S, O, R>(
+    _condition: impl FnOnce() -> bool,
+    input: &'a mut [I],
+    chunk_size: usize,
+    thread_limit: Option<usize>,
+    new_thread_state: impl Fn(usize) -> S + Send + Clone,
+    consume: impl Fn(&'a mut I, &mut S) -> Option<O> + Send + Clone,
+    reducer: R,
+) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>
+where
+    R: Reduce<Input = O>,
+    I: Send,
+    O: Send,
+{
+    serial::in_parallel_chunks(input, chunk_size, thread_limit, new_thread_state, consume, reducer)
+}
 /// Run [`in_parallel()`] only if the given `condition()` returns true when eagerly evaluated.
 ///
 /// For parameters, see the documentation of [`in_parallel()`]

gix-features/src/parallel/serial.rs

@@ -133,3 +133,35 @@ where
     }
     reducer.finalize()
 }
+
+/// Read items from `input` and `consume` them in a single thread, producing an output to be collected by a `reducer`,
+/// whose task is to aggregate these outputs into the final result returned by this function.
+///
+/// * `new_thread_state(thread_number) -> State` produces thread-local state once per thread to be based to `consume`
+/// * `consume(Item, &mut State) -> Output` produces an output given an input along with mutable state.
+/// * For `reducer`, see the [`Reduce`] trait
+/// * if `thread_limit` has no effect as everything is run on the main thread, but is present to keep the signature
+///   similar to the parallel version.
+///
+/// **This serial version performing all calculations on the current thread.**
+pub fn in_parallel_chunks<'a, I, S, O, R>(
+    input: &'a mut [I],
+    _chunk_size: usize,
+    _thread_limit: Option<usize>,
+    new_thread_state: impl Fn(usize) -> S + Send + Clone,
+    consume: impl Fn(&'a mut I, &mut S) -> Option<O> + Send + Clone,
+    mut reducer: R,
+) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>
+where
+    R: Reduce<Input = O>,
+    I: Send,
+    O: Send,
+{
+    let mut state = new_thread_state(0);
+    for item in input {
+        if let Some(res) = consume(item, &mut state) {
+            drop(reducer.feed(res)?);
+        }
+    }
+    reducer.finalize()
+}

gix-index/src/lib.rs

@@ -93,9 +93,10 @@ pub struct State {
     /// same timestamp as this as potentially changed, checking more thoroughly if a change actually happened.
     timestamp: FileTime,
     version: Version,
-    entries: Vec<Entry>,
+    ///
+    pub entries: Vec<Entry>,
     /// A memory area keeping all index paths, in full length, independently of the index version.
-    path_backing: PathStorage,
+    pub path_backing: PathStorage,
     /// True if one entry in the index has a special marker mode
     is_sparse: bool,
 

gix-worktree/src/index/status.rs

@@ -46,7 +46,7 @@ pub enum Status<T = ()> {
 }
 
 ///
-pub trait Collector<'index> {
+pub trait Collector<'index>: Send {
     /// Data generated by comparing two files/entries
     type Diff;
     ///

gix-worktree/src/index/status/recorder.rs

@@ -11,7 +11,7 @@ pub struct Recorder<'index, T = ()> {
     pub records: Vec<(&'index BStr, Status<T>, bool)>,
 }
 
-impl<'index, T> Collector<'index> for Recorder<'index, T> {
+impl<'index, T: Send> Collector<'index> for Recorder<'index, T> {
     type Diff = T;
 
     fn visit_entry(