indicator/rayon/
ticked.rs

1use crate::{Operator, TickValue, Tickable};
2use rayon::prelude::*;
3use std::collections::HashMap;
4use std::hash::Hash;
5
6/// [`Facet`] combinator.
7#[derive(Debug, Clone, Copy)]
8pub struct Facet<I, P1, P2>(
9    pub(super) P1,
10    pub(super) P2,
11    pub(super) core::marker::PhantomData<fn() -> I>,
12);
13
14/// Combine two ticked operators into a [`Facet`] ticked operator.
15pub fn facet_t<I, P1, P2>(op1: P1, op2: P2) -> Facet<I, P1, P2> {
16    Facet(op1, op2, core::marker::PhantomData)
17}
18
19impl<I, P1, P2> Operator<I> for Facet<I, P1, P2>
20where
21    I: Tickable + Clone + Send,
22    P1: Operator<I> + Send,
23    P2: Operator<I> + Send,
24    P1::Output: Tickable + Send,
25    P2::Output: Tickable + Send,
26    <P1::Output as Tickable>::Value: Send,
27    <P2::Output as Tickable>::Value: Send,
28{
29    type Output = TickValue<(
30        <P1::Output as Tickable>::Value,
31        <P2::Output as Tickable>::Value,
32    )>;
33
34    fn next(&mut self, input: I) -> Self::Output {
35        let tick = input.tick();
36        let i1 = input.clone();
37        let (o1, o2) = rayon::join(
38            || self.0.next(i1).into_tick_value().value,
39            || self.1.next(input).into_tick_value().value,
40        );
41        TickValue {
42            tick,
43            value: (o1, o2),
44        }
45    }
46}
47
48/// [`FacetMap`] combinator.
49#[derive(Debug, Clone)]
50pub struct FacetMap<I, Q, P>(HashMap<Q, P>, core::marker::PhantomData<fn() -> I>);
51
52impl<I, Q, P> Operator<I> for FacetMap<I, Q, P>
53where
54    I: Tickable + Clone + Sync,
55    Q: Eq + Hash + Clone + Sync + Send,
56    P: Operator<I> + Send,
57    P::Output: Tickable,
58    <P::Output as Tickable>::Value: Send,
59{
60    type Output = TickValue<HashMap<Q, <P::Output as Tickable>::Value>>;
61
62    fn next(&mut self, input: I) -> Self::Output {
63        let tick = input.tick();
64        let value = self
65            .0
66            .par_iter_mut()
67            .map(|(k, p)| {
68                let o = p.next(input.clone()).into_tick_value().value;
69                (k.clone(), o)
70            })
71            .collect();
72        TickValue { tick, value }
73    }
74}
75
76/// Create an operator that apply different operators to the same input,
77/// and return the collections of outputs as its output.
78pub fn facet_map_t<I, It, Q, P>(ops: It) -> FacetMap<I, Q, P>
79where
80    It: IntoIterator<Item = (Q, P)>,
81    I: Tickable + Clone + Sync,
82    Q: Eq + Hash + Clone + Sync + Send,
83    P: Operator<I> + Send,
84    P::Output: Tickable,
85    <P::Output as Tickable>::Value: Send,
86{
87    FacetMap(ops.into_iter().collect(), core::marker::PhantomData)
88}
indicator/rayon/ticked.rs

indicator/rayon/
ticked.rs
