Implemented 'position_any', 'position_first' and 'position_last' operation

asparit/src/core/iterator.rs

@@ -31,6 +31,7 @@ use crate::{
         min::{Min, MinBy, MinByKey},
         panic_fuse::PanicFuse,
         partition::{Partition, PartitionMap},
+        position::Position,
         product::Product,
         reduce::{Reduce, ReduceWith},
         skip::Skip,
@@ -2117,4 +2118,91 @@ pub trait IndexedParallelIterator<'a>: ParallelIterator<'a> {
     fn take(self, n: usize) -> Take<Self> {
         Take::new(self, n)
     }
+
+    /// Searches for **some** item in the parallel iterator that
+    /// matches the given operation, and returns its index.  Like
+    /// `ParallelIterator::find_any`, the parallel search will not
+    /// necessarily find the **first** match, and once a match is
+    /// found we'll attempt to stop processing any more.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use rayon::prelude::*;
+    ///
+    /// let a = [1, 2, 3, 3];
+    ///
+    /// let i = a.par_iter().position_any(|&x| x == 3).expect("found");
+    /// assert!(i == 2 || i == 3);
+    ///
+    /// assert_eq!(a.par_iter().position_any(|&x| x == 100), None);
+    /// ```
+    fn position_any<O>(self, operation: O) -> Position<Self, O>
+    where
+        O: Fn(Self::Item) -> bool + Clone + Send + 'a,
+    {
+        Position::new(self, operation, FindMatch::Any)
+    }
+
+    /// Searches for the sequentially **first** item in the parallel iterator
+    /// that matches the given operation, and returns its index.
+    ///
+    /// Like `ParallelIterator::find_first`, once a match is found,
+    /// all attempts to the right of the match will be stopped, while
+    /// attempts to the left must continue in case an earlier match
+    /// is found.
+    ///
+    /// Note that not all parallel iterators have a useful order, much like
+    /// sequential `HashMap` iteration, so "first" may be nebulous.  If you
+    /// just want the first match that discovered anywhere in the iterator,
+    /// `position_any` is a better choice.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use rayon::prelude::*;
+    ///
+    /// let a = [1, 2, 3, 3];
+    ///
+    /// assert_eq!(a.par_iter().position_first(|&x| x == 3), Some(2));
+    ///
+    /// assert_eq!(a.par_iter().position_first(|&x| x == 100), None);
+    /// ```
+    fn position_first<O>(self, operation: O) -> Position<Self, O>
+    where
+        O: Fn(Self::Item) -> bool + Clone + Send + 'a,
+    {
+        Position::new(self, operation, FindMatch::First)
+    }
+
+    /// Searches for the sequentially **last** item in the parallel iterator
+    /// that matches the given operation, and returns its index.
+    ///
+    /// Like `ParallelIterator::find_last`, once a match is found,
+    /// all attempts to the left of the match will be stopped, while
+    /// attempts to the right must continue in case a later match
+    /// is found.
+    ///
+    /// Note that not all parallel iterators have a useful order, much like
+    /// sequential `HashMap` iteration, so "last" may be nebulous.  When the
+    /// order doesn't actually matter to you, `position_any` is a better
+    /// choice.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use rayon::prelude::*;
+    ///
+    /// let a = [1, 2, 3, 3];
+    ///
+    /// assert_eq!(a.par_iter().position_last(|&x| x == 3), Some(3));
+    ///
+    /// assert_eq!(a.par_iter().position_last(|&x| x == 100), None);
+    /// ```
+    fn position_last<O>(self, operation: O) -> Position<Self, O>
+    where
+        O: Fn(Self::Item) -> bool + Clone + Send + 'a,
+    {
+        Position::new(self, operation, FindMatch::Last)
+    }
 }

asparit/src/iter/find.rs

@@ -232,7 +232,13 @@ where
     }
 
     fn is_full(&self) -> bool {
-        self.found.load(Ordering::Relaxed) > 0
+        let found = self.found.load(Ordering::Relaxed);
+
+        match self.find_match {
+            FindMatch::Any => found != 0,
+            FindMatch::First => found != 0 && found < self.lower_bound,
+            FindMatch::Last => found != 0 && found > self.upper_bound,
+        }
     }
 }
 
@@ -298,7 +304,23 @@ where
     }
 
     fn is_full(&self) -> bool {
-        self.found.load(Ordering::Relaxed) > 0
+        let found_best_in_range = match self.find_match {
+            FindMatch::Any => self.item.is_some(),
+            FindMatch::First => self.item.is_some(),
+            FindMatch::Last => false,
+        };
+
+        if found_best_in_range {
+            return true;
+        }
+
+        let found = self.found.load(Ordering::Relaxed);
+
+        match self.find_match {
+            FindMatch::Any => found != 0,
+            FindMatch::First => found != 0 && found < self.lower_bound,
+            FindMatch::Last => found != 0 && found > self.upper_bound,
+        }
     }
 }
 

asparit/src/iter/mod.rs

@@ -23,6 +23,7 @@ pub mod min;
 pub mod noop;
 pub mod panic_fuse;
 pub mod partition;
+pub mod position;
 pub mod product;
 pub mod reduce;
 pub mod skip;
@@ -44,14 +45,9 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_for_each() {
-        let x = vec![0usize, 1, 2, 3, 4, 5, 6, 7, 8, 9]
+        let x = vec![0usize, 1, 2, 3, 4, 5]
             .into_par_iter()
-            .skip(1)
-            .step_by(3)
-            .enumerate()
-            .for_each(|x| {
-                dbg!(x);
-            })
+            .position_any(|x| x % 2 == 1)
             .exec()
             .await;
 

asparit/src/iter/position.rs

@@ -0,0 +1,41 @@
+use crate::{Driver, Executor, IndexedParallelIterator, WithIndexedProducer};
+
+use super::find::{Find, FindMatch};
+
+pub struct Position<X, O> {
+    base: X,
+    operation: O,
+    find_match: FindMatch,
+}
+
+impl<X, O> Position<X, O> {
+    pub fn new(base: X, operation: O, find_match: FindMatch) -> Self {
+        Self {
+            base,
+            operation,
+            find_match,
+        }
+    }
+}
+
+impl<'a, X, O, I> Driver<'a, Option<usize>, Option<(usize, bool)>> for Position<X, O>
+where
+    X: IndexedParallelIterator<'a, Item = I> + WithIndexedProducer<'a, Item = I>,
+    O: Fn(I) -> bool + Clone + Send + 'a,
+    I: Send + 'a,
+{
+    fn exec_with<E>(self, executor: E) -> E::Result
+    where
+        E: Executor<'a, Option<usize>, Option<(usize, bool)>>,
+    {
+        let executor = executor.into_inner();
+        let iterator = Find::new(
+            self.base.map(self.operation).enumerate(),
+            |(_, x): &(usize, bool)| -> bool { *x },
+            self.find_match,
+        );
+        let inner = iterator.exec_with(executor);
+
+        E::map(inner, |x| x.map(|(i, _)| i))
+    }
+}