Skip to content

Commit 5f1f6ac

Browse files
MingweiSamuelMingweiSamuel
committed
refactor(dfir_lang): remove stratum, add push codegen, test persist_mut, persist_mut_keyed
PR: #2969
1 parent 341b774 commit 5f1f6ac

12 files changed

Lines changed: 173 additions & 147 deletions

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

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

33
use super::{
4-
DelayType, OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
4+
OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
55
OperatorWriteOutput, Persistence, RANGE_0, RANGE_1, WriteContextArgs,
66
};
77
use crate::diagnostic::{Diagnostic, Level};
@@ -41,13 +41,14 @@ pub const PERSIST_MUT: OperatorConstraints = OperatorConstraints {
4141
flo_type: None,
4242
ports_inn: None,
4343
ports_out: None,
44-
input_delaytype_fn: |_| Some(DelayType::Stratum),
44+
input_delaytype_fn: |_| None,
4545
write_fn: |wc @ &WriteContextArgs {
4646
root,
4747
op_span,
4848
work_fn_async,
4949
ident,
5050
inputs,
51+
outputs,
5152
is_pull,
5253
op_name,
5354
op_inst:
@@ -61,8 +62,6 @@ pub const PERSIST_MUT: OperatorConstraints = OperatorConstraints {
6162
..
6263
},
6364
diagnostics| {
64-
assert!(is_pull);
65-
6665
if [Persistence::Mutable] != persistence_args[..] {
6766
diagnostics.push(Diagnostic::spanned(
6867
op_span,
@@ -80,7 +79,7 @@ pub const PERSIST_MUT: OperatorConstraints = OperatorConstraints {
8079
let mut #persistdata_ident = #root::util::sparse_vec::SparseVec::default();
8180
};
8281

83-
let write_iterator = {
82+
let write_iterator = if is_pull {
8483
let input = &inputs[0];
8584
quote_spanned! {op_span=>
8685
let #ident = {
@@ -107,6 +106,23 @@ pub const PERSIST_MUT: OperatorConstraints = OperatorConstraints {
107106
#root::dfir_pipes::pull::iter(iter)
108107
};
109108
}
109+
} else {
110+
let output = &outputs[0];
111+
quote_spanned! {op_span=>
112+
let #ident = #root::dfir_pipes::push::fold(
113+
&mut #persistdata_ident,
114+
|state: &mut #root::util::sparse_vec::SparseVec<_>, item| {
115+
match item {
116+
#root::util::Persistence::Persist(v) => state.push(v),
117+
#root::util::Persistence::Delete(v) => state.delete(&v),
118+
}
119+
},
120+
#root::dfir_pipes::push::flat_map(
121+
|state: &mut #root::util::sparse_vec::SparseVec<_>| state.iter().cloned().collect::<::std::vec::Vec<_>>(),
122+
#output,
123+
),
124+
);
125+
}
110126
};
111127

112128
Ok(OperatorWriteOutput {

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

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

33
use super::{
4-
DelayType, OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
4+
OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
55
OperatorWriteOutput, Persistence, RANGE_0, RANGE_1, WriteContextArgs,
66
};
77
use crate::diagnostic::{Diagnostic, Level};
@@ -41,13 +41,14 @@ pub const PERSIST_MUT_KEYED: OperatorConstraints = OperatorConstraints {
4141
flo_type: None,
4242
ports_inn: None,
4343
ports_out: None,
44-
input_delaytype_fn: |_| Some(DelayType::Stratum),
44+
input_delaytype_fn: |_| None,
4545
write_fn: |wc @ &WriteContextArgs {
4646
root,
4747
op_span,
4848
work_fn_async,
4949
ident,
5050
inputs,
51+
outputs,
5152
is_pull,
5253
op_name,
5354
op_inst:
@@ -61,8 +62,6 @@ pub const PERSIST_MUT_KEYED: OperatorConstraints = OperatorConstraints {
6162
..
6263
},
6364
diagnostics| {
64-
assert!(is_pull);
65-
6665
if [Persistence::Mutable] != persistence_args[..] {
6766
diagnostics.push(Diagnostic::spanned(
6867
op_span,
@@ -81,7 +80,7 @@ pub const PERSIST_MUT_KEYED: OperatorConstraints = OperatorConstraints {
8180
#root::rustc_hash::FxHashMap::<_, #root::util::sparse_vec::SparseVec<_>>::default();
8281
};
8382

84-
let write_iterator = {
83+
let write_iterator = if is_pull {
8584
let input = &inputs[0];
8685
quote_spanned! {op_span=>
8786
let #ident = {
@@ -116,6 +115,51 @@ pub const PERSIST_MUT_KEYED: OperatorConstraints = OperatorConstraints {
116115
#root::dfir_pipes::pull::iter(iter)
117116
};
118117
}
118+
} else {
119+
let output = &outputs[0];
120+
quote_spanned! {op_span=>
121+
let #ident = {
122+
#[inline(always)]
123+
fn check_push<'a, Next, K, V>(
124+
persistdata: &'a mut #root::rustc_hash::FxHashMap<K, #root::util::sparse_vec::SparseVec<V>>,
125+
next: Next,
126+
)
127+
-> impl 'a + #root::dfir_pipes::push::Push<#root::util::PersistenceKeyed::<K, V>, ()>
128+
where
129+
Next: 'a + #root::dfir_pipes::push::Push<(K, V), ()>,
130+
K: ::std::clone::Clone + ::std::cmp::Eq + ::std::hash::Hash,
131+
V: ::std::clone::Clone + ::std::cmp::Eq + ::std::hash::Hash,
132+
{
133+
#root::dfir_pipes::push::fold(
134+
persistdata,
135+
|state: &mut #root::rustc_hash::FxHashMap<K, #root::util::sparse_vec::SparseVec<V>>, item| {
136+
match item {
137+
#root::util::PersistenceKeyed::Persist(k, v) => {
138+
state.entry(k).or_default().push(v);
139+
},
140+
#root::util::PersistenceKeyed::Delete(k) => {
141+
state.remove(&k);
142+
}
143+
}
144+
},
145+
#root::dfir_pipes::push::flat_map(
146+
#[allow(clippy::clone_on_copy)]
147+
#[allow(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
148+
|state: &mut #root::rustc_hash::FxHashMap<K, #root::util::sparse_vec::SparseVec<V>>| {
149+
state
150+
.iter()
151+
.flat_map(|(k, vec)| {
152+
vec.iter().map(move |v| (k.clone(), v.clone()))
153+
})
154+
},
155+
next,
156+
),
157+
)
158+
}
159+
160+
check_push(&mut #persistdata_ident, #output)
161+
};
162+
}
119163
};
120164

121165
Ok(OperatorWriteOutput {

‎dfir_rs/tests/snapshots/surface_persist__persist_mut@graphvis_dot.snap‎

Lines changed: 7 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -10,49 +10,29 @@ digraph {
1010
n3v1 [label="(n3v1) tee()", shape=house, fillcolor="#ffff88"]
1111
n4v1 [label="(n4v1) for_each(|v| pull_tx.send(v).unwrap())", shape=house, fillcolor="#ffff88"]
1212
n5v1 [label="(n5v1) flat_map(|x| if x == 3 { vec![Persist(x), Delete(x)] } else { vec![Persist(x)] })", shape=house, fillcolor="#ffff88"]
13-
n6v1 [label="(n6v1) persist_mut::<'mutable>()", shape=invhouse, fillcolor="#88aaff"]
13+
n6v1 [label="(n6v1) persist_mut::<'mutable>()", shape=house, fillcolor="#ffff88"]
1414
n7v1 [label="(n7v1) for_each(|v| push_tx.send(v).unwrap())", shape=house, fillcolor="#ffff88"]
15-
n8v1 [label="(n8v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
16-
n9v1 [label="(n9v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
1715
n2v1 -> n3v1
18-
n1v1 -> n8v1
16+
n1v1 -> n2v1
1917
n3v1 -> n4v1
2018
n6v1 -> n7v1
21-
n5v1 -> n9v1
19+
n5v1 -> n6v1
2220
n3v1 -> n5v1
23-
n8v1 -> n2v1 [color=red]
24-
n9v1 -> n6v1 [color=red]
2521
subgraph sg_1v1 {
2622
cluster=true
2723
fillcolor="#dddddd"
2824
style=filled
2925
label = "sg_1v1"
30-
subgraph sg_1v1_var_my_tee {
31-
cluster=true
32-
label="var my_tee"
33-
n1v1
34-
}
35-
}
36-
subgraph sg_2v1 {
37-
cluster=true
38-
fillcolor="#dddddd"
39-
style=filled
40-
label = "sg_2v1"
4126
n4v1
4227
n5v1
43-
subgraph sg_2v1_var_my_tee {
28+
n6v1
29+
n7v1
30+
subgraph sg_1v1_var_my_tee {
4431
cluster=true
4532
label="var my_tee"
33+
n1v1
4634
n2v1
4735
n3v1
4836
}
4937
}
50-
subgraph sg_3v1 {
51-
cluster=true
52-
fillcolor="#dddddd"
53-
style=filled
54-
label = "sg_3v1"
55-
n6v1
56-
n7v1
57-
}
5838
}

‎dfir_rs/tests/snapshots/surface_persist__persist_mut@graphvis_mermaid.snap‎

Lines changed: 7 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -13,32 +13,22 @@ linkStyle default stroke:#aaa
1313
3v1[/"(3v1) <code>tee()</code>"\]:::pushClass
1414
4v1[/"(4v1) <code>for_each(|v| pull_tx.send(v).unwrap())</code>"\]:::pushClass
1515
5v1[/"(5v1) <code>flat_map(|x| if x == 3 { vec![Persist(x), Delete(x)] } else { vec![Persist(x)] })</code>"\]:::pushClass
16-
6v1[\"(6v1) <code>persist_mut::&lt;'mutable&gt;()</code>"/]:::pullClass
16+
6v1[/"(6v1) <code>persist_mut::&lt;'mutable&gt;()</code>"\]:::pushClass
1717
7v1[/"(7v1) <code>for_each(|v| push_tx.send(v).unwrap())</code>"\]:::pushClass
18-
8v1["(8v1) <code>handoff</code>"]:::otherClass
19-
9v1["(9v1) <code>handoff</code>"]:::otherClass
2018
2v1-->3v1
21-
1v1-->8v1
19+
1v1-->2v1
2220
3v1-->4v1
2321
6v1-->7v1
24-
5v1-->9v1
22+
5v1-->6v1
2523
3v1-->5v1
26-
8v1--x2v1; linkStyle 6 stroke:red
27-
9v1--x6v1; linkStyle 7 stroke:red
2824
subgraph sg_1v1 ["sg_1v1"]
29-
subgraph sg_1v1_var_my_tee ["var <tt>my_tee</tt>"]
30-
1v1
31-
end
32-
end
33-
subgraph sg_2v1 ["sg_2v1"]
3425
4v1
3526
5v1
36-
subgraph sg_2v1_var_my_tee ["var <tt>my_tee</tt>"]
27+
6v1
28+
7v1
29+
subgraph sg_1v1_var_my_tee ["var <tt>my_tee</tt>"]
30+
1v1
3731
2v1
3832
3v1
3933
end
4034
end
41-
subgraph sg_3v1 ["sg_3v1"]
42-
6v1
43-
7v1
44-
end

‎dfir_rs/tests/snapshots/surface_persist__persist_mut_keyed@graphvis_dot.snap‎

Lines changed: 7 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -10,49 +10,29 @@ digraph {
1010
n3v1 [label="(n3v1) tee()", shape=house, fillcolor="#ffff88"]
1111
n4v1 [label="(n4v1) for_each(|(_k, v)| pull_tx.send(v).unwrap())", shape=house, fillcolor="#ffff88"]
1212
n5v1 [label="(n5v1) flat_map(|(k, v)| {\l if v == 3 { vec![Persist(k, v), Delete(k)] } else { vec![Persist(k, v)] }\l})\l", shape=house, fillcolor="#ffff88"]
13-
n6v1 [label="(n6v1) persist_mut_keyed::<'mutable>()", shape=invhouse, fillcolor="#88aaff"]
13+
n6v1 [label="(n6v1) persist_mut_keyed::<'mutable>()", shape=house, fillcolor="#ffff88"]
1414
n7v1 [label="(n7v1) for_each(|(_k, v)| push_tx.send(v).unwrap())", shape=house, fillcolor="#ffff88"]
15-
n8v1 [label="(n8v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
16-
n9v1 [label="(n9v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
1715
n2v1 -> n3v1
18-
n1v1 -> n8v1
16+
n1v1 -> n2v1
1917
n3v1 -> n4v1
2018
n6v1 -> n7v1
21-
n5v1 -> n9v1
19+
n5v1 -> n6v1
2220
n3v1 -> n5v1
23-
n8v1 -> n2v1 [color=red]
24-
n9v1 -> n6v1 [color=red]
2521
subgraph sg_1v1 {
2622
cluster=true
2723
fillcolor="#dddddd"
2824
style=filled
2925
label = "sg_1v1"
30-
subgraph sg_1v1_var_my_tee {
31-
cluster=true
32-
label="var my_tee"
33-
n1v1
34-
}
35-
}
36-
subgraph sg_2v1 {
37-
cluster=true
38-
fillcolor="#dddddd"
39-
style=filled
40-
label = "sg_2v1"
4126
n4v1
4227
n5v1
43-
subgraph sg_2v1_var_my_tee {
28+
n6v1
29+
n7v1
30+
subgraph sg_1v1_var_my_tee {
4431
cluster=true
4532
label="var my_tee"
33+
n1v1
4634
n2v1
4735
n3v1
4836
}
4937
}
50-
subgraph sg_3v1 {
51-
cluster=true
52-
fillcolor="#dddddd"
53-
style=filled
54-
label = "sg_3v1"
55-
n6v1
56-
n7v1
57-
}
5838
}

‎dfir_rs/tests/snapshots/surface_persist__persist_mut_keyed@graphvis_mermaid.snap‎

Lines changed: 7 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -13,32 +13,22 @@ linkStyle default stroke:#aaa
1313
3v1[/"(3v1) <code>tee()</code>"\]:::pushClass
1414
4v1[/"(4v1) <code>for_each(|(_k, v)| pull_tx.send(v).unwrap())</code>"\]:::pushClass
1515
5v1[/"<div style=text-align:center>(5v1)</div> <code>flat_map(|(k, v)| {<br> if v == 3 { vec![Persist(k, v), Delete(k)] } else { vec![Persist(k, v)] }<br>})</code>"\]:::pushClass
16-
6v1[\"(6v1) <code>persist_mut_keyed::&lt;'mutable&gt;()</code>"/]:::pullClass
16+
6v1[/"(6v1) <code>persist_mut_keyed::&lt;'mutable&gt;()</code>"\]:::pushClass
1717
7v1[/"(7v1) <code>for_each(|(_k, v)| push_tx.send(v).unwrap())</code>"\]:::pushClass
18-
8v1["(8v1) <code>handoff</code>"]:::otherClass
19-
9v1["(9v1) <code>handoff</code>"]:::otherClass
2018
2v1-->3v1
21-
1v1-->8v1
19+
1v1-->2v1
2220
3v1-->4v1
2321
6v1-->7v1
24-
5v1-->9v1
22+
5v1-->6v1
2523
3v1-->5v1
26-
8v1--x2v1; linkStyle 6 stroke:red
27-
9v1--x6v1; linkStyle 7 stroke:red
2824
subgraph sg_1v1 ["sg_1v1"]
29-
subgraph sg_1v1_var_my_tee ["var <tt>my_tee</tt>"]
30-
1v1
31-
end
32-
end
33-
subgraph sg_2v1 ["sg_2v1"]
3425
4v1
3526
5v1
36-
subgraph sg_2v1_var_my_tee ["var <tt>my_tee</tt>"]
27+
6v1
28+
7v1
29+
subgraph sg_1v1_var_my_tee ["var <tt>my_tee</tt>"]
30+
1v1
3731
2v1
3832
3v1
3933
end
4034
end
41-
subgraph sg_3v1 ["sg_3v1"]
42-
6v1
43-
7v1
44-
end

‎dfir_rs/tests/surface_push_blocking.rs‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,3 +141,37 @@ pub fn test_reduce_no_replay_push() {
141141
df.run_tick_sync();
142142
assert_eq!(&[13], &*collect_ready::<Vec<_>, _>(&mut out_recv));
143143
}
144+
145+
/// persist_mut on push side: source -> tee -> persist_mut -> for_each
146+
#[multiplatform_test]
147+
pub fn test_persist_mut_push() {
148+
use dfir_rs::util::Persistence::*;
149+
150+
let (out_send, mut out_recv) = dfir_rs::util::unbounded_channel::<usize>();
151+
let mut df = dfir_syntax! {
152+
my_tee = source_iter([Persist(1), Persist(2), Persist(3), Delete(2)]) -> tee();
153+
my_tee -> persist_mut::<'mutable>() -> for_each(|v| out_send.send(v).unwrap());
154+
my_tee -> null();
155+
};
156+
df.run_available_sync();
157+
let mut out = collect_ready::<Vec<_>, _>(&mut out_recv);
158+
out.sort();
159+
assert_eq!(&[1, 3], &*out);
160+
}
161+
162+
/// persist_mut_keyed on push side: source -> tee -> persist_mut_keyed -> for_each
163+
#[multiplatform_test]
164+
pub fn test_persist_mut_keyed_push() {
165+
use dfir_rs::util::PersistenceKeyed::*;
166+
167+
let (out_send, mut out_recv) = dfir_rs::util::unbounded_channel::<(i32, i32)>();
168+
let mut df = dfir_syntax! {
169+
my_tee = source_iter([Persist(1, 10), Persist(2, 20), Persist(3, 30), Delete(2)]) -> tee();
170+
my_tee -> persist_mut_keyed::<'mutable>() -> for_each(|v| out_send.send(v).unwrap());
171+
my_tee -> null();
172+
};
173+
df.run_available_sync();
174+
let mut out = collect_ready::<Vec<_>, _>(&mut out_recv);
175+
out.sort();
176+
assert_eq!(&[(1, 10), (3, 30)], &*out);
177+
}

0 commit comments

Comments
 (0)