orx-parallel 4.0.0

Performant parallel computations with an expressive iterator API.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

use crate::ParExtend;
use crate::infallible::recursive::utils;
use crate::{Par, ParDrain, ParUse, Params, ThreadPool, infallible::Xap, runner::ParRunner};
use alloc::vec::Vec;

struct Local<I, O> {
    input: Vec<I>,
    output: Vec<O>,
}

impl<I, O> Local<I, O> {
    fn new() -> Self {
        Self {
            input: Default::default(),
            output: Default::default(),
        }
    }
}

pub fn collect_arb<R, C, X, I, E, P>(
    mut runner: R,
    params: Params,
    iter: C,
    xap: X,
    extend: E,
    dst: &mut P,
) where
    R: ParRunner,
    C: IntoIterator,
    X: Xap<I = C::Item>,
    I: IntoIterator<Item = X::I>,
    E: Fn(&X::I) -> I + Send + Copy,
    X::O: Send,
    X::I: Send,
    P: ParExtend<X::O>,
{
    let max_threads: usize = runner.pool().max_num_threads().into();

    let mut data: Vec<_> = (0..max_threads).map(|_| Local::new()).collect();

    let mut inputs: Vec<_> = iter.into_iter().collect();

    let par = inputs.par_drain(..).runner(&mut runner);
    let par = params.apply(par).use_slice(&mut data);

    par.for_each(move |u, i| {
        u.input.extend(extend(&i));
        u.output.extend(xap.xap(i));
    });
    utils::into_outer_par(&mut inputs, &mut data, |x| &mut x.input, &mut runner);
    dst.extend(data.iter_mut().flat_map(|x| x.output.drain(..)));

    while !inputs.is_empty() {
        let par = inputs.par_drain(..).runner(&mut runner);
        let par = params.apply(par).use_slice(&mut data);

        par.for_each(move |u, i| {
            u.input.extend(extend(&i));
            u.output.extend(xap.xap(i));
        });

        utils::into_outer_par(&mut inputs, &mut data, |x| &mut x.input, &mut runner);
        dst.extend(data.iter_mut().flat_map(|x| x.output.drain(..)));
    }
}