Skip to content

Commit 93d6298

Browse files
MingweiSamuelMingweiSamuel
committed
refactor(dfir_lang): remove stratum, add push codegen, test reduce, reduce_keyed, reduce_no_replay 
1 parent ecf3876 commit 93d6298

6 files changed

Lines changed: 122 additions & 19 deletions

File tree

‎dfir_lang/src/graph/ops/reduce.rs‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use quote::quote_spanned;
22

33
use super::{
4-
DelayType, OperatorCategory, OperatorConstraints, OperatorWriteOutput, Persistence, RANGE_0,
4+
OperatorCategory, OperatorConstraints, OperatorWriteOutput, Persistence, RANGE_0,
55
RANGE_1, WriteContextArgs,
66
};
77

@@ -44,14 +44,15 @@ pub const REDUCE: OperatorConstraints = OperatorConstraints {
4444
flo_type: None,
4545
ports_inn: None,
4646
ports_out: None,
47-
input_delaytype_fn: |_| Some(DelayType::Stratum),
47+
input_delaytype_fn: |_| None,
4848
write_fn: |wc @ &WriteContextArgs {
4949
root,
5050
op_span,
5151
work_fn,
5252
work_fn_async,
5353
ident,
5454
inputs,
55+
outputs,
5556
is_pull,
5657
arguments,
5758
..
@@ -117,15 +118,30 @@ pub const REDUCE: OperatorConstraints = OperatorConstraints {
117118
)
118119
);
119120
}
120-
} else {
121-
// Is only push when used as a singleton, so no need to push to `outputs[0]`.
121+
} else if outputs.is_empty() {
122+
// Terminal push: reduce is a singleton reference target with no downstream.
122123
quote_spanned! {op_span=>
123124
let #ident = #root::dfir_pipes::push::for_each(|#item_ident| {
124125
#assign_accum_ident
125126

126127
#foreach_body
127128
});
128129
}
130+
} else {
131+
let output = &outputs[0];
132+
quote_spanned! {op_span=>
133+
let #ident = #root::dfir_pipes::push::reduce_ref(
134+
&mut #singleton_output_ident,
135+
|#accumulator_ident: &mut _, #item_ident| {
136+
#[allow(clippy::redundant_closure_call)]
137+
(#func)(#accumulator_ident, #item_ident);
138+
},
139+
#root::dfir_pipes::push::map(
140+
|__val: &mut _| ::std::clone::Clone::clone(&*__val),
141+
#output,
142+
),
143+
);
144+
}
129145
};
130146

