Rename make_handoff to make_edge

benches/benches/fork_join.rs

@@ -15,7 +15,7 @@ fn benchmark_hydroflow(c: &mut Criterion) {
         b.iter(|| {
             let mut df = Hydroflow::new();
 
-            let (start_send, start_recv) = df.make_handoff::<VecHandoff<usize>>();
+            let (start_send, start_recv) = df.make_edge::<VecHandoff<usize>>();
 
             let mut sent = false;
             df.add_subgraph_source(start_send, move |_ctx, send| {
@@ -25,8 +25,8 @@ fn benchmark_hydroflow(c: &mut Criterion) {
                 }
             });
 
-            let (send1, mut recv1) = df.make_handoff::<VecHandoff<_>>();
-            let (send2, mut recv2) = df.make_handoff::<VecHandoff<_>>();
+            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() {
@@ -39,8 +39,8 @@ fn benchmark_hydroflow(c: &mut Criterion) {
             });
 
             for _ in 0..NUM_OPS {
-                let (send1, next_recv1) = df.make_handoff();
-                let (send2, next_recv2) = df.make_handoff();
+                let (send1, next_recv1) = df.make_edge();
+                let (send2, next_recv2) = df.make_edge();
 
                 df.add_subgraph_2in_2out(
                     recv1,

benches/benches/identity.rs

@@ -154,7 +154,7 @@ fn benchmark_hydroflow(c: &mut Criterion) {
         b.iter(|| {
             let mut df = Hydroflow::new();
 
-            let (next_send, mut next_recv) = df.make_handoff::<VecHandoff<usize>>();
+            let (next_send, mut next_recv) = df.make_edge::<VecHandoff<usize>>();
 
             let mut sent = false;
             df.add_subgraph_source(next_send, move |_ctx, send| {
@@ -164,7 +164,7 @@ fn benchmark_hydroflow(c: &mut Criterion) {
                 }
             });
             for _ in 0..NUM_OPS {
-                let (next_send, next_next_recv) = df.make_handoff();
+                let (next_send, next_next_recv) = df.make_edge();
 
                 df.add_subgraph_in_out(next_recv, next_send, |_ctx, recv, send| {
                     send.give(Iter(recv.take_inner().into_iter()));

benches/benches/reachability.rs

@@ -130,12 +130,12 @@ fn benchmark_hydroflow_scheduled(c: &mut Criterion) {
             let mut df = Hydroflow::new();
 
             type Hoff = VecHandoff<usize>;
-            let (reachable_out, merge_lhs) = df.make_handoff::<Hoff>();
-            let (neighbors_out, merge_rhs) = df.make_handoff::<Hoff>();
-            let (merge_out, distinct_in) = df.make_handoff::<Hoff>();
-            let (distinct_out, tee_in) = df.make_handoff::<Hoff>();
-            let (tee_out1, neighbors_in) = df.make_handoff::<Hoff>();
-            let (tee_out2, sink_in) = df.make_handoff::<Hoff>();
+            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));
@@ -207,9 +207,9 @@ fn benchmark_hydroflow(c: &mut Criterion) {
             // A dataflow that represents graph reachability.
             let mut df = Hydroflow::new();
 
-            let (reachable_out, origins_in) = df.make_handoff::<VecHandoff<usize>>();
-            let (did_reach_out, possible_reach_in) = df.make_handoff::<VecHandoff<usize>>();
-            let (output_out, sink_in) = df.make_handoff::<VecHandoff<usize>>();
+            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));

chat/src/server.rs

@@ -41,8 +41,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_handoff::<VecHandoff<String>, Option<String>>();
+    let (memberships_push, memberships_pull) = hf.make_edge::<VecHandoff<String>, Option<String>>();
     // and now the other start_tee
     let member_join_input = hf
         .start_tee()

covid_tracing/src/main.rs

@@ -33,8 +33,8 @@ fn main() {
         df.add_channel_input::<_, VecHandoff<(Pid, (DateTime, DateTime))>>();
     let (people_send, people_recv) = df.add_channel_input::<_, VecHandoff<(Pid, (Name, Phone))>>();
 
-    let (loop_send, loop_recv) = df.make_handoff::<VecHandoff<(Pid, DateTime)>>();
-    let (notifs_send, notifs_recv) = df.make_handoff::<VecHandoff<(Pid, DateTime)>>();
+    let (loop_send, loop_recv) = df.make_edge::<VecHandoff<(Pid, DateTime)>>();
+    let (notifs_send, notifs_recv) = df.make_edge::<VecHandoff<(Pid, DateTime)>>();
 
     type MyJoinState = RefCell<JoinState<&'static str, (usize, usize), (&'static str, usize)>>;
     let state_handle = df.add_state(MyJoinState::default());

covid_tracing_dist/src/database.rs

@@ -17,9 +17,9 @@ pub(crate) async fn run_database(opts: Opts) {
 
     let mut df = Hydroflow::new();
 
-    let (notifs, notif_sink) = df.make_handoff::<VecHandoff<(String, usize)>>();
-    let (encode_contacts_out, contacts_merge) = df.make_handoff::<VecHandoff<Message>>();
-    let (encode_diagnoses_out, diagnoses_merge) = df.make_handoff::<VecHandoff<Message>>();
+    let (notifs, notif_sink) = df.make_edge::<VecHandoff<(String, usize)>>();
+    let (encode_contacts_out, contacts_merge) = df.make_edge::<VecHandoff<Message>>();
+    let (encode_diagnoses_out, diagnoses_merge) = df.make_edge::<VecHandoff<Message>>();
 
     let (contacts_in, contacts_out) =
         df.add_channel_input::<_, VecHandoff<(&'static str, &'static str, usize)>>();

covid_tracing_dist/src/tracker.rs

@@ -15,10 +15,10 @@ pub(crate) async fn run_tracker(opts: Opts) {
     let stream = TcpStream::connect(opts.addr).await.unwrap();
     let (network_out, network_in) = df.add_tcp_stream(stream);
 
-    let (contacts, contacts_in) = df.make_handoff::<VecHandoff<(String, String, usize)>>();
-    let (diagnoses, diagnosed_in) = df.make_handoff::<VecHandoff<(String, (usize, usize))>>();
-    let (loop_out, loop_in) = df.make_handoff::<VecHandoff<(Pid, DateTime)>>();
-    let (notifs_out, encoder_in) = df.make_handoff::<VecHandoff<(Pid, DateTime)>>();
+    let (contacts, contacts_in) = df.make_edge::<VecHandoff<(String, String, usize)>>();
+    let (diagnoses, diagnosed_in) = df.make_edge::<VecHandoff<(String, (usize, usize))>>();
+    let (loop_out, loop_in) = df.make_edge::<VecHandoff<(Pid, DateTime)>>();
+    let (notifs_out, encoder_in) = df.make_edge::<VecHandoff<(Pid, DateTime)>>();
 
     df.add_subgraph(
         tl!(network_in),

hydroflow/src/builder/hydroflow_builder.rs

@@ -25,13 +25,11 @@ pub struct HydroflowBuilder {
 impl HydroflowBuilder {
     /// Creates a handoff, returning push and pull ends which can be chained
     /// using the Surface API.
-    pub fn make_handoff<H, T>(
-        &mut self,
-    ) -> (HandoffPushSurfaceReversed<H, T>, HandoffPullSurface<H>)
+    pub fn make_edge<H, T>(&mut self) -> (HandoffPushSurfaceReversed<H, T>, HandoffPullSurface<H>)
     where
         H: Handoff + CanReceive<T>,
     {
-        let (send, recv) = self.hydroflow.make_handoff();
+        let (send, recv) = self.hydroflow.make_edge();
         let push = HandoffPushSurfaceReversed::new(send);
         let pull = HandoffPullSurface::new(recv);
         (push, pull)

hydroflow/src/builder/mod.rs

@@ -194,8 +194,8 @@ fn test_covid() {
 
     let mut builder = HydroflowBuilder::default();
 
-    let (loop_send, loop_recv) = builder.make_handoff::<VecHandoff<(Pid, DateTime)>, _>();
-    let (notifs_send, notifs_recv) = builder.make_handoff::<VecHandoff<(Pid, DateTime)>, _>();
+    let (loop_send, loop_recv) = builder.make_edge::<VecHandoff<(Pid, DateTime)>, _>();
+    let (notifs_send, notifs_recv) = builder.make_edge::<VecHandoff<(Pid, DateTime)>, _>();
 
     let (diagnosed_send, diagnosed) =
         builder.add_channel_input::<Option<(Pid, (DateTime, DateTime))>, VecHandoff<_>>();

hydroflow/src/builder/surface/exchange.rs

@@ -73,7 +73,7 @@ where
         Other: PullSurface<ItemOut = (Key, Val)>,
     {
         let (local_inputs_send, local_inputs_recv) =
-            builder.make_handoff::<VecHandoff<(Key, Val)>, Option<(Key, Val)>>();
+            builder.make_edge::<VecHandoff<(Key, Val)>, Option<(Key, Val)>>();
 
         let num_participants: u64 = address_book.len().try_into().unwrap();
 
@@ -118,7 +118,7 @@ where
         Other: PullSurface<ItemOut = U>,
     {
         let (local_inputs_send, local_inputs_recv) =
-            builder.make_handoff::<VecHandoff<U>, Option<U>>();
+            builder.make_edge::<VecHandoff<U>, Option<U>>();
 
         builder.add_subgraph(
             IterPullSurface::new(addresses.into_iter())

hydroflow/src/scheduled/graph.rs

@@ -232,7 +232,7 @@ impl Hydroflow {
         sg_id
     }
 
-    pub fn make_handoff<H>(&mut self) -> (InputPort<H>, OutputPort<H>)
+    pub fn make_edge<H>(&mut self) -> (InputPort<H>, OutputPort<H>)
     where
         H: 'static + Handoff,
     {

hydroflow/src/scheduled/graph_ext.rs

@@ -129,7 +129,7 @@ impl GraphExt for Hydroflow {
         use std::sync::mpsc;
 
         let (sender, receiver) = mpsc::sync_channel(8000);
-        let (send_port, recv_port) = self.make_handoff();
+        let (send_port, recv_port) = self.make_edge();
         let sg_id = self.add_subgraph_source::<_, W>(send_port, move |_ctx, send| {
             for x in receiver.try_iter() {
                 send.give(x);
@@ -145,7 +145,7 @@ impl GraphExt for Hydroflow {
     {
         let input = super::input::Buffer::default();
         let inner_input = input.clone();
-        let (send_port, recv_port) = self.make_handoff::<W>();
+        let (send_port, recv_port) = self.make_edge::<W>();
         let sg_id = self.add_subgraph_source::<_, W>(send_port, move |_ctx, send| {
             for x in (*inner_input.0).borrow_mut().drain(..) {
                 send.give(x);

hydroflow/src/scheduled/mod.rs

@@ -1,11 +1,11 @@
 pub mod context;
 pub mod graph;
-pub mod port;
 pub mod graph_ext;
 pub mod handoff;
 #[cfg(feature = "variadic_generics")]
 pub mod input;
 pub mod net;
+pub mod port;
 pub mod query;
 pub mod reactor;
 pub mod state;

hydroflow/src/scheduled/net/mod.rs

@@ -103,7 +103,7 @@ impl Hydroflow {
         reader: OwnedReadHalf,
     ) -> OutputPort<VecHandoff<Message>> {
         let reader = FramedRead::new(reader, LengthDelimitedCodec::new());
-        let (send_port, recv_port) = self.make_handoff();
+        let (send_port, recv_port) = self.make_edge();
         self.add_input_from_stream(
             send_port,
             reader.map(|buf| Some(<Message>::decode(buf.unwrap().into()))),
@@ -118,7 +118,7 @@ impl Hydroflow {
         let mut writer = FramedWrite::new(writer, LengthDelimitedCodec::new());
         let mut message_queue = VecDeque::new();
 
-        let (input_port, output_port) = self.make_handoff::<VecHandoff<Message>>();
+        let (input_port, output_port) = self.make_edge::<VecHandoff<Message>>();
         self.add_subgraph_sink(output_port, move |ctx, recv| {
             let waker = ctx.waker();
             let mut cx = std::task::Context::from_waker(&waker);

hydroflow/src/scheduled/net/network_vertex.rs

@@ -79,7 +79,7 @@ impl Hydroflow {
             }
         });
 
-        let (send_port, recv_port) = self.make_handoff();
+        let (send_port, recv_port) = self.make_edge();
         self.add_input_from_stream(send_port, incoming_messages.map(Some));
 
         (port, recv_port)
@@ -199,7 +199,7 @@ impl Hydroflow {
 
         let mut buffered_messages = Vec::new();
         let mut next_messages = Vec::new();
-        let (input_port, output_port) = self.make_handoff();
+        let (input_port, output_port) = self.make_edge();
         self.add_subgraph_sink(output_port, move |_ctx, recv| {
             buffered_messages.extend(recv.take_inner());
             for msg in buffered_messages.drain(..) {

hydroflow/src/scheduled/query.rs

@@ -25,7 +25,7 @@ impl Query {
     {
         let mut df = self.df.borrow_mut();
 
-        let (send_port, recv_port) = df.make_handoff();
+        let (send_port, recv_port) = df.make_edge();
         df.add_subgraph_source(send_port, f);
 
         Operator {
@@ -40,7 +40,7 @@ impl Query {
     {
         let mut df = self.df.borrow_mut();
 
-        let (send_port, recv_port) = df.make_handoff();
+        let (send_port, recv_port) = df.make_edge();
         df.add_subgraph_n_m(
             ops.into_iter().map(|op| op.recv_port).collect(),
             vec![send_port],
@@ -81,7 +81,7 @@ where
     {
         let mut df = self.df.borrow_mut();
 
-        let (send_port, recv_port) = df.make_handoff();
+        let (send_port, recv_port) = df.make_edge();
         df.add_subgraph_in_out(self.recv_port, send_port, move |_ctx, recv, send| {
             send.give(Iter(recv.take_inner().into_iter().map(&mut f)));
         });
@@ -100,7 +100,7 @@ where
     {
         let mut df = self.df.borrow_mut();
 
-        let (send_port, recv_port) = df.make_handoff();
+        let (send_port, recv_port) = df.make_edge();
         df.add_subgraph_in_out(self.recv_port, send_port, move |_ctx, recv, send| {
             send.give(Iter(recv.take_inner().into_iter().filter(&mut f)));
         });
@@ -118,7 +118,7 @@ where
 
         let mut df = self.df.borrow_mut();
 
-        let (send_port, recv_port) = df.make_handoff::<VecHandoff<T>>();
+        let (send_port, recv_port) = df.make_edge::<VecHandoff<T>>();
         df.add_subgraph_2in_out(
             self.recv_port,
             other.recv_port,
@@ -163,7 +163,7 @@ impl<T: Clone> Operator<T> {
         let mut sends = Vec::with_capacity(n);
         let mut recvs = Vec::with_capacity(n);
         for _ in 0..n {
-            let (send_port, recv_port) = df.make_handoff::<VecHandoff<T>>();
+            let (send_port, recv_port) = df.make_edge::<VecHandoff<T>>();
             sends.push(send_port);
             recvs.push(Operator {
                 df: self.df.clone(),

hydroflow/tests/test.rs

@@ -22,9 +22,9 @@ fn map_filter() {
     // A simple dataflow with one source feeding into one sink with some processing in the middle.
     let mut df = Hydroflow::new();
 
-    let (source, map_in) = df.make_handoff::<VecHandoff<i32>>();
-    let (map_out, filter_in) = df.make_handoff::<VecHandoff<i32>>();
-    let (filter_out, sink) = df.make_handoff::<VecHandoff<i32>>();
+    let (source, map_in) = df.make_edge::<VecHandoff<i32>>();
+    let (map_out, filter_in) = df.make_edge::<VecHandoff<i32>>();
+    let (filter_out, sink) = df.make_edge::<VecHandoff<i32>>();
 
     let data = [1, 2, 3, 4];
     df.add_subgraph(tl!(), tl!(source), move |_ctx, tl!(), tl!(send)| {
@@ -67,7 +67,7 @@ fn map_filter() {
 #[test]
 fn test_basic_variadic() {
     let mut df = Hydroflow::new();
-    let (source_send, sink_recv) = df.make_handoff::<VecHandoff<usize>>();
+    let (source_send, sink_recv) = df.make_edge::<VecHandoff<usize>>();
     df.add_subgraph_source(source_send, move |_ctx, send| {
         send.give(Some(5));
     });
@@ -91,7 +91,7 @@ fn test_basic_variadic() {
 fn test_basic_n_m() {
     let mut df = Hydroflow::new();
 
-    let (source_send, sink_recv) = df.make_handoff::<VecHandoff<usize>>();
+    let (source_send, sink_recv) = df.make_edge::<VecHandoff<usize>>();
 
     df.add_subgraph_n_m(
         vec![],
@@ -140,12 +140,12 @@ fn test_cycle() {
 
     let mut df = Hydroflow::new();
 
-    let (reachable, merge_lhs) = df.make_handoff::<VecHandoff<usize>>();
-    let (neighbors_out, merge_rhs) = df.make_handoff::<VecHandoff<usize>>();
-    let (merge_out, distinct_in) = df.make_handoff::<VecHandoff<usize>>();
-    let (distinct_out, tee_in) = df.make_handoff::<VecHandoff<usize>>();
-    let (tee_out1, neighbors_in) = df.make_handoff::<VecHandoff<usize>>();
-    let (tee_out2, sink_in) = df.make_handoff::<VecHandoff<usize>>();
+    let (reachable, merge_lhs) = df.make_edge::<VecHandoff<usize>>();
+    let (neighbors_out, merge_rhs) = df.make_edge::<VecHandoff<usize>>();
+    let (merge_out, distinct_in) = df.make_edge::<VecHandoff<usize>>();
+    let (distinct_out, tee_in) = df.make_edge::<VecHandoff<usize>>();
+    let (tee_out1, neighbors_in) = df.make_edge::<VecHandoff<usize>>();
+    let (tee_out2, sink_in) = df.make_edge::<VecHandoff<usize>>();
 
     let mut initially_reachable = vec![1];
     df.add_subgraph_source(reachable, move |_ctx, send| {
@@ -343,7 +343,7 @@ fn test_input_channel() {
             let done_inner = done.clone();
             let mut df = Hydroflow::new();
 
-            let (in_chan, input) = df.make_handoff();
+            let (in_chan, input) = df.make_edge();
             df.add_input_from_stream::<_, VecHandoff<usize>, _>(in_chan, receiver);
             df.add_subgraph_sink(input, move |_ctx, recv| {
                 for v in recv.take_inner() {

relalg/src/runtime.rs

@@ -31,7 +31,7 @@ pub(crate) fn run_dataflow(r: RelExpr) -> Vec<Vec<Datum>> {
 }
 
 fn render_relational(df: &mut Hydroflow, r: RelExpr) -> OutputPort<VecHandoff<Vec<Datum>>> {
-    let (send_port, recv_port) = df.make_handoff();
+    let (send_port, recv_port) = df.make_edge();
     match r {
         RelExpr::Values(mut v) => {
             // TODO: drip-feed data?