Implemented 'while_some' operation

asparit/src/core/iterator.rs

@@ -32,6 +32,7 @@ use crate::{
         try_for_each::{TryForEach, TryForEachInit, TryForEachWith},
         try_reduce::{TryReduce, TryReduceWith},
         update::Update,
+        while_some::WhileSome,
     },
     misc::Try,
 };
@@ -1498,6 +1499,36 @@ pub trait ParallelIterator<'a>: Sized + Send {
         All::new(self, operation)
     }
 
+    /// Creates an iterator over the `Some` items of this iterator, halting
+    /// as soon as any `None` is found.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use rayon::prelude::*;
+    /// use std::sync::atomic::{AtomicUsize, Ordering};
+    ///
+    /// let counter = AtomicUsize::new(0);
+    /// let value = (0_i32..2048)
+    ///     .into_par_iter()
+    ///     .map(|x| {
+    ///              counter.fetch_add(1, Ordering::SeqCst);
+    ///              if x < 1024 { Some(x) } else { None }
+    ///          })
+    ///     .while_some()
+    ///     .max();
+    ///
+    /// assert!(value = Some(1023));
+    /// assert!(counter.load(Ordering::SeqCst) < 2048); // should not have visited every single one
+    /// ```
+    fn while_some<T>(self) -> WhileSome<Self>
+    where
+        Self: ParallelIterator<'a, Item = Option<T>>,
+        T: Send + 'a,
+    {
+        WhileSome::new(self)
+    }
+
     /// Creates a fresh collection containing all the elements produced
     /// by this parallel iterator.
     ///

asparit/src/inner/mod.rs

@@ -23,6 +23,7 @@ pub mod try_fold;
 pub mod try_for_each;
 pub mod try_reduce;
 pub mod update;
