Require handoffs and subgraphs to have friendly names

Author: MingweiSamuel

benches/benches/fork_join.rs

@@ -15,34 +15,41 @@ fn benchmark_hydroflow(c: &mut Criterion) {
         b.iter(|| {
             let mut df = Hydroflow::new();
 
-            let (start_send, start_recv) = df.make_edge::<VecHandoff<usize>>();
+            let (start_send, start_recv) = df.make_edge::<VecHandoff<usize>>("start".into());
 
             let mut sent = false;
-            df.add_subgraph_source(start_send, move |_ctx, send| {
+            df.add_subgraph_source("source".into(), start_send, move |_ctx, send| {
                 if !sent {
                     sent = true;
                     send.give(Iter(0..NUM_INTS));
                 }
             });
 
-            let (send1, mut recv1) = df.make_edge::<VecHandoff<_>>();
-            let (send2, mut recv2) = df.make_edge::<VecHandoff<_>>();
-
-            df.add_subgraph_in_2out(start_recv, send1, send2, |_ctx, recv, send1, send2| {
-                for v in recv.take_inner().into_iter() {
-                    if v % 2 == 0 {
-                        send1.give(Some(v));
-                    } else {
-                        send2.give(Some(v));
+            let (send1, mut recv1) = df.make_edge::<VecHandoff<_>>("1".into());
+            let (send2, mut recv2) = df.make_edge::<VecHandoff<_>>("2".into());
+
+            df.add_subgraph_in_2out(
+                "fork".into(),
+                start_recv,
+                send1,
+                send2,
+                |_ctx, recv, send1, send2| {
+                    for v in recv.take_inner().into_iter() {
+                        if v % 2 == 0 {
+                            send1.give(Some(v));
+                        } else {
+                            send2.give(Some(v));
+                        }
                     }
-                }
-            });
+                },
+            );
 
             for _ in 0..NUM_OPS {
-                let (send1, next_recv1) = df.make_edge();
-                let (send2, next_recv2) = df.make_edge();
+                let (send1, next_recv1) = df.make_edge("1".into());
+                let (send2, next_recv2) = df.make_edge("2".into());
 
                 df.add_subgraph_2in_2out(
+                    "join-fork".into(),
                     recv1,
                     recv2,
                     send1,
@@ -66,7 +73,7 @@ fn benchmark_hydroflow(c: &mut Criterion) {
                 recv2 = next_recv2;
             }
 
-            df.add_subgraph_2sink(recv1, recv2, |_ctx, recv1, recv2| {
+            df.add_subgraph_2sink("join (merge)".into(), recv1, recv2, |_ctx, recv1, recv2| {
                 for x in recv1.take_inner() {
                     black_box(x);
                 }

benches/benches/identity.rs

@@ -154,26 +154,31 @@ fn benchmark_hydroflow(c: &mut Criterion) {
         b.iter(|| {
             let mut df = Hydroflow::new();
 
-            let (next_send, mut next_recv) = df.make_edge::<VecHandoff<usize>>();
+            let (next_send, mut next_recv) = df.make_edge::<VecHandoff<usize>>("end".into());
 
             let mut sent = false;
-            df.add_subgraph_source(next_send, move |_ctx, send| {
+            df.add_subgraph_source("source".into(), next_send, move |_ctx, send| {
                 if !sent {
                     sent = true;
                     send.give(Iter(0..NUM_INTS));
                 }
             });
             for _ in 0..NUM_OPS {
-                let (next_send, next_next_recv) = df.make_edge();
+                let (next_send, next_next_recv) = df.make_edge("handoff".into());
 
-                df.add_subgraph_in_out(next_recv, next_send, |_ctx, recv, send| {
-                    send.give(Iter(recv.take_inner().into_iter()));
-                });
+                df.add_subgraph_in_out(
+                    "identity".into(),
+                    next_recv,
+                    next_send,
+                    |_ctx, recv, send| {
+                        send.give(Iter(recv.take_inner().into_iter()));
+                    },
+                );
 
                 next_recv = next_next_recv;
             }
 
-            df.add_subgraph_sink(next_recv, |_ctx, recv| {
+            df.add_subgraph_sink("sink".into(), next_recv, |_ctx, recv| {
                 for x in recv.take_inner() {
                     black_box(x);
                 }

benches/benches/reachability.rs

@@ -130,19 +130,26 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
             let mut df = Hydroflow::new();
 
             type Hoff = VecHandoff<usize>;
-            let (reachable_out, merge_lhs) = df.make_edge::<Hoff>();
-            let (neighbors_out, merge_rhs) = df.make_edge::<Hoff>();
-            let (merge_out, distinct_in) = df.make_edge::<Hoff>();
-            let (distinct_out, tee_in) = df.make_edge::<Hoff>();
-            let (tee_out1, neighbors_in) = df.make_edge::<Hoff>();
-            let (tee_out2, sink_in) = df.make_edge::<Hoff>();
-
-            df.add_subgraph_source(reachable_out, move |_ctx, send| {
-                send.give(Some(1));
-            });
+            let (reachable_out, merge_lhs) =
+                df.make_edge::<Hoff>("reachable_out -> merge_lhs".into());
+            let (neighbors_out, merge_rhs) =
+                df.make_edge::<Hoff>("neighbors_out -> merge_rhs".into());
+            let (merge_out, distinct_in) = df.make_edge::<Hoff>("merge_out -> distinct_in".into());
+            let (distinct_out, tee_in) = df.make_edge::<Hoff>("distinct_out -> tee_in".into());
+            let (tee_out1, neighbors_in) = df.make_edge::<Hoff>("tee_out1 -> neighbors_in".into());
+            let (tee_out2, sink_in) = df.make_edge::<Hoff>("tee_out2 -> sink_in".into());
+
+            df.add_subgraph_source(
+                "initially reachable source".into(),
+                reachable_out,
+                move |_ctx, send| {
+                    send.give(Some(1));
+                },
+            );
 
             let seen_handle = df.add_state::<RefCell<HashSet<usize>>>(Default::default());
             df.add_subgraph(
+                "distinct".into(),
                 tl!(distinct_in),
                 tl!(distinct_out),
                 move |context, tl!(recv), tl!(send)| {
@@ -156,6 +163,7 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
             );
 
             df.add_subgraph_2in_out(
+                "merge".into(),
                 merge_lhs,
                 merge_rhs,
                 merge_out,
@@ -165,24 +173,35 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
                 },
             );
 
-            df.add_subgraph_in_out(neighbors_in, neighbors_out, move |_ctx, recv, send| {
-                for v in recv.take_inner() {
-                    if let Some(neighbors) = edges.get(&v) {
-                        send.give(Iter(neighbors.iter().copied()));
+            df.add_subgraph_in_out(
+                "get neighbors".into(),
+                neighbors_in,
+                neighbors_out,
+                move |_ctx, recv, send| {
+                    for v in recv.take_inner() {
+                        if let Some(neighbors) = edges.get(&v) {
+                            send.give(Iter(neighbors.iter().copied()));
+                        }
                     }
-                }
-            });
+                },
+            );
 
-            df.add_subgraph_in_2out(tee_in, tee_out1, tee_out2, |_ctx, recv, send1, send2| {
-                for v in recv.take_inner() {
-                    send1.give(Some(v));
-                    send2.give(Some(v));
-                }
-            });
+            df.add_subgraph_in_2out(
+                "tee".into(),
+                tee_in,
+                tee_out1,
+                tee_out2,
+                |_ctx, recv, send1, send2| {
+                    for v in recv.take_inner() {
+                        send1.give(Some(v));
+                        send2.give(Some(v));
+                    }
+                },
+            );
 
             let reachable_verts = Rc::new(RefCell::new(HashSet::new()));
             let reachable_inner = reachable_verts.clone();
-            df.add_subgraph_sink(sink_in, move |_ctx, recv| {
+            df.add_subgraph_sink("output sink".into(), sink_in, move |_ctx, recv| {
                 (*reachable_inner).borrow_mut().extend(recv.take_inner());
             });
 
@@ -207,17 +226,24 @@ fn benchmark_hydroflow(c: &mut Criterion) {
             // A dataflow that represents graph reachability.
             let mut df = Hydroflow::new();
 
-            let (reachable_out, origins_in) = df.make_edge::<VecHandoff<usize>>();
-            let (did_reach_out, possible_reach_in) = df.make_edge::<VecHandoff<usize>>();
-            let (output_out, sink_in) = df.make_edge::<VecHandoff<usize>>();
-
-            df.add_subgraph_source(reachable_out, move |_ctx, send| {
-                send.give(Some(1));
-            });
+            let (reachable_out, origins_in) =
+                df.make_edge::<VecHandoff<usize>>("reachable -> origins".into());
+            let (did_reach_out, possible_reach_in) =
+                df.make_edge::<VecHandoff<usize>>("did_reach -> possible_reach".into());
+            let (output_out, sink_in) = df.make_edge::<VecHandoff<usize>>("output -> sink".into());
+
+            df.add_subgraph_source(
+                "initially reachable source".into(),
+                reachable_out,
+                move |_ctx, send| {
+                    send.give(Some(1));
+                },
+            );
 
             let seen_handle = df.add_state::<RefCell<HashSet<usize>>>(Default::default());
 
             df.add_subgraph(
+                "main".into(),
                 tl!(origins_in, possible_reach_in),
                 tl!(did_reach_out, output_out),
                 move |context, tl!(origins, did_reach_recv), tl!(did_reach_send, output)| {
@@ -249,7 +275,7 @@ fn benchmark_hydroflow(c: &mut Criterion) {
 
             let reachable_verts = Rc::new(RefCell::new(HashSet::new()));
             let reachable_inner = reachable_verts.clone();
-            df.add_subgraph_sink(sink_in, move |_ctx, recv| {
+            df.add_subgraph_sink("output sink".into(), sink_in, move |_ctx, recv| {
                 (*reachable_inner).borrow_mut().extend(recv.take_inner());
             });
 

hydroflow/examples/chat/client.rs

@@ -30,7 +30,7 @@ pub(crate) async fn run_client(opts: Opts) {
         let reader = tokio::io::BufReader::new(tokio::io::stdin());
         let lines = tokio_stream::wrappers::LinesStream::new(reader.lines())
             .map(|result| Some(result.expect("Failed to read stdin as UTF-8.")));
-        df.add_input_from_stream::<_, VecHandoff<String>, _>(lines)
+        df.add_input_from_stream::<_, VecHandoff<String>, _>("stdin".into(), lines)
     };
 
     // format addresses
@@ -39,10 +39,11 @@ pub(crate) async fn run_client(opts: Opts) {
     let messages_addr = format!("localhost:{}", messages_port);
 
     // set up the flow for requesting to be a member
+    // TODO(mingwei): use surface API instead of `wrap_input` here.
     let (my_info_send, my_info_recv) = df
         .hydroflow
-        .make_edge::<VecHandoff<(String, MemberRequest)>>();
-    let my_info_set = df.hydroflow.add_input(my_info_send);
+        .make_edge::<VecHandoff<(String, MemberRequest)>>("my_info".into());
+    let my_info_set = df.hydroflow.add_input("my_info input".into(), my_info_send);
     let my_info_get = df.wrap_input(my_info_recv);
     my_info_set.give(Some((
         addr,
@@ -52,7 +53,10 @@ pub(crate) async fn run_client(opts: Opts) {
             messages_addr,
         },
     )));
-    df.add_subgraph(my_info_get.pull_to_push().push_to(connect_req));
+    df.add_subgraph(
+        "my_info 2".into(),
+        my_info_get.pull_to_push().push_to(connect_req),
+    );
 
     let nickname = opts.name.clone();
     let nick2 = nickname.clone();
@@ -73,10 +77,11 @@ pub(crate) async fn run_client(opts: Opts) {
         })
         .map(Some)
         .push_to(messages_send);
-    df.add_subgraph(sg);
+    df.add_subgraph("sending messages".into(), sg);
 
     // set up the flow for receiving messages
     df.add_subgraph(
+        "receiving messages".into(),
         messages_recv
             .flatten()
             .filter(move |x| x.nickname != nick2)

hydroflow/examples/chat/server.rs

@@ -41,7 +41,8 @@ pub(crate) async fn run_server(opts: Opts) {
 
     // 2. feed new members into the join
     // But first, we need a buffer to turn push into pull for cross_join.
-    let (memberships_push, memberships_pull) = hf.make_edge::<VecHandoff<String>, Option<String>>();
+    let (memberships_push, memberships_pull) =
+        hf.make_edge::<VecHandoff<String>, Option<String>>("memberships".into());
     // and now the other start_tee
     let member_join_input = hf
         .start_tee()
@@ -54,7 +55,7 @@ pub(crate) async fn run_server(opts: Opts) {
         .flatten()
         .pull_to_push()
         .tee(membership_response, member_join_input);
-    hf.add_subgraph(sg);
+    hf.add_subgraph("tee members".into(), sg);
 
     // And assemble the cross-join of msgs_in and members_in, flowing to members_out
     let msgs_in = msgs_in.flatten();
@@ -75,7 +76,7 @@ pub(crate) async fn run_server(opts: Opts) {
         .pull_to_push()
         .push_to(messages_out);
 
-    hf.add_subgraph(sg);
+    hf.add_subgraph("main cross-join".into(), sg);
 
     let mut hf = hf.build();
 

hydroflow/examples/covid_tracing/main.rs

@@ -27,21 +27,26 @@ fn main() {
 
     let mut df = Hydroflow::new();
 
-    let (contacts_send, contacts_recv) = df.make_edge::<VecHandoff<(Pid, Pid, DateTime)>>();
-    let contacts_send = df.add_channel_input(contacts_send);
+    let (contacts_send, contacts_recv) =
+        df.make_edge::<VecHandoff<(Pid, Pid, DateTime)>>("contacts".into());
+    let contacts_send = df.add_channel_input("contacts input".into(), contacts_send);
+
     let (diagnosed_send, diagnosed_recv) =
-        df.make_edge::<VecHandoff<(Pid, (DateTime, DateTime))>>();
-    let diagnosed_send = df.add_channel_input(diagnosed_send);
-    let (people_send, people_recv) = df.make_edge::<VecHandoff<(Pid, (Name, Phone))>>();
-    let people_send = df.add_channel_input(people_send);
+        df.make_edge::<VecHandoff<(Pid, (DateTime, DateTime))>>("diagnosed".into());
+    let diagnosed_send = df.add_channel_input("diagnosed input".into(), diagnosed_send);
+
+    let (people_send, people_recv) =
+        df.make_edge::<VecHandoff<(Pid, (Name, Phone))>>("people".into());
+    let people_send = df.add_channel_input("people input".into(), people_send);
 
-    let (loop_send, loop_recv) = df.make_edge::<VecHandoff<(Pid, DateTime)>>();
-    let (notifs_send, notifs_recv) = df.make_edge::<VecHandoff<(Pid, DateTime)>>();
+    let (loop_send, loop_recv) = df.make_edge::<VecHandoff<(Pid, DateTime)>>("loop".into());
+    let (notifs_send, notifs_recv) = df.make_edge::<VecHandoff<(Pid, DateTime)>>("notifs".into());
 
     type MyJoinState = RefCell<JoinState<&'static str, (usize, usize), (&'static str, usize)>>;
     let state_handle = df.add_state(MyJoinState::default());
 
     df.add_subgraph(
+        "main".into(),
         tl!(contacts_recv, diagnosed_recv, loop_recv),
         tl!(notifs_send, loop_send),
         move |context,
@@ -90,6 +95,7 @@ fn main() {
     let mut people_exposure = Default::default();
 
     df.add_subgraph(
+        "join people and notifs".into(),
         tl!(people_recv, notifs_recv),
         tl!(),
         move |_ctx, tl!(peoples, exposures), ()| {