use crate::ParExtend;
use crate::infallible::{FilMapOf, FilOf, FlatMapOf, FlattenOf, InsOf, MapOf, Xap};
use crate::option::XapOptionIter;
use crate::option::par::ParOption;
use crate::option::par_core::ParOptionCore;
use crate::option::par_runner::ParRunnerOpt;
use crate::parameters::{ChunkSize, IterationOrder, NumThreads, Params};
use crate::runner::{DefaultRunner, ParRunner};
use crate::sizes::SizePair;
use orx_concurrent_iter::ConcurrentIter;
pub struct ParOptionIter<I, M, X1, X2, S, R = DefaultRunner>
where
I: ConcurrentIter,
X1: Xap<I = I::Item, O = Option<M>>,
X2: Xap<I = M>,
S: SizePair<S1 = X1::Size, S2 = X2::Size>,
R: ParRunner,
{
iter: I,
x1: X1,
x2: X2,
exe: R,
params: Params,
s: S,
}
impl<I, M, X1, X2, S, R> ParOptionIter<I, M, X1, X2, S, R>
where
I: ConcurrentIter,
X1: Xap<I = I::Item, O = Option<M>>,
X2: Xap<I = M>,
S: SizePair<S1 = X1::Size, S2 = X2::Size>,
R: ParRunner,
{
pub(crate) fn new(iter: I, x1: X1, x2: X2, exe: R, params: Params) -> Self {
Self {
iter,
x1,
x2,
exe,
params,
s: Default::default(),
}
}
fn with_xap2<Y2, T>(self, x2: Y2) -> ParOptionIter<I, M, X1, Y2, T, R>
where
Y2: Xap<I = M>,
T: SizePair<S1 = X1::Size, S2 = Y2::Size>,
{
ParOptionIter::new(self.iter, self.x1, x2, self.exe, self.params)
}
}
impl<I, M, X1, X2, S, R> IntoIterator for ParOptionIter<I, M, X1, X2, S, R>
where
I: ConcurrentIter,
X1: Xap<I = I::Item, O = Option<M>>,
X2: Xap<I = M>,
S: SizePair<S1 = X1::Size, S2 = X2::Size>,
R: ParRunner,
{
type Item = Option<X2::O>;
type IntoIter = XapOptionIter<I::SequentialIter, M, X1, X2, S>;
fn into_iter(self) -> Self::IntoIter {
XapOptionIter::new(self.iter.into_seq_iter(), self.x1, self.x2)
}
}
impl<I, M, X1, X2, S, R> ParOptionCore for ParOptionIter<I, M, X1, X2, S, R>
where
I: ConcurrentIter,
X1: Xap<I = I::Item, O = Option<M>>,
X2: Xap<I = M>,
S: SizePair<S1 = X1::Size, S2 = X2::Size>,
R: ParRunner,
{
type Elem = X2::O;
type Runner = R;
type Input = I;
type M = M;
type Xap1 = X1;
type Xap2 = X2;
type Size = S;
fn destruct(
self,
) -> (
Self::Input,
Self::Xap1,
Self::Xap2,
Self::Runner,
Self::Size,
Params,
) {
(self.iter, self.x1, self.x2, self.exe, self.s, self.params)
}
}
impl<I, M, X1, X2, S, R> ParOption for ParOptionIter<I, M, X1, X2, S, R>
where
I: ConcurrentIter,
X1: Xap<I = I::Item, O = Option<M>>,
X2: Xap<I = M>,
S: SizePair<S1 = X1::Size, S2 = X2::Size>,
R: ParRunner,
{
fn runner<Q: ParRunner>(
self,
runner: Q,
) -> impl ParOption<
Elem = Self::Elem,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = Self::Xap2,
Input = Self::Input,
Size = Self::Size,
> {
let (iter, x1, x2, _, s, params) = self.destruct();
ParOptionIter {
iter,
x1,
x2,
exe: runner,
s,
params,
}
}
#[cfg(feature = "std")]
fn runner_with_diagnostics(
self,
) -> impl ParOption<
Elem = Self::Elem,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = Self::Xap2,
Input = Self::Input,
Size = Self::Size,
> {
let (iter, x1, x2, exe, s, params) = self.destruct();
ParOptionIter {
iter,
x1,
x2,
exe: exe.with_diagnostics(),
s,
params,
}
}
fn num_threads(mut self, num_threads: impl Into<NumThreads>) -> Self {
self.params = self.params.with_num_threads(num_threads);
self
}
fn chunk_size(mut self, chunk_size: impl Into<ChunkSize>) -> Self {
self.params = self.params.with_chunk_size(chunk_size);
self
}
fn iteration_order(mut self, collect: IterationOrder) -> Self {
self.params = self.params.with_collect_ordering(collect);
self
}
fn map<Q, H>(
self,
h: H,
) -> impl ParOption<
Elem = Q,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = MapOf<Self::Xap2, Q, H>,
Input = Self::Input,
Size = Self::Size,
>
where
H: Fn(X2::O) -> Q + Copy + Send,
{
let x2 = self.x2.map(h);
self.with_xap2(x2)
}
fn inspect<H>(
self,
h: H,
) -> impl ParOption<
Elem = Self::Elem,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = InsOf<Self::Xap2, H>,
Input = Self::Input,
Size = Self::Size,
>
where
H: Fn(&X2::O) + Copy + Send,
{
let x2 = self.x2.inspect(h);
self.with_xap2(x2)
}
fn filter<H>(
self,
h: H,
) -> impl ParOption<
Elem = Self::Elem,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = FilOf<Self::Xap2, H>,
Input = Self::Input,
Size = <Self::Size as SizePair>::ThenBin,
>
where
H: Fn(&X2::O) -> bool + Copy + Send,
{
let x2 = self.x2.filter(h);
self.with_xap2(x2)
}
fn filter_map<Q, H>(
self,
h: H,
) -> impl ParOption<
Elem = Q,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = FilMapOf<Self::Xap2, Q, H>,
Input = Self::Input,
Size = <Self::Size as SizePair>::ThenBin,
>
where
H: Fn(X2::O) -> Option<Q> + Copy + Send,
{
let x2 = self.x2.filter_map(h);
self.with_xap2(x2)
}
fn flat_map<V, H>(
self,
h: H,
) -> impl ParOption<
Elem = V::Item,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = FlatMapOf<Self::Xap2, V, H>,
Input = Self::Input,
Size = <Self::Size as SizePair>::ThenMany,
>
where
V: IntoIterator,
H: Fn(X2::O) -> V + Copy + Send,
{
let x2 = self.x2.flat_map(h);
self.with_xap2(x2)
}
fn flatten(
self,
) -> impl ParOption<
Elem = <Self::Elem as IntoIterator>::Item,
Xap1 = Self::Xap1,
M = Self::M,
Xap2 = FlattenOf<Self::Xap2>,
Input = Self::Input,
Size = <Self::Size as SizePair>::ThenMany,
>
where
Self::Elem: IntoIterator,
{
let x2 = self.x2.flatten();
self.with_xap2(x2)
}
fn size_hint(&self) -> (usize, Option<usize>) {
<S as SizePair>::transformed_size_hint(self.iter.size_hint())
}
fn first(self) -> Option<Option<X2::O>>
where
X2::O: Send,
{
let (iter, x1, x2, mut exe, s, params) = self.destruct();
match params.iteration_order {
IterationOrder::Ordered => exe.next(s, params, iter, x1, x2).map(|x| x.map(|x| x.val)),
IterationOrder::Arbitrary => exe.next_any(s, params, iter, x1, x2),
}
}
fn reduce<F>(self, f: F) -> Option<Option<X2::O>>
where
F: Fn(X2::O, X2::O) -> X2::O + Send + Copy,
X2::O: Send,
{
let (iter, x1, x2, mut exe, s, params) = self.destruct();
exe.reduce(s, params, iter, x1, x2, f)
}
fn collect_into<P>(self, dst: &mut P) -> Option<()>
where
P: ParExtend<X2::O>,
X2::O: Send,
{
let (iter, x1, x2, mut exe, s, params) = self.destruct();
match params.iteration_order {
IterationOrder::Ordered => exe.collect(s, params, iter, x1, x2, dst),
IterationOrder::Arbitrary => exe.collect_arb(s, params, iter, x1, x2, dst),
}
}
}