Switch builder API to port-identified iterators

Author: frankmcsherry

timely/src/dataflow/operators/core/feedback.rs

@@ -113,7 +113,7 @@ impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
         let summary = handle.summary;
         let mut output = handle.output;
 
-        let mut input = builder.new_input_connection(self, Pipeline, vec![Antichain::from_elem(summary.clone())]);
+        let mut input = builder.new_input_connection(self, Pipeline, [(0, Antichain::from_elem(summary.clone()))]);
 
         builder.build(move |_capability| move |_frontier| {
             let mut output = output.activate();

timely/src/dataflow/operators/generic/builder_raw.rs

@@ -12,7 +12,7 @@ use crate::scheduling::{Schedule, Activations};
 
 use crate::progress::{Source, Target};
 use crate::progress::{Timestamp, Operate, operate::SharedProgress, Antichain};
-use crate::progress::operate::Connectivity;
+use crate::progress::operate::{Connectivity, PortConnectivity};
 use crate::Container;
 use crate::dataflow::{StreamCore, Scope};
 use crate::dataflow::channels::pushers::Tee;
@@ -105,48 +105,53 @@ impl<G: Scope> OperatorBuilder<G> {
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
     pub fn new_input<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P) -> P::Puller
-        where
-            P: ParallelizationContract<G::Timestamp, C> {
-        let connection = vec![Antichain::from_elem(Default::default()); self.shape.outputs];
+    where
+        P: ParallelizationContract<G::Timestamp, C>
+    {
+        let connection = (0 .. self.shape.outputs).map(|o| (o, Antichain::from_elem(Default::default())));
         self.new_input_connection(stream, pact, connection)
     }
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
-    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> P::Puller
+    pub fn new_input_connection<C: Container, P, I>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: I) -> P::Puller
     where
-        P: ParallelizationContract<G::Timestamp, C> {
-
+        P: ParallelizationContract<G::Timestamp, C>,
+        I: IntoIterator<Item = (usize, Antichain<<G::Timestamp as Timestamp>::Summary>)>,
+    {
         let channel_id = self.scope.new_identifier();
         let logging = self.scope.logging();
         let (sender, receiver) = pact.connect(&mut self.scope, channel_id, Rc::clone(&self.address), logging);
         let target = Target::new(self.index, self.shape.inputs);
         stream.connect_to(target, sender, channel_id);
 
         self.shape.inputs += 1;
-        assert_eq!(self.shape.outputs, connection.len());
-        self.summary.push(connection.into());
+        let connectivity: PortConnectivity<_> = connection.into_iter().collect();
+        assert!(connectivity.iter_ports().all(|(o,_)| o < self.shape.outputs));
+        self.summary.push(connectivity);
 
         receiver
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
     pub fn new_output<C: Container>(&mut self) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
 
-        let connection = vec![Antichain::from_elem(Default::default()); self.shape.inputs];
+        let connection = (0 .. self.shape.inputs).map(|i| (i, Antichain::from_elem(Default::default())));
         self.new_output_connection(connection)
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
-    pub fn new_output_connection<C: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
+    pub fn new_output_connection<C: Container, I>(&mut self, connection: I) -> (Tee<G::Timestamp, C>, StreamCore<G, C>)
+    where
+        I: IntoIterator<Item = (usize, Antichain<<G::Timestamp as Timestamp>::Summary>)>,
+    {
         let new_output = self.shape.outputs;
         self.shape.outputs += 1;
         let (targets, registrar) = Tee::<G::Timestamp,C>::new();
         let source = Source::new(self.index, new_output);
         let stream = StreamCore::new(source, registrar, self.scope.clone());
 
-        assert_eq!(self.shape.inputs, connection.len());
-        for (summary, entry) in self.summary.iter_mut().zip(connection.into_iter()) {
-            summary.add_port(new_output, entry);
+        for (input, entry) in connection {
+            self.summary[input].add_port(new_output, entry);
         }
 
         (targets, stream)

timely/src/dataflow/operators/generic/builder_rc.rs

@@ -64,7 +64,7 @@ impl<G: Scope> OperatorBuilder<G> {
     where
         P: ParallelizationContract<G::Timestamp, C> {
 
-        let connection = (0..self.builder.shape().outputs()).map(|_| Antichain::from_elem(Default::default())).collect();
+        let connection = (0..self.builder.shape().outputs()).map(|o| (o, Antichain::from_elem(Default::default())));
         self.new_input_connection(stream, pact, connection)
     }
 
@@ -76,25 +76,26 @@ impl<G: Scope> OperatorBuilder<G> {
     ///
     /// Commonly the connections are either the unit summary, indicating the same timestamp might be produced as output, or an empty
     /// antichain indicating that there is no connection from the input to the output.
-    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> InputHandleCore<G::Timestamp, C, P::Puller>
-        where
-            P: ParallelizationContract<G::Timestamp, C> {
-
+    pub fn new_input_connection<C: Container, P, I>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: I) -> InputHandleCore<G::Timestamp, C, P::Puller>
+    where
+        P: ParallelizationContract<G::Timestamp, C>,
+        I: IntoIterator<Item = (usize, Antichain<<G::Timestamp as Timestamp>::Summary>)> + Clone,
+    {
         let puller = self.builder.new_input_connection(stream, pact, connection.clone());
 
         let input = PullCounter::new(puller);
         self.frontier.push(MutableAntichain::new());
         self.consumed.push(Rc::clone(input.consumed()));
 
-        let shared_summary = Rc::new(RefCell::new(connection.into()));
+        let shared_summary = Rc::new(RefCell::new(connection.into_iter().collect()));
         self.summaries.push(Rc::clone(&shared_summary));
 
         new_input_handle(input, Rc::clone(&self.internal), shared_summary, self.logging.clone())
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
     pub fn new_output<CB: ContainerBuilder>(&mut self) -> (OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>, StreamCore<G, CB::Container>) {
-        let connection = (0..self.builder.shape().inputs()).map(|_| Antichain::from_elem(Default::default())).collect();
+        let connection = (0..self.builder.shape().inputs()).map(|i| (i, Antichain::from_elem(Default::default())));
         self.new_output_connection(connection)
     }
 
@@ -106,13 +107,13 @@ impl<G: Scope> OperatorBuilder<G> {
     ///
     /// Commonly the connections are either the unit summary, indicating the same timestamp might be produced as output, or an empty
     /// antichain indicating that there is no connection from the input to the output.
-    pub fn new_output_connection<CB: ContainerBuilder>(
-        &mut self,
-        connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>
-    ) -> (
+    pub fn new_output_connection<CB: ContainerBuilder, I>(&mut self, connection: I) -> (
         OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
         StreamCore<G, CB::Container>
-    ) {
+    )
+    where
+        I: IntoIterator<Item = (usize, Antichain<<G::Timestamp as Timestamp>::Summary>)> + Clone,
+    {
         let new_output = self.shape().outputs();
         let (tee, stream) = self.builder.new_output_connection(connection.clone());
 
@@ -122,8 +123,8 @@ impl<G: Scope> OperatorBuilder<G> {
         let mut buffer = PushBuffer::new(PushCounter::new(tee));
         self.produced.push(Rc::clone(buffer.inner().produced()));
 
-        for (summary, connection) in self.summaries.iter().zip(connection.into_iter()) {
-            summary.borrow_mut().add_port(new_output, connection.clone());
+        for (input, entry) in connection {
+            self.summaries[input].borrow_mut().add_port(new_output, entry);
         }
 
         (OutputWrapper::new(buffer, internal), stream)

timely/src/progress/operate.rs

@@ -101,9 +101,9 @@ impl<TS> PortConnectivity<TS> {
     }
 }
 
-impl<TS> From<Vec<Antichain<TS>>> for PortConnectivity<TS> {
-    fn from(list: Vec<Antichain<TS>>) -> Self {
-        Self { tree: list.into_iter().enumerate().filter(|(_,p)| !p.is_empty()).collect() }
+impl<TS> FromIterator<(usize, Antichain<TS>)> for PortConnectivity<TS> {
+    fn from_iter<T>(iter: T) -> Self where T: IntoIterator<Item = (usize, Antichain<TS>)> {
+        Self { tree: iter.into_iter().filter(|(_,p)| !p.is_empty()).collect() }
     }
 }
 

timely/src/progress/reachability.rs

@@ -17,9 +17,9 @@
 //! let mut builder = Builder::<usize>::new();
 //!
 //! // Each node with one input connected to one output.
-//! builder.add_node(0, 1, 1, vec![vec![Antichain::from_elem(0)].into()]);
-//! builder.add_node(1, 1, 1, vec![vec![Antichain::from_elem(0)].into()]);
-//! builder.add_node(2, 1, 1, vec![vec![Antichain::from_elem(1)].into()]);
+//! builder.add_node(0, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
+//! builder.add_node(1, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
+//! builder.add_node(2, 1, 1, vec![[(0, Antichain::from_elem(1))].into_iter().collect()]);
 //!
 //! // Connect nodes in sequence, looping around to the first from the last.
 //! builder.add_edge(Source::new(0, 0), Target::new(1, 0));
@@ -113,9 +113,9 @@ use crate::progress::timestamp::PathSummary;
 /// let mut builder = Builder::<usize>::new();
 ///
 /// // Each node with one input connected to one output.
-/// builder.add_node(0, 1, 1, vec![vec![Antichain::from_elem(0)].into()]);
-/// builder.add_node(1, 1, 1, vec![vec![Antichain::from_elem(0)].into()]);
-/// builder.add_node(2, 1, 1, vec![vec![Antichain::from_elem(1)].into()]);
+/// builder.add_node(0, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
+/// builder.add_node(1, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
+/// builder.add_node(2, 1, 1, vec![[(0, Antichain::from_elem(1))].into_iter().collect()]);
 ///
 /// // Connect nodes in sequence, looping around to the first from the last.
 /// builder.add_edge(Source::new(0, 0), Target::new(1, 0));
@@ -224,9 +224,9 @@ impl<T: Timestamp> Builder<T> {
     /// let mut builder = Builder::<usize>::new();
     ///
     /// // Each node with one input connected to one output.
-    /// builder.add_node(0, 1, 1, vec![vec![Antichain::from_elem(0)].into()]);
-    /// builder.add_node(1, 1, 1, vec![vec![Antichain::from_elem(0)].into()]);
-    /// builder.add_node(2, 1, 1, vec![vec![Antichain::from_elem(0)].into()]);
+    /// builder.add_node(0, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
+    /// builder.add_node(1, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
+    /// builder.add_node(2, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
     ///
     /// // Connect nodes in sequence, looping around to the first from the last.
     /// builder.add_edge(Source::new(0, 0), Target::new(1, 0));
@@ -253,8 +253,8 @@ impl<T: Timestamp> Builder<T> {
     ///
     /// // Two inputs and outputs, only one of which advances.
     /// builder.add_node(0, 2, 2, vec![
-    ///     vec![Antichain::from_elem(0),Antichain::new(),].into(),
-    ///     vec![Antichain::new(),Antichain::from_elem(1),].into(),
+    ///     [(0,Antichain::from_elem(0)),(1,Antichain::new())].into_iter().collect(),
+    ///     [(0,Antichain::new()),(1,Antichain::from_elem(1))].into_iter().collect(),
     /// ]);
     ///
     /// // Connect each output to the opposite input.