Skip to main content

orx_parallel/infallible/recursive/
par.rs

1use crate::infallible::recursive::par_core::ParRecCore;
2use crate::infallible::xap::FlattenOf;
3use crate::infallible::{FilMapOf, FilOf, FlatMapOf, InsOf, MapOf};
4use crate::runner::ParRunner;
5use crate::{ChunkSize, IterationOrder, NumThreads};
6use crate::{ParExtend, Sum};
7use alloc::vec::Vec;
8use core::cmp::Ordering;
9
10/// Infallible parallel recursive iterator.
11///
12/// `ParRec` is the central trait for describing recursive parallel computations as iterator
13/// pipelines. It mirrors common sequential iterator operations (`map`,
14/// `filter`, `flat_map`, `collect`, `reduce`, ...) while allowing runtime
15/// configuration of execution details such as number of threads, chunk size,
16/// iteration order, and runner/pool selection.
17///
18/// Recursive traversal can be deterministic: with [`IterationOrder::Ordered`] (the default),
19/// order-sensitive operations use breadth-first order, level by level and left-to-right following
20/// input and child generation order.
21///
22/// [`IterationOrder::Ordered`]: crate::IterationOrder::Ordered
23///
24/// Related traits:
25/// - [`ParUse`](crate::ParUse) for worker-local mutable state,
26/// - [`ParOption`](crate::ParOption) for `Option`-based fallibility,
27/// - [`ParResult`](crate::ParResult) for `Result`-based fallibility.
28///
29/// # Examples
30///
31/// ```
32/// use orx_parallel::*;
33///
34/// // A small rooted tree represented as adjacency lists; node 0 is the root.
35/// let children: Vec<Vec<usize>> = vec![vec![1, 2], vec![3, 4], vec![5], vec![], vec![], vec![]];
36///
37/// let sum_of_even_squares: usize = par_recursive([0usize], |node| children[*node].iter().copied())
38///     .map(|x| x * x)
39///     .filter(|x| x % 2 == 0)
40///     .sum();
41///
42/// assert_eq!(sum_of_even_squares, 20);
43/// ```
44pub trait ParRec: Sized + ParRecCore {
45    // configuration
46
47    /// Replaces the current parallel runner with `runner`.
48    ///
49    /// This allows per-computation control over execution strategy.
50    ///
51    /// Please see [`Runner`] for parallel runners implemented in this crate.
52    ///
53    /// [`Runner`]: crate::Runner
54    ///
55    /// # Examples
56    ///
57    /// ```
58    /// use orx_parallel::*;
59    ///
60    /// let children: Vec<Vec<usize>> = vec![vec![1, 2], vec![3, 4], vec![5], vec![], vec![], vec![]];
61    ///
62    /// let baseline: usize = par_recursive([0usize], |node| children[*node].iter().copied()).sum();
63    ///
64    /// let par = par_recursive([0usize], |node| children[*node].iter().copied());
65    ///
66    /// let par = par.runner(Runner::fixed());
67    ///
68    /// let configured: usize = par.sum();
69    /// assert_eq!(baseline, configured);
70    /// ```
71    fn runner<Q: ParRunner>(
72        self,
73        runner: Q,
74    ) -> impl ParRec<Item = Self::Item, Xap = Self::Xap, Input = Self::Input>;
75
76    /// Wraps the current parallel runner with a diagnostics-enabled runner.
77    ///
78    /// The returned iterator behaves the same, but additionally reports runtime
79    /// diagnostics at the end of the computation.
80    ///
81    /// # Examples
82    ///
83    /// ```
84    /// # #[cfg(feature = "std")]
85    /// # fn main() {
86    /// use orx_parallel::*;
87    ///
88    /// let par = par_recursive([1i32], |&x| (x < 10_000).then_some(x + 1))
89    ///     .num_threads(4);
90    ///
91    /// #[cfg(feature = "std")]
92    /// let par = par.runner_with_diagnostics();
93    ///
94    /// let sum = par.sum::<i32>();
95    /// assert_eq!(sum, 50005000);
96    /// # }
97    /// ```
98    ///
99    /// This will print a summary report which currently looks like the following:
100    ///
101    /// ```console
102    /// │ # Parallel Executor Diagnostics
103    /// │
104    /// │   Available threads : 4
105    /// │   Used threads      : 4
106    /// │   Wall time         : 1.15 ms
107    /// │
108    /// │ ## Summary Table
109    /// │   thread  num_chunks   num_tasks  min_chunk  avg_chunk  max_chunk    util%
110    /// │   ------  ----------  ----------  ---------  ---------  ---------  -------
111    /// │        0          35       27335        781        781        781   100.0%
112    /// │        1          32       24992        781        781        781    91.5%
113    /// │        2          30       23430        781        781        781    85.9%
114    /// │        3          28       21868        781        781        781    77.8%
115    /// │
116    /// │ ## Workload Balance
117    /// │   max/min task ratio  : 1.25x  (1.00 = perfect balance)
118    /// │   coeff. of variation : 8.3%  (lower is better)
119    /// │
120    /// │ ## Thread Active Timeline  (each block ≈ 0.02 ms)
121    /// │   [ 0] ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇
122    /// │   [ 1]     ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇
123    /// │   [ 2]         ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇
124    /// │   [ 3]             ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇
125    /// │
126    /// │ ## Thread Task Distribution  (bar length ∝ tasks processed)
127    /// │   [ 0] ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇  (27335)
128    /// │   [ 1] ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇  (24992)
129    /// │   [ 2] ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇  (23430)
130    /// │   [ 3] ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇  (21868)
131    /// ```
132    #[cfg(feature = "std")]
133    fn runner_with_diagnostics(
134        self,
135    ) -> impl ParRec<Item = Self::Item, Xap = Self::Xap, Input = Self::Input>;
136
137    /// Sets the maximum number of worker threads for this computation.
138    ///
139    /// This method configures the **computation layer** of the thread count decision.
140    /// The actual number of threads used is determined by combining:
141    ///
142    /// 1. **Pool constraint** (from `pool()` method or default pool)
143    ///    - Already includes `ORX_NUM_THREADS` environment variable constraint
144    /// 2. **Computation constraint** (this method)
145    ///    - Your per-computation thread preference
146    /// 3. **Input size constraint**
147    ///    - Cannot spawn more threads than input elements
148    ///
149    /// The actual thread count is the **minimum** of all these constraints.
150    ///
151    /// # Parameter Interpretation
152    ///
153    /// Integer values map as follows:
154    /// - `0` => `NumThreads::Auto` (use all available threads, spawn only as needed)
155    /// - `n > 0` => `NumThreads::Max(n)` (cap at `n` threads)
156    ///
157    /// # Thread Count Decision Logic
158    ///
159    /// ```text
160    /// available = pool.max_num_threads()      // Pool maximum (includes env variable)
161    ///
162    /// requested = match num_threads {
163    ///     0 | Auto => input_size.max(1),      // Limited by input size
164    ///     Max(n) => min(input_size, n),       // Limited by input size and this param
165    /// };
166    ///
167    /// actual_threads = min(requested, available)
168    /// ```
169    ///
170    /// # Examples
171    ///
172    /// ```ignore
173    /// use orx_parallel::*;
174    ///
175    /// // Sequential execution
176    /// let sum: usize = par_recursive([1usize], |&x| (x < 10).then_some(x + 1))
177    ///     .num_threads(1)
178    ///     .sum();
179    /// assert_eq!(sum, 55);
180    ///
181    /// // Cap at 4 threads
182    /// let sum: usize = par_recursive([1usize], |&x| (x < 1000).then_some(x + 1))
183    ///     .num_threads(4)
184    ///     .sum();
185    ///
186    /// // Auto: uses available threads (respects ORX_NUM_THREADS)
187    /// let sum: usize = par_recursive([1usize], |&x| (x < 10).then_some(x + 1))
188    ///     .num_threads(0)
189    ///     .sum();
190    /// ```
191    ///
192    /// # See Also
193    ///
194    /// - [`NumThreads`](crate::NumThreads) - Type for thread configuration
195    /// - [`thread_usage.md`](https://github.com/orxfun/orx-parallel/blob/main/docs/thread_usage.md) - Complete threading guide
196    fn num_threads(self, num_threads: impl Into<NumThreads>) -> Self;
197
198    /// Sets chunk size used when pulling items from the concurrent input.
199    ///
200    /// Integer values map as follows:
201    /// - `0` => automatic (default)
202    /// - `n > 0` => exact chunk size `n`
203    ///
204    /// # Examples
205    ///
206    /// ```
207    /// use orx_parallel::*;
208    ///
209    /// let values: Vec<_> = par_recursive([0usize], |&x| (x < 31).then_some(x + 1))
210    ///     .chunk_size(8)
211    ///     .map(|x| x + 1)
212    ///     .collect();
213    ///
214    /// assert_eq!(values.len(), 32);
215    /// assert_eq!(values[0], 1);
216    /// assert_eq!(values[31], 32);
217    /// ```
218    ///
219    /// # Rules of Thumb
220    ///
221    /// * Automatic chunk size (default) is efficient in general.
222    ///   Parallel runner aims to find best chunk sizes to balance between minimizing parallelization overhead
223    ///   and maximizing resource utilization.
224    /// * While tuning a specific computation, we aim to find the smallest chunk size that is large enough
225    ///   to mitigate the impact of parallelization overhead.
226    /// * If the individual tasks are large enough, parallelization overhead becomes insignificant making
227    ///   `chunk_size = 1` the optimal choice.
228    fn chunk_size(self, chunk_size: impl Into<ChunkSize>) -> Self;
229
230    /// Sets iteration order semantics for operations sensitive to ordering.
231    ///
232    /// `Ordered` (default) preserves positional meaning (for example, `first` returns the
233    /// earliest matching element in input order). `Arbitrary` allows any matching
234    /// element that is reached first in parallel execution.
235    ///
236    /// # Examples
237    ///
238    /// ```
239    /// use orx_parallel::*;
240    ///
241    /// let ordered = par_recursive([1i32], |&x| (x < 9_999).then_some(x + 1))
242    ///     .iteration_order(IterationOrder::Ordered)
243    ///     .find(|x| x % 3421 == 0);
244    /// assert_eq!(ordered, Some(3421));
245    ///
246    /// let any = par_recursive([1i32], |&x| (x < 9_999).then_some(x + 1))
247    ///     .iteration_order(IterationOrder::Arbitrary)
248    ///     .find(|x| x % 3421 == 0)
249    ///     .unwrap();
250    /// assert!([3421, 6842].contains(&any));
251    /// ```
252    fn iteration_order(self, collect: IterationOrder) -> Self;
253
254    // transformations
255
256    /// Maps each element with closure `h`.
257    ///
258    /// # Examples
259    ///
260    /// ```
261    /// use orx_parallel::*;
262    ///
263    /// let doubled: Vec<_> = par_recursive([1i32], |&x| (x < 3).then_some(x + 1))
264    ///     .map(|x| 2 * x)
265    ///     .collect();
266    /// assert_eq!(doubled, vec![2, 4, 6]);
267    /// ```
268    fn map<Q, H>(
269        self,
270        h: H,
271    ) -> impl ParRec<Item = Q, Xap = MapOf<Self::Xap, Q, H>, Input = Self::Input>
272    where
273        H: Fn(Self::Item) -> Q + Copy + Send;
274
275    /// Runs `h` on each element and forwards the item unchanged.
276    ///
277    /// Useful for logging or debugging pipelines.
278    ///
279    /// # Examples
280    ///
281    /// ```
282    /// use orx_parallel::*;
283    ///
284    /// let out: Vec<_> = par_recursive([1i32], |&x| (x < 4).then_some(x + 1))
285    ///     .inspect(|x| {
286    ///         println!("observed {x}");
287    ///     })
288    ///     .collect();
289    ///
290    /// assert_eq!(out, vec![1, 2, 3, 4]);
291    /// ```
292    fn inspect<H>(
293        self,
294        h: H,
295    ) -> impl ParRec<Item = Self::Item, Xap = InsOf<Self::Xap, H>, Input = Self::Input>
296    where
297        H: Fn(&Self::Item) + Copy + Send;
298
299    /// Keeps only elements satisfying predicate `h`.
300    ///
301    /// # Examples
302    ///
303    /// ```
304    /// use orx_parallel::*;
305    ///
306    /// let odds: Vec<_> = par_recursive([1i32], |&x| (x < 6).then_some(x + 1))
307    ///     .filter(|x| x % 2 == 1)
308    ///     .collect();
309    /// assert_eq!(odds, vec![1, 3, 5]);
310    /// ```
311    fn filter<H>(
312        self,
313        h: H,
314    ) -> impl ParRec<Item = Self::Item, Xap = FilOf<Self::Xap, H>, Input = Self::Input>
315    where
316        H: Fn(&Self::Item) -> bool + Copy + Send;
317
318    /// Maps and filters in a single pass.
319    ///
320    /// Returns mapped values for elements where `h` returns `Some(_)`.
321    ///
322    /// # Examples
323    ///
324    /// ```
325    /// use orx_parallel::*;
326    ///
327    /// let numbers: Vec<_> = par_recursive(["1", "x", "5"], |_: &&str| None::<&str>)
328    ///     .filter_map(|s| s.parse::<usize>().ok())
329    ///     .collect();
330    ///
331    /// assert_eq!(numbers, vec![1, 5]);
332    /// ```
333    fn filter_map<Q, H>(
334        self,
335        h: H,
336    ) -> impl ParRec<Item = Q, Xap = FilMapOf<Self::Xap, Q, H>, Input = Self::Input>
337    where
338        H: Fn(Self::Item) -> Option<Q> + Copy + Send;
339
340    /// Maps each element to an iterator and flattens one level.
341    ///
342    /// # Examples
343    ///
344    /// ```
345    /// use orx_parallel::*;
346    ///
347    /// let out: Vec<_> = par_recursive([1i32], |&x| (x < 3).then_some(x + 1))
348    ///     .flat_map(|x| [x, x + 10])
349    ///     .collect();
350    /// assert_eq!(out, vec![1, 11, 2, 12, 3, 13]);
351    /// ```
352    fn flat_map<V, H>(
353        self,
354        h: H,
355    ) -> impl ParRec<Item = V::Item, Xap = FlatMapOf<Self::Xap, V, H>, Input = Self::Input>
356    where
357        V: IntoIterator,
358        H: Fn(Self::Item) -> V + Copy + Send;
359
360    /// Flattens one level of nested iterables.
361    ///
362    /// # Examples
363    ///
364    /// ```
365    /// use orx_parallel::*;
366    ///
367    /// let nested = vec![vec![1, 2], vec![3, 4]];
368    /// let mut flat: Vec<_> = par_recursive(nested, |_: &Vec<i32>| None::<Vec<i32>>)
369    ///     .flatten()
370    ///     .collect();
371    /// flat.sort();
372    ///
373    /// assert_eq!(flat, vec![1, 2, 3, 4]);
374    /// ```
375    fn flatten(
376        self,
377    ) -> impl ParRec<
378        Item = <Self::Item as IntoIterator>::Item,
379        Xap = FlattenOf<Self::Xap>,
380        Input = Self::Input,
381    >
382    where
383        Self::Item: IntoIterator;
384
385    // compute
386
387    /// Returns an item, or `None` if empty.
388    ///
389    /// When [`IterationOrder::Ordered`] (default) is set, returns the first item in deterministic
390    /// breadth-first order (level by level, left-to-right following input and child generation order).
391    ///
392    /// Setting [`IterationOrder::Arbitrary`] may provide speed improvements when ordering is not
393    /// important; however, ordered traversal is also optimized so the performance difference
394    /// is generally small.
395    ///
396    /// This operation is short-circuiting: once a first candidate is determined,
397    /// remaining work is cancelled.
398    ///
399    /// [`IterationOrder::Ordered`]: crate::IterationOrder::Ordered
400    /// [`IterationOrder::Arbitrary`]: crate::IterationOrder::Arbitrary
401    ///
402    /// # Examples
403    ///
404    /// ```
405    /// use orx_parallel::*;
406    ///
407    /// let empty = par_recursive(Vec::<usize>::new(), |_: &usize| None::<usize>).first();
408    /// assert_eq!(empty, None);
409    ///
410    /// let first = par_recursive([1usize], |&x| (x < 3).then_some(x + 1))
411    ///     .first();
412    /// assert_eq!(first, Some(1));
413    /// ```
414    fn first(self) -> Option<Self::Item>
415    where
416        Self::Item: Send,
417        <Self::Input as IntoIterator>::Item: Send;
418
419    /// Reduces items into one value using associative reducer `f`.
420    ///
421    /// Returns `None` for an empty iterator.
422    ///
423    /// # Examples
424    ///
425    /// ```
426    /// use orx_parallel::*;
427    ///
428    /// let reduced = par_recursive([1i32], |&x| (x < 5).then_some(x + 1))
429    ///     .reduce(|a, b| a + b);
430    /// assert_eq!(reduced, Some(15));
431    /// ```
432    fn reduce<F>(self, f: F) -> Option<Self::Item>
433    where
434        F: Fn(Self::Item, Self::Item) -> Self::Item + Send + Copy,
435        Self::Item: Send,
436        <Self::Input as IntoIterator>::Item: Send;
437
438    /// Collects all items into `dst`.
439    ///
440    /// When [`IterationOrder::Ordered`] (default) is set, items are collected in a deterministic
441    /// breadth-first order (level by level, left-to-right following input and child generation order).
442    ///
443    /// Setting [`IterationOrder::Arbitrary`] may provide speed improvements when ordering is not
444    /// important; however, ordered collection is also optimized so the performance difference
445    /// is generally small.
446    ///
447    /// [`IterationOrder::Ordered`]: crate::IterationOrder::Ordered
448    /// [`IterationOrder::Arbitrary`]: crate::IterationOrder::Arbitrary
449    ///
450    /// # Examples
451    ///
452    /// ```
453    /// use orx_parallel::*;
454    ///
455    /// let mut dst = vec![10];
456    /// par_recursive([0i32], |&x| (x < 2).then_some(x + 1))
457    ///     .collect_into(&mut dst);
458    /// assert_eq!(dst, vec![10, 0, 1, 2]);
459    /// ```
460    fn collect_into<P>(self, dst: &mut P)
461    where
462        P: ParExtend<Self::Item>,
463        Self::Item: Send,
464        <Self::Input as IntoIterator>::Item: Send;
465
466    /// Collects all items into a new collection.
467    ///
468    /// When [`IterationOrder::Ordered`] (default) is set, items are collected in a deterministic
469    /// breadth-first order (level by level, left-to-right following input and child generation order).
470    ///
471    /// Setting [`IterationOrder::Arbitrary`] may provide speed improvements when ordering is not
472    /// important; however, ordered collection is also optimized so the performance difference
473    /// is generally small.
474    ///
475    /// [`IterationOrder::Ordered`]: crate::IterationOrder::Ordered
476    /// [`IterationOrder::Arbitrary`]: crate::IterationOrder::Arbitrary
477    ///
478    /// # Examples
479    ///
480    /// ```
481    /// use orx_parallel::*;
482    ///
483    /// let out: Vec<_> = par_recursive([1i32], |&x| (x < 3).then_some(x + 1))
484    ///     .map(|x| x * 2)
485    ///     .collect();
486    /// assert_eq!(out, vec![2, 4, 6]);
487    /// ```
488    fn collect<P>(self) -> P
489    where
490        P: ParExtend<Self::Item> + Default,
491        Self::Item: Send,
492        <Self::Input as IntoIterator>::Item: Send,
493    {
494        let mut dst = P::default();
495        self.collect_into(&mut dst);
496        dst
497    }
498
499    // compute - derived
500
501    /// Returns `true` if all items satisfy predicate `f`.
502    ///
503    /// Empty iterators return `true`.
504    ///
505    /// This operation is short-circuiting: evaluation stops as soon as one item
506    /// fails the predicate.
507    ///
508    /// # Examples
509    ///
510    /// ```
511    /// use orx_parallel::*;
512    ///
513    /// assert!(par_recursive([1i32], |&x| (x < 4).then_some(x + 1))
514    ///     .all(|x| x > &0));
515    /// assert!(!par_recursive([1i32], |&x| (x < 4).then_some(x + 1))
516    ///     .all(|x| x % 2 == 0));
517    /// ```
518    fn all<F>(self, f: F) -> bool
519    where
520        F: Fn(&Self::Item) -> bool + Copy + Send,
521        <Self::Input as IntoIterator>::Item: Send,
522    {
523        self.map(move |x| f(&x)).find(|x| !*x).is_none()
524    }
525
526    /// Returns `true` if any item satisfies predicate `f`.
527    ///
528    /// Empty iterators return `false`.
529    ///
530    /// This operation is short-circuiting: evaluation stops as soon as one item
531    /// satisfies the predicate.
532    ///
533    /// # Examples
534    ///
535    /// ```
536    /// use orx_parallel::*;
537    ///
538    /// assert!(par_recursive([1i32], |&x| (x < 4).then_some(x + 1))
539    ///     .any(|x| x % 2 == 0));
540    /// assert!(!par_recursive([1i32], |&x| (x < 4).then_some(x + 1))
541    ///     .any(|x| x > &10));
542    /// ```
543    fn any<F>(self, f: F) -> bool
544    where
545        F: Fn(&Self::Item) -> bool + Copy + Send,
546        <Self::Input as IntoIterator>::Item: Send,
547    {
548        self.map(move |x| f(&x)).find(|x| *x).is_some()
549    }
550
551    /// Counts elements.
552    ///
553    /// # Examples
554    ///
555    /// ```
556    /// use orx_parallel::*;
557    ///
558    /// let n = par_recursive([1i32], |&x| (x < 10).then_some(x + 1))
559    ///     .filter(|x| x % 3 == 0)
560    ///     .count();
561    /// assert_eq!(n, 3);
562    /// ```
563    fn count(self) -> usize
564    where
565        <Self::Input as IntoIterator>::Item: Send,
566    {
567        self.map(|_| 1).reduce(|a, b| a + b).unwrap_or(0)
568    }
569
570    /// Finds the first item satisfying predicate `f`, or `None` if none match.
571    ///
572    /// When [`IterationOrder::Ordered`] (default) is set, returns the first matching item in
573    /// deterministic breadth-first order (level by level, left-to-right following input and child
574    /// generation order).
575    ///
576    /// Setting [`IterationOrder::Arbitrary`] may provide speed improvements when ordering is not
577    /// important; however, ordered traversal is also optimized so the performance difference
578    /// is generally small.
579    ///
580    /// This is equivalent to `self.filter(f).first()`.
581    ///
582    /// This operation is short-circuiting: once a matching item is found,
583    /// remaining work is cancelled.
584    ///
585    /// [`IterationOrder::Ordered`]: crate::IterationOrder::Ordered
586    /// [`IterationOrder::Arbitrary`]: crate::IterationOrder::Arbitrary
587    ///
588    /// # Examples
589    ///
590    /// ```
591    /// use orx_parallel::*;
592    ///
593    /// let found = par_recursive([1i32], |&x| (x < 100).then_some(x + 1))
594    ///     .find(|x| x % 17 == 0);
595    /// assert_eq!(found, Some(17));
596    /// ```
597    fn find<F>(self, f: F) -> Option<Self::Item>
598    where
599        Self::Item: Send,
600        F: Fn(&Self::Item) -> bool + Copy + Send,
601        <Self::Input as IntoIterator>::Item: Send,
602    {
603        self.filter(f).first()
604    }
605
606    /// Folds elements into per-thread accumulators and returns them.
607    ///
608    /// The output contains one accumulator for each participating worker.
609    ///
610    /// # Examples
611    ///
612    /// ```
613    /// use orx_parallel::*;
614    ///
615    /// let partials: Vec<usize> = par_recursive([1usize], |&x| (x < 5).then_some(x + 1))
616    ///     .num_threads(2)
617    ///     .fold(|| 0usize, |acc, x| *acc += x);
618    ///
619    /// assert!(!partials.is_empty());
620    ///
621    /// assert_eq!(partials.iter().sum::<usize>(), 15);
622    /// ```
623    fn fold<B, I, F>(self, init: I, f: F) -> Vec<B>
624    where
625        B: Send,
626        I: Fn() -> B,
627        F: Fn(&mut B, Self::Item) + Copy + Send,
628        <Self::Input as IntoIterator>::Item: Send;
629
630    /// Executes `f` for each item.
631    ///
632    /// # Examples
633    ///
634    /// ```
635    /// use core::sync::atomic::{AtomicUsize, Ordering};
636    /// use orx_parallel::*;
637    ///
638    /// let total = AtomicUsize::new(0);
639    ///
640    /// par_recursive([1usize], |&x| (x < 4).then_some(x + 1))
641    ///     .for_each(|x| {
642    ///         total.fetch_add(x, Ordering::Relaxed);
643    ///     });
644    ///
645    /// assert_eq!(total.load(Ordering::Relaxed), 10);
646    /// ```
647    fn for_each<F>(self, f: F)
648    where
649        F: Fn(Self::Item) + Send + Copy,
650        <Self::Input as IntoIterator>::Item: Send,
651    {
652        let _ = self.map(f).reduce(|_, _| {});
653    }
654
655    /// Returns maximum element, or `None` if empty.
656    ///
657    /// # Examples
658    ///
659    /// ```
660    /// use orx_parallel::*;
661    ///
662    /// let max = par_recursive([1i32], |&x| (x < 4).then_some(x + 1))
663    ///     .max();
664    /// assert_eq!(max, Some(4));
665    ///
666    /// let empty = par_recursive(Vec::<usize>::new(), |_: &usize| None::<usize>)
667    ///     .max();
668    /// assert_eq!(empty, None);
669    /// ```
670    fn max(self) -> Option<Self::Item>
671    where
672        Self::Item: Ord + Send,
673        <Self::Input as IntoIterator>::Item: Send,
674    {
675        self.reduce(Ord::max)
676    }
677
678    /// Returns element considered maximum by comparator `f`.
679    ///
680    /// # Examples
681    ///
682    /// ```
683    /// use orx_parallel::*;
684    ///
685    /// let x = par_recursive(vec![-3_i32, 0, 1, 5, -10], |_: &i32| None::<i32>)
686    ///     .max_by(|a, b| a.cmp(b));
687    /// assert_eq!(x, Some(5));
688    /// ```
689    fn max_by<F>(self, f: F) -> Option<Self::Item>
690    where
691        Self::Item: Send,
692        F: Fn(&Self::Item, &Self::Item) -> Ordering + Copy + Send,
693        <Self::Input as IntoIterator>::Item: Send,
694    {
695        let reduce = move |x, y| match f(&x, &y) {
696            Ordering::Greater | Ordering::Equal => x,
697            Ordering::Less => y,
698        };
699        self.reduce(reduce)
700    }
701
702    /// Returns element with maximum key value.
703    ///
704    /// # Examples
705    ///
706    /// ```
707    /// use orx_parallel::*;
708    ///
709    /// let x = par_recursive(vec![-3_i32, 0, 1, 5, -10], |_: &i32| None::<i32>)
710    ///     .max_by_key(|x| x.abs());
711    /// assert_eq!(x, Some(-10));
712    /// ```
713    fn max_by_key<B, F>(self, f: F) -> Option<Self::Item>
714    where
715        Self::Item: Send,
716        B: Ord,
717        F: Fn(&Self::Item) -> B + Copy + Send,
718        <Self::Input as IntoIterator>::Item: Send,
719    {
720        let reduce = move |x, y| match f(&x).cmp(&f(&y)) {
721            Ordering::Greater | Ordering::Equal => x,
722            Ordering::Less => y,
723        };
724        self.reduce(reduce)
725    }
726
727    /// Returns minimum element, or `None` if empty.
728    ///
729    /// # Examples
730    ///
731    /// ```
732    /// use orx_parallel::*;
733    ///
734    /// let min = par_recursive([1i32], |&x| (x < 4).then_some(x + 1))
735    ///     .min();
736    /// assert_eq!(min, Some(1));
737    ///
738    /// let empty = par_recursive(Vec::<usize>::new(), |_: &usize| None::<usize>)
739    ///     .min();
740    /// assert_eq!(empty, None);
741    /// ```
742    fn min(self) -> Option<Self::Item>
743    where
744        Self::Item: Ord + Send,
745        <Self::Input as IntoIterator>::Item: Send,
746    {
747        self.reduce(Ord::min)
748    }
749
750    /// Returns element considered minimum by comparator `f`.
751    ///
752    /// # Examples
753    ///
754    /// ```
755    /// use orx_parallel::*;
756    ///
757    /// let x = par_recursive(vec![-3_i32, 0, 1, 5, -10], |_: &i32| None::<i32>)
758    ///     .min_by(|a, b| a.cmp(b));
759    /// assert_eq!(x, Some(-10));
760    /// ```
761    fn min_by<F>(self, f: F) -> Option<Self::Item>
762    where
763        Self::Item: Send,
764        F: Fn(&Self::Item, &Self::Item) -> Ordering + Copy + Send,
765        <Self::Input as IntoIterator>::Item: Send,
766    {
767        let reduce = move |x, y| match f(&x, &y) {
768            Ordering::Less | Ordering::Equal => x,
769            Ordering::Greater => y,
770        };
771        self.reduce(reduce)
772    }
773
774    /// Returns element with minimum key value.
775    ///
776    /// # Examples
777    ///
778    /// ```
779    /// use orx_parallel::*;
780    ///
781    /// let x = par_recursive(vec![-3_i32, 0, 1, 5, -10], |_: &i32| None::<i32>)
782    ///     .min_by_key(|x| x.abs());
783    /// assert_eq!(x, Some(0));
784    /// ```
785    fn min_by_key<B, F>(self, f: F) -> Option<Self::Item>
786    where
787        Self::Item: Send,
788        B: Ord,
789        F: Fn(&Self::Item) -> B + Copy + Send,
790        <Self::Input as IntoIterator>::Item: Send,
791    {
792        let reduce = move |x, y| match f(&x).cmp(&f(&y)) {
793            Ordering::Less | Ordering::Equal => x,
794            Ordering::Greater => y,
795        };
796        self.reduce(reduce)
797    }
798
799    /// Sums elements using [`Sum`] implementation of the item type.
800    ///
801    /// Empty iterators return additive identity (`zero`).
802    ///
803    /// # Examples
804    ///
805    /// ```
806    /// use orx_parallel::*;
807    ///
808    /// let sum: usize = par_recursive([1usize], |&x| (x < 4).then_some(x + 1))
809    ///     .sum();
810    /// assert_eq!(sum, 10);
811    /// ```
812    fn sum<S>(self) -> S
813    where
814        Self::Item: Sum<S>,
815        S: Send,
816        <Self::Input as IntoIterator>::Item: Send,
817    {
818        self.map(Self::Item::owned)
819            .reduce(Self::Item::add)
820            .unwrap_or(Self::Item::zero())
821    }
822}