parallel_task 0.6.0

A fast data parallelism library for Rust
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//! ParallelMap is a structure type that captures the Map object and function necessary to run the values within the AtomicIterator in parallel.

use std::marker::PhantomData;
use crate::task_queue::TaskQueue;
use crate::utils;
use crate::worker_thread::WorkerThreads;
use super::iterators::iterator::*;

use super::collector::*;

/// ParallelMapIter allows calling the .map(f) on type implementing AtomicIterator to get a ParallelMap object on which collect may be called.
/// ```
/// use parallel_task::prelude::*;
/// 
/// let res = (0..100_000).collect::<Vec<i32>>().parallel_iter().map(|val|*val).collect::<Vec<i32>>();
/// assert_eq!(res.len(),100_000)
/// ```
/// 
pub trait ParallelMapIter<I, V,F,T>
where Self: AtomicIterator<AtomicItem = V> + Send + Sized,
F: Fn(V) -> T + Send + Sync,
V: Send + Sync,
T:Send + Sync
{
    fn map(self,f:F) -> ParallelMap<V,F,T,Self>{
        ParallelMap::new(self,f)
    }
}

impl<I,V,F,T> ParallelMapIter<I, V,F,T> for I 
where I: AtomicIterator<AtomicItem = V> + Send + Sized,
F: Fn(V) -> T + Send + Sync,
V: Send + Sync,
T:Send + Sync {}

pub struct ParallelMap<V,F,T,I>
where I: AtomicIterator<AtomicItem = V> + Send + Sized,
F: Fn(V) -> T + Send + Sync,
V: Send,
T:Send 
{
    pub iter: TaskQueue<I,V>,
    pub f:F,
    pub num_threads:usize,
    pub v: PhantomData<V>,
    pub t: PhantomData<T>,
}

/// ParallelMap is a object that allows map function to be run on AtomicIterators.
/// The results may be then collected in types implementing Collector trait.
/// 
/// ```
/// use parallel_task::prelude::*;
/// 
/// let res = (0..100_000).collect::<Vec<i32>>().parallel_iter().map(|val|*val).collect::<Vec<i32>>();
/// assert_eq!(res.len(),100_000)
/// ```
/// 
#[allow(dead_code)]
impl<I,V,F,T> ParallelMap<V,F,T,I>
where I:AtomicIterator<AtomicItem = V> + Send + Sized,
F: Fn(V) -> T + Send + Sync,
V: Send + Sync,
T:Send + Sync
{
    pub fn new(iter:I,f:F) -> Self
    {                   
        Self {
            iter: TaskQueue { iter },
            f,
            num_threads: Self::max_threads(),
            v: PhantomData,
            t: PhantomData    
        }
    }

    /// Available_parallelism() function gives an idea of the CPUs available. 
    fn max_threads() -> usize {        
        utils::max_threads()
    }

    /// Set the thread pool size for running the parallel tasker.    
    pub fn threads(mut self, nthreads:usize) -> Self {
        self.num_threads = usize::min(Self::max_threads(),nthreads);
        self
    }

    /// Collect the results of the Map in a type implementing Collector trait    
    pub fn collect<C>(self) -> C
    where C: Collector<T>
    {                
        let num_threads = self.num_threads;        

        WorkerThreads { nthreads: num_threads }
        .collect(self)      
        
    }
}