131147
Ok(OperatorWriteOutput {

‎dfir_lang/src/graph/ops/reduce_keyed.rs‎

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use quote::{ToTokens, quote_spanned};
22

33
use super::{
4-
DelayType, OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
4+
OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
55
OperatorWriteOutput, Persistence, RANGE_1, WriteContextArgs,
66
};
77

@@ -68,15 +68,15 @@ pub const REDUCE_KEYED: OperatorConstraints = OperatorConstraints {
6868
flo_type: None,
6969
ports_inn: None,
7070
ports_out: None,
71-
input_delaytype_fn: |_| Some(DelayType::Stratum),
71+
input_delaytype_fn: |_| None,
7272
write_fn: |wc @ &WriteContextArgs {
7373
op_span,
7474
ident,
7575
inputs,
76+
outputs,
7677
is_pull,
7778
work_fn_async,
7879
root,
79-
op_name,
8080
op_inst:
8181
OperatorInstance {
8282
generics: OpInstGenerics { type_args, .. },
@@ -86,8 +86,6 @@ pub const REDUCE_KEYED: OperatorConstraints = OperatorConstraints {
8686
..
8787
},
8888
diagnostics| {
89-
assert!(is_pull, "TODO(mingwei): `{}` only supports pull.", op_name);
90-
9189
let [persistence] = wc.persistence_args_disallow_mutable(diagnostics);
9290

9391
let generic_type_args = [
@@ -118,7 +116,16 @@ pub const REDUCE_KEYED: OperatorConstraints = OperatorConstraints {
118116
_ => Default::default(),
119117
};
120118

121-
let write_iterator = {
119+
let write_iterator = if !is_pull {
120+
let output = &outputs[0];
121+
quote_spanned! {op_span=>
122+
let #ident = #root::dfir_pipes::push::ReduceKeyed::new(
123+
&mut #singleton_output_ident,
124+
#aggfn,
125+
#output,
126+
);
127+
}
128+
} else {
122129
let iter_expr = match persistence {
123130
Persistence::None | Persistence::Tick => quote_spanned! {op_span=>
124131
#hashtable_ident.drain()

‎dfir_lang/src/graph/ops/reduce_no_replay.rs‎

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
use quote::quote_spanned;
22

33
use super::{
4-
DelayType, OperatorCategory, OperatorConstraints, OperatorWriteOutput, Persistence, RANGE_0,
5-
RANGE_1, WriteContextArgs,
4+
OperatorCategory, OperatorConstraints, OperatorWriteOutput, Persistence, RANGE_0, RANGE_1,
5+
WriteContextArgs,
66
};
77

88
/// > 1 input stream, 1 output stream
@@ -27,7 +27,7 @@ pub const REDUCE_NO_REPLAY: OperatorConstraints = OperatorConstraints {
2727
flo_type: None,
2828
ports_inn: None,
2929
ports_out: None,
30-
input_delaytype_fn: |_| Some(DelayType::Stratum),
30+
input_delaytype_fn: |_| None,
3131
write_fn: |wc @ &WriteContextArgs {
3232
root,
3333
context,
@@ -36,6 +36,7 @@ pub const REDUCE_NO_REPLAY: OperatorConstraints = OperatorConstraints {
3636
work_fn_async,
3737
ident,
3838
inputs,
39+
outputs,
3940
is_pull,
4041
arguments,
4142
..
@@ -108,15 +109,40 @@ pub const REDUCE_NO_REPLAY: OperatorConstraints = OperatorConstraints {
108109
)
109110
};
110111
}
111-
} else {
112-
// Is only push when used as a singleton, so no need to push to `outputs[0]`.
112+
} else if outputs.is_empty() {
113+
// Terminal push: reduce_no_replay is a singleton reference target with no downstream.
113114
quote_spanned! {op_span=>
114115
let #ident = #root::dfir_pipes::push::for_each(|#item_ident| {
115116
#assign_accum_ident
116117

117118
#foreach_body
118119
});
119120
}
121+
} else {
122+
let output = &outputs[0];
123+
let was_updated_ident = wc.make_ident("was_updated");
124+
quote_spanned! {op_span=>
125+
let #was_updated_ident = ::std::cell::Cell::new(false);
126+
let #ident = #root::dfir_pipes::push::reduce_ref(
127+
&mut #singleton_output_ident,
128+
|#accumulator_ident: &mut _, #item_ident| {
129+
#was_updated_ident.set(true);
130+
#[allow(clippy::redundant_closure_call)]
131+
(#func)(#accumulator_ident, #item_ident);
132+
},
133+
#root::dfir_pipes::push::filter(
134+
{
135+
let __was_updated = &#was_updated_ident;
136+
let __context: &_ = #context;
137+
move |_| __was_updated.get() || __context.current_tick().0 == 0
138+
},
139+
#root::dfir_pipes::push::map(
140+
|__val: &mut _| ::std::clone::Clone::clone(&*__val),
141+
#output,
142+
),
143+
),
144+
);
145+
}
120146
};
121147

122148
Ok(OperatorWriteOutput {

‎dfir_rs/tests/compile-fail/stable/surface_reduce_keyed_badtype_int.stderr‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
error[E0271]: type mismatch resolving `<Iter<Drain<'_, '_, {integer}>> as Pull>::Item == (_, _)`
1+
error[E0271]: type mismatch resolving `<impl Pull<Item = {integer}, Meta = (), CanPend = <Iter<&mut impl Iterator<Item = {integer}>> as Pull>::CanPend, CanEnd = <Iter<&mut impl Iterator<Item = {integer}>> as Pull>::CanEnd> as Pull>::Item == (_, _)`
22
--> tests/compile-fail/stable/surface_reduce_keyed_badtype_int.rs:3:9
33
|
44
3 | source_iter(0..1)
@@ -14,7 +14,7 @@ note: required by a bound in `check_input`
1414
4 | -> fold_keyed(|| 0, |old: &mut u32, val: u32| { *old += val; })
1515
| ^^^^^^^^^^ required by this bound in `check_input`
1616

17-
error[E0271]: type mismatch resolving `<Iter<Drain<'_, '_, {integer}>> as Pull>::Item == (_, _)`
17+
error[E0271]: type mismatch resolving `<impl Pull<Item = {integer}, Meta = (), CanPend = <Iter<&mut impl Iterator<Item = {integer}>> as Pull>::CanPend, CanEnd = <Iter<&mut impl Iterator<Item = {integer}>> as Pull>::CanEnd> as Pull>::Item == (_, _)`
1818
--> tests/compile-fail/stable/surface_reduce_keyed_badtype_int.rs:4:16
1919
|
2020
4 | -> fold_keyed(|| 0, |old: &mut u32, val: u32| { *old += val; })

‎dfir_rs/tests/compile-fail/stable/surface_reduce_keyed_badtype_option.stderr‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
error[E0271]: type mismatch resolving `<Iter<Drain<'_, '_, Option<{integer}>>> as Pull>::Item == (_, _)`
1+
error[E0271]: type mismatch resolving `<impl Pull<Item = Option<{integer}>, Meta = (), CanPend = <Iter<&mut impl Iterator<Item = Option<{integer}>>> as Pull>::CanPend, CanEnd = <Iter<&mut impl Iterator<Item = Option<{integer}>>> as Pull>::CanEnd> as Pull>::Item == (_, _)`
22
--> tests/compile-fail/stable/surface_reduce_keyed_badtype_option.rs:3:9
33
|
44
3 | source_iter([ Some(5), None, Some(12) ])
@@ -14,7 +14,7 @@ note: required by a bound in `check_input`
1414
4 | -> fold_keyed(|| 0, |old: &mut u32, val: u32| { *old += val; })
1515
| ^^^^^^^^^^ required by this bound in `check_input`
1616

17-
error[E0271]: type mismatch resolving `<Iter<Drain<'_, '_, Option<{integer}>>> as Pull>::Item == (_, _)`
17+
error[E0271]: type mismatch resolving `<impl Pull<Item = Option<{integer}>, Meta = (), CanPend = <Iter<&mut impl Iterator<Item = Option<{integer}>>> as Pull>::CanPend, CanEnd = <Iter<&mut impl Iterator<Item = Option<{integer}>>> as Pull>::CanEnd> as Pull>::Item == (_, _)`
1818
--> tests/compile-fail/stable/surface_reduce_keyed_badtype_option.rs:4:16
1919
|
2020
4 | -> fold_keyed(|| 0, |old: &mut u32, val: u32| { *old += val; })

‎dfir_rs/tests/surface_push_blocking.rs‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,19 @@ use dfir_rs::dfir_syntax;
55
use dfir_rs::util::collect_ready;
66
use multiplatform_test::multiplatform_test;
77

8+
/// reduce on push side: source -> tee -> reduce -> for_each
9+
#[multiplatform_test]
10+
pub fn test_reduce_push() {
11+
let (out_send, mut out_recv) = dfir_rs::util::unbounded_channel::<i32>();
12+
let mut df = dfir_syntax! {
13+
my_tee = source_iter([1, 2, 3]) -> tee();
14+
my_tee -> reduce(|a: &mut _, b| *a += b) -> for_each(|v| out_send.send(v).unwrap());
15+
my_tee -> null();
16+
};
17+
df.run_available_sync();
18+
assert_eq!(&[6], &*collect_ready::<Vec<_>, _>(&mut out_recv));
19+
}
20+
821
/// fold_keyed on push side
922
#[multiplatform_test]
1023
pub fn test_fold_keyed_push() {
@@ -20,6 +33,21 @@ pub fn test_fold_keyed_push() {
2033
assert_eq!(&[(1, 30), (2, 30)], &*out);
2134
}
2235

36+
/// reduce_keyed on push side
37+
#[multiplatform_test]
38+
pub fn test_reduce_keyed_push() {
39+
let (out_send, mut out_recv) = dfir_rs::util::unbounded_channel::<(i32, i32)>();
40+
let mut df = dfir_syntax! {
41+
my_tee = source_iter([(1, 10), (1, 20), (2, 30)]) -> tee();
42+
my_tee -> reduce_keyed(|a: &mut _, b| *a += b) -> for_each(|v| out_send.send(v).unwrap());
43+
my_tee -> null();
44+
};
45+
df.run_available_sync();
46+
let mut out = collect_ready::<Vec<_>, _>(&mut out_recv);
47+
out.sort();
48+
assert_eq!(&[(1, 30), (2, 30)], &*out);
49+
}
50+
2351
/// fold on push side: source -> tee -> fold -> for_each
2452
#[multiplatform_test]
2553
pub fn test_fold_push() {
@@ -59,3 +87,29 @@ pub fn test_fold_no_replay_push() {
5987
assert_eq!(&[13], &*collect_ready::<Vec<_>, _>(&mut out_recv));
6088
}
6189

90+
/// reduce_no_replay on push side: source -> tee -> reduce_no_replay -> for_each
91+
#[multiplatform_test]
92+
pub fn test_reduce_no_replay_push() {
93+
let (items_send, items_recv) = dfir_rs::util::unbounded_channel::<i32>();
94+
let (out_send, mut out_recv) = dfir_rs::util::unbounded_channel::<i32>();
95+
let mut df = dfir_syntax! {
96+
my_tee = source_stream(items_recv) -> tee();
97+
my_tee -> reduce_no_replay::<'static>(|a: &mut _, b| *a += b) -> for_each(|v| out_send.send(v).unwrap());
98+
my_tee -> null();
99+
};
100+
101+
items_send.send(1).unwrap();
102+
items_send.send(2).unwrap();
103+
df.run_tick_sync();
104+
assert_eq!(&[3], &*collect_ready::<Vec<_>, _>(&mut out_recv));
105+
106+
// No new input: reduce_no_replay should NOT emit.
107+
df.run_tick_sync();
108+
assert_eq!(&[] as &[i32], &*collect_ready::<Vec<_>, _>(&mut out_recv));
109+
110+
// New input arrives: should emit updated accumulator.
111+
items_send.send(10).unwrap();
112+
df.run_tick_sync();
113+
assert_eq!(&[13], &*collect_ready::<Vec<_>, _>(&mut out_recv));
114+
}
115+

0 commit comments

Comments
 (0)