Use Name: Into<Cow<'static, str>> generic to avoid .into()s

Author: MingweiSamuel

benches/benches/fork_join.rs

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

benches/benches/identity.rs

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

benches/benches/reachability.rs

@@ -130,17 +130,15 @@ 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>("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());
+            let (reachable_out, merge_lhs) = df.make_edge::<_, Hoff>("reachable_out -> merge_lhs");
+            let (neighbors_out, merge_rhs) = df.make_edge::<_, Hoff>("neighbors_out -> merge_rhs");
+            let (merge_out, distinct_in) = df.make_edge::<_, Hoff>("merge_out -> distinct_in");
+            let (distinct_out, tee_in) = df.make_edge::<_, Hoff>("distinct_out -> tee_in");
+            let (tee_out1, neighbors_in) = df.make_edge::<_, Hoff>("tee_out1 -> neighbors_in");
+            let (tee_out2, sink_in) = df.make_edge::<_, Hoff>("tee_out2 -> sink_in");
 
             df.add_subgraph_source(
-                "initially reachable source".into(),
+                "initially reachable source",
                 reachable_out,
                 move |_ctx, send| {
                     send.give(Some(1));
@@ -149,7 +147,7 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
 
             let seen_handle = df.add_state::<RefCell<HashSet<usize>>>(Default::default());
             df.add_subgraph(
-                "distinct".into(),
+                "distinct",
                 tl!(distinct_in),
                 tl!(distinct_out),
                 move |context, tl!(recv), tl!(send)| {
@@ -163,7 +161,7 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
             );
 
             df.add_subgraph_2in_out(
-                "merge".into(),
+                "merge",
                 merge_lhs,
                 merge_rhs,
                 merge_out,
@@ -174,7 +172,7 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
             );
 
             df.add_subgraph_in_out(
-                "get neighbors".into(),
+                "get neighbors",
                 neighbors_in,
                 neighbors_out,
                 move |_ctx, recv, send| {
@@ -187,7 +185,7 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
             );
 
             df.add_subgraph_in_2out(
-                "tee".into(),
+                "tee",
                 tee_in,
                 tee_out1,
                 tee_out2,
@@ -201,7 +199,7 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
 
             let reachable_verts = Rc::new(RefCell::new(HashSet::new()));
             let reachable_inner = reachable_verts.clone();
-            df.add_subgraph_sink("output sink".into(), sink_in, move |_ctx, recv| {
+            df.add_subgraph_sink("output sink", sink_in, move |_ctx, recv| {
                 (*reachable_inner).borrow_mut().extend(recv.take_inner());
             });
 
@@ -227,13 +225,13 @@ fn benchmark_hydroflow(c: &mut Criterion) {
             let mut df = Hydroflow::new();
 
             let (reachable_out, origins_in) =
-                df.make_edge::<VecHandoff<usize>>("reachable -> origins".into());
+                df.make_edge::<_, VecHandoff<usize>>("reachable -> origins");
             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.make_edge::<_, VecHandoff<usize>>("did_reach -> possible_reach");
+            let (output_out, sink_in) = df.make_edge::<_, VecHandoff<usize>>("output -> sink");
 
             df.add_subgraph_source(
-                "initially reachable source".into(),
+                "initially reachable source",
                 reachable_out,
                 move |_ctx, send| {
                     send.give(Some(1));
@@ -243,7 +241,7 @@ fn benchmark_hydroflow(c: &mut Criterion) {
             let seen_handle = df.add_state::<RefCell<HashSet<usize>>>(Default::default());
 
             df.add_subgraph(
-                "main".into(),
+                "main",
                 tl!(origins_in, possible_reach_in),
                 tl!(did_reach_out, output_out),
                 move |context, tl!(origins, did_reach_recv), tl!(did_reach_send, output)| {
@@ -275,7 +273,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("output sink".into(), sink_in, move |_ctx, recv| {
+            df.add_subgraph_sink("output sink", 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>, _>("stdin".into(), lines)
+        df.add_input_from_stream::<_, _, VecHandoff<String>, _>("stdin", lines)
     };
 
     // format addresses
@@ -42,8 +42,8 @@ pub(crate) async fn run_client(opts: Opts) {
     // TODO(mingwei): use surface API instead of `wrap_input` here.
     let (my_info_send, my_info_recv) = df
         .hydroflow
-        .make_edge::<VecHandoff<(String, MemberRequest)>>("my_info".into());
-    let my_info_set = df.hydroflow.add_input("my_info input".into(), my_info_send);
+        .make_edge::<_, VecHandoff<(String, MemberRequest)>>("my_info");
+    let my_info_set = df.hydroflow.add_input("my_info input", my_info_send);
     let my_info_get = df.wrap_input(my_info_recv);
     my_info_set.give(Some((
         addr,
@@ -53,10 +53,7 @@ pub(crate) async fn run_client(opts: Opts) {
             messages_addr,
         },
     )));
-    df.add_subgraph(
-        "my_info 2".into(),
-        my_info_get.pull_to_push().push_to(connect_req),
-    );
+    df.add_subgraph("my_info 2", my_info_get.pull_to_push().push_to(connect_req));
 
     let nickname = opts.name.clone();
     let nick2 = nickname.clone();
@@ -77,11 +74,11 @@ pub(crate) async fn run_client(opts: Opts) {
         })
         .map(Some)
         .push_to(messages_send);
-    df.add_subgraph("sending messages".into(), sg);
+    df.add_subgraph("sending messages", sg);
 
     // set up the flow for receiving messages
     df.add_subgraph(
-        "receiving messages".into(),
+        "receiving messages",
         messages_recv
             .flatten()
             .filter(move |x| x.nickname != nick2)

hydroflow/examples/chat/server.rs

@@ -42,7 +42,7 @@ 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>>("memberships".into());
+        hf.make_edge::<_, VecHandoff<String>, Option<String>>("memberships");
     // and now the other start_tee
     let member_join_input = hf
         .start_tee()
@@ -55,7 +55,7 @@ pub(crate) async fn run_server(opts: Opts) {
         .flatten()
         .pull_to_push()
         .tee(membership_response, member_join_input);
-    hf.add_subgraph("tee members".into(), sg);
+    hf.add_subgraph("tee members", sg);
 
     // And assemble the cross-join of msgs_in and members_in, flowing to members_out
     let msgs_in = msgs_in.flatten();
@@ -76,7 +76,7 @@ pub(crate) async fn run_server(opts: Opts) {
         .pull_to_push()
         .push_to(messages_out);
 
-    hf.add_subgraph("main cross-join".into(), sg);
+    hf.add_subgraph("main cross-join", sg);
 
     let mut hf = hf.build();
 

hydroflow/examples/covid_tracing/main.rs

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

hydroflow/examples/covid_tracing_dist/database.rs

@@ -17,21 +17,21 @@ pub(crate) async fn run_database(opts: Opts) {
 
     let mut df = Hydroflow::new();
 
-    let (notifs, notif_sink) = df.make_edge::<VecHandoff<(String, usize)>>("notifs".into());
+    let (notifs, notif_sink) = df.make_edge::<_, VecHandoff<(String, usize)>>("notifs");
     let (encode_contacts_out, contacts_merge) =
-        df.make_edge::<VecHandoff<Message>>("encoded contacts".into());
+        df.make_edge::<_, VecHandoff<Message>>("encoded contacts");
     let (encode_diagnoses_out, diagnoses_merge) =
-        df.make_edge::<VecHandoff<Message>>("encoded diagnoses".into());
+        df.make_edge::<_, VecHandoff<Message>>("encoded diagnoses");
 
     let (contacts_send, contacts_recv) =
-        df.make_edge::<VecHandoff<(&'static str, &'static str, usize)>>("contacts".into());
-    let contacts_send = df.add_channel_input("contacts input".into(), contacts_send);
+        df.make_edge::<_, VecHandoff<(&'static str, &'static str, usize)>>("contacts");
+    let contacts_send = df.add_channel_input("contacts input", contacts_send);
     let (diagnosed_send, diagnosed_recv) =
-        df.make_edge::<VecHandoff<(&'static str, (usize, usize))>>("diagnosed".into());
-    let diagnosed_send = df.add_channel_input("diagnosed input".into(), diagnosed_send);
+        df.make_edge::<_, VecHandoff<(&'static str, (usize, usize))>>("diagnosed");
+    let diagnosed_send = df.add_channel_input("diagnosed input", diagnosed_send);
     let (people_send, people_recv) =
-        df.make_edge::<VecHandoff<(String, (String, String))>>("people".into());
-    let people_send = df.add_channel_input("people input".into(), people_send);
+        df.make_edge::<_, VecHandoff<(String, (String, String))>>("people");
+    let people_send = df.add_channel_input("people input", people_send);
 
     let stream = TcpListener::bind(format!("localhost:{}", opts.port))
         .await
@@ -41,7 +41,7 @@ pub(crate) async fn run_database(opts: Opts) {
     let (network_in, network_out) = df.add_tcp_stream(stream);
 
     df.add_subgraph_in_out(
-        "decode messages".into(),
+        "decode messages",
         network_out,
         notifs,
         |_ctx, recv, send| {
@@ -89,7 +89,7 @@ pub(crate) async fn run_database(opts: Opts) {
     });
 
     df.add_subgraph_2in_out(
-        "merge contacts and diagnoses".into(),
+        "merge contacts and diagnoses",
         contacts_merge,
         diagnoses_merge,
         network_in,
@@ -101,7 +101,7 @@ pub(crate) async fn run_database(opts: Opts) {
     );
 
     df.add_subgraph_in_out(
-        "encode contacts".into(),
+        "encode contacts",
         contacts_recv,
         encode_contacts_out,
         |_ctx, recv, send| {
@@ -115,7 +115,7 @@ pub(crate) async fn run_database(opts: Opts) {
     );
 
     df.add_subgraph_in_out(
-        "encode diagnoses".into(),
+        "encode diagnoses",
         diagnosed_recv,
         encode_diagnoses_out,
         |_ctx, recv, send| {
@@ -130,7 +130,7 @@ pub(crate) async fn run_database(opts: Opts) {
 
     let mut join_state = Default::default();
     df.add_subgraph(
-        "join people and notifs".into(),
+        "join people and notifs",
         tl!(notif_sink, people_recv),
         tl!(),
         move |_ctx, tl!(notifs, people), tl!()| {