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

//! ParallelForEach object is implemented for AtomicIterator and hence for types implementing
//! Fetch trait. This allows an FnMut function to be run on each value of the collection implementing
//! Fetch.

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

/// ParallelForEachIter allows calling the .for_each(f) to run a Fn function on type implementing AtomicIterator
/// ```
/// use parallel_task::prelude::*;
/// 
/// (0..100_000).collect::<Vec<i32>>().parallel_iter().for_each(|val|{ print!(" {}",val);});
/// assert_eq!(1,1)
/// ```
/// 
pub trait ParallelForEachIter<I, V,F>
where Self: AtomicIterator<AtomicItem = V> + Send + Sized,
F: Fn(V) + Send + Sync,
V: Send + Sync,
{
    fn for_each(self,f:F) {
        ParallelForEach::new(self,f).run()
    }
}

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

/// ParallelForEach is a structure type that captures the information necessary to run the values within the Iterator in parallel
/// Its the result of parallel_task that can be run on any Iterator implementing type.
/// ```
/// use parallel_task::prelude::*;
/// 
/// (0..100_000).collect::<Vec<i32>>().parallel_iter().for_each(|val|{ print!(" {}",val);});
/// assert_eq!(1,1)
/// ```
/// 
pub struct ParallelForEach<V,F,I>
where I: AtomicIterator<AtomicItem = V> + Send + Sized,
F: Fn(V) + Send + Sync,
V: Send
{
    pub iter: TaskQueue<I,V>,
    pub f:F,
    pub num_threads:usize,
    pub v: PhantomData<V>,    
}

#[allow(dead_code)]
impl<I,V,F> ParallelForEach<V,F,I>
where I:AtomicIterator<AtomicItem = V> + Send + Sized,
F: Fn(V) + Send + Sync,
V: Send + Sync,
{
    pub fn new(iter:I,f:F) -> Self
    {                   
        Self {
            iter: TaskQueue { iter },
            f,
            num_threads: Self::max_threads(),
            v: PhantomData,            
        }
    }

    fn max_threads() -> usize {
        // available_parallelism() function gives an idea of the CPUs available 
        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
    }
        
    pub fn run(self)    
    {                
        let num_threads = self.num_threads;        

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