use super::plumbing::*; use super::*; use std::fmt::{self, Debug}; impl Fold where I: ParallelIterator, F: Fn(U, I::Item) -> U + Sync + Send, ID: Fn() -> U + Sync + Send, U: Send, { pub(super) fn new(base: I, identity: ID, fold_op: F) -> Self { Fold { base, identity, fold_op, } } } /// `Fold` is an iterator that applies a function over an iterator producing a single value. /// This struct is created by the [`fold()`] method on [`ParallelIterator`] /// /// [`fold()`]: trait.ParallelIterator.html#method.fold /// [`ParallelIterator`]: trait.ParallelIterator.html #[must_use = "iterator adaptors are lazy and do nothing unless consumed"] #[derive(Clone)] pub struct Fold { base: I, identity: ID, fold_op: F, } impl Debug for Fold { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Fold").field("base", &self.base).finish() } } impl ParallelIterator for Fold where I: ParallelIterator, F: Fn(U, I::Item) -> U + Sync + Send, ID: Fn() -> U + Sync + Send, U: Send, { type Item = U; fn drive_unindexed(self, consumer: C) -> C::Result where C: UnindexedConsumer, { let consumer1 = FoldConsumer { base: consumer, fold_op: &self.fold_op, identity: &self.identity, }; self.base.drive_unindexed(consumer1) } } struct FoldConsumer<'c, C, ID, F> { base: C, fold_op: &'c F, identity: &'c ID, } impl<'r, U, T, C, ID, F> Consumer for FoldConsumer<'r, C, ID, F> where C: Consumer, F: Fn(U, T) -> U + Sync, ID: Fn() -> U + Sync, U: Send, { type Folder = FoldFolder<'r, C::Folder, U, F>; 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); ( FoldConsumer { base: left, ..self }, FoldConsumer { base: right, ..self }, reducer, ) } fn into_folder(self) -> Self::Folder { FoldFolder { base: self.base.into_folder(), item: (self.identity)(), fold_op: self.fold_op, } } fn full(&self) -> bool { self.base.full() } } impl<'r, U, T, C, ID, F> UnindexedConsumer for FoldConsumer<'r, C, ID, F> where C: UnindexedConsumer, F: Fn(U, T) -> U + Sync, ID: Fn() -> U + Sync, U: Send, { fn split_off_left(&self) -> Self { FoldConsumer { base: self.base.split_off_left(), ..*self } } fn to_reducer(&self) -> Self::Reducer { self.base.to_reducer() } } struct FoldFolder<'r, C, ID, F> { base: C, fold_op: &'r F, item: ID, } impl<'r, C, ID, F, T> Folder for FoldFolder<'r, C, ID, F> where C: Folder, F: Fn(ID, T) -> ID + Sync, { type Result = C::Result; fn consume(self, item: T) -> Self { let item = (self.fold_op)(self.item, item); FoldFolder { base: self.base, fold_op: self.fold_op, item, } } fn consume_iter(self, iter: I) -> Self where I: IntoIterator, { fn not_full(base: &C) -> impl Fn(&T) -> bool + '_ where C: Folder, { move |_| !base.full() } let base = self.base; let item = iter .into_iter() // stop iterating if another thread has finished .take_while(not_full(&base)) .fold(self.item, self.fold_op); FoldFolder { base, item, fold_op: self.fold_op, } } fn complete(self) -> C::Result { self.base.consume(self.item).complete() } fn full(&self) -> bool { self.base.full() } } // /////////////////////////////////////////////////////////////////////////// impl FoldWith where I: ParallelIterator, F: Fn(U, I::Item) -> U + Sync + Send, U: Send + Clone, { pub(super) fn new(base: I, item: U, fold_op: F) -> Self { FoldWith { base, item, fold_op, } } } /// `FoldWith` is an iterator that applies a function over an iterator producing a single value. /// This struct is created by the [`fold_with()`] method on [`ParallelIterator`] /// /// [`fold_with()`]: trait.ParallelIterator.html#method.fold_with /// [`ParallelIterator`]: trait.ParallelIterator.html #[must_use = "iterator adaptors are lazy and do nothing unless consumed"] #[derive(Clone)] pub struct FoldWith { base: I, item: U, fold_op: F, } impl Debug for FoldWith { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("FoldWith") .field("base", &self.base) .field("item", &self.item) .finish() } } impl ParallelIterator for FoldWith where I: ParallelIterator, F: Fn(U, I::Item) -> U + Sync + Send, U: Send + Clone, { type Item = U; fn drive_unindexed(self, consumer: C) -> C::Result where C: UnindexedConsumer, { let consumer1 = FoldWithConsumer { base: consumer, item: self.item, fold_op: &self.fold_op, }; self.base.drive_unindexed(consumer1) } } struct FoldWithConsumer<'c, C, U, F> { base: C, item: U, fold_op: &'c F, } impl<'r, U, T, C, F> Consumer for FoldWithConsumer<'r, C, U, F> where C: Consumer, F: Fn(U, T) -> U + Sync, U: Send + Clone, { type Folder = FoldFolder<'r, C::Folder, U, F>; 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); ( FoldWithConsumer { base: left, item: self.item.clone(), ..self }, FoldWithConsumer { base: right, ..self }, reducer, ) } fn into_folder(self) -> Self::Folder { FoldFolder { base: self.base.into_folder(), item: self.item, fold_op: self.fold_op, } } fn full(&self) -> bool { self.base.full() } } impl<'r, U, T, C, F> UnindexedConsumer for FoldWithConsumer<'r, C, U, F> where C: UnindexedConsumer, F: Fn(U, T) -> U + Sync, U: Send + Clone, { fn split_off_left(&self) -> Self { FoldWithConsumer { base: self.base.split_off_left(), item: self.item.clone(), ..*self } } fn to_reducer(&self) -> Self::Reducer { self.base.to_reducer() } }