orx_parallel/infallible/recursive/
par_iter.rs

1use crate::ParExtend;
2use crate::infallible::Xap;
3use crate::infallible::recursive::execution;
4use crate::infallible::recursive::par::ParRec;
5use crate::infallible::recursive::par_core::ParRecCore;
6use crate::infallible::xap::{FilMapOf, FilOf, FlatMapOf, FlattenOf, InsOf, MapOf};
7use crate::parameters::{ChunkSize, IterationOrder, NumThreads, Params};
8use crate::runner::{DefaultRunner, ParRunner};
9use alloc::vec::Vec;
10
11/// Parallel iterator.
12pub struct ParRecIter<I, X, Ix, Ex, R = DefaultRunner>
13where
14    I: IntoIterator,
15    X: Xap<I = I::Item>,
16    R: ParRunner,
17    Ix: IntoIterator<Item = X::I>,
18    Ex: Fn(&I::Item) -> Ix + Send + Copy,
19{
20    iter: I,
21    xap: X,
22    exe: R,
23    params: Params,
24    extend: Ex,
25}
26
27impl<I, X, Ix, Ex, R> ParRecIter<I, X, Ix, Ex, R>
28where
29    I: IntoIterator,
30    X: Xap<I = I::Item>,
31    R: ParRunner,
32    Ix: IntoIterator<Item = X::I>,
33    Ex: Fn(&I::Item) -> Ix + Send + Copy,
34{
35    pub(crate) fn new(iter: I, xap: X, exe: R, params: Params, extend: Ex) -> Self {
36        Self {
37            iter,
38            xap,
39            exe,
40            params,
41            extend,
42        }
43    }
44
45    pub(super) fn with_xap<Y: Xap<I = I::Item>>(self, xap: Y) -> ParRecIter<I, Y, Ix, Ex, R> {
46        ParRecIter::new(self.iter, xap, self.exe, self.params, self.extend)
47    }
48
49    fn destruct_x(self) -> (I, X, R, Params, Ex) {
50        (self.iter, self.xap, self.exe, self.params, self.extend)
51    }
52}
53
54impl<I, X, Ix, Ex, R> ParRecCore for ParRecIter<I, X, Ix, Ex, R>
55where
56    I: IntoIterator,
57    X: Xap<I = I::Item>,
58    R: ParRunner,
59    Ix: IntoIterator<Item = X::I>,
60    Ex: Fn(&I::Item) -> Ix + Send + Copy,
61{
62    type Item = X::O;
63
64    type Runner = R;
65
66    type Input = I;
67
68    type Xap = X;
69
70    fn destruct(self) -> (Self::Input, Self::Xap, Self::Runner, Params) {
71        (self.iter, self.xap, self.exe, self.params)
72    }
73}
74
75impl<I, X, Ix, Ex, R> ParRec for ParRecIter<I, X, Ix, Ex, R>
76where
77    I: IntoIterator,
78    X: Xap<I = I::Item>,
79    R: ParRunner,
80    Ix: IntoIterator<Item = X::I>,
81    Ex: Fn(&I::Item) -> Ix + Send + Copy,
82{
83    // configuration
84
85    fn runner<Q: ParRunner>(
86        self,
87        runner: Q,
88    ) -> impl ParRec<Item = Self::Item, Xap = Self::Xap, Input = Self::Input> {
89        let (iter, xap, _, params, extend) = self.destruct_x();
90        ParRecIter::new(iter, xap, runner, params, extend)
91    }
92
93    #[cfg(feature = "std")]
94    fn runner_with_diagnostics(
95        self,
96    ) -> impl ParRec<Item = Self::Item, Xap = Self::Xap, Input = Self::Input> {
97        let (iter, xap, exe, params, extend) = self.destruct_x();
98        ParRecIter::new(iter, xap, exe.with_diagnostics(), params, extend)
99    }
100
101    fn num_threads(mut self, num_threads: impl Into<NumThreads>) -> Self {
102        self.params = self.params.with_num_threads(num_threads);
103        self
104    }
105
106    fn chunk_size(mut self, chunk_size: impl Into<ChunkSize>) -> Self {
107        self.params = self.params.with_chunk_size(chunk_size);
108        self
109    }
110
111    fn iteration_order(mut self, collect: IterationOrder) -> Self {
112        self.params = self.params.with_collect_ordering(collect);
113        self
114    }
115
116    // transformations
117
118    fn map<Q, H>(
119        self,
120        h: H,
121    ) -> impl ParRec<Item = Q, Xap = MapOf<Self::Xap, Q, H>, Input = Self::Input>
122    where
123        H: Fn(Self::Item) -> Q + Copy + Send,
124    {
125        let xap = self.xap.map(h);
126        self.with_xap(xap)
127    }
128
129    fn inspect<H>(
130        self,
131        h: H,
132    ) -> impl ParRec<Item = Self::Item, Xap = InsOf<Self::Xap, H>, Input = Self::Input>
133    where
134        H: Fn(&Self::Item) + Copy + Send,
135    {
136        let xap = self.xap.inspect(h);
137        self.with_xap(xap)
138    }
139
140    fn filter<H>(
141        self,
142        h: H,
143    ) -> impl ParRec<Item = Self::Item, Xap = FilOf<Self::Xap, H>, Input = Self::Input>
144    where
145        H: Fn(&Self::Item) -> bool + Copy + Send,
146    {
147        let xap = self.xap.filter(h);
148        self.with_xap(xap)
149    }
150
151    fn filter_map<Q, H>(
152        self,
153        h: H,
154    ) -> impl ParRec<Item = Q, Xap = FilMapOf<Self::Xap, Q, H>, Input = Self::Input>
155    where
156        H: Fn(Self::Item) -> Option<Q> + Copy + Send,
157    {
158        let xap = self.xap.filter_map(h);
159        self.with_xap(xap)
160    }
161
162    fn flat_map<V, H>(
163        self,
164        h: H,
165    ) -> impl ParRec<Item = V::Item, Xap = FlatMapOf<Self::Xap, V, H>, Input = Self::Input>
166    where
167        V: IntoIterator,
168        H: Fn(Self::Item) -> V + Copy + Send,
169    {
170        let xap = self.xap.flat_map(h);
171        self.with_xap(xap)
172    }
173
174    fn flatten(
175        self,
176    ) -> impl ParRec<
177        Item = <Self::Item as IntoIterator>::Item,
178        Xap = FlattenOf<Self::Xap>,
179        Input = Self::Input,
180    >
181    where
182        Self::Item: IntoIterator,
183    {
184        let xap = self.xap.flatten();
185        self.with_xap(xap)
186    }
187
188    // compute
189
190    fn first(self) -> Option<Self::Item>
191    where
192        Self::Item: Send,
193        <Self::Input as IntoIterator>::Item: Send,
194    {
195        let (iter, x, exe, params, extend) = self.destruct_x();
196
197        match params.iteration_order {
198            IterationOrder::Ordered => execution::next(exe, params, iter, x, extend),
199            IterationOrder::Arbitrary => execution::next_any(exe, params, iter, x, extend),
200        }
201    }
202
203    fn reduce<F>(self, f: F) -> Option<Self::Item>
204    where
205        F: Fn(Self::Item, Self::Item) -> Self::Item + Send + Copy,
206        Self::Item: Send,
207        <Self::Input as IntoIterator>::Item: Send,
208    {
209        let (iter, x, exe, params, extend) = self.destruct_x();
210        execution::reduce(exe, params, iter, x, extend, f)
211    }
212
213    fn fold<B, Id, F>(self, init: Id, f: F) -> Vec<B>
214    where
215        B: Send,
216        Id: Fn() -> B,
217        F: Fn(&mut B, Self::Item) + Copy + Send,
218        <Self::Input as IntoIterator>::Item: Send,
219    {
220        let (iter, x, exe, params, extend) = self.destruct_x();
221        execution::fold(exe, params, iter, x, extend, init, f)
222    }
223
224    fn collect_into<C>(self, dst: &mut C)
225    where
226        C: ParExtend<Self::Item>,
227        Self::Item: Send,
228        <Self::Input as IntoIterator>::Item: Send,
229    {
230        let (iter, x, exe, params, extend) = self.destruct_x();
231        match params.iteration_order {
232            IterationOrder::Ordered => execution::collect(exe, params, iter, x, extend, dst),
233            IterationOrder::Arbitrary => execution::collect_arb(exe, params, iter, x, extend, dst),
234        }
235    }
236}
orx_parallel/infallible/recursive/par_iter.rs

orx_parallel/infallible/recursive/
par_iter.rs
