Skip to main content

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