Consuming iterators for PortConnectivity and less cloning.

timely/src/progress/operate.rs

@@ -153,6 +153,13 @@ impl<TS> Default for PortConnectivity<TS> {
     }
 }
 
+impl<TS> IntoIterator for PortConnectivity<TS> {
+    type Item = (usize, Antichain<TS>);
+    type IntoIter = std::vec::IntoIter<(usize, Antichain<TS>)>;
+    /// Consumes the connectivity, yielding each port and its antichain.
+    fn into_iter(self) -> Self::IntoIter { self.entries.into_iter() }
+}
+
 impl<TS> PortConnectivity<TS> {
     /// Borrowing iterator of port identifiers and antichains.
     pub fn iter_ports(&self) -> impl Iterator<Item = (usize, &Antichain<TS>)> {

timely/src/progress/reachability.rs

@@ -313,7 +313,7 @@
             for outputs in summary.iter() {
                 for (output, summaries) in outputs.iter_ports() {
                     let source = Location::new_source(index, output);
                     for summary in summaries.elements().iter() {
                         if summary == &Default::default() {
                             in_degree[index_of(&source)] += 1;
                         }
@@ -579,10 +579,10 @@
         }
 
         // Build columnar nodes: Vecs<Vecs<Vec<(usize, T::Summary)>>>.
-        let nodes = build_nested_vecs(builder.nodes.iter().map(|connectivity| {
-            connectivity.iter().map(|port_conn| {
-                port_conn.iter_ports().flat_map(|(port, antichain)| {
-                    antichain.elements().iter().map(move |s| (port, s.clone()))
+        let nodes = build_nested_vecs(builder.nodes.into_iter().map(|connectivity| {
+            connectivity.into_iter().map(|port_conn| {
+                port_conn.into_iter().flat_map(|(port, antichain)| {
+                    antichain.into_iter().map(move |s| (port, s))
                 })
             })
         }));
@@ -593,17 +593,17 @@
         }));
 
         // Build columnar target and source summaries.
-        let target_summaries = build_nested_vecs(target_sum.iter().map(|ports| {
-            ports.iter().map(|port_conn| {
-                port_conn.iter_ports().flat_map(|(port, antichain)| {
-                    antichain.elements().iter().map(move |s| (port, s.clone()))
+        let target_summaries = build_nested_vecs(target_sum.into_iter().map(|ports| {
+            ports.into_iter().map(|port_conn| {
+                port_conn.into_iter().flat_map(|(port, antichain)| {
+                    antichain.into_iter().map(move |s| (port, s))
                 })
             })
         }));
-        let source_summaries = build_nested_vecs(source_sum.iter().map(|ports| {
-            ports.iter().map(|port_conn| {
-                port_conn.iter_ports().flat_map(|(port, antichain)| {
-                    antichain.elements().iter().map(move |s| (port, s.clone()))
+        let source_summaries = build_nested_vecs(source_sum.into_iter().map(|ports| {
+            ports.into_iter().map(|port_conn| {
+                port_conn.into_iter().flat_map(|(port, antichain)| {
+                    antichain.into_iter().map(move |s| (port, s))
                 })
             })
         }));
@@ -728,7 +728,7 @@
                             .implications
                             .update_iter(Some((time, diff)));
 
                         for (time, diff) in changes {
                             for &(output_port, ref summary) in (&self.nodes).get(location.node).get(port_index).into_index_iter() {
                                 if let Some(new_time) = summary.results_in(&time) {
                                     let source = Location { node: location.node, port: Port::Source(output_port) };
@@ -748,7 +748,7 @@
                             .implications
                             .update_iter(Some((time, diff)));
 
                         for (time, diff) in changes {
                             for new_target in (&self.edges).get(location.node).get(port_index).into_index_iter() {
                                 self.worklist.push(Reverse((
                                     time.clone(),
@@ -887,7 +887,7 @@
             }
         }
     }
     reverse_internal.sort_unstable_by(|x, y| (x.0, x.1).cmp(&(y.0, y.1)));
 
     // Accumulated summaries to scope outputs, keyed by `(location, output)`.
     let mut accumulated: BinaryRuns<(Location, usize), Antichain<T::Summary>> = BinaryRuns::default();

timely/src/progress/subgraph.rs

@@ -566,9 +566,9 @@ where
         // with each element containing `self.outputs()` antichains regardless
         // of how long `self.scope_summary` is
         let mut internal_summary = vec![PortConnectivityBuilder::default(); self.inputs()];
-        for (input_idx, input) in self.scope_summary.iter().enumerate() {
-            for (output_idx, output) in input.iter_ports() {
-                for outer in output.elements().iter().cloned().map(TInner::summarize) {
+        for (input_idx, input) in std::mem::take(&mut self.scope_summary).into_iter().enumerate() {
+            for (output_idx, output) in input {
+                for outer in output.into_iter().map(TInner::summarize) {
                     internal_summary[input_idx].insert(output_idx, outer);
                 }
             }