Make PortConnectivity a struct, with sufficient methods

Author: frankmcsherry

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

@@ -124,7 +124,7 @@ impl<G: Scope> OperatorBuilder<G> {
 
         self.shape.inputs += 1;
         assert_eq!(self.shape.outputs, connection.len());
-        self.summary.push(connection);
+        self.summary.push(connection.into());
 
         receiver
     }
@@ -146,7 +146,7 @@ impl<G: Scope> OperatorBuilder<G> {
         self.shape.outputs += 1;
         assert_eq!(self.shape.inputs, connection.len());
         for (summary, entry) in self.summary.iter_mut().zip(connection.into_iter()) {
-            summary.push(entry);
+            summary.add_port(entry);
         }
 
         (targets, stream)

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

@@ -86,7 +86,7 @@ impl<G: Scope> OperatorBuilder<G> {
         self.frontier.push(MutableAntichain::new());
         self.consumed.push(Rc::clone(input.consumed()));
 
-        let shared_summary = Rc::new(RefCell::new(connection));
+        let shared_summary = Rc::new(RefCell::new(connection.into()));
         self.summaries.push(Rc::clone(&shared_summary));
 
         new_input_handle(input, Rc::clone(&self.internal), shared_summary, self.logging.clone())
@@ -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().push(connection.clone());
+            summary.borrow_mut().add_port(connection.clone());
         }
 
         (OutputWrapper::new(buffer, internal), stream)

timely/src/progress/operate.rs

@@ -61,8 +61,63 @@ pub trait Operate<T: Timestamp> : Schedule {
 /// Operator internal connectivity, from inputs to outputs.
 pub type Connectivity<TS> = Vec<PortConnectivity<TS>>;
 /// Internal connectivity from one port to any number of opposing ports.
-pub type PortConnectivity<TS> = Vec<Antichain<TS>>;
+#[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Debug, Clone, Eq, PartialEq)]
+pub struct PortConnectivity<TS> {
+    list: Vec<Antichain<TS>>,
+}
+
+impl<TS> Default for PortConnectivity<TS> {
+    fn default() -> Self {
+        Self { list: Vec::new() }
+    }
+}
+
+impl<TS> PortConnectivity<TS> {
+    /// Introduces default summaries for `0 .. count` ports.
+    pub fn default_for(count: usize) -> Self {
+        let mut list = Vec::with_capacity(count);
+        for _ in 0 .. count { list.push(Default::default()) }
+        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()); }
+        &mut self.list[index]
+    }
+    /// 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>) {
+        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()
+    }
+}
 
+impl<TS> From<Vec<Antichain<TS>>> for PortConnectivity<TS> {
+    fn from(list: Vec<Antichain<TS>>) -> Self {
+        Self { list }
+    }
+}
+
+impl<TS> std::ops::Index<usize> for PortConnectivity<TS> {
+    type Output = Antichain<TS>;
+    fn index(&self, index: usize) -> &Self::Output {
+        &self.list[index]
+    }
+}
 
 /// Progress information shared between parent and child.
 #[derive(Debug)]

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)]]);
-//! builder.add_node(1, 1, 1, vec![vec![Antichain::from_elem(0)]]);
-//! builder.add_node(2, 1, 1, vec![vec![Antichain::from_elem(1)]]);
+//! 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()]);
 //!
 //! // Connect nodes in sequence, looping around to the first from the last.
 //! builder.add_edge(Source::new(0, 0), Target::new(1, 0));
@@ -80,7 +80,7 @@ use crate::progress::{Source, Target};
 use crate::progress::ChangeBatch;
 use crate::progress::{Location, Port};
 use crate::progress::operate::{Connectivity, PortConnectivity};
-use crate::progress::frontier::{Antichain, MutableAntichain};
+use crate::progress::frontier::MutableAntichain;
 use crate::progress::timestamp::PathSummary;
 
 
@@ -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)]]);
-/// builder.add_node(1, 1, 1, vec![vec![Antichain::from_elem(0)]]);
-/// builder.add_node(2, 1, 1, vec![vec![Antichain::from_elem(1)]]);
+/// 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()]);
 ///
 /// // Connect nodes in sequence, looping around to the first from the last.
 /// builder.add_edge(Source::new(0, 0), Target::new(1, 0));
@@ -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.len()); }
+        for x in summary.iter() { debug_assert_eq!(outputs, x.next_port()); }
 
         while self.nodes.len() <= index {
             self.nodes.push(Vec::new());
@@ -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)]]);
-    /// builder.add_node(1, 1, 1, vec![vec![Antichain::from_elem(0)]]);
-    /// builder.add_node(2, 1, 1, vec![vec![Antichain::from_elem(0)]]);
+    /// 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()]);
     ///
     /// // 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(),],
-    ///     vec![Antichain::new(),Antichain::from_elem(1),],
+    ///     vec![Antichain::from_elem(0),Antichain::new(),].into(),
+    ///     vec![Antichain::new(),Antichain::from_elem(1),].into(),
     /// ]);
     ///
     /// // Connect each output to the opposite input.
