Revision control
Copy as Markdown
Other Tools
use super::plumbing::*;
use super::*;
use std::sync::atomic::{AtomicBool, Ordering};
pub(super) fn find<I, P>(pi: I, find_op: P) -> Option<I::Item>
where
I: ParallelIterator,
P: Fn(&I::Item) -> bool + Sync,
{
let found = AtomicBool::new(false);
let consumer = FindConsumer::new(&find_op, &found);
pi.drive_unindexed(consumer)
}
struct FindConsumer<'p, P> {
find_op: &'p P,
found: &'p AtomicBool,
}
impl<'p, P> FindConsumer<'p, P> {
fn new(find_op: &'p P, found: &'p AtomicBool) -> Self {
FindConsumer { find_op, found }
}
}
impl<'p, T, P: 'p> Consumer<T> for FindConsumer<'p, P>
where
T: Send,
P: Fn(&T) -> bool + Sync,
{
type Folder = FindFolder<'p, T, P>;
type Reducer = FindReducer;
type Result = Option<T>;
fn split_at(self, _index: usize) -> (Self, Self, Self::Reducer) {
(self.split_off_left(), self, FindReducer)
}
fn into_folder(self) -> Self::Folder {
FindFolder {
find_op: self.find_op,
found: self.found,
item: None,
}
}
fn full(&self) -> bool {
self.found.load(Ordering::Relaxed)
}
}
impl<'p, T, P: 'p> UnindexedConsumer<T> for FindConsumer<'p, P>
where
T: Send,
P: Fn(&T) -> bool + Sync,
{
fn split_off_left(&self) -> Self {
FindConsumer::new(self.find_op, self.found)
}
fn to_reducer(&self) -> Self::Reducer {
FindReducer
}
}
struct FindFolder<'p, T, P> {
find_op: &'p P,
found: &'p AtomicBool,
item: Option<T>,
}
impl<'p, T, P> Folder<T> for FindFolder<'p, T, P>
where
P: Fn(&T) -> bool + 'p,
{
type Result = Option<T>;
fn consume(mut self, item: T) -> Self {
if (self.find_op)(&item) {
self.found.store(true, Ordering::Relaxed);
self.item = Some(item);
}
self
}
fn consume_iter<I>(mut self, iter: I) -> Self
where
I: IntoIterator<Item = T>,
{
fn not_full<T>(found: &AtomicBool) -> impl Fn(&T) -> bool + '_ {
move |_| !found.load(Ordering::Relaxed)
}
self.item = iter
.into_iter()
// stop iterating if another thread has found something
.take_while(not_full(self.found))
.find(self.find_op);
if self.item.is_some() {
self.found.store(true, Ordering::Relaxed)
}
self
}
fn complete(self) -> Self::Result {
self.item
}
fn full(&self) -> bool {
self.found.load(Ordering::Relaxed)
}
}
struct FindReducer;
impl<T> Reducer<Option<T>> for FindReducer {
fn reduce(self, left: Option<T>, right: Option<T>) -> Option<T> {
left.or(right)
}
}