Merge #464

464: Add "parallel" iter::empty() and iter::once() r=cuviper a=cuviper

These are simplistic producers, mirroring their `std` namesakes.  They
are essentially just optimized equivalents to iterating `None` and
`Some(x)`, where `empty()` is a ZST, `once()` doesn't need the enum tag
for an `Option`, and neither have the associated enum branches.

src/iter/empty.rs

@@ -0,0 +1,93 @@
+use iter::internal::*;
+use iter::*;
+
+use std;
+use std::fmt;
+use std::marker::PhantomData;
+
+/// Creates a parallel iterator that produces nothing.
+///
+/// This admits no parallelism on its own, but it could be used for code that
+/// deals with generic parallel iterators.
+///
+/// # Examples
+///
+/// ```
+/// use rayon::prelude::*;
+/// use rayon::iter::empty;
+///
+/// let pi = (0..1234).into_par_iter()
+///     .chain(empty())
+///     .chain(1234..10_000);
+///
+/// assert_eq!(pi.count(), 10_000);
+/// ```
+pub fn empty<T: Send>() -> Empty<T> {
+    Empty { marker: PhantomData }
+}
+
+/// Iterator adaptor for [the `empty()` function](fn.empty.html).
+pub struct Empty<T: Send> {
+    marker: PhantomData<T>,
+}
+
+impl<T: Send> Clone for Empty<T> {
+    fn clone(&self) -> Self {
+        empty()
+    }
+}
+
+impl<T: Send> fmt::Debug for Empty<T> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.pad("Empty")
+    }
+}
+
+impl<T: Send> ParallelIterator for Empty<T> {
+    type Item = T;
+
+    fn drive_unindexed<C>(self, consumer: C) -> C::Result
+        where C: UnindexedConsumer<Self::Item>
+    {
+        self.drive(consumer)
+    }
+
+    fn opt_len(&mut self) -> Option<usize> {
+        Some(0)
+    }
+}
+
+impl<T: Send> IndexedParallelIterator for Empty<T> {
+    fn drive<C>(self, consumer: C) -> C::Result
+        where C: Consumer<Self::Item>
+    {
+        consumer.into_folder().complete()
+    }
+
+    fn len(&mut self) -> usize {
+        0
+    }
+
+    fn with_producer<CB>(self, callback: CB) -> CB::Output
+        where CB: ProducerCallback<Self::Item>
+    {
+        callback.callback(EmptyProducer(PhantomData))
+    }
+}
+
+/// Private empty producer
+struct EmptyProducer<T: Send>(PhantomData<T>);
+
+impl<T: Send> Producer for EmptyProducer<T> {
+    type Item = T;
+    type IntoIter = std::iter::Empty<T>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        std::iter::empty()
+    }
+
+    fn split_at(self, index: usize) -> (Self, Self) {
+        debug_assert_eq!(index, 0);
+        (self, EmptyProducer(PhantomData))
+    }
+}

src/iter/mod.rs

@@ -87,6 +87,11 @@ mod repeat;
 pub use self::repeat::{Repeat, repeat};
 pub use self::repeat::{RepeatN, repeatn};
 
+mod empty;
+pub use self::empty::{Empty, empty};
+mod once;
+pub use self::once::{Once, once};
+
 #[cfg(test)]
 mod test;
 

src/iter/once.rs

@@ -0,0 +1,65 @@
+use iter::internal::*;
+use iter::*;
+
+/// Creates a parallel iterator that produces an element exactly once.
+///
+/// This admits no parallelism on its own, but it could be chained to existing
+/// parallel iterators to extend their contents, or otherwise used for any code
+/// that deals with generic parallel iterators.
+///
+/// # Examples
+///
+/// ```
+/// use rayon::prelude::*;
+/// use rayon::iter::once;
+///
+/// let pi = (0..1234).into_par_iter()
+///     .chain(once(-1))
+///     .chain(1234..10_000);
+///
+/// assert_eq!(pi.clone().count(), 10_001);
+/// assert_eq!(pi.clone().filter(|&x| x < 0).count(), 1);
+/// assert_eq!(pi.position_any(|x| x < 0), Some(1234));
+/// ```
+pub fn once<T: Send>(item: T) -> Once<T> {
+    Once { item: item }
+}
+
+/// Iterator adaptor for [the `once()` function](fn.once.html).
+#[derive(Clone, Debug)]
+pub struct Once<T: Send> {
+    item: T,
+}
+
+impl<T: Send> ParallelIterator for Once<T> {
+    type Item = T;
+
+    fn drive_unindexed<C>(self, consumer: C) -> C::Result
+        where C: UnindexedConsumer<Self::Item>
+    {
+        self.drive(consumer)
+    }
+
+    fn opt_len(&mut self) -> Option<usize> {
+        Some(1)
+    }
+}
+
+impl<T: Send> IndexedParallelIterator for Once<T> {
+    fn drive<C>(self, consumer: C) -> C::Result
+        where C: Consumer<Self::Item>
+    {
+        consumer.into_folder().consume(self.item).complete()
+    }
+
+    fn len(&mut self) -> usize {
+        1
+    }
+
+    fn with_producer<CB>(self, callback: CB) -> CB::Output
+        where CB: ProducerCallback<Self::Item>
+    {
+        // Let `OptionProducer` handle it.
+        Some(self.item).into_par_iter().with_producer(callback)
+    }
+}

