use crate::parallel::{num_threads, Reducer};use crossbeam_utils::thread;/// Runs `left` and `right` in parallel, returning their output when both are done.pub fn join<O1: Send, O2: Send>(left: impl FnOnce() -> O1 + Send, right: impl FnOnce() -> O2 + Send) -> (O1, O2) { thread::scope(|s| { let left = s.spawn(|_| left()); let right = s.spawn(|_| right()); (left.join().unwrap(), right.join().unwrap()) }) .unwrap()}/// Read items from `input` and `consume` them, producing an output to be collected by a `reducer`, whose task is to/// aggregate these outputs into the final result returned by this function.////// * 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 [`Reducer`] traitpub fn in_parallel<I, S, O, R>( input: impl Iterator<Item = I> + Send, thread_limit: Option<usize>, new_thread_state: impl Fn(usize) -> S + Send + Sync, consume: impl Fn(I, &mut S) -> O + Send + Sync, mut reducer: R,) -> Result<<R as Reducer>::Output, <R as Reducer>::Error>where R: Reducer<Input = O>, I: Send, O: Send,{ let num_threads = num_threads(thread_limit); let new_thread_state = &new_thread_state; let consume = &consume; thread::scope(move |s| { let receive_result = { let (send_input, receive_input) = crossbeam_channel::bounded::<I>(num_threads); let (send_result, receive_result) = std::sync::mpsc::sync_channel::<O>(num_threads); for thread_id in 0..num_threads { s.spawn({ let send_result = send_result.clone(); let receive_input = receive_input.clone(); move |_| { let mut state = new_thread_state(thread_id); for item in receive_input { if send_result.send(consume(item, &mut state)).is_err() { break; } } } }); } s.spawn(move |_| { for item in input { if send_input.send(item).is_err() { break; } } }); receive_result }; for item in receive_result { reducer.feed(item)?; } reducer.finalize() }) .unwrap()}
use crate::parallel::{num_threads, Reducer};
use crossbeam_utils::thread;
/// Runs `left` and `right` in parallel, returning their output when both are done.
pub fn join<O1: Send, O2: Send>(left: impl FnOnce() -> O1 + Send, right: impl FnOnce() -> O2 + Send) -> (O1, O2) {
thread::scope(|s| {
let left = s.spawn(|_| left());
let right = s.spawn(|_| right());
(left.join().unwrap(), right.join().unwrap())
})
.unwrap()
}
/// Read items from `input` and `consume` them, producing an output to be collected by a `reducer`, whose task is to
/// aggregate these outputs into the final result returned by this function.
///
/// * 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 [`Reducer`] trait
pub fn in_parallel<I, S, O, R>(
input: impl Iterator<Item = I> + Send,
thread_limit: Option<usize>,
new_thread_state: impl Fn(usize) -> S + Send + Sync,
consume: impl Fn(I, &mut S) -> O + Send + Sync,
mut reducer: R,
) -> Result<<R as Reducer>::Output, <R as Reducer>::Error>
where
R: Reducer<Input = O>,
I: Send,
O: Send,
{
let num_threads = num_threads(thread_limit);
let new_thread_state = &new_thread_state;
let consume = &consume;
thread::scope(move |s| {
let receive_result = {
let (send_input, receive_input) = crossbeam_channel::bounded::<I>(num_threads);
let (send_result, receive_result) = std::sync::mpsc::sync_channel::<O>(num_threads);
for thread_id in 0..num_threads {
s.spawn({
let send_result = send_result.clone();
let receive_input = receive_input.clone();
move |_| {
let mut state = new_thread_state(thread_id);
for item in receive_input {
if send_result.send(consume(item, &mut state)).is_err() {
break;
}
}
}
});
}
s.spawn(move |_| {
for item in input {
if send_input.send(item).is_err() {
break;
}
}
});
receive_result
};
for item in receive_result {
reducer.feed(item)?;
}
reducer.finalize()
})
.unwrap()
}