Repository: rayon-rs/rayon
Pull request #970 (2022-08-19 11:20:14)

add fold_chunks and fold_chunks_with parallel iterators
This adds two iterators to allow working with fixed-size chunks in a similar way to `chunks()` but without incurring the overhead of allocating a `Vec<T>` for each chunk.

Something similar was discussed in #773.

A call such as:
```
par_iter.fold_chunks(chunk_len, || init(), |acc, x| foo(acc, x))
```
would be functionally equivalent to:
```
par_iter.chunks(chunk_len)
    .map(|chunk|
        chunk.into_iter()
            .fold(init(), |acc, x| foo(acc, x|))
```
A final `fold()` (or `for_each()`, `sum()`, etc) across all mapped chunks can be chained by user code if needed, like this I think it's more flexible.


Patch:
diff --git a/rayon-core/src/lib.rs b/rayon-core/src/lib.rs
index ac95156..d218f68 100644
--- a/rayon-core/src/lib.rs
+++ b/rayon-core/src/lib.rs
@@ -270,7 +270,7 @@ impl ThreadPoolBuilder {
     /// The threads in this pool will start by calling `wrapper`, which should
     /// do initialization and continue by calling `ThreadBuilder::run()`.
     ///
-    /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.8/crossbeam/fn.scope.html
+    /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.7/crossbeam/fn.scope.html
     ///
     /// # Examples
     ///
@@ -340,7 +340,7 @@ impl<S> ThreadPoolBuilder<S> {
     /// if the pool is leaked. Furthermore, the global thread pool doesn't terminate
     /// until the entire process exits!
     ///
-    /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.8/crossbeam/fn.scope.html
+    /// [`crossbeam::scope`]: https://docs.rs/crossbeam/0.7/crossbeam/fn.scope.html
     ///
     /// # Examples
     ///
@@ -385,39 +385,6 @@ impl<S> ThreadPoolBuilder<S> {
     ///     Ok(())
     /// }
     /// ```
-    ///
-    /// This can also be used for a pool of scoped threads like [`crossbeam::scope`],
-    /// or [`std::thread::scope`] introduced in Rust 1.63, which is encapsulated in
-    /// [`build_scoped`](#method.build_scoped).
-    ///
-    /// [`std::thread::scope`]: https://doc.rust-lang.org/std/thread/fn.scope.html
-    ///
-    /// ```
-    /// # use rayon_core as rayon;
-    /// fn main() -> Result<(), rayon::ThreadPoolBuildError> {
-    ///     std::thread::scope(|scope| {
-    ///         let pool = rayon::ThreadPoolBuilder::new()
-    ///             .spawn_handler(|thread| {
-    ///                 let mut builder = std::thread::Builder::new();
-    ///                 if let Some(name) = thread.name() {
-    ///                     builder = builder.name(name.to_string());
-    ///                 }
-    ///                 if let Some(size) = thread.stack_size() {
-    ///                     builder = builder.stack_size(size);
-    ///                 }
-    ///                 builder.spawn_scoped(scope, || {
-    ///                     // Add any scoped initialization here, then run!
-    ///                     thread.run()
-    ///                 })?;
-    ///                 Ok(())
-    ///             })
-    ///             .build()?;
-    ///
-    ///         pool.install(|| println!("Hello from my custom scoped thread!"));
-    ///         Ok(())
-    ///     })
-    /// }
-    /// ```
     pub fn spawn_handler<F>(self, spawn: F) -> ThreadPoolBuilder<CustomSpawn<F>>
     where
         F: FnMut(ThreadBuilder) -> io::Result<()>,
diff --git a/rayon-demo/src/factorial/mod.rs b/rayon-demo/src/factorial/mod.rs
index 16ef999..84f9ba9 100644
--- a/rayon-demo/src/factorial/mod.rs
+++ b/rayon-demo/src/factorial/mod.rs
@@ -33,6 +33,37 @@ fn factorial_par_iter(b: &mut test::Bencher) {
     b.iter(|| assert_eq!(fact(test::black_box(N)), f));
 }
 
+#[bench]
+/// Compute the Factorial using rayon::fold_with.
+fn factorial_fold_with(b: &mut test::Bencher) {
+    fn fact(n: u32) -> BigUint {
+        (1..n + 1)
+            .into_par_iter()
+            .with_min_len(64) // for fair comparison with factorial_fold_chunks_with()
+            .fold_with(BigUint::from(1_u32), |acc, x| acc.mul(x))
+            .reduce_with(Mul::mul)
+            .unwrap()
+    }
+
+    let f = factorial(N);
+    b.iter(|| assert_eq!(fact(test::black_box(N)), f));
+}
+
+#[bench]
+/// Compute the Factorial using rayon::fold_chunks_with.
+fn factorial_fold_chunks_with(b: &mut test::Bencher) {
+    fn fact(n: u32) -> BigUint {
+        (1..n + 1)
+            .into_par_iter()
+            .fold_chunks_with(64, BigUint::from(1_u32), |acc, x| acc.mul(x))
+            .reduce_with(Mul::mul)
+            .unwrap()
+    }
+
+    let f = factorial(N);
+    b.iter(|| assert_eq!(fact(test::black_box(N)), f));
+}
+
 #[bench]
 /// Compute the Factorial using divide-and-conquer serial recursion.
 fn factorial_recursion(b: &mut test::Bencher) {
diff --git a/src/compile_fail/must_use.rs b/src/compile_fail/must_use.rs
index ac50a62..b4b3d77 100644
--- a/src/compile_fail/must_use.rs
+++ b/src/compile_fail/must_use.rs
@@ -33,6 +33,8 @@ must_use! {
     step_by             /** v.par_iter().step_by(2); */
     chain               /** v.par_iter().chain(&v); */
     chunks              /** v.par_iter().chunks(2); */
+    fold_chunks         /** v.par_iter().fold_chunks(2, || 0, |x, _| x); */
+    fold_chunks_with    /** v.par_iter().fold_chunks_with(2, 0, |x, _| x); */
     cloned              /** v.par_iter().cloned(); */
     copied              /** v.par_iter().copied(); */
     enumerate           /** v.par_iter().enumerate(); */
diff --git a/src/iter/fold_chunks.rs b/src/iter/fold_chunks.rs
new file mode 100644
index 0000000..6eefd3e
--- /dev/null
+++ b/src/iter/fold_chunks.rs
@@ -0,0 +1,375 @@
+use std::cmp::min;
+use std::fmt::{self, Debug};
+
+use super::plumbing::*;
+use super::*;
+use crate::math::div_round_up;
+
+/// `FoldChunks` is an iterator that groups elements of an underlying iterator and applies a
+/// function over them, producing a single value for each group.
+///
+/// This struct is created by the [`fold_chunks()`] method on [`IndexedParallelIterator`]
+///
+/// [`fold_chunks()`]: trait.IndexedParallelIterator.html#method.fold_chunks
+/// [`IndexedParallelIterator`]: trait.IndexedParallelIterator.html
+#[must_use = "iterator adaptors are lazy and do nothing unless consumed"]
+#[derive(Clone)]
+pub struct FoldChunks<I, ID, F>
+where
+    I: IndexedParallelIterator,
+{
+    base: I,
+    chunk_size: usize,
+    fold_op: F,
+    identity: ID,
+}
+
+impl<I: IndexedParallelIterator + Debug, ID, F> Debug for FoldChunks<I, ID, F> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Fold")
+            .field("base", &self.base)
+            .field("chunk_size", &self.chunk_size)
+            .finish()
+    }
+}
+
+impl<I, ID, U, F> FoldChunks<I, ID, F>
+where
+    I: IndexedParallelIterator,
+    ID: Fn() -> U + Send + Sync,
+    F: Fn(U, I::Item) -> U + Send + Sync,
+    U: Send,
+{
+    /// Creates a new `FoldChunks` iterator
+    pub(super) fn new(base: I, chunk_size: usize, identity: ID, fold_op: F) -> Self {
+        FoldChunks {
+            base,
+            chunk_size,
+            identity,
+            fold_op,
+        }
+    }
+}
+
+impl<I, ID, U, F> ParallelIterator for FoldChunks<I, ID, F>
+where
+    I: IndexedParallelIterator,
+    ID: Fn() -> U + Send + Sync,
+    F: Fn(U, I::Item) -> U + Send + Sync,
+    U: Send,
+{
+    type Item = U;
+
+    fn drive_unindexed<C>(self, consumer: C) -> C::Result
+    where
+        C: Consumer<U>,
+    {
+        bridge(self, consumer)
+    }
+
+    fn opt_len(&self) -> Option<usize> {
+        Some(self.len())
+    }
+}
+
+impl<I, ID, U, F> IndexedParallelIterator for FoldChunks<I, ID, F>
+where
+    I: IndexedParallelIterator,
+    ID: Fn() -> U + Send + Sync,
+    F: Fn(U, I::Item) -> U + Send + Sync,
+    U: Send,
+{
+    fn len(&self) -> usize {
+        div_round_up(self.base.len(), self.chunk_size)
+    }
+
+    fn drive<C>(self, consumer: C) -> C::Result
+    where
+        C: Consumer<Self::Item>,
+    {
+        bridge(self, consumer)
+    }
+
+    fn with_producer<CB>(self, callback: CB) -> CB::Output
+    where
+        CB: ProducerCallback<Self::Item>,
+    {
+        let len = self.base.len();
+        return self.base.with_producer(Callback {
+            chunk_size: self.chunk_size,
+            len,
+            identity: self.identity,
+            fold_op: self.fold_op,
+            callback,
+        });
+
+        struct Callback<CB, ID, F> {
+            chunk_size: usize,
+            len: usize,
+            identity: ID,
+            fold_op: F,
+            callback: CB,
+        }
+
+        impl<T, CB, ID, U, F> ProducerCallback<T> for Callback<CB, ID, F>
+        where
+            CB: ProducerCallback<U>,
+            ID: Fn() -> U + Send + Sync,
+            F: Fn(U, T) -> U + Send + Sync,
+        {
+            type Output = CB::Output;
+
+            fn callback<P>(self, base: P) -> CB::Output
+            where
+                P: Producer<Item = T>,
+            {
+                self.callback.callback(FoldChunksProducer {
+                    chunk_size: self.chunk_size,
+                    len: self.len,
+                    identity: &self.identity,
+                    fold_op: &self.fold_op,
+                    base,
+                })
+            }
+        }
+    }
+}
+
+struct FoldChunksProducer<'f, P, ID, F>
+where
+    P: Producer,
+{
+    chunk_size: usize,
+    len: usize,
+    identity: &'f ID,
+    fold_op: &'f F,
+    base: P,
+}
+
+impl<'f, P, ID, U, F> Producer for FoldChunksProducer<'f, P, ID, F>
+where
+    P: Producer,
+    ID: Fn() -> U + Send + Sync,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    type Item = F::Output;
+    type IntoIter = FoldChunksSeq<'f, P, ID, F>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        FoldChunksSeq {
+            chunk_size: self.chunk_size,
+            len: self.len,
+            identity: self.identity,
+            fold_op: self.fold_op,
+            inner: if self.len > 0 { Some(self.base) } else { None },
+        }
+    }
+
+    fn min_len(&self) -> usize {
+        div_round_up(self.base.min_len(), self.chunk_size)
+    }
+
+    fn max_len(&self) -> usize {
+        self.base.max_len() / self.chunk_size
+    }
+
+    fn split_at(self, index: usize) -> (Self, Self) {
+        let elem_index = min(index * self.chunk_size, self.len);
+        let (left, right) = self.base.split_at(elem_index);
+        (
+            FoldChunksProducer {
+                chunk_size: self.chunk_size,
+                len: elem_index,
+                identity: self.identity,
+                fold_op: self.fold_op,
+                base: left,
+            },
+            FoldChunksProducer {
+                chunk_size: self.chunk_size,
+                len: self.len - elem_index,
+                identity: self.identity,
+                fold_op: self.fold_op,
+                base: right,
+            },
+        )
+    }
+}
+
+struct FoldChunksSeq<'f, P, ID, F> {
+    chunk_size: usize,
+    len: usize,
+    identity: &'f ID,
+    fold_op: &'f F,
+    inner: Option<P>,
+}
+
+impl<'f, P, ID, U, F> Iterator for FoldChunksSeq<'f, P, ID, F>
+where
+    P: Producer,
+    ID: Fn() -> U + Send + Sync,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    type Item = U;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let producer = self.inner.take()?;
+        if self.len > self.chunk_size {
+            let (left, right) = producer.split_at(self.chunk_size);
+            self.inner = Some(right);
+            self.len -= self.chunk_size;
+            let chunk = left.into_iter();
+            Some(chunk.fold((self.identity)(), self.fold_op))
+        } else {
+            debug_assert!(self.len > 0);
+            self.len = 0;
+            let chunk = producer.into_iter();
+            Some(chunk.fold((self.identity)(), self.fold_op))
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = self.len();
+        (len, Some(len))
+    }
+}
+
+impl<'f, P, ID, U, F> ExactSizeIterator for FoldChunksSeq<'f, P, ID, F>
+where
+    P: Producer,
+    ID: Fn() -> U + Send + Sync,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    #[inline]
+    fn len(&self) -> usize {
+        div_round_up(self.len, self.chunk_size)
+    }
+}
+
+impl<'f, P, ID, U, F> DoubleEndedIterator for FoldChunksSeq<'f, P, ID, F>
+where
+    P: Producer,
+    ID: Fn() -> U + Send + Sync,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    fn next_back(&mut self) -> Option<Self::Item> {
+        let producer = self.inner.take()?;
+        if self.len > self.chunk_size {
+            let mut size = self.len % self.chunk_size;
+            if size == 0 {
+                size = self.chunk_size;
+            }
+            let (left, right) = producer.split_at(self.len - size);
+            self.inner = Some(left);
+            self.len -= size;
+            let chunk = right.into_iter();
+            Some(chunk.fold((self.identity)(), self.fold_op))
+        } else {
+            debug_assert!(self.len > 0);
+            self.len = 0;
+            let chunk = producer.into_iter();
+            Some(chunk.fold((self.identity)(), self.fold_op))
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use std::ops::Add;
+
+    #[test]
+    fn check_fold_chunks() {
+        let words = "bishbashbosh!"
+            .chars()
+            .collect::<Vec<_>>()
+            .into_par_iter()
+            .fold_chunks(4, String::new, |mut s, c| {
+                s.push(c);
+                s
+            })
+            .collect::<Vec<_>>();
+
+        assert_eq!(words, vec!["bish", "bash", "bosh", "!"]);
+    }
+
+    // 'closure' values for tests below
+    fn id() -> i32 {
+        0
+    }
+    fn sum<T, U>(x: T, y: U) -> T
+    where
+        T: Add<U, Output = T>,
+    {
+        x + y
+    }
+
+    #[test]
+    #[should_panic(expected = "chunk_size must not be zero")]
+    fn check_fold_chunks_zero_size() {
+        let _: Vec<i32> = vec![1, 2, 3]
+            .into_par_iter()
+            .fold_chunks(0, id, sum)
+            .collect();
+    }
+
+    #[test]
+    fn check_fold_chunks_even_size() {
+        assert_eq!(
+            vec![1 + 2 + 3, 4 + 5 + 6, 7 + 8 + 9],
+            (1..10)
+                .into_par_iter()
+                .fold_chunks(3, id, sum)
+                .collect::<Vec<i32>>()
+        );
+    }
+
+    #[test]
+    fn check_fold_chunks_empty() {
+        let v: Vec<i32> = vec![];
+        let expected: Vec<i32> = vec![];
+        assert_eq!(
+            expected,
+            v.into_par_iter()
+                .fold_chunks(2, id, sum)
+                .collect::<Vec<i32>>()
+        );
+    }
+
+    #[test]
+    fn check_fold_chunks_len() {
+        assert_eq!(4, (0..8).into_par_iter().fold_chunks(2, id, sum).len());
+        assert_eq!(3, (0..9).into_par_iter().fold_chunks(3, id, sum).len());
+        assert_eq!(3, (0..8).into_par_iter().fold_chunks(3, id, sum).len());
+        assert_eq!(1, (&[1]).par_iter().fold_chunks(3, id, sum).len());
+        assert_eq!(0, (0..0).into_par_iter().fold_chunks(3, id, sum).len());
+    }
+
+    #[test]
+    fn check_fold_chunks_uneven() {
+        let cases: Vec<(Vec<u32>, usize, Vec<u32>)> = vec![
+            ((0..5).collect(), 3, vec![0 + 1 + 2, 3 + 4]),
+            (vec![1], 5, vec![1]),
+            ((0..4).collect(), 3, vec![0 + 1 + 2, 3]),
+        ];
+
+        for (i, (v, n, expected)) in cases.into_iter().enumerate() {
+            let mut res: Vec<u32> = vec![];
+            v.par_iter()
+                .fold_chunks(n, || 0, sum)
+                .collect_into_vec(&mut res);
+            assert_eq!(expected, res, "Case {} failed", i);
+
+            res.truncate(0);
+            v.into_par_iter()
+                .fold_chunks(n, || 0, sum)
+                .rev()
+                .collect_into_vec(&mut res);
+            assert_eq!(
+                expected.into_iter().rev().collect::<Vec<u32>>(),
+                res,
+                "Case {} reversed failed",
+                i
+            );
+        }
+    }
+}
diff --git a/src/iter/fold_chunks_with.rs b/src/iter/fold_chunks_with.rs
new file mode 100644
index 0000000..a506254
--- /dev/null
+++ b/src/iter/fold_chunks_with.rs
@@ -0,0 +1,370 @@
+use std::cmp::min;
+use std::fmt::{self, Debug};
+
+use super::plumbing::*;
+use super::*;
+use crate::math::div_round_up;
+
+/// `FoldChunksWith` is an iterator that groups elements of an underlying iterator and applies a
+/// function over them, producing a single value for each group.
+///
+/// This struct is created by the [`fold_chunks_with()`] method on [`IndexedParallelIterator`]
+///
+/// [`fold_chunks_with()`]: trait.IndexedParallelIterator.html#method.fold_chunks
+/// [`IndexedParallelIterator`]: trait.IndexedParallelIterator.html
+#[must_use = "iterator adaptors are lazy and do nothing unless consumed"]
+#[derive(Clone)]
+pub struct FoldChunksWith<I, U, F>
+where
+    I: IndexedParallelIterator,
+{
+    base: I,
+    chunk_size: usize,
+    item: U,
+    fold_op: F,
+}
+
+impl<I: IndexedParallelIterator + Debug, U: Debug, F> Debug for FoldChunksWith<I, U, F> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Fold")
+            .field("base", &self.base)
+            .field("chunk_size", &self.chunk_size)
+            .field("item", &self.item)
+            .finish()
+    }
+}
+
+impl<I, U, F> FoldChunksWith<I, U, F>
+where
+    I: IndexedParallelIterator,
+    U: Send + Clone,
+    F: Fn(U, I::Item) -> U + Send + Sync,
+{
+    /// Creates a new `FoldChunksWith` iterator
+    pub(super) fn new(base: I, chunk_size: usize, item: U, fold_op: F) -> Self {
+        FoldChunksWith {
+            base,
+            chunk_size,
+            item,
+            fold_op,
+        }
+    }
+}
+
+impl<I, U, F> ParallelIterator for FoldChunksWith<I, U, F>
+where
+    I: IndexedParallelIterator,
+    U: Send + Clone,
+    F: Fn(U, I::Item) -> U + Send + Sync,
+{
+    type Item = U;
+
+    fn drive_unindexed<C>(self, consumer: C) -> C::Result
+    where
+        C: Consumer<U>,
+    {
+        bridge(self, consumer)
+    }
+
+    fn opt_len(&self) -> Option<usize> {
+        Some(self.len())
+    }
+}
+
+impl<I, U, F> IndexedParallelIterator for FoldChunksWith<I, U, F>
+where
+    I: IndexedParallelIterator,
+    U: Send + Clone,
+    F: Fn(U, I::Item) -> U + Send + Sync,
+{
+    fn len(&self) -> usize {
+        div_round_up(self.base.len(), self.chunk_size)
+    }
+
+    fn drive<C>(self, consumer: C) -> C::Result
+    where
+        C: Consumer<Self::Item>,
+    {
+        bridge(self, consumer)
+    }
+
+    fn with_producer<CB>(self, callback: CB) -> CB::Output
+    where
+        CB: ProducerCallback<Self::Item>,
+    {
+        let len = self.base.len();
+        return self.base.with_producer(Callback {
+            chunk_size: self.chunk_size,
+            len,
+            item: self.item,
+            fold_op: self.fold_op,
+            callback,
+        });
+
+        struct Callback<CB, T, F> {
+            chunk_size: usize,
+            len: usize,
+            item: T,
+            fold_op: F,
+            callback: CB,
+        }
+
+        impl<T, U, F, CB> ProducerCallback<T> for Callback<CB, U, F>
+        where
+            CB: ProducerCallback<U>,
+            U: Send + Clone,
+            F: Fn(U, T) -> U + Send + Sync,
+        {
+            type Output = CB::Output;
+
+            fn callback<P>(self, base: P) -> CB::Output
+            where
+                P: Producer<Item = T>,
+            {
+                self.callback.callback(FoldChunksWithProducer {
+                    chunk_size: self.chunk_size,
+                    len: self.len,
+                    item: self.item,
+                    fold_op: &self.fold_op,
+                    base,
+                })
+            }
+        }
+    }
+}
+
+struct FoldChunksWithProducer<'f, P, U, F>
+where
+    P: Producer,
+{
+    chunk_size: usize,
+    len: usize,
+    item: U,
+    fold_op: &'f F,
+    base: P,
+}
+
+impl<'f, P, U, F> Producer for FoldChunksWithProducer<'f, P, U, F>
+where
+    P: Producer,
+    U: Send + Clone,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    type Item = F::Output;
+    type IntoIter = FoldChunksWithSeq<'f, P, U, F>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        FoldChunksWithSeq {
+            chunk_size: self.chunk_size,
+            len: self.len,
+            item: self.item,
+            fold_op: self.fold_op,
+            inner: if self.len > 0 { Some(self.base) } else { None },
+        }
+    }
+
+    fn min_len(&self) -> usize {
+        div_round_up(self.base.min_len(), self.chunk_size)
+    }
+
+    fn max_len(&self) -> usize {
+        self.base.max_len() / self.chunk_size
+    }
+
+    fn split_at(self, index: usize) -> (Self, Self) {
+        let elem_index = min(index * self.chunk_size, self.len);
+        let (left, right) = self.base.split_at(elem_index);
+        (
+            FoldChunksWithProducer {
+                chunk_size: self.chunk_size,
+                len: elem_index,
+                item: self.item.clone(),
+                fold_op: self.fold_op,
+                base: left,
+            },
+            FoldChunksWithProducer {
+                chunk_size: self.chunk_size,
+                len: self.len - elem_index,
+                item: self.item,
+                fold_op: self.fold_op,
+                base: right,
+            },
+        )
+    }
+}
+
+struct FoldChunksWithSeq<'f, P, U, F> {
+    chunk_size: usize,
+    len: usize,
+    item: U,
+    fold_op: &'f F,
+    inner: Option<P>,
+}
+
+impl<'f, P, U, F> Iterator for FoldChunksWithSeq<'f, P, U, F>
+where
+    P: Producer,
+    U: Send + Clone,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    type Item = U;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let producer = self.inner.take()?;
+        if self.len > self.chunk_size {
+            let (left, right) = producer.split_at(self.chunk_size);
+            self.inner = Some(right);
+            self.len -= self.chunk_size;
+            let chunk = left.into_iter();
+            Some(chunk.fold(self.item.clone(), self.fold_op))
+        } else {
+            debug_assert!(self.len > 0);
+            self.len = 0;
+            let chunk = producer.into_iter();
+            Some(chunk.fold(self.item.clone(), self.fold_op))
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = self.len();
+        (len, Some(len))
+    }
+}
+
+impl<'f, P, U, F> ExactSizeIterator for FoldChunksWithSeq<'f, P, U, F>
+where
+    P: Producer,
+    U: Send + Clone,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    #[inline]
+    fn len(&self) -> usize {
+        div_round_up(self.len, self.chunk_size)
+    }
+}
+
+impl<'f, P, U, F> DoubleEndedIterator for FoldChunksWithSeq<'f, P, U, F>
+where
+    P: Producer,
+    U: Send + Clone,
+    F: Fn(U, P::Item) -> U + Send + Sync,
+{
+    fn next_back(&mut self) -> Option<Self::Item> {
+        let producer = self.inner.take()?;
+        if self.len > self.chunk_size {
+            let mut size = self.len % self.chunk_size;
+            if size == 0 {
+                size = self.chunk_size;
+            }
+            let (left, right) = producer.split_at(self.len - size);
+            self.inner = Some(left);
+            self.len -= size;
+            let chunk = right.into_iter();
+            Some(chunk.fold(self.item.clone(), self.fold_op))
+        } else {
+            debug_assert!(self.len > 0);
+            self.len = 0;
+            let chunk = producer.into_iter();
+            Some(chunk.fold(self.item.clone(), self.fold_op))
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use std::ops::Add;
+
+    #[test]
+    fn check_fold_chunks_with() {
+        let words = "bishbashbosh!"
+            .chars()
+            .collect::<Vec<_>>()
+            .into_par_iter()
+            .fold_chunks_with(4, String::new(), |mut s, c| {
+                s.push(c);
+                s
+            })
+            .collect::<Vec<_>>();
+
+        assert_eq!(words, vec!["bish", "bash", "bosh", "!"]);
+    }
+
+    // 'closure' value for tests below
+    fn sum<T, U>(x: T, y: U) -> T
+    where
+        T: Add<U, Output = T>,
+    {
+        x + y
+    }
+
+    #[test]
+    #[should_panic(expected = "chunk_size must not be zero")]
+    fn check_fold_chunks_zero_size() {
+        let _: Vec<i32> = vec![1, 2, 3]
+            .into_par_iter()
+            .fold_chunks_with(0, 0, sum)
+            .collect();
+    }
+
+    #[test]
+    fn check_fold_chunks_even_size() {
+        assert_eq!(
+            vec![1 + 2 + 3, 4 + 5 + 6, 7 + 8 + 9],
+            (1..10)
+                .into_par_iter()
+                .fold_chunks_with(3, 0, sum)
+                .collect::<Vec<i32>>()
+        );
+    }
+
+    #[test]
+    fn check_fold_chunks_with_empty() {
+        let v: Vec<i32> = vec![];
+        let expected: Vec<i32> = vec![];
+        assert_eq!(
+            expected,
+            v.into_par_iter()
+                .fold_chunks_with(2, 0, sum)
+                .collect::<Vec<i32>>()
+        );
+    }
+
+    #[test]
+    fn check_fold_chunks_len() {
+        assert_eq!(4, (0..8).into_par_iter().fold_chunks_with(2, 0, sum).len());
+        assert_eq!(3, (0..9).into_par_iter().fold_chunks_with(3, 0, sum).len());
+        assert_eq!(3, (0..8).into_par_iter().fold_chunks_with(3, 0, sum).len());
+        assert_eq!(1, (&[1]).par_iter().fold_chunks_with(3, 0, sum).len());
+        assert_eq!(0, (0..0).into_par_iter().fold_chunks_with(3, 0, sum).len());
+    }
+
+    #[test]
+    fn check_fold_chunks_uneven() {
+        let cases: Vec<(Vec<u32>, usize, Vec<u32>)> = vec![
+            ((0..5).collect(), 3, vec![0 + 1 + 2, 3 + 4]),
+            (vec![1], 5, vec![1]),
+            ((0..4).collect(), 3, vec![0 + 1 + 2, 3]),
+        ];
+
+        for (i, (v, n, expected)) in cases.into_iter().enumerate() {
+            let mut res: Vec<u32> = vec![];
+            v.par_iter()
+                .fold_chunks_with(n, 0, sum)
+                .collect_into_vec(&mut res);
+            assert_eq!(expected, res, "Case {} failed", i);
+
+            res.truncate(0);
+            v.into_par_iter()
+                .fold_chunks_with(n, 0, sum)
+                .rev()
+                .collect_into_vec(&mut res);
+            assert_eq!(
+                expected.into_iter().rev().collect::<Vec<u32>>(),
+                res,
+                "Case {} reversed failed",
+                i
+            );
+        }
+    }
+}
diff --git a/src/iter/mod.rs b/src/iter/mod.rs
index 50cd4de..02afbd3 100644
--- a/src/iter/mod.rs
+++ b/src/iter/mod.rs
@@ -119,6 +119,8 @@ mod flat_map_iter;
 mod flatten;
 mod flatten_iter;
 mod fold;
+mod fold_chunks;
+mod fold_chunks_with;
 mod for_each;
 mod from_par_iter;
 mod inspect;
@@ -165,6 +167,8 @@ pub use self::{
     flatten::Flatten,
     flatten_iter::FlattenIter,
     fold::{Fold, FoldWith},
+    fold_chunks::FoldChunks,
+    fold_chunks_with::FoldChunksWith,
     inspect::Inspect,
     interleave::Interleave,
     interleave_shortest::InterleaveShortest,
@@ -2417,6 +2421,89 @@ pub trait IndexedParallelIterator: ParallelIterator {
         Chunks::new(self, chunk_size)
     }
 
+    /// Splits an iterator into fixed-size chunks, performing a sequential [`fold()`] on
+    /// each chunk.
+    ///
+    /// Returns an iterator that produces a folded result for each chunk of items
+    /// produced by this iterator.
+    ///
+    /// This works essentially like:
+    ///
+    /// ```text
+    /// iter.chunks(chunk_size)
+    ///     .map(|chunk|
+    ///         chunk.into_iter()
+    ///             .fold(identity, fold_op)
+    ///     )
+    /// ```
+    ///
+    /// except there is no per-chunk allocation overhead.
+    ///
+    /// [`fold()`]: std::iter::Iterator#method.fold
+    ///
+    /// **Panics** if `chunk_size` is 0.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use rayon::prelude::*;
+    /// let nums = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
+    /// let chunk_sums = nums.into_par_iter().fold_chunks(2, || 0, |a, n| a + n).collect::<Vec<_>>();
+    /// assert_eq!(chunk_sums, vec![3, 7, 11, 15, 19]);
+    /// ```
+    #[track_caller]
+    fn fold_chunks<T, ID, F>(
+        self,
+        chunk_size: usize,
+        identity: ID,
+        fold_op: F,
+    ) -> FoldChunks<Self, ID, F>
+    where
+        ID: Fn() -> T + Send + Sync,
+        F: Fn(T, Self::Item) -> T + Send + Sync,
+        T: Send,
+    {
+        assert!(chunk_size != 0, "chunk_size must not be zero");
+        FoldChunks::new(self, chunk_size, identity, fold_op)
+    }
+
+    /// Splits an iterator into fixed-size chunks, performing a sequential [`fold()`] on
+    /// each chunk.
+    ///
+    /// Returns an iterator that produces a folded result for each chunk of items
+    /// produced by this iterator.
+    ///
+    /// This works essentially like `fold_chunks(chunk_size, || init.clone(), fold_op)`,
+    /// except it doesn't require the `init` type to be `Sync`, nor any other form of
+    /// added synchronization.
+    ///
+    /// [`fold()`]: std::iter::Iterator#method.fold
+    ///
+    /// **Panics** if `chunk_size` is 0.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use rayon::prelude::*;
+    /// let nums = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
+    /// let chunk_sums = nums.into_par_iter().fold_chunks_with(2, 0, |a, n| a + n).collect::<Vec<_>>();
+    /// assert_eq!(chunk_sums, vec![3, 7, 11, 15, 19]);
+    /// ```
+    #[track_caller]
+    fn fold_chunks_with<T, F>(
+        self,
+        chunk_size: usize,
+        init: T,
+        fold_op: F,
+    ) -> FoldChunksWith<Self, T, F>
+    where
+        T: Send + Clone,
+        F: Fn(T, Self::Item) -> T + Send + Sync,
+    {
+        assert!(chunk_size != 0, "chunk_size must not be zero");
+        FoldChunksWith::new(self, chunk_size, init, fold_op)
+    }
+
     /// Lexicographically compares the elements of this `ParallelIterator` with those of
     /// another.
     ///


Test patch:
diff --git a/tests/clones.rs b/tests/clones.rs
index 2f512ca..2b78f09 100644
--- a/tests/clones.rs
+++ b/tests/clones.rs
@@ -130,6 +130,12 @@ fn clone_adaptors() {
     check(v.par_iter().flatten_iter());
     check(v.par_iter().with_max_len(1).fold(|| 0, |x, _| x));
     check(v.par_iter().with_max_len(1).fold_with(0, |x, _| x));
+    check(v.par_iter().with_max_len(1).fold_chunks(1, || 0, |x, _| x));
+    check(
+        v.par_iter()
+            .with_max_len(1)
+            .fold_chunks_with(1, 0, |x, _| x),
+    );
     check(v.par_iter().with_max_len(1).try_fold(|| 0, |_, &x| x));
     check(v.par_iter().with_max_len(1).try_fold_with(0, |_, &x| x));
     check(v.par_iter().inspect(|_| ()));
diff --git a/tests/debug.rs b/tests/debug.rs
index d107b13..1cbf4e6 100644
--- a/tests/debug.rs
+++ b/tests/debug.rs
@@ -155,6 +155,8 @@ fn debug_adaptors() {
     check(v.par_iter().map(Some).flatten_iter());
     check(v.par_iter().fold(|| 0, |x, _| x));
     check(v.par_iter().fold_with(0, |x, _| x));
+    check(v.par_iter().fold_chunks(3, || 0, |x, _| x));
+    check(v.par_iter().fold_chunks_with(3, 0, |x, _| x));
     check(v.par_iter().try_fold(|| 0, |x, _| Some(x)));
     check(v.par_iter().try_fold_with(0, |x, _| Some(x)));
     check(v.par_iter().inspect(|_| ()));