src/iter/test.rs

@@ -1888,3 +1888,33 @@ fn check_repeatn_zip_right() {
         assert_eq!(item, (4, 4));
     }
 }
+
+#[test]
+fn check_empty() {
+    // drive_unindexed
+    let mut v: Vec<i32> = empty().filter(|_| unreachable!()).collect();
+    assert!(v.is_empty());
+
+    // drive (indexed)
+    empty().collect_into(&mut v);
+    assert!(v.is_empty());
+
+    // with_producer
+    let v: Vec<(i32, i32)> = empty().zip(1..10).collect();
+    assert!(v.is_empty());
+}
+
+#[test]
+fn check_once() {
+    // drive_unindexed
+    let mut v: Vec<i32> = once(42).filter(|_| true).collect();
+    assert_eq!(v, &[42]);
+
+    // drive (indexed)
+    once(42).collect_into(&mut v);
+    assert_eq!(v, &[42]);
+
+    // with_producer
+    let v: Vec<(i32, i32)> = once(42).zip(1..10).collect();
+    assert_eq!(v, &[(42, 1)]);
+}

src/option.rs

@@ -29,7 +29,7 @@ impl<T: Send> ParallelIterator for IntoIter<T> {
     fn drive_unindexed<C>(self, consumer: C) -> C::Result
         where C: UnindexedConsumer<Self::Item>
     {
-        bridge(self, consumer)
+        self.drive(consumer)
     }
 
     fn opt_len(&mut self) -> Option<usize> {
@@ -41,7 +41,11 @@ impl<T: Send> IndexedParallelIterator for IntoIter<T> {
     fn drive<C>(self, consumer: C) -> C::Result
         where C: Consumer<Self::Item>
     {
-        bridge(self, consumer)
+        let mut folder = consumer.into_folder();
+        if let Some(item) = self.opt {
+            folder = folder.consume(item);
+        }
+        folder.complete()
     }
 
     fn len(&mut self) -> usize {
@@ -121,6 +125,7 @@ impl<T: Send> Producer for OptionProducer<T> {
     }
 
     fn split_at(self, index: usize) -> (Self, Self) {
+        debug_assert!(index <= 1);
         let none = OptionProducer { opt: None };
         if index == 0 {
             (none, self)

tests/clones.rs

@@ -132,6 +132,16 @@ fn clone_adaptors() {
     check(v.par_iter().zip_eq(&v));
 }
 
+#[test]
+fn clone_empty() {
+    check(rayon::iter::empty::<i32>());
+}
+
+#[test]
+fn clone_once() {
+    check(rayon::iter::once(10));
+}
+
 #[test]
 fn clone_repeat() {
     let x: Option<i32> = None;

tests/debug.rs

@@ -141,6 +141,16 @@ fn debug_adaptors() {
     check(v.par_iter().zip_eq(&v));
 }
 
+#[test]
+fn debug_empty() {
+    check(rayon::iter::empty::<i32>());
+}
+
+#[test]
+fn debug_once() {
+    check(rayon::iter::once(10));
+}
+
 #[test]
 fn debug_repeat() {
     let x: Option<i32> = None;

tests/producer_split_at.rs

@@ -96,6 +96,18 @@ fn check_len<I: ExactSizeIterator>(iter: &I, len: usize) {
 
 // **** Base Producers ****
 
+#[test]
+fn empty() {
+    let v = vec![42];
+    check(&v[..0], || rayon::iter::empty());
+}
+
+#[test]
+fn once() {
+    let v = vec![42];
+    check(&v, || rayon::iter::once(42));
+}
+
 #[test]
 fn option() {
     let v = vec![42];