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}