@@ -285,7 +285,7 @@ impl<T: Timestamp> Builder<T> {
             for (input, outputs) in summary.iter().enumerate() {
                 let target = Location::new_target(index, input);
                 in_degree.entry(target).or_insert(0);
-                for (output, summaries) in outputs.iter().enumerate() {
+                for (output, summaries) in outputs.iter_ports() {
                     let source = Location::new_source(index, output);
                     for summary in summaries.elements().iter() {
                         if summary == &Default::default() {
@@ -322,7 +322,7 @@ impl<T: Timestamp> Builder<T> {
                     }
                 },
                 Port::Target(port) => {
-                    for (output, summaries) in self.nodes[node][port].iter().enumerate() {
+                    for (output, summaries) in self.nodes[node][port].iter_ports() {
                         let source = Location::new_source(node, output);
                         for summary in summaries.elements().iter() {
                             if summary == &Default::default() {
@@ -442,7 +442,7 @@ impl<T: Timestamp> PortInformation<T> {
         PortInformation {
             pointstamps: MutableAntichain::new(),
             implications: MutableAntichain::new(),
-            output_summaries: Vec::new(),
+            output_summaries: PortConnectivity::default(),
         }
     }
 
@@ -514,7 +514,7 @@ impl<T:Timestamp> Tracker<T> {
             .collect::<Vec<_>>();
 
         // Summary of scope inputs to scope outputs.
-        let mut builder_summary = vec![vec![]; builder.shape[0].1];
+        let mut builder_summary = vec![PortConnectivity::default(); builder.shape[0].1];
 
         // Compile summaries from each location to each scope output.
         let output_summaries = summarize_outputs::<T>(&builder.nodes, &builder.edges);
@@ -598,7 +598,7 @@ impl<T:Timestamp> Tracker<T> {
 
             for (time, diff) in changes {
                 self.total_counts += diff;
-                for (output, summaries) in operator.output_summaries.iter().enumerate() {
+                for (output, summaries) in operator.output_summaries.iter_ports() {
                     let output_changes = &mut self.output_changes[output];
                     summaries
                         .elements()
@@ -617,7 +617,7 @@ impl<T:Timestamp> Tracker<T> {
 
             for (time, diff) in changes {
                 self.total_counts += diff;
-                for (output, summaries) in operator.output_summaries.iter().enumerate() {
+                for (output, summaries) in operator.output_summaries.iter_ports() {
                     let output_changes = &mut self.output_changes[output];
                     summaries
                         .elements()
@@ -658,7 +658,7 @@ impl<T:Timestamp> Tracker<T> {
 
                         for (time, diff) in changes {
                             let nodes = &self.nodes[location.node][port_index];
-                            for (output_port, summaries) in nodes.iter().enumerate() {
+                            for (output_port, summaries) in nodes.iter_ports() {
                                 let source = Location { node: location.node, port: Port::Source(output_port) };
                                 for summary in summaries.elements().iter() {
                                     if let Some(new_time) = summary.results_in(&time) {
@@ -778,17 +778,14 @@ 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)
-                        .and_modify(|antichains| antichains.reserve(output))
-                        .or_insert_with(|| Vec::with_capacity(output));
-
-                    while antichains.len() <= output { antichains.push(Antichain::new()); }
+                    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() {
                         if let Some(combined) = operator_summary.followed_by(&summary) {
-                            if antichains[output].insert(combined.clone()) {
+                            if antichains.insert_ref(output, &combined) {
                                 worklist.push_back((location, output, combined));
                             }
                         }
@@ -803,14 +800,11 @@ 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)
-                        .and_modify(|antichains| antichains.reserve(output))
-                        .or_insert_with(|| Vec::with_capacity(output));
-
-                    while antichains.len() <= output { antichains.push(Antichain::new()); }
+                    let antichains = results.entry(source).or_default();
+                    // TODO: This is redundant with `insert_ref` below.
+                    antichains.ensure(output);
 
-                    if antichains[output].insert(summary.clone()) {
+                    if antichains.insert_ref(output, &summary) {
                         worklist.push_back((source, output, summary.clone()));
                     }
                 }

timely/src/progress/subgraph.rs

@@ -16,10 +16,10 @@ use crate::logging::TimelySummaryLogger as SummaryLogger;
 use crate::scheduling::Schedule;
 use crate::scheduling::activate::Activations;
 
-use crate::progress::frontier::{Antichain, MutableAntichain, MutableAntichainFilter};
+use crate::progress::frontier::{MutableAntichain, MutableAntichainFilter};
 use crate::progress::{Timestamp, Operate, operate::SharedProgress};
 use crate::progress::{Location, Port, Source, Target};
-use crate::progress::operate::Connectivity;
+use crate::progress::operate::{Connectivity, PortConnectivity};
 use crate::progress::ChangeBatch;
 use crate::progress::broadcast::Progcaster;
 use crate::progress::reachability;
@@ -168,7 +168,7 @@ where
         let mut builder = reachability::Builder::new();
 
         // Child 0 has `inputs` outputs and `outputs` inputs, not yet connected.
-        let summary = (0..outputs).map(|_| (0..inputs).map(|_| Antichain::new()).collect()).collect();
+        let summary = (0..outputs).map(|_| PortConnectivity::default_for(inputs)).collect();
         builder.add_node(0, outputs, inputs, summary);
         for (index, child) in self.children.iter().enumerate().skip(1) {
             builder.add_node(index, child.inputs, child.outputs, child.internal_summary.clone());
@@ -555,10 +555,10 @@ where
         // Note that we need to have `self.inputs()` elements in the summary
         // with each element containing `self.outputs()` antichains regardless
         // of how long `self.scope_summary` is
-        let mut internal_summary = vec![vec![Antichain::new(); self.outputs()]; self.inputs()];
+        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().enumerate() {
-                let antichain = &mut internal_summary[input_idx][output_idx];
+            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));
             }
@@ -570,7 +570,7 @@ where
             "the internal summary should have as many elements as there are inputs",
         );
         debug_assert!(
-            internal_summary.iter().all(|summary| summary.len() == self.outputs()),
+            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",
         );
 
@@ -671,7 +671,7 @@ impl<T: Timestamp> PerOperatorState<T> {
             inputs,
         );
         assert!(
-            !internal_summary.iter().any(|x| x.len() != outputs),
+            !internal_summary.iter().any(|x| x.next_port() != outputs),
             "operator summary had too few outputs",
         );