orx_parallel/result/par.rs
1#![allow(clippy::type_complexity)]
2
3use crate::infallible::fun::{FnCloned, FnCopied};
4use crate::infallible::{FilMapOf, FilOf, FlatMapOf, FlattenOf, InsOf, MapOf, MappedOf, Xap};
5use crate::infallible_use::xap_variants::IdUse;
6use crate::result::ParResultIter;
7use crate::result::par_core::ParResultCore;
8use crate::result_use::ParUseResultIter;
9use crate::runner::ParRunner;
10use crate::sizes::{OneOne, SizePair};
11use crate::use_var::{UseSlice, UseVec};
12use crate::{ChunkSize, IterationOrder, NumThreads, ParExtend, ParUseResult, Sum};
13use alloc::vec::Vec;
14use core::cmp::Ordering;
15use orx_concurrent_iter::ExactSizeConcurrentIter;
16
17/// Fallible parallel iterator over `Result` values.
18///
19/// `ParResult` represents pipelines where each element may fail with an error.
20/// It is commonly created from [`Par`](crate::Par) with
21/// [`into_fallible`](crate::Par::into_fallible).
22///
23/// Conceptually, this is similar to using the `?` operator in Rust:
24/// both let you keep logic on the success path while failures short-circuit.
25/// In `ParResult`, the success path works with plain `T` values (instead of
26/// `Result<T, E>`), and the parallel computation stops when an error is
27/// observed.
28///
29/// Related traits:
30/// - [`Par`](crate::Par) for infallible pipelines,
31/// - [`ParUseResult`](crate::ParUseResult) for the same fallibility model with worker-local state.
32///
33/// # Examples
34///
35/// Parse and validate records in parallel.
36///
37/// ```
38/// use orx_parallel::*;
39///
40/// let records = ["3", "8", "21", "34"];
41///
42/// let validated: Result<Vec<usize>, _> = records
43/// .into_par()
44/// .map(|s| s.parse::<usize>())
45/// .into_fallible()
46/// .map(|x| x * 2)
47/// .filter(|x| *x <= 70)
48/// .collect();
49///
50/// assert_eq!(validated, Ok(vec![6, 16, 42, 68]));
51///
52/// let with_failure: Result<Vec<usize>, _> = ["3", "bad", "21", "34"]
53/// .into_par()
54/// .map(|s| s.parse::<usize>())
55/// .into_fallible()
56/// .map(|x| x * 2)
57/// .collect();
58///
59/// assert!(with_failure.is_err());
60/// ```
61pub trait ParResult: Sized + ParResultCore {
62 // configuration
63
64 /// Replaces the current parallel runner with `runner`.
65 ///
66 /// # Examples
67 ///
68 /// ```
69 /// use orx_parallel::*;
70 ///
71 /// let par = ["1", "2", "3"]
72 /// .into_par()
73 /// .map(|s| s.parse::<usize>())
74 /// .into_fallible();
75 ///
76 /// let par = par.runner(Runner::fixed());
77 ///
78 /// let out: Result<Vec<_>, _> = par.collect();
79 /// assert_eq!(out, Ok(vec![1, 2, 3]));
80 /// ```
81 fn runner<Q: ParRunner>(
82 self,
83 runner: Q,
84 ) -> impl ParResult<
85 Elem = Self::Elem,
86 Error = Self::Error,
87 Xap1 = Self::Xap1,
88 M = Self::M,
89 Xap2 = Self::Xap2,
90 Input = Self::Input,
91 Size = Self::Size,
92 >;
93
94 #[cfg(feature = "std")]
95 /// Wraps the current runner with diagnostics-enabled execution.
96 ///
97 /// # Examples
98 ///
99 /// ```
100 /// use orx_parallel::*;
101 ///
102 /// let par = ["1", "2", "3"]
103 /// .into_par()
104 /// .map(|s| s.parse::<usize>())
105 /// .into_fallible();
106 ///
107 /// #[cfg(feature = "std")]
108 /// let par = par.runner_with_diagnostics();
109 ///
110 /// let out: Result<Vec<_>, _> = par.collect();
111 /// assert_eq!(out, Ok(vec![1, 2, 3]));
112 /// ```
113 fn runner_with_diagnostics(
114 self,
115 ) -> impl ParResult<
116 Elem = Self::Elem,
117 Error = Self::Error,
118 Xap1 = Self::Xap1,
119 M = Self::M,
120 Xap2 = Self::Xap2,
121 Input = Self::Input,
122 Size = Self::Size,
123 >;
124
125 /// Sets the maximum number of worker threads for this computation.
126 ///
127 /// This method configures the **computation layer** of the thread count decision.
128 /// The actual number of threads used is determined by combining:
129 ///
130 /// 1. **Pool constraint** (from `pool()` method or default pool)
131 /// - Already includes `ORX_NUM_THREADS` environment variable constraint
132 /// 2. **Computation constraint** (this method)
133 /// - Your per-computation thread preference
134 /// 3. **Input size constraint**
135 /// - Cannot spawn more threads than input elements
136 ///
137 /// The actual thread count is the **minimum** of all these constraints.
138 ///
139 /// # Parameter Interpretation
140 ///
141 /// Integer values map as follows:
142 /// - `0` => `NumThreads::Auto` (use all available threads, spawn only as needed)
143 /// - `n > 0` => `NumThreads::Max(n)` (cap at `n` threads)
144 ///
145 /// # Thread Count Decision Logic
146 ///
147 /// ```text
148 /// available = pool.max_num_threads() // Pool maximum (includes env variable)
149 ///
150 /// requested = match num_threads {
151 /// 0 | Auto => input_size.max(1), // Limited by input size
152 /// Max(n) => min(input_size, n), // Limited by input size and this param
153 /// };
154 ///
155 /// actual_threads = min(requested, available)
156 /// ```
157 ///
158 /// # Examples
159 ///
160 /// ```ignore
161 /// use orx_parallel::*;
162 ///
163 /// // Sequential execution
164 /// let out: Result<Vec<_>, _> = ["1", "2", "3"]
165 /// .into_par()
166 /// .map(|s| s.parse::<usize>())
167 /// .into_fallible()
168 /// .num_threads(1)
169 /// .collect();
170 /// assert_eq!(out, Ok(vec![1, 2, 3]));
171 ///
172 /// // Cap at 4 threads
173 /// let out: Result<Vec<_>, _> = (1..1001)
174 /// .into_par()
175 /// .map(Ok::<_, String>)
176 /// .into_fallible()
177 /// .num_threads(4)
178 /// .collect();
179 /// ```
180 ///
181 /// # See Also
182 ///
183 /// - [`NumThreads`](crate::NumThreads) - Type for thread configuration
184 /// - [`thread_usage.md`](https://github.com/orxfun/orx-parallel/blob/main/docs/thread_usage.md) - Complete threading guide
185 fn num_threads(self, num_threads: impl Into<NumThreads>) -> Self;
186
187 /// Sets chunk size used when pulling items from the concurrent input.
188 ///
189 /// # Examples
190 ///
191 /// ```
192 /// use orx_parallel::*;
193 ///
194 /// let out: Result<Vec<_>, _> = ["1", "2", "3", "4"]
195 /// .into_par()
196 /// .map(|s| s.parse::<usize>())
197 /// .into_fallible()
198 /// .chunk_size(2)
199 /// .collect();
200 ///
201 /// assert_eq!(out, Ok(vec![1, 2, 3, 4]));
202 /// ```
203 fn chunk_size(self, chunk_size: impl Into<ChunkSize>) -> Self;
204
205 /// Sets iteration-order semantics for order-sensitive operations.
206 ///
207 /// # Examples
208 ///
209 /// ```
210 /// use orx_parallel::*;
211 ///
212 /// let ordered = (1..10_000)
213 /// .into_par()
214 /// .map(Ok::<_, &'static str>)
215 /// .into_fallible()
216 /// .iteration_order(IterationOrder::Ordered)
217 /// .find(|x| x % 3421 == 0);
218 /// assert_eq!(ordered, Ok(Some(3421)));
219 /// ```
220 fn iteration_order(self, collect: IterationOrder) -> Self;
221
222 // kind transformations
223
224 /// Creates one mutable `Use` value per participating worker.
225 ///
226 /// # Examples
227 ///
228 /// ```
229 /// use orx_parallel::*;
230 /// use rand::prelude::*;
231 /// use rand_chacha::ChaCha8Rng;
232 ///
233 /// let out: Result<Vec<_>, _> = ["0", "1", "2", "3", "4", "5", "6", "7"]
234 /// .into_par()
235 /// .map(|s| s.parse::<usize>())
236 /// .into_fallible()
237 /// .use_new(|thread_idx| ChaCha8Rng::seed_from_u64(10 + thread_idx as u64))
238 /// .map(|rng, x| x + rng.random_range(0..10))
239 /// .collect();
240 ///
241 /// assert_eq!(out.as_ref().map(Vec::len), Ok(8));
242 /// ```
243 fn use_new<U, F>(
244 self,
245 f: F,
246 ) -> impl ParUseResult<
247 Elem = Self::Elem,
248 Error = Self::Error,
249 Use = U,
250 Xap1 = IdUse<Self::Xap1, U>,
251 M = Self::M,
252 Xap2 = IdUse<Self::Xap2, U>,
253 Input = Self::Input,
254 Size = Self::Size,
255 >
256 where
257 U: Send,
258 F: Fn(usize) -> U + Sync,
259 {
260 let (iter, x1, x2, exe, _, params) = self.destruct();
261 let x1 = IdUse::<_, U>::new(x1);
262 let x2 = IdUse::<_, U>::new(x2);
263 let u = UseVec::new(f);
264 ParUseResultIter::new(u, iter, x1, x2, exe, params)
265 }
266
267 /// Uses an externally-owned [`UseVec`](crate::UseVec) as worker-local state.
268 ///
269 /// # Examples
270 ///
271 /// ```
272 /// use orx_parallel::*;
273 ///
274 /// let mut sums = UseVec::new(|_| 0usize);
275 ///
276 /// let result = ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]
277 /// .into_par()
278 /// .map(|s| s.parse::<usize>())
279 /// .into_fallible()
280 /// .use_vec(&mut sums)
281 /// .for_each(|local, x| *local += x);
282 ///
283 /// assert_eq!(result, Ok(()));
284 /// assert_eq!(sums.into_vec().into_iter().sum::<usize>(), 55);
285 /// ```
286 fn use_vec<U, F>(
287 self,
288 use_vec: &mut UseVec<U, F>,
289 ) -> impl ParUseResult<
290 Elem = Self::Elem,
291 Error = Self::Error,
292 Use = U,
293 Xap1 = IdUse<Self::Xap1, U>,
294 M = Self::M,
295 Xap2 = IdUse<Self::Xap2, U>,
296 Input = Self::Input,
297 Size = Self::Size,
298 >
299 where
300 U: Send,
301 F: Fn(usize) -> U + Sync,
302 {
303 let (iter, x1, x2, exe, _, params) = self.destruct();
304 let x1 = IdUse::<_, U>::new(x1);
305 let x2 = IdUse::<_, U>::new(x2);
306 ParUseResultIter::new(use_vec, iter, x1, x2, exe, params)
307 }
308
309 /// Uses a caller-provided mutable slice as worker-local mutable state.
310 ///
311 /// # Examples
312 ///
313 /// ```
314 /// use orx_parallel::*;
315 ///
316 /// let mut sums = vec![0usize; 4];
317 /// let result = ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]
318 /// .into_par()
319 /// .map(|s| s.parse::<usize>())
320 /// .into_fallible()
321 /// .use_slice(&mut sums)
322 /// .for_each(|local, x| *local += x);
323 ///
324 /// assert_eq!(result, Ok(()));
325 /// assert_eq!(sums.into_iter().sum::<usize>(), 55);
326 /// ```
327 ///
328 /// # Panics
329 ///
330 /// Panics if the input produces at least one element but `slice` is empty.
331 fn use_slice<'a, U>(
332 self,
333 slice: &'a mut [U],
334 ) -> impl ParUseResult<
335 Elem = Self::Elem,
336 Error = Self::Error,
337 Use = U,
338 Xap1 = IdUse<Self::Xap1, U>,
339 M = Self::M,
340 Xap2 = IdUse<Self::Xap2, U>,
341 Input = Self::Input,
342 Size = Self::Size,
343 >
344 where
345 U: Send + 'a,
346 {
347 let (iter, x1, x2, exe, _, params) = self.destruct();
348 let x1 = IdUse::<_, U>::new(x1);
349 let x2 = IdUse::<_, U>::new(x2);
350 let u = UseSlice::new(slice);
351 ParUseResultIter::new(u, iter, x1, x2, exe, params)
352 }
353
354 /// Copies elements of a reference iterator on the success path.
355 ///
356 /// # Examples
357 ///
358 /// ```
359 /// use orx_parallel::*;
360 ///
361 /// let data = vec![1, 2, 3];
362 /// let out: Result<Vec<_>, &'static str> = data
363 /// .par()
364 /// .map(Ok::<_, &'static str>)
365 /// .into_fallible()
366 /// .copied()
367 /// .collect();
368 ///
369 /// assert_eq!(out, Ok(vec![1, 2, 3]));
370 /// ```
371 fn copied<'a, O>(
372 self,
373 ) -> impl ParResult<
374 Elem = O,
375 Error = Self::Error,
376 Xap1 = Self::Xap1,
377 M = Self::M,
378 Xap2 = MappedOf<Self::Xap2, FnCopied<'a, O>>,
379 Input = Self::Input,
380 Size = Self::Size,
381 >
382 where
383 Self: ParResult<Elem = &'a O>,
384 O: Copy + 'a,
385 {
386 let (iter, x1, x2, exe, _, params) = self.destruct();
387 ParResultIter::new(iter, x1, x2.mapped(FnCopied::new()), exe, params)
388 }
389
390 /// Clones elements of a reference iterator on the success path.
391 ///
392 /// # Examples
393 ///
394 /// ```
395 /// use orx_parallel::*;
396 ///
397 /// let data = vec!["a".to_string(), "b".to_string()];
398 /// let out: Result<Vec<_>, &'static str> = data
399 /// .par()
400 /// .map(Ok::<_, &'static str>)
401 /// .into_fallible()
402 /// .cloned()
403 /// .collect();
404 ///
405 /// assert_eq!(out, Ok(vec!["a".to_string(), "b".to_string()]));
406 /// ```
407 fn cloned<'a, O>(
408 self,
409 ) -> impl ParResult<
410 Elem = O,
411 Error = Self::Error,
412 Xap1 = Self::Xap1,
413 M = Self::M,
414 Xap2 = MappedOf<Self::Xap2, FnCloned<'a, O>>,
415 Input = Self::Input,
416 Size = Self::Size,
417 >
418 where
419 Self: ParResult<Elem = &'a O>,
420 O: Clone + 'a,
421 {
422 let (iter, x1, x2, exe, _, params) = self.destruct();
423 ParResultIter::new(iter, x1, x2.mapped(FnCloned::new()), exe, params)
424 }
425
426 // transformations
427
428 /// Maps each successful element with closure `h`.
429 ///
430 /// # Examples
431 ///
432 /// ```
433 /// use orx_parallel::*;
434 ///
435 /// let out: Result<Vec<_>, _> = ["1", "2", "3"]
436 /// .into_par()
437 /// .map(|s| s.parse::<usize>())
438 /// .into_fallible()
439 /// .map(|x| 2 * x)
440 /// .collect();
441 ///
442 /// assert_eq!(out, Ok(vec![2, 4, 6]));
443 /// ```
444 fn map<Q, H>(
445 self,
446 h: H,
447 ) -> impl ParResult<
448 Elem = Q,
449 Error = Self::Error,
450 Xap1 = Self::Xap1,
451 M = Self::M,
452 Xap2 = MapOf<Self::Xap2, Q, H>,
453 Input = Self::Input,
454 Size = Self::Size,
455 >
456 where
457 H: Fn(Self::Elem) -> Q + Copy + Send;
458
459 /// Runs `h` on each successful element and forwards it unchanged.
460 ///
461 /// # Examples
462 ///
463 /// ```
464 /// use core::sync::atomic::{AtomicUsize, Ordering};
465 /// use orx_parallel::*;
466 ///
467 /// let seen = AtomicUsize::new(0);
468 /// let out: Result<Vec<_>, _> = ["1", "2", "3", "4"]
469 /// .into_par()
470 /// .map(|s| s.parse::<usize>())
471 /// .into_fallible()
472 /// .inspect(|_| { seen.fetch_add(1, Ordering::Relaxed); })
473 /// .collect();
474 ///
475 /// assert_eq!(out, Ok(vec![1, 2, 3, 4]));
476 /// assert_eq!(seen.load(Ordering::Relaxed), 4);
477 /// ```
478 fn inspect<H>(
479 self,
480 h: H,
481 ) -> impl ParResult<
482 Elem = Self::Elem,
483 Error = Self::Error,
484 Xap1 = Self::Xap1,
485 M = Self::M,
486 Xap2 = InsOf<Self::Xap2, H>,
487 Input = Self::Input,
488 Size = Self::Size,
489 >
490 where
491 H: Fn(&Self::Elem) + Copy + Send;
492
493 /// Keeps successful elements satisfying predicate `h`.
494 ///
495 /// # Examples
496 ///
497 /// ```
498 /// use orx_parallel::*;
499 ///
500 /// let out: Result<Vec<_>, _> = ["1", "2", "3", "4", "5", "6"]
501 /// .into_par()
502 /// .map(|s| s.parse::<usize>())
503 /// .into_fallible()
504 /// .filter(|x| x % 2 == 1)
505 /// .collect();
506 ///
507 /// assert_eq!(out, Ok(vec![1, 3, 5]));
508 /// ```
509 fn filter<H>(
510 self,
511 h: H,
512 ) -> impl ParResult<
513 Elem = Self::Elem,
514 Error = Self::Error,
515 Xap1 = Self::Xap1,
516 M = Self::M,
517 Xap2 = FilOf<Self::Xap2, H>,
518 Input = Self::Input,
519 Size = <Self::Size as SizePair>::ThenBin,
520 >
521 where
522 H: Fn(&Self::Elem) -> bool + Copy + Send;
523
524 /// Maps and filters successful elements in a single pass.
525 ///
526 /// # Examples
527 ///
528 /// ```
529 /// use orx_parallel::*;
530 ///
531 /// let out: Result<Vec<_>, &'static str> = ["1", "x", "5"]
532 /// .into_par()
533 /// .map(Ok::<_, &'static str>)
534 /// .into_fallible()
535 /// .filter_map(|s| s.parse::<usize>().ok())
536 /// .collect();
537 ///
538 /// assert_eq!(out, Ok(vec![1, 5]));
539 /// ```
540 fn filter_map<Q, H>(
541 self,
542 h: H,
543 ) -> impl ParResult<
544 Elem = Q,
545 Error = Self::Error,
546 Xap1 = Self::Xap1,
547 M = Self::M,
548 Xap2 = FilMapOf<Self::Xap2, Q, H>,
549 Input = Self::Input,
550 Size = <Self::Size as SizePair>::ThenBin,
551 >
552 where
553 H: Fn(Self::Elem) -> Option<Q> + Copy + Send;
554
555 /// Maps each successful element to an iterator and flattens one level.
556 ///
557 /// # Examples
558 ///
559 /// ```
560 /// use orx_parallel::*;
561 ///
562 /// let out: Result<Vec<_>, _> = ["1", "2", "3"]
563 /// .into_par()
564 /// .map(|s| s.parse::<usize>())
565 /// .into_fallible()
566 /// .flat_map(|x| [x, x + 10])
567 /// .collect();
568 ///
569 /// assert_eq!(out, Ok(vec![1, 11, 2, 12, 3, 13]));
570 /// ```
571 fn flat_map<V, H>(
572 self,
573 h: H,
574 ) -> impl ParResult<
575 Elem = V::Item,
576 Error = Self::Error,
577 Xap1 = Self::Xap1,
578 M = Self::M,
579 Xap2 = FlatMapOf<Self::Xap2, V, H>,
580 Input = Self::Input,
581 Size = <Self::Size as SizePair>::ThenMany,
582 >
583 where
584 V: IntoIterator,
585 H: Fn(Self::Elem) -> V + Copy + Send;
586
587 /// Flattens one level of nested iterables on the success path.
588 ///
589 /// # Examples
590 ///
591 /// ```
592 /// use orx_parallel::*;
593 ///
594 /// let nested = vec![vec![1, 2], vec![3, 4]];
595 /// let out: Result<Vec<_>, &'static str> = nested
596 /// .into_par()
597 /// .map(Ok::<_, &'static str>)
598 /// .into_fallible()
599 /// .flatten()
600 /// .collect();
601 ///
602 /// assert_eq!(out, Ok(vec![1, 2, 3, 4]));
603 /// ```
604 fn flatten(
605 self,
606 ) -> impl ParResult<
607 Elem = <Self::Elem as IntoIterator>::Item,
608 Error = Self::Error,
609 Xap1 = Self::Xap1,
610 M = Self::M,
611 Xap2 = FlattenOf<Self::Xap2>,
612 Input = Self::Input,
613 Size = <Self::Size as SizePair>::ThenMany,
614 >
615 where
616 Self::Elem: IntoIterator;
617
618 /// Returns a lower and optional upper bound on the number of successful output items.
619 ///
620 /// The bounds follow the usual [`Iterator::size_hint`] convention.
621 /// The upper bound includes items that may be removed by filtering or short-circuiting.
622 ///
623 /// # Examples
624 ///
625 /// ```
626 /// use orx_parallel::*;
627 ///
628 /// let values = (0..4)
629 /// .into_par()
630 /// .map(Ok::<_, ()>)
631 /// .into_fallible()
632 /// .filter(|x| x % 2 == 0);
633 ///
634 /// assert_eq!(values.size_hint(), (0, Some(4)));
635 /// ```
636 fn size_hint(&self) -> (usize, Option<usize>);
637
638 /// Returns the exact number of output items.
639 fn len(&self) -> usize
640 where
641 Self::Input: ExactSizeConcurrentIter,
642 Self: ParResult<Size = OneOne>,
643 {
644 self.size_hint().0
645 }
646
647 /// Returns `true` when the parallel iterator has no output items.
648 fn is_empty(&self) -> bool
649 where
650 Self::Input: ExactSizeConcurrentIter,
651 Self: ParResult<Size = OneOne>,
652 {
653 self.len() == 0
654 }
655
656 // compute
657
658 /// Returns the first successful item according to iteration order.
659 ///
660 /// # Examples
661 ///
662 /// ```
663 /// use orx_parallel::*;
664 ///
665 /// assert_eq!(["1", "2", "3"].into_par().map(|s| s.parse::<usize>()).into_fallible().first(), Ok(Some(1)));
666 /// assert_eq!(Vec::<&str>::new().into_par().map(|s| s.parse::<usize>()).into_fallible().first(), Ok(None));
667 /// ```
668 fn first(self) -> Result<Option<Self::Elem>, Self::Error>
669 where
670 Self::Elem: Send,
671 Self::Error: Send;
672
673 /// Reduces successful items into one value using `f`.
674 ///
675 /// # Examples
676 ///
677 /// ```
678 /// use orx_parallel::*;
679 ///
680 /// let ok = (1..6)
681 /// .into_par()
682 /// .map(Ok::<_, &'static str>)
683 /// .into_fallible()
684 /// .reduce(|a, b| a + b);
685 /// assert_eq!(ok, Ok(Some(15)));
686 ///
687 /// let fail = ["1", "x", "3"]
688 /// .into_par()
689 /// .map(|s| s.parse::<usize>().map_err(|_| "parse"))
690 /// .into_fallible()
691 /// .reduce(|a, b| a + b);
692 /// assert_eq!(fail, Err("parse"));
693 /// ```
694 fn reduce<F>(self, f: F) -> Result<Option<Self::Elem>, Self::Error>
695 where
696 F: Fn(Self::Elem, Self::Elem) -> Self::Elem + Send + Copy,
697 Self::Elem: Send,
698 Self::Error: Send;
699
700 /// Collects successful items into `dst`.
701 ///
702 /// # Examples
703 ///
704 /// ```
705 /// use orx_parallel::*;
706 ///
707 /// let mut dst = vec![10usize];
708 /// let ok = ["0", "1", "2"]
709 /// .into_par()
710 /// .map(|s| s.parse::<usize>())
711 /// .into_fallible()
712 /// .collect_into(&mut dst);
713 /// assert_eq!(ok, Ok(()));
714 /// assert_eq!(dst, vec![10, 0, 1, 2]);
715 /// ```
716 fn collect_into<P>(self, dst: &mut P) -> Result<(), Self::Error>
717 where
718 P: ParExtend<Self::Elem>,
719 Self::Elem: Send,
720 Self::Error: Send;
721
722 /// Collects successful items into a new collection.
723 ///
724 /// # Examples
725 ///
726 /// ```
727 /// use orx_parallel::*;
728 ///
729 /// let out: Result<Vec<_>, _> = ["1", "2", "3"]
730 /// .into_par()
731 /// .map(|s| s.parse::<usize>())
732 /// .into_fallible()
733 /// .collect();
734 ///
735 /// assert_eq!(out, Ok(vec![1, 2, 3]));
736 /// ```
737 fn collect<P>(self) -> Result<P, Self::Error>
738 where
739 P: ParExtend<Self::Elem> + Default,
740 Self::Elem: Send,
741 Self::Error: Send,
742 {
743 let mut dst = P::default();
744 self.collect_into(&mut dst)?;
745 Ok(dst)
746 }
747
748 // compute - derived
749
750 /// Returns `Ok(true)` if all successful items satisfy `f`.
751 ///
752 /// Returns `Err(e)` on short-circuit failure.
753 ///
754 /// # Examples
755 ///
756 /// ```
757 /// use orx_parallel::*;
758 ///
759 /// let ok = ["1", "2", "3", "4"]
760 /// .into_par()
761 /// .map(|s| s.parse::<usize>())
762 /// .into_fallible()
763 /// .all(|x| x > &0);
764 /// assert_eq!(ok, Ok(true));
765 /// ```
766 fn all<F>(self, f: F) -> Result<bool, Self::Error>
767 where
768 Self::Elem: Send,
769 F: Fn(&Self::Elem) -> bool + Sync,
770 Self::Error: Send,
771 {
772 self.map(|x| f(&x))
773 .find(|x| !*x)
774 .map(|x| x.map(|_| false).unwrap_or(true))
775 }
776
777 /// Returns `Ok(true)` if any successful item satisfies `f`.
778 ///
779 /// Returns `Err(e)` on short-circuit failure.
780 ///
781 /// # Examples
782 ///
783 /// ```
784 /// use orx_parallel::*;
785 ///
786 /// let ok = ["1", "2", "3", "4"]
787 /// .into_par()
788 /// .map(|s| s.parse::<usize>())
789 /// .into_fallible()
790 /// .any(|x| x % 2 == 0);
791 /// assert_eq!(ok, Ok(true));
792 /// ```
793 fn any<F>(self, f: F) -> Result<bool, Self::Error>
794 where
795 Self::Elem: Send,
796 F: Fn(&Self::Elem) -> bool + Sync,
797 Self::Error: Send,
798 {
799 self.map(|x| f(&x)).find(|x| *x).map(|x| x.is_some())
800 }
801
802 /// Counts successful elements.
803 ///
804 /// Returns `Err(e)` on short-circuit failure.
805 ///
806 /// # Examples
807 ///
808 /// ```
809 /// use orx_parallel::*;
810 ///
811 /// let n = ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]
812 /// .into_par()
813 /// .map(|s| s.parse::<usize>())
814 /// .into_fallible()
815 /// .filter(|x| x % 3 == 0)
816 /// .count();
817 /// assert_eq!(n, Ok(3));
818 /// ```
819 fn count(self) -> Result<usize, Self::Error>
820 where
821 Self::Elem: Send,
822 Self::Error: Send,
823 {
824 self.map(|_| 1).reduce(|a, b| a + b).map(|x| x.unwrap_or(0))
825 }
826
827 /// Finds first (ordered) or any (arbitrary) successful item satisfying `f`.
828 ///
829 /// Returns `Err(e)` on short-circuit failure.
830 ///
831 /// # Examples
832 ///
833 /// ```
834 /// use orx_parallel::*;
835 ///
836 /// let numbers = (1..101).map(|x| x.to_string()).collect::<Vec<_>>();
837 /// let found = numbers
838 /// .into_par()
839 /// .map(|s| s.parse::<usize>())
840 /// .into_fallible()
841 /// .find(|x| x % 17 == 0);
842 /// assert_eq!(found, Ok(Some(17)));
843 /// ```
844 fn find<F>(self, f: F) -> Result<Option<Self::Elem>, Self::Error>
845 where
846 Self::Elem: Send,
847 F: Fn(&Self::Elem) -> bool + Sync,
848 Self::Error: Send,
849 {
850 self.filter(&f).first()
851 }
852
853 /// Folds successful elements into per-thread accumulators.
854 ///
855 /// Returns `Err(e)` on short-circuit failure.
856 ///
857 /// # Examples
858 ///
859 /// ```
860 /// use orx_parallel::*;
861 ///
862 /// let partials = ["1", "2", "3", "4", "5"]
863 /// .into_par()
864 /// .map(|s| s.parse::<usize>())
865 /// .into_fallible()
866 /// .fold(|| 0usize, |acc, x| *acc += x)
867 /// .unwrap();
868 ///
869 /// assert_eq!(partials.iter().sum::<usize>(), 15);
870 /// ```
871 fn fold<B, I, F>(self, init: I, f: F) -> Result<Vec<B>, Self::Error>
872 where
873 B: Send,
874 I: Fn() -> B + Sync,
875 F: Fn(&mut B, Self::Elem) + Copy + Send,
876 Self::Error: Send,
877 {
878 let mut use_vec = UseVec::new(|_| init());
879 let par_use = self.use_vec(&mut use_vec);
880 let result = par_use.for_each(move |u: &mut B, x| f(u, x));
881 result.map(|_| use_vec.into_vec())
882 }
883
884 /// Executes `f` for each successful element.
885 ///
886 /// Returns `Err(e)` on short-circuit failure.
887 ///
888 /// # Examples
889 ///
890 /// ```
891 /// use core::sync::atomic::{AtomicUsize, Ordering};
892 /// use orx_parallel::*;
893 ///
894 /// let total = AtomicUsize::new(0);
895 /// let ok = ["1", "2", "3", "4"]
896 /// .into_par()
897 /// .map(|s| s.parse::<usize>())
898 /// .into_fallible()
899 /// .for_each(|x| {
900 /// total.fetch_add(x, Ordering::Relaxed);
901 /// });
902 ///
903 /// assert_eq!(ok, Ok(()));
904 /// assert_eq!(total.load(Ordering::Relaxed), 10);
905 /// ```
906 fn for_each<F>(self, f: F) -> Result<(), Self::Error>
907 where
908 F: Fn(Self::Elem) + Send + Copy,
909 Self::Error: Send,
910 {
911 self.map(f).reduce(|_, _| {}).map(|_| ())
912 }
913
914 /// Returns maximum successful element.
915 ///
916 /// Returns `Err(e)` on short-circuit failure.
917 ///
918 /// # Examples
919 ///
920 /// ```
921 /// use orx_parallel::*;
922 ///
923 /// let m = ["1", "2", "3", "4"]
924 /// .into_par()
925 /// .map(|s| s.parse::<usize>())
926 /// .into_fallible()
927 /// .max();
928 /// assert_eq!(m, Ok(Some(4)));
929 /// ```
930 fn max(self) -> Result<Option<Self::Elem>, Self::Error>
931 where
932 Self::Elem: Ord + Send,
933 Self::Error: Send,
934 {
935 self.reduce(Ord::max)
936 }
937
938 /// Returns successful element considered maximum by comparator `f`.
939 ///
940 /// Returns `Err(e)` on short-circuit failure.
941 ///
942 /// # Examples
943 ///
944 /// ```
945 /// use orx_parallel::*;
946 ///
947 /// let x = vec![-3_i32, 0, 1, 5, -10]
948 /// .into_par()
949 /// .map(Ok::<_, &'static str>)
950 /// .into_fallible()
951 /// .max_by(|a, b| a.cmp(b));
952 /// assert_eq!(x, Ok(Some(5)));
953 /// ```
954 fn max_by<F>(self, f: F) -> Result<Option<Self::Elem>, Self::Error>
955 where
956 Self::Elem: Send,
957 F: Fn(&Self::Elem, &Self::Elem) -> Ordering + Sync,
958 Self::Error: Send,
959 {
960 let reduce = |x, y| match f(&x, &y) {
961 Ordering::Greater | Ordering::Equal => x,
962 Ordering::Less => y,
963 };
964 self.reduce(reduce)
965 }
966
967 /// Returns successful element with maximum key value.
968 ///
969 /// Returns `Err(e)` on short-circuit failure.
970 ///
971 /// # Examples
972 ///
973 /// ```
974 /// use orx_parallel::*;
975 ///
976 /// let x = vec![-3_i32, 0, 1, 5, -10]
977 /// .into_par()
978 /// .map(Ok::<_, &'static str>)
979 /// .into_fallible()
980 /// .max_by_key(|x| x.abs());
981 /// assert_eq!(x, Ok(Some(-10)));
982 /// ```
983 fn max_by_key<B, F>(self, f: F) -> Result<Option<Self::Elem>, Self::Error>
984 where
985 Self::Elem: Send,
986 B: Ord,
987 F: Fn(&Self::Elem) -> B + Sync,
988 Self::Error: Send,
989 {
990 let reduce = |x, y| match f(&x).cmp(&f(&y)) {
991 Ordering::Greater | Ordering::Equal => x,
992 Ordering::Less => y,
993 };
994 self.reduce(reduce)
995 }
996
997 /// Returns minimum successful element.
998 ///
999 /// Returns `Err(e)` on short-circuit failure.
1000 ///
1001 /// # Examples
1002 ///
1003 /// ```
1004 /// use orx_parallel::*;
1005 ///
1006 /// let m = ["1", "2", "3", "4"]
1007 /// .into_par()
1008 /// .map(|s| s.parse::<usize>())
1009 /// .into_fallible()
1010 /// .min();
1011 /// assert_eq!(m, Ok(Some(1)));
1012 /// ```
1013 fn min(self) -> Result<Option<Self::Elem>, Self::Error>
1014 where
1015 Self::Elem: Ord + Send,
1016 Self::Error: Send,
1017 {
1018 self.reduce(Ord::min)
1019 }
1020
1021 /// Returns successful element considered minimum by comparator `f`.
1022 ///
1023 /// Returns `Err(e)` on short-circuit failure.
1024 ///
1025 /// # Examples
1026 ///
1027 /// ```
1028 /// use orx_parallel::*;
1029 ///
1030 /// let x = vec![-3_i32, 0, 1, 5, -10]
1031 /// .into_par()
1032 /// .map(Ok::<_, &'static str>)
1033 /// .into_fallible()
1034 /// .min_by(|a, b| a.cmp(b));
1035 /// assert_eq!(x, Ok(Some(-10)));
1036 /// ```
1037 fn min_by<F>(self, f: F) -> Result<Option<Self::Elem>, Self::Error>
1038 where
1039 Self::Elem: Send,
1040 F: Fn(&Self::Elem, &Self::Elem) -> Ordering + Sync,
1041 Self::Error: Send,
1042 {
1043 let reduce = |x, y| match f(&x, &y) {
1044 Ordering::Less | Ordering::Equal => x,
1045 Ordering::Greater => y,
1046 };
1047 self.reduce(reduce)
1048 }
1049
1050 /// Returns successful element with minimum key value.
1051 ///
1052 /// Returns `Err(e)` on short-circuit failure.
1053 ///
1054 /// # Examples
1055 ///
1056 /// ```
1057 /// use orx_parallel::*;
1058 ///
1059 /// let x = vec![-3_i32, 0, 1, 5, -10]
1060 /// .into_par()
1061 /// .map(Ok::<_, &'static str>)
1062 /// .into_fallible()
1063 /// .min_by_key(|x| x.abs());
1064 /// assert_eq!(x, Ok(Some(0)));
1065 /// ```
1066 fn min_by_key<B, F>(self, f: F) -> Result<Option<Self::Elem>, Self::Error>
1067 where
1068 Self::Elem: Send,
1069 B: Ord,
1070 F: Fn(&Self::Elem) -> B + Sync,
1071 Self::Error: Send,
1072 {
1073 let reduce = |x, y| match f(&x).cmp(&f(&y)) {
1074 Ordering::Less | Ordering::Equal => x,
1075 Ordering::Greater => y,
1076 };
1077 self.reduce(reduce)
1078 }
1079
1080 /// Sums successful elements using [`Sum`] implementation.
1081 ///
1082 /// Returns `Err(e)` on short-circuit failure.
1083 ///
1084 /// # Examples
1085 ///
1086 /// ```
1087 /// use orx_parallel::*;
1088 ///
1089 /// let ok: Result<usize, _> = ["1", "2", "3", "4"]
1090 /// .into_par()
1091 /// .map(|s| s.parse::<usize>())
1092 /// .into_fallible()
1093 /// .sum();
1094 /// assert_eq!(ok, Ok(10));
1095 /// ```
1096 fn sum<S>(self) -> Result<S, Self::Error>
1097 where
1098 Self::Elem: Sum<S>,
1099 S: Send,
1100 Self::Error: Send,
1101 {
1102 self.map(Self::Elem::owned)
1103 .reduce(Self::Elem::add)
1104 .map(|x| x.unwrap_or(Self::Elem::zero()))
1105 }
1106}