dfir_lang/graph/ops/
lattice_reduce.rs

Help
1use quote::quote_spanned;
2use syn::parse_quote_spanned;
3
4use super::{
5    DelayType, OperatorCategory, OperatorConstraints, OperatorWriteOutput, RANGE_0, RANGE_1,
6    WriteContextArgs,
7};
8
9/// > 1 input stream, 1 output stream
10///
11/// A specialized operator for merging lattices together into an accumulated value. Like [`reduce()`](#reduce)
12/// but specialized for lattice types. `lattice_reduce()` is equivalent to `reduce(dfir_rs::lattices::Merge::merge)`.
13///
14/// `lattice_reduce` can also be provided with one generic lifetime persistence argument, either
15/// `'tick` or `'static`, to specify how data persists. With `'tick`, values will only be collected
16/// within the same tick. With `'static`, values will be remembered across ticks and will be
17/// aggregated with pairs arriving in later ticks. When not explicitly specified persistence
18/// defaults to `'tick`.
19///
20/// `lattice_reduce` is differentiated from `lattice_fold` in that `lattice_reduce` does not require the accumulating type to have a sensible default value.
21/// But it also means that the accumulating function inputs and the accumulating type must be the same.
22///
23/// ```dfir
24/// use dfir_rs::lattices::Max;
25///
26/// source_iter([1, 2, 3, 4, 5])
27///     -> map(Max::new)
28///     -> lattice_reduce()
29///     -> assert_eq([Max::new(5)]);
30/// ```
31pub const LATTICE_REDUCE: OperatorConstraints = OperatorConstraints {
32    name: "lattice_reduce",
33    categories: &[OperatorCategory::LatticeFold],
34    hard_range_inn: RANGE_1,
35    soft_range_inn: RANGE_1,
36    hard_range_out: RANGE_1,
37    soft_range_out: RANGE_1,
38    num_args: 0,
39    persistence_args: &(0..=1),
40    type_args: RANGE_0,
41    is_external_input: false,
42    has_singleton_output: false,
43    flo_type: None,
44    ports_inn: None,
45    ports_out: None,
46    input_delaytype_fn: |_| Some(DelayType::MonotoneAccum),
47    write_fn: |wc @ &WriteContextArgs {
48                   root,
49                   inputs,
50                   op_span,
51                   is_pull,
52                   ..
53               },
54               diagnostics| {
55        assert!(is_pull);
56
57        let arguments = &parse_quote_spanned! {op_span=>
58            |acc, item| { #root::lattices::Merge::<_>::merge(acc, item); }
59        };
60        let wc = WriteContextArgs {
61            arguments,
62            ..wc.clone()
63        };
64
65        let OperatorWriteOutput {
66            write_prologue,
67            write_prologue_after,
68            write_iterator,
69            write_iterator_after,
70        } = (super::reduce::REDUCE.write_fn)(&wc, diagnostics)?;
71
72        assert_eq!(1, inputs.len());
73        let input = &inputs[0];
74
75        let write_iterator = quote_spanned! {op_span=>
76            let #input = {
77                #[inline(always)]
78                fn check_inputs<Lat>(
79                    input: impl ::std::iter::Iterator<Item = Lat>
80                ) -> impl ::std::iter::Iterator<Item = Lat>
81                where
82                    Lat: #root::lattices::Merge<Lat>,
83                {
84                    input
85                }
86                check_inputs::<_>(#input)
87            };
88            #write_iterator
89        };
90        Ok(OperatorWriteOutput {
91            write_prologue,
92            write_prologue_after,
93            write_iterator,
94            write_iterator_after,
95        })
96    },
97};
dfir_lang/graph/ops/lattice_reduce.rs

dfir_lang/graph/ops/
lattice_reduce.rs