+pub mod while_some;
 
 #[cfg(test)]
 mod tests {
@@ -53,9 +54,11 @@ mod tests {
             .chain(b)
             .update(|x| x.push(0))
             .flatten_iter()
+            .map(Some)
+            .while_some()
             .map_init(
                 move || i.fetch_add(1, Ordering::Relaxed),
-                |init, item| (item, *init),
+                |init, item| Some((item, *init)),
             )
             .try_fold_with(String::new(), |s, item| -> Result<String, ()> {
                 Ok(format!("{} + {:?}", s, item))

asparit/src/inner/while_some.rs

@@ -0,0 +1,385 @@
+use std::iter::{DoubleEndedIterator, ExactSizeIterator, Iterator};
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
+
+use crate::{
+    Consumer, Executor, Folder, IndexedParallelIterator, IndexedProducer, IndexedProducerCallback,
+    ParallelIterator, Producer, ProducerCallback, Reducer,
+};
+
+/* WhileSome */
+
+pub struct WhileSome<X> {
+    base: X,
+}
+
+impl<X> WhileSome<X> {
+    pub fn new(base: X) -> Self {
+        Self { base }
+    }
+}
+
+impl<'a, X, T> ParallelIterator<'a> for WhileSome<X>
+where
+    X: ParallelIterator<'a, Item = Option<T>>,
+    T: Send + 'a,
+{
+    type Item = T;
+
+    fn drive<E, C, D, R>(self, executor: E, base: C) -> E::Result
+    where
+        E: Executor<'a, D>,
+        C: Consumer<Self::Item, Result = D, Reducer = R> + 'a,
+        D: Send + 'a,
+        R: Reducer<D> + Send + 'a,
+    {
+        let consumer = WhileSomeConsumer {
+            base,
+            is_full: Arc::new(AtomicBool::new(false)),
+        };
+
+        self.base.drive(executor, consumer)
+    }
+
+    fn with_producer<CB>(self, base: CB) -> CB::Output
+    where
+        CB: ProducerCallback<'a, Self::Item>,
+    {
+        self.base.with_producer(WhileSomeCallback { base })
+    }
+
+    fn len_hint_opt(&self) -> Option<usize> {
+        self.base.len_hint_opt()
+    }
+}
+
+impl<'a, X, T> IndexedParallelIterator<'a> for WhileSome<X>
+where
+    X: IndexedParallelIterator<'a, Item = Option<T>>,
+    T: Send + 'a,
+{
+    fn drive_indexed<E, C, D, R>(self, executor: E, base: C) -> E::Result
+    where
+        E: Executor<'a, D>,
+        C: Consumer<Self::Item, Result = D, Reducer = R> + 'a,
+        D: Send + 'a,
+        R: Reducer<D> + Send + 'a,
+    {
+        let consumer = WhileSomeConsumer {
+            base,
+            is_full: Arc::new(AtomicBool::new(false)),
+        };
+
+        self.base.drive_indexed(executor, consumer)
+    }
+
+    fn with_producer_indexed<CB>(self, base: CB) -> CB::Output
+    where
+        CB: IndexedProducerCallback<'a, Self::Item>,
+    {
+        self.base.with_producer_indexed(WhileSomeCallback { base })
+    }
+
+    fn len_hint(&self) -> usize {
+        self.base.len_hint()
+    }
+}
+
+/* WhileSomeCallback */
+
+struct WhileSomeCallback<CB> {
+    base: CB,
+}
+
+impl<'a, CB, T> ProducerCallback<'a, Option<T>> for WhileSomeCallback<CB>
+where
+    CB: ProducerCallback<'a, T>,
+{
+    type Output = CB::Output;
+
+    fn callback<P>(self, base: P) -> Self::Output
+    where
+        P: Producer<Item = Option<T>> + 'a,
+    {
+        self.base.callback(WhileSomeProducer {
+            base,
+            is_full: Arc::new(AtomicBool::new(false)),
+        })
+    }
+}
+
+impl<'a, CB, T> IndexedProducerCallback<'a, Option<T>> for WhileSomeCallback<CB>
+where
+    CB: IndexedProducerCallback<'a, T>,
+{
+    type Output = CB::Output;
+
+    fn callback<P>(self, base: P) -> Self::Output
+    where
+        P: IndexedProducer<Item = Option<T>> + 'a,
+    {
+        self.base.callback(WhileSomeProducer {
+            base,
+            is_full: Arc::new(AtomicBool::new(false)),
+        })
+    }
+}
+
+/* WhileSomeProducer */
+
+struct WhileSomeProducer<P> {
+    base: P,
+    is_full: Arc<AtomicBool>,
+}
+
+impl<P, T> Producer for WhileSomeProducer<P>
+where
+    P: Producer<Item = Option<T>>,
+{
+    type Item = T;
+    type IntoIter = WhileSomeIter<P::IntoIter>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        unimplemented!()
+    }
+
+    fn split(self) -> (Self, Option<Self>) {
+        let is_full = self.is_full;
+        let (left, right) = self.base.split();
+
+        let left = Self {
+            base: left,
+            is_full: is_full.clone(),
+        };
+        let right = right.map(|x| Self { base: x, is_full });
+
+        (left, right)
+    }
+
+    fn splits(&self) -> Option<usize> {
+        self.base.splits()
+    }
+}
+
+impl<P, T> IndexedProducer for WhileSomeProducer<P>
+where
+    P: IndexedProducer<Item = Option<T>>,
+{
+    type Item = T;
+    type IntoIter = WhileSomeIter<P::IntoIter>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        WhileSomeIter {
+            base: self.base.into_iter(),
+            is_full: self.is_full,
+        }
+    }
+
+    fn len(&self) -> usize {
+        self.base.len()
+    }
+
+    fn split_at(self, index: usize) -> (Self, Self) {
+        let (left, right) = self.base.split_at(index);
+
+        let left = Self {
+            base: left,
+            is_full: self.is_full.clone(),
+        };
+        let right = Self {
+            base: right,
+            is_full: self.is_full,
+        };
+
+        (left, right)
+    }
+
+    fn splits(&self) -> Option<usize> {
+        self.base.splits()
+    }
+
+    fn min_len(&self) -> Option<usize> {
+        self.base.min_len()
+    }
+
+    fn max_len(&self) -> Option<usize> {
+        self.base.max_len()
+    }
+}
+
+/* WhileSomeIter */
+
+struct WhileSomeIter<I> {
+    base: I,
+    is_full: Arc<AtomicBool>,
+}
+
+impl<I, T> Iterator for WhileSomeIter<I>
+where
+    I: Iterator<Item = Option<T>>,
+{
+    type Item = T;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.is_full.load(Ordering::Relaxed) {
+            return None;
+        }
+
+        match self.base.next()? {
+            Some(next) => Some(next),
+            None => {
+                self.is_full.store(true, Ordering::Relaxed);
+
+                None
+            }
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if self.is_full.load(Ordering::Relaxed) {
+            (0, Some(0))
+        } else {
+            let (_min, max) = self.base.size_hint();
+
+            (0, max)
+        }
+    }
+}
+
+impl<I, T> DoubleEndedIterator for WhileSomeIter<I>
+where
+    I: DoubleEndedIterator<Item = Option<T>>,
+{
+    fn next_back(&mut self) -> Option<Self::Item> {
+        if self.is_full.load(Ordering::Relaxed) {
+            return None;
+        }
+
+        match self.base.next()? {
+            Some(next) => Some(next),
+            None => {
+                self.is_full.store(true, Ordering::Relaxed);
+
+                None
+            }
+        }
+    }
+}
+
+impl<I, T> ExactSizeIterator for WhileSomeIter<I> where I: ExactSizeIterator<Item = Option<T>> {}
+
+/* WhileSomeConsumer */
+
+struct WhileSomeConsumer<C> {
+    base: C,
+    is_full: Arc<AtomicBool>,
+}
+
+impl<C, T> Consumer<Option<T>> for WhileSomeConsumer<C>
+where
+    C: Consumer<T>,
+    T: Send,
+{
+    type Folder = WhileSomeFolder<C::Folder>;
+    type Reducer = C::Reducer;
+    type Result = C::Result;
+
+    fn split(self) -> (Self, Self, Self::Reducer) {
+        let (left, right, reducer) = self.base.split();
+
+        let left = Self {
+            base: left,
+            is_full: self.is_full.clone(),
+        };
+        let right = Self {
+            base: right,
+            is_full: self.is_full,
+        };
+
+        (left, right, reducer)
+    }
+
+    fn split_at(self, index: usize) -> (Self, Self, Self::Reducer) {
+        let (left, right, reducer) = self.base.split_at(index);
+
+        let left = Self {
+            base: left,
+            is_full: self.is_full.clone(),
+        };
+        let right = Self {
+            base: right,
+            is_full: self.is_full,
+        };
+
+        (left, right, reducer)
+    }
+
+    fn into_folder(self) -> Self::Folder {
+        let base = self.base.into_folder();
+
+        WhileSomeFolder {
+            base,
+            is_full: self.is_full,
+        }
+    }
+
+    fn is_full(&self) -> bool {
+        self.is_full.load(Ordering::Relaxed) || self.base.is_full()
+    }
+}
+
+/* WhileSomeFolder */
+
+struct WhileSomeFolder<F> {
+    base: F,
+    is_full: Arc<AtomicBool>,
+}
+
+impl<F, I> Folder<Option<I>> for WhileSomeFolder<F>
+where
+    F: Folder<I>,
+{
+    type Result = F::Result;
+
+    fn consume(mut self, item: Option<I>) -> Self {
+        match item {
+            Some(item) => self.base = self.base.consume(item),
+            None => self.is_full.store(true, Ordering::Relaxed),
+        }
+
+        self
+    }
+
+    fn consume_iter<X>(mut self, iter: X) -> Self
+    where
+        X: IntoIterator<Item = Option<I>>,
+    {
+        let is_full = self.is_full.clone();
+
+        let iter = iter
+            .into_iter()
+            .take_while(|x| match *x {
+                Some(_) => !is_full.load(Ordering::Relaxed),
+                None => {
+                    is_full.store(true, Ordering::Relaxed);
+
+                    false
+                }
+            })
+            .map(Option::unwrap);
+
+        self.base = self.base.consume_iter(iter);
+
+        self
+    }
+
+    fn complete(self) -> Self::Result {
+        self.base.complete()
+    }
+
+    fn is_full(&self) -> bool {
+        self.is_full.load(Ordering::Relaxed) || self.base.is_full()
+    }
+}