avx_parallel/
parallel_vec.rs

1//! High-performance parallel iterators with true parallelism
2//!
3//! This module provides parallel iterator implementations that actually
4//! execute in parallel using the thread pool.
5
6use crate::executor;
7use std::marker::PhantomData;
8
9/// A parallel iterator that executes operations in parallel
10pub struct ParallelVec<'a, T: Sync> {
11    data: &'a [T],
12}
13
14impl<'a, T: Sync> ParallelVec<'a, T> {
15    /// Create a new parallel vector iterator
16    pub fn new(data: &'a [T]) -> Self {
17        Self { data }
18    }
19
20    /// Map operation in parallel
21    pub fn map<R, F>(self, f: F) -> ParallelMap<'a, T, R, F>
22    where
23        R: Send,
24        F: Fn(&T) -> R + Send + Sync,
25    {
26        ParallelMap {
27            data: self.data,
28            f,
29            _phantom: PhantomData,
30        }
31    }
32
33    /// Filter operation in parallel
34    pub fn filter<F>(self, f: F) -> ParallelFilter<'a, T, F>
35    where
36        F: Fn(&T) -> bool + Send + Sync,
37    {
38        ParallelFilter {
39            data: self.data,
40            f,
41        }
42    }
43
44    /// For each operation in parallel
45    pub fn for_each<F>(self, f: F)
46    where
47        F: Fn(&T) + Send + Sync,
48    {
49        executor::parallel_for_each(self.data, f);
50    }
51
52    /// Sum operation in parallel
53    pub fn sum(self) -> T
54    where
55        T: Clone + Send + std::iter::Sum,
56    {
57        executor::parallel_sum(self.data)
58    }
59
60    /// Reduce operation in parallel
61    pub fn reduce<F>(self, f: F) -> Option<T>
62    where
63        T: Clone + Send,
64        F: Fn(T, T) -> T + Send + Sync,
65    {
66        executor::parallel_reduce(self.data, f)
67    }
68
69    /// Collect into a Vec
70    pub fn collect(self) -> Vec<&'a T> {
71        self.data.iter().collect()
72    }
73}
74
75/// Parallel map operation
76pub struct ParallelMap<'a, T, R, F> {
77    data: &'a [T],
78    f: F,
79    _phantom: PhantomData<R>,
80}
81
82impl<'a, T, R, F> ParallelMap<'a, T, R, F>
83where
84    T: Sync,
85    R: Send + 'static,
86    F: Fn(&T) -> R + Send + Sync,
87{
88    /// Collect the mapped results
89    pub fn collect(self) -> Vec<R> {
90        executor::parallel_map(self.data, self.f)
91    }
92
93    /// Sum the mapped results
94    pub fn sum(self) -> R
95    where
96        R: std::iter::Sum,
97    {
98        let results = self.collect();
99        results.into_iter().sum()
100    }
101}
102
103/// Parallel filter operation
104pub struct ParallelFilter<'a, T, F> {
105    data: &'a [T],
106    f: F,
107}
108
109impl<'a, T, F> ParallelFilter<'a, T, F>
110where
111    T: Sync,
112    F: Fn(&T) -> bool + Send + Sync,
113{
114    /// Collect the filtered results
115    pub fn collect(self) -> Vec<&'a T> {
116        executor::parallel_filter(self.data, self.f)
117    }
118
119    /// Map the filtered results
120    pub fn map<R, F2>(self, f2: F2) -> Vec<R>
121    where
122        R: Send + 'static,
123        F2: Fn(&T) -> R + Send + Sync,
124    {
125        let filtered = self.collect();
126        executor::parallel_map(&filtered, |&item| f2(item))
127    }
128}
129
130/// Extension trait to enable parallel iteration
131pub trait IntoParallelVec<T: Sync> {
132    /// Convert into a parallel vector iterator
133    fn par_vec(&self) -> ParallelVec<'_, T>;
134}
135
136impl<T: Sync> IntoParallelVec<T> for Vec<T> {
137    fn par_vec(&self) -> ParallelVec<'_, T> {
138        ParallelVec::new(self)
139    }
140}
141
142impl<T: Sync> IntoParallelVec<T> for [T] {
143    fn par_vec(&self) -> ParallelVec<'_, T> {
144        ParallelVec::new(self)
145    }
146}
147
148#[cfg(test)]
149mod tests {
150    use super::*;
151
152    #[test]
153    fn test_parallel_vec_map() {
154        let data: Vec<i32> = (0..10000).collect();
155        let results = data.par_vec().map(|&x| x * 2).collect();
156
157        assert_eq!(results.len(), 10000);
158        assert_eq!(results[0], 0);
159        assert_eq!(results[9999], 19998);
160    }
161
162    #[test]
163    fn test_parallel_vec_filter() {
164        let data: Vec<i32> = (0..10000).collect();
165        let results = data.par_vec().filter(|&x| x % 2 == 0).collect();
166
167        assert_eq!(results.len(), 5000);
168    }
169
170    #[test]
171    fn test_parallel_vec_sum() {
172        let data: Vec<i32> = (1..=100).collect();
173        let result = data.par_vec().sum();
174
175        assert_eq!(result, 5050);
176    }
177
178    #[test]
179    fn test_parallel_vec_reduce() {
180        let data: Vec<i32> = (1..=10).collect();
181        let result = data.par_vec().reduce(|a, b| a + b);
182
183        assert_eq!(result, Some(55));
184    }
185
186    #[test]
187    fn test_parallel_vec_chain() {
188        let data: Vec<i32> = (0..1000).collect();
189        let result: i32 = data.par_vec()
190            .map(|&x| x * 2)
191            .sum();
192
193        let expected: i32 = (0..1000).map(|x| x * 2).sum();
194        assert_eq!(result, expected);
195    }
196}
197
198
199
200
201
avx_parallel/parallel_vec.rs

avx_parallel/
parallel_vec.rs
