Make PortConnectivity API more explicit

Author: frankmcsherry

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

@@ -138,15 +138,15 @@ impl<G: Scope> OperatorBuilder<G> {
 
     /// 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>) {
-
+        let new_output = self.shape.outputs;
+        self.shape.outputs += 1;
         let (targets, registrar) = Tee::<G::Timestamp,C>::new();
-        let source = Source::new(self.index, self.shape.outputs);
+        let source = Source::new(self.index, new_output);
         let stream = StreamCore::new(source, registrar, self.scope.clone());
 
-        self.shape.outputs += 1;
         assert_eq!(self.shape.inputs, connection.len());
         for (summary, entry) in self.summary.iter_mut().zip(connection.into_iter()) {
-            summary.add_port(entry);
+            summary.add_port(new_output, entry);
         }
 
         (targets, stream)

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

@@ -113,7 +113,7 @@ impl<G: Scope> OperatorBuilder<G> {
         OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
         StreamCore<G, CB::Container>
     ) {
-
+        let new_output = self.shape().outputs();
         let (tee, stream) = self.builder.new_output_connection(connection.clone());
 
         let internal = Rc::new(RefCell::new(ChangeBatch::new()));
@@ -123,7 +123,7 @@ impl<G: Scope> OperatorBuilder<G> {
         self.produced.push(Rc::clone(buffer.inner().produced()));
 
         for (summary, connection) in self.summaries.iter().zip(connection.into_iter()) {
-            summary.borrow_mut().add_port(connection.clone());
+            summary.borrow_mut().add_port(new_output, connection.clone());
         }
 
         (OutputWrapper::new(buffer, internal), stream)

timely/src/progress/operate.rs

@@ -80,29 +80,30 @@ impl<TS> PortConnectivity<TS> {
         Self { list }
     }
     /// Ensures an entry exists at `index` and returns a mutable reference to it.
-    pub fn ensure(&mut self, index: usize) -> &mut Antichain<TS> {
-        while self.next_port() <= index { self.add_port(Antichain::new()); }
+    fn ensure(&mut self, index: usize) -> &mut Antichain<TS> {
+        while self.list.len() <= index { self.add_port(self.list.len(), Antichain::new()); }
         &mut self.list[index]
     }
     /// Inserts an element by reference, ensuring that the index exists.
+    pub fn insert(&mut self, index: usize, element: TS) -> bool where TS : crate::PartialOrder + Clone {
+        self.ensure(index).insert(element)
+    }
+    /// Inserts an element by reference, ensuring that the index exists.
     pub fn insert_ref(&mut self, index: usize, element: &TS) -> bool where TS : crate::PartialOrder + Clone {
         self.ensure(index).insert_ref(element)
     }
-    /// Introduces a summary for the port `self.next_port()`.
-    pub fn add_port(&mut self, summary: Antichain<TS>) {
+    /// Introduces a summary for `port`. Panics if a summary already exists.
+    pub fn add_port(&mut self, port: usize, summary: Antichain<TS>) {
+        assert_eq!(self.list.len(), port);
         self.list.push(summary);
     }
     /// Borrowing iterator of port identifiers and antichains.
     pub fn iter_ports(&self) -> impl Iterator<Item = (usize, &Antichain<TS>)> {
         self.list.iter().enumerate()
     }
-    /// Owning iterator of port identifiers and antichains.
-    pub fn into_iter_ports(self) -> impl Iterator<Item = (usize, Antichain<TS>)> {
-        self.list.into_iter().enumerate()
-    }
-    /// Announces the next output port identifier.
-    pub fn next_port(&self) -> usize {
-        self.list.len()
+    /// Announces the largest port identifier, largely for debug asserts.
+    pub fn max_port(&self) -> usize {
+        self.list.len() - 1
     }
 }
 

timely/src/progress/reachability.rs

@@ -160,7 +160,7 @@ impl<T: Timestamp> Builder<T> {
 
         // Assert that all summaries exist.
         debug_assert_eq!(inputs, summary.len());
-        for x in summary.iter() { debug_assert_eq!(outputs, x.next_port()); }
+        debug_assert!(summary.iter().all(|os| os.iter_ports().all(|(o,_)| o < outputs)));
 
         while self.nodes.len() <= index {
             self.nodes.push(Vec::new());
@@ -779,8 +779,6 @@ fn summarize_outputs<T: Timestamp>(
                     // Determine the current path summaries from the input port.
                     let location = Location { node: location.node, port: Port::Target(input_port) };
                     let antichains = results.entry(location).or_default();
-                    // TODO: This is redundant with `insert_ref` below.
-                    antichains.ensure(output);
 
                     // Combine each operator-internal summary to the output with `summary`.
                     for operator_summary in summaries[output_port].elements().iter() {
@@ -801,8 +799,6 @@ fn summarize_outputs<T: Timestamp>(
                 // Each target should have (at most) one source.
                 if let Some(&source) = reverse.get(&location) {
                     let antichains = results.entry(source).or_default();
-                    // TODO: This is redundant with `insert_ref` below.
-                    antichains.ensure(output);
 
                     if antichains.insert_ref(output, &summary) {
                         worklist.push_back((source, output, summary.clone()));

timely/src/progress/subgraph.rs

@@ -558,9 +558,9 @@ where
         let mut internal_summary = vec![PortConnectivity::default_for(self.outputs); self.inputs()];
         for (input_idx, input) in self.scope_summary.iter().enumerate() {
             for (output_idx, output) in input.iter_ports() {
-                let antichain = internal_summary[input_idx].ensure(output_idx);
-                antichain.reserve(output.elements().len());
-                antichain.extend(output.elements().iter().cloned().map(TInner::summarize));
+                for outer in output.elements().iter().cloned().map(TInner::summarize) {
+                    internal_summary[input_idx].insert(output_idx, outer);
+                }
             }
         }
 
@@ -570,8 +570,8 @@ where
             "the internal summary should have as many elements as there are inputs",
         );
         debug_assert!(
-            internal_summary.iter().all(|summary| summary.next_port() == self.outputs()),
-            "each element of the internal summary should have as many elements as there are outputs",
+            internal_summary.iter().all(|os| os.iter_ports().all(|(o,_)| o < self.outputs())),
+            "each element of the internal summary should only reference valid outputs",
         );
 
         // Each child has expressed initial capabilities (their `shared_progress.internals`).
@@ -671,8 +671,8 @@ impl<T: Timestamp> PerOperatorState<T> {
             inputs,
         );
         assert!(
-            !internal_summary.iter().any(|x| x.next_port() != outputs),
-            "operator summary had too few outputs",
+            internal_summary.iter().all(|os| os.iter_ports().all(|(o,_)| o < outputs)),
+            "operator summary references invalid output port",
         );
 
         PerOperatorState {