orx-parallel 3.4.0

High performance, configurable and expressive parallel computation library.
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

use crate::Params;
use crate::generic_values::Values;
use crate::generic_values::runner_results::Fallibility;
use crate::runner::{ComputationKind, NumSpawned, ParallelRunner, SharedStateOf, ThreadRunnerOf};
use crate::using::executor::thread_compute as th;
use crate::using::using_variants::Using;
use orx_concurrent_iter::ConcurrentIter;

pub fn m<U, C, I, O, M1, Red>(
    using: U,
    mut orchestrator: C,
    params: Params,
    iter: I,
    map1: M1,
    reduce: Red,
) -> (NumSpawned, Option<O>)
where
    U: Using,
    C: ParallelRunner,
    I: ConcurrentIter,
    M1: Fn(&mut U::Item, I::Item) -> O + Sync,
    Red: Fn(&mut U::Item, O, O) -> O + Sync,
    O: Send,
{
    let thread_map =
        |nt: NumSpawned, iter: &I, state: &SharedStateOf<C>, thread_runner: ThreadRunnerOf<C>| {
            let u = using.create(nt.into_inner());
            Ok(th::reduce::m(u, thread_runner, iter, state, &map1, &reduce))
        };
    let (num_spawned, result) =
        orchestrator.map_infallible(params, iter, ComputationKind::Collect, thread_map);

    let mut u = using.into_inner();
    let acc = match result {
        Ok(results) => results
            .into_iter()
            .flatten()
            .reduce(|a, b| reduce(&mut u, a, b)),
    };

    (num_spawned, acc)
}

type ResultReduce<Vo> =
    Result<Option<<Vo as Values>::Item>, <<Vo as Values>::Fallibility as Fallibility>::Error>;

pub fn x<U, C, I, Vo, X1, Red>(
    using: U,
    mut orchestrator: C,
    params: Params,
    iter: I,
    xap1: X1,
    reduce: Red,
) -> (NumSpawned, ResultReduce<Vo>)
where
    U: Using,
    C: ParallelRunner,
    I: ConcurrentIter,
    Vo: Values,
    Vo::Item: Send,
    X1: Fn(&mut U::Item, I::Item) -> Vo + Sync,
    Red: Fn(&mut U::Item, Vo::Item, Vo::Item) -> Vo::Item + Sync,
{
    let thread_map =
        |nt: NumSpawned, iter: &I, state: &SharedStateOf<C>, thread_runner: ThreadRunnerOf<C>| {
            let u = using.create(nt.into_inner());
            th::reduce::x(u, thread_runner, iter, state, &xap1, &reduce).into_result()
        };
    let (num_spawned, result) = orchestrator.map_all::<Vo::Fallibility, _, _, _>(
        params,
        iter,
        ComputationKind::Collect,
        thread_map,
    );
    let mut u = using.into_inner();
    let acc = result.map(|results| {
        results
            .into_iter()
            .flatten()
            .reduce(|a, b| reduce(&mut u, a, b))
    });
    (num_spawned, acc)
}