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
use quote::quote_spanned;
use syn::parse_quote;
use super::{
DelayType, OperatorCategory, OperatorConstraints, OperatorWriteOutput, RANGE_0, RANGE_1,
WriteContextArgs,
};
/// > 2 input streams, 1 output stream, no arguments.
///
/// Defers streaming input and releases it downstream when a signal is delivered. The order of input is preserved. This allows for buffering data and delivering it at a later, chosen, tick.
///
/// There are two inputs to `defer_signal`, 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 `defer_signal` 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
/// gate = defer_signal();
///
/// source_iter([1, 2, 3]) -> [input]gate;
/// source_iter([()]) -> [signal]gate;
///
/// gate -> assert_eq([1, 2, 3]);
/// ```
pub const DEFER_SIGNAL: OperatorConstraints = OperatorConstraints {
name: "defer_signal",
categories: &[OperatorCategory::Persistence],
persistence_args: RANGE_0,
type_args: RANGE_0,
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(|| super::PortListSpec::Fixed(parse_quote! { input, signal })),
ports_out: None,
input_delaytype_fn: |_| Some(DelayType::Stratum),
write_fn: |wc @ &WriteContextArgs {
root,
context,
df_ident,
ident,
op_span,
work_fn_async,
inputs,
is_pull,
..
},
_| {
assert!(is_pull);
let buffer_ident = wc.make_ident("buffer");
let borrow_ident = wc.make_ident("borrow");
let signal_ident = wc.make_ident("signal");
// TODO(mingwei): different lifetimes? `'tick`?
let write_prologue = quote_spanned! {op_span=>
let #buffer_ident = #df_ident.add_state(::std::cell::RefCell::new(::std::vec::Vec::new()));
};
let input = &inputs[0];
let signal = &inputs[1];
let write_iterator = {
quote_spanned! {op_span=>
let mut #borrow_ident = unsafe {
// SAFETY: handle from `#df_ident.add_state(..)`.
#context.state_ref_unchecked(#buffer_ident)
}.borrow_mut();
// Eagerly consume input to ensure updated state.
{
let fut = #root::dfir_pipes::pull::Pull::for_each(#input, |item| {
::std::vec::Vec::push(&mut *#borrow_ident, item);
});
let () = #work_fn_async(fut).await;
}
let #signal_ident = {
// Short-circuit after first signal message.
let fut = #root::dfir_pipes::pull::Pull::next(#signal);
#work_fn_async(fut).await.is_some()
};
let #ident = #root::dfir_pipes::pull::iter(if #signal_ident {
#borrow_ident.drain(..)
} else {
#borrow_ident.drain(..0) // Hack for empty.
});
}
};
Ok(OperatorWriteOutput {
write_prologue,
write_iterator,
write_iterator_after: Default::default(),
..Default::default()
})
},
};