Add TakeAny

Author: yuyuyureka

src/iter/mod.rs

@@ -145,6 +145,7 @@ mod splitter;
 mod step_by;
 mod sum;
 mod take;
+mod take_any;
 mod try_fold;
 mod try_reduce;
 mod try_reduce_with;
@@ -188,6 +189,7 @@ pub use self::{
     splitter::{split, Split},
     step_by::StepBy,
     take::Take,
+    take_any::TakeAny,
     try_fold::{TryFold, TryFoldWith},
     update::Update,
     while_some::WhileSome,
@@ -2194,6 +2196,25 @@ pub trait ParallelIterator: Sized + Send {
         Intersperse::new(self, element)
     }
 
+    /// Creates an iterator that yields the first `n` elements.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use rayon::prelude::*;
+    ///
+    /// let result: Vec<_> = (0..100)
+    ///     .into_par_iter()
+    ///     .filter(|&x| x % 2 == 0)
+    ///     .take_any(5)
+    ///     .collect();
+    ///
+    /// assert_eq!(result.len(), 5);
+    /// ```
+    fn take_any(self, n: usize) -> TakeAny<Self> {
+        TakeAny::new(self, n)
+    }
+
     /// Internal method used to define the behavior of this parallel
     /// iterator. You should not need to call this directly.
     ///

src/iter/take_any.rs

@@ -0,0 +1,148 @@
+use super::plumbing::*;
+use super::*;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+/// `TakeAny` is an iterator that iterates over the first `n` elements.
+/// This struct is created by the [`take_any()`] method on [`ParallelIterator`]
+///
+/// [`take_any()`]: trait.ParallelIterator.html#method.take_any
+/// [`ParallelIterator`]: trait.ParallelIterator.html
+#[must_use = "iterator adaptors are lazy and do nothing unless consumed"]
+#[derive(Debug)]
+pub struct TakeAny<I: ParallelIterator> {
+    base: I,
+    count: AtomicUsize,
+}
+
+impl<I> TakeAny<I>
+where
+    I: ParallelIterator,
+{
+    /// Creates a new `TakeAny` iterator.
+    pub(super) fn new(base: I, count: usize) -> Self {
+        TakeAny {
+            base,
+            count: AtomicUsize::new(count),
+        }
+    }
+}
+
+impl<I, T> ParallelIterator for TakeAny<I>
+where
+    I: ParallelIterator<Item = T>,
+    T: Send,
+{
+    type Item = T;
+
+    fn drive_unindexed<C>(self, consumer: C) -> C::Result
+    where
+        C: UnindexedConsumer<Self::Item>,
+    {
+        let consumer1 = TakeAnyConsumer {
+            base: consumer,
+            count: &self.count,
+        };
+        self.base.drive_unindexed(consumer1)
+    }
+}
+
+/// ////////////////////////////////////////////////////////////////////////
+/// Consumer implementation
+
+struct TakeAnyConsumer<'f, C> {
+    base: C,
+    count: &'f AtomicUsize,
+}
+
+impl<'f, T, C> Consumer<T> for TakeAnyConsumer<'f, C>
+where
+    C: Consumer<T>,
+    T: Send,
+{
+    type Folder = TakeAnyFolder<'f, C::Folder>;
+    type Reducer = C::Reducer;
+    type Result = C::Result;
+
+    fn split_at(self, index: usize) -> (Self, Self, Self::Reducer) {
+        let (left, right, reducer) = self.base.split_at(index);
+        (
+            TakeAnyConsumer { base: left, ..self },
+            TakeAnyConsumer {
+                base: right,
+                ..self
+            },
+            reducer,
+        )
+    }
+
+    fn into_folder(self) -> Self::Folder {
+        TakeAnyFolder {
+            base: self.base.into_folder(),
+            count: self.count,
+        }
+    }
+
+    fn full(&self) -> bool {
+        self.count.load(Ordering::Relaxed) == 0 || self.base.full()
+    }
+}
+
+impl<'f, T, C> UnindexedConsumer<T> for TakeAnyConsumer<'f, C>
+where
+    C: UnindexedConsumer<T>,
+    T: Send,
+{
+    fn split_off_left(&self) -> Self {
+        TakeAnyConsumer {
+            base: self.base.split_off_left(),
+            ..*self
+        }
+    }
+
+    fn to_reducer(&self) -> Self::Reducer {
+        self.base.to_reducer()
+    }
+}
+
+struct TakeAnyFolder<'f, C> {
+    base: C,
+    count: &'f AtomicUsize,
+}
+
+fn checked_decrement(u: &AtomicUsize) -> bool {
+    u.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |u| u.checked_sub(1))
+        .is_ok()
+}
+
+impl<'f, T, C> Folder<T> for TakeAnyFolder<'f, C>
+where
+    C: Folder<T>,
+{
+    type Result = C::Result;
+
+    fn consume(mut self, item: T) -> Self {
+        if checked_decrement(self.count) {
+            self.base = self.base.consume(item);
+        }
+        self
+    }
+
+    fn consume_iter<I>(mut self, iter: I) -> Self
+    where
+        I: IntoIterator<Item = T>,
+    {
+        self.base = self.base.consume_iter(
+            iter.into_iter()
+                .take_while(move |_| checked_decrement(self.count)),
+        );
+        self
+    }
+
+    fn complete(self) -> C::Result {
+        self.base.complete()
+    }
+
+    fn full(&self) -> bool {
+        self.count.load(Ordering::Relaxed) == 0 || self.base.full()
+    }
+}