1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

use quote::{quote_spanned, ToTokens};
use syn::parse_quote;

use super::{
    DelayType, OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
    OperatorWriteOutput, PortListSpec, WriteContextArgs, RANGE_0, RANGE_1,
};

/// > 2 input streams, 1 output stream, no arguments.
///
/// Batches streaming input and releases it downstream when a signal is delivered. This allows for buffering data and delivering it later while also folding it into a single lattice data structure.
/// This operator is similar to `defer_signal` in that it batches input and releases it when a signal is given. It is also similar to `lattice_fold` in that it folds the input into a single lattice.
/// So, `_lattice_fold_batch` is a combination of both `defer_signal` and `lattice_fold`. This operator is useful when trying to combine a sequence of `defer_signal` and `lattice_fold` operators without unnecessary memory consumption.
///
/// There are two inputs to `_lattice_fold_batch`, they are `input` and `signal`.
/// `input` is the input data flow. Data that is delivered on this input is collected in order inside of the `_lattice_fold_batch` operator.
/// When anything is sent to `signal` the collected data is released downstream. The entire `signal` input is consumed each tick, so sending 5 things on `signal` will not release inputs on the next 5 consecutive ticks.
///
/// ```dfir
/// use lattices::set_union::SetUnionHashSet;
/// use lattices::set_union::SetUnionSingletonSet;
///
/// source_iter([1, 2, 3])
///     -> map(SetUnionSingletonSet::new_from)
///     -> [input]batcher;
///
/// source_iter([()])
///     -> [signal]batcher;
///
/// batcher = _lattice_fold_batch::<SetUnionHashSet<usize>>()
///     -> assert_eq([SetUnionHashSet::new_from([1, 2, 3])]);
/// ```
pub const _LATTICE_FOLD_BATCH: OperatorConstraints = OperatorConstraints {
    name: "_lattice_fold_batch",
    categories: &[OperatorCategory::CompilerFusionOperator],
    persistence_args: RANGE_0,
    type_args: &(0..=1),
    hard_range_inn: &(2..=2),
    soft_range_inn: &(2..=2),
    hard_range_out: RANGE_1,
    soft_range_out: RANGE_1,
    num_args: 0,
    is_external_input: false,
    has_singleton_output: false,
    flo_type: None,
    ports_inn: Some(|| PortListSpec::Fixed(parse_quote! { input, signal })),
    ports_out: None,
    input_delaytype_fn: |_| Some(DelayType::MonotoneAccum),
    write_fn: |wc @ &WriteContextArgs {
                   context,
                   hydroflow,
                   ident,
                   op_span,
                   root,
                   inputs,
                   is_pull,
                   op_inst:
                       OperatorInstance {
                           generics: OpInstGenerics { type_args, .. },
                           ..
                       },
                   ..
               },
               _| {
        assert!(is_pull);

        let lattice_type = type_args
            .first()
            .map(ToTokens::to_token_stream)
            .unwrap_or(quote_spanned!(op_span=> _));

        let lattice_ident = wc.make_ident("lattice");

        let write_prologue = quote_spanned! {op_span=>
            let #lattice_ident = #hydroflow.add_state(::std::cell::RefCell::new(<#lattice_type as ::std::default::Default>::default()));
        };

        let input = &inputs[0];
        let signal = &inputs[1];

        let write_iterator = {
            quote_spanned! {op_span=>

                {
                    let mut __lattice = #context.state_ref(#lattice_ident).borrow_mut();

                    for __item in #input {
                        #root::lattices::Merge::merge(&mut *__lattice, __item);
                    }
                }

                let #ident = if #signal.count() > 0 {
                    ::std::option::Option::Some(#context.state_ref(#lattice_ident).take())
                } else {
                    ::std::option::Option::None
                }.into_iter();
            }
        };

        Ok(OperatorWriteOutput {
            write_prologue,
            write_iterator,
            write_iterator_after: Default::default(),
            ..Default::default()
        })
    },
};