/// This trait abstracts the different ways we can "unzip" one parallel /// iterator into two distinct consumers, which we can handle almost /// identically apart from how to process the individual items. trait UnzipOp<T>: Sync + Send { /// The type of item expected by the left consumer. type Left: Send;
/// The type of item expected by the right consumer. type Right: Send;
/// Consumes one item and feeds it to one or both of the underlying folders. fn consume<FA, FB>(&self, item: T, left: FA, right: FB) -> (FA, FB) where
FA: Folder<Self::Left>,
FB: Folder<Self::Right>;
/// Reports whether this op may support indexed consumers. /// - e.g. true for `unzip` where the item count passed through directly. /// - e.g. false for `partition` where the sorting is not yet known. fn indexable() -> bool { false
}
}
/// Runs an unzip-like operation into default `ParallelExtend` collections. fn execute<I, OP, FromA, FromB>(pi: I, op: OP) -> (FromA, FromB) where
I: ParallelIterator,
OP: UnzipOp<I::Item>,
FromA: Default + Send + ParallelExtend<OP::Left>,
FromB: Default + Send + ParallelExtend<OP::Right>,
{ letmut a = FromA::default(); letmut b = FromB::default();
execute_into(&mut a, &mut b, pi, op);
(a, b)
}
/// Runs an unzip-like operation into `ParallelExtend` collections. fn execute_into<I, OP, FromA, FromB>(a: &mut FromA, b: &mut FromB, pi: I, op: OP) where
I: ParallelIterator,
OP: UnzipOp<I::Item>,
FromA: Send + ParallelExtend<OP::Left>,
FromB: Send + ParallelExtend<OP::Right>,
{ // We have no idea what the consumers will look like for these // collections' `par_extend`, but we can intercept them in our own // `drive_unindexed`. Start with the left side, type `A`: let iter = UnzipA { base: pi, op, b };
a.par_extend(iter);
}
/// Unzips the items of a parallel iterator into a pair of arbitrary /// `ParallelExtend` containers. /// /// This is called by `ParallelIterator::unzip`. pub(super) fn unzip<I, A, B, FromA, FromB>(pi: I) -> (FromA, FromB) where
I: ParallelIterator<Item = (A, B)>,
FromA: Default + Send + ParallelExtend<A>,
FromB: Default + Send + ParallelExtend<B>,
A: Send,
B: Send,
{
execute(pi, Unzip)
}
/// Unzips an `IndexedParallelIterator` into two arbitrary `Consumer`s. /// /// This is called by `super::collect::unzip_into_vecs`. pub(super) fn unzip_indexed<I, A, B, CA, CB>(pi: I, left: CA, right: CB) -> (CA::Result, CB::Result) where
I: IndexedParallelIterator<Item = (A, B)>,
CA: Consumer<A>,
CB: Consumer<B>,
A: Send,
B: Send,
{ let consumer = UnzipConsumer {
op: &Unzip,
left,
right,
};
pi.drive(consumer)
}
/// An `UnzipOp` that splits a tuple directly into the two consumers. struct Unzip;
impl<A: Send, B: Send> UnzipOp<(A, B)> for Unzip { type Left = A; type Right = B;
/// Partitions the items of a parallel iterator into a pair of arbitrary /// `ParallelExtend` containers. /// /// This is called by `ParallelIterator::partition`. pub(super) fn partition<I, A, B, P>(pi: I, predicate: P) -> (A, B) where
I: ParallelIterator,
A: Default + Send + ParallelExtend<I::Item>,
B: Default + Send + ParallelExtend<I::Item>,
P: Fn(&I::Item) -> bool + Sync + Send,
{
execute(pi, Partition { predicate })
}
/// An `UnzipOp` that routes items depending on a predicate function. struct Partition<P> {
predicate: P,
}
impl<P, T> UnzipOp<T> for Partition<P> where
P: Fn(&T) -> bool + Sync + Send,
T: Send,
{ type Left = T; type Right = T;
/// A fake iterator to intercept the `Consumer` for type `A`. struct UnzipA<'b, I, OP, FromB> {
base: I,
op: OP,
b: &'b mut FromB,
}
impl<'b, I, OP, FromB> ParallelIterator for UnzipA<'b, I, OP, FromB> where
I: ParallelIterator,
OP: UnzipOp<I::Item>,
FromB: Send + ParallelExtend<OP::Right>,
{ type Item = OP::Left;
fn drive_unindexed<C>(self, consumer: C) -> C::Result where
C: UnindexedConsumer<Self::Item>,
{ letmut result = None;
{ // Now it's time to find the consumer for type `B` let iter = UnzipB {
base: self.base,
op: self.op,
left_consumer: consumer,
left_result: &mut result,
}; self.b.par_extend(iter);
} // NB: If for some reason `b.par_extend` doesn't actually drive the // iterator, then we won't have a result for the left side to return // at all. We can't fake an arbitrary consumer's result, so panic.
result.expect("unzip consumers didn't execute!")
}
/// A fake iterator to intercept the `Consumer` for type `B`. struct UnzipB<'r, I, OP, CA> where
I: ParallelIterator,
OP: UnzipOp<I::Item>,
CA: UnindexedConsumer<OP::Left>,
CA::Result: 'r,
{
base: I,
op: OP,
left_consumer: CA,
left_result: &'r mut Option<CA::Result>,
}
impl<'r, I, OP, CA> ParallelIterator for UnzipB<'r, I, OP, CA> where
I: ParallelIterator,
OP: UnzipOp<I::Item>,
CA: UnindexedConsumer<OP::Left>,
{ type Item = OP::Right;
fn drive_unindexed<C>(self, consumer: C) -> C::Result where
C: UnindexedConsumer<Self::Item>,
{ // Now that we have two consumers, we can unzip the real iterator. let consumer = UnzipConsumer {
op: &self.op,
left: self.left_consumer,
right: consumer,
};
let result = self.base.drive_unindexed(consumer);
*self.left_result = Some(result.0);
result.1
}
/// `Consumer` that unzips into two other `Consumer`s struct UnzipConsumer<'a, OP, CA, CB> {
op: &'a OP,
left: CA,
right: CB,
}
impl<'a, T, OP, CA, CB> Consumer<T> for UnzipConsumer<'a, OP, CA, CB> where
OP: UnzipOp<T>,
CA: Consumer<OP::Left>,
CB: Consumer<OP::Right>,
{ type Folder = UnzipFolder<'a, OP, CA::Folder, CB::Folder>; type Reducer = UnzipReducer<CA::Reducer, CB::Reducer>; type Result = (CA::Result, CB::Result);
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.