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

use crate::ParExtend;
use crate::infallible::xap::Xap;
use crate::runner::ParRunner;
use orx_concurrent_iter::{ChunkPuller, ConcurrentIter};

pub fn collect_arb<Q, I, X, P>(th_idx: usize, state: &Q::State, iter: &I, x: X) -> P::ThreadValues
where
    Q: ParRunner,
    I: ConcurrentIter,
    X: Xap<I = I::Item>,
    P: ParExtend<X::O>,
{
    let mut collected = P::new_thread_values();
    let out = &mut collected;

    let mut chunk_puller = iter.chunk_puller_by(0, th_idx);

    loop {
        let chunk_size = Q::next_chunk_size(state, iter.size_hint());
        let chunk_state = Q::begin_chunk(th_idx, chunk_size);

        match chunk_size {
            0 | 1 => match iter.next_by(th_idx) {
                Some(i) => {
                    let values = x.xap(i);
                    P::add_thread_values(out, values);
                }
                None if iter.is_completed_when_none_returned() => break,
                None => {}
            },
            c => {
                chunk_puller.resize_for_chunk_size(c);

                match chunk_puller.pull() {
                    Some(chunk) => {
                        let values = chunk.flat_map(|i| x.xap(i));
                        P::add_thread_values(out, values);
                    }
                    None if iter.is_completed_when_none_returned() => break,
                    None => {}
                }
            }
        }

        Q::complete_chunk(state, chunk_state);
    }

    collected
}