Adjust reachability logic to be more linear

Author: frankmcsherry

timely/src/progress/reachability.rs

@@ -749,6 +749,19 @@ fn summarize_outputs<T: Timestamp>(
         }
     }
 
+    // A reverse map from operator outputs to inputs, along their internal summaries.
+    let mut reverse_internal: HashMap<_, Vec<_>> = HashMap::new();
+    for (node, connectivity) in nodes.iter().enumerate() {
+        for (input, outputs) in connectivity.iter().enumerate() {
+            for (output, summary) in outputs.iter_ports() {
+                reverse_internal
+                    .entry(Location::new_source(node, output))
+                    .or_default()
+                    .push((input, summary));
+            }
+        }
+    }
+    
     let mut results: HashMap<Location, PortConnectivity<T::Summary>> = HashMap::new();
     let mut worklist = VecDeque::<(Location, usize, T::Summary)>::new();
 
@@ -771,28 +784,47 @@ fn summarize_outputs<T: Timestamp>(
 
             // This is an output port of an operator, or a scope input.
             // We want to crawl up the operator, to its inputs.
-            Port::Source(output_port) => {
-
-                // Consider each input port of the associated operator.
-                for (input_port, summaries) in nodes[location.node].iter().enumerate() {
-
-                    // 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();
-
-                    // Combine each operator-internal summary to the output with `summary`.
-                    if let Some(connection) = summaries.get(output_port) {
-                        for operator_summary in connection.elements().iter() {
-                            if let Some(combined) = operator_summary.followed_by(&summary) {
-                                if antichains.insert_ref(output, &combined) {
+            // Port::Source(output_port) => {
+
+            //     // Consider each input port of the associated operator.
+            //     for (input_port, summaries) in nodes[location.node].iter().enumerate() {
+
+            //         // 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();
+
+            //         // Combine each operator-internal summary to the output with `summary`.
+            //         if let Some(connection) = summaries.get(output_port) {
+            //             for operator_summary in connection.elements().iter() {
+            //                 if let Some(combined) = operator_summary.followed_by(&summary) {
+            //                     if antichains.insert_ref(output, &combined) {
+            //                         worklist.push_back((location, output, combined));
+            //                     }
+            //                 }
+            //             }
+            //         }
+            //     }
+
+            // },
+
+            Port::Source(_output_port) => {
+                if let Some(inputs) = reverse_internal.get(&location) {
+                    for (input_port, operator_summary) in inputs.iter() {
+                        let location = Location::new_target(location.node, *input_port);
+                        for summary in operator_summary.elements().iter() {
+                            if let Some(combined) = summary.followed_by(&summary) {
+                                let update =
+                                results.entry(location)
+                                    .or_default()
+                                    .insert_ref(output, &combined);
+                                if update {
                                     worklist.push_back((location, output, combined));
                                 }
                             }
                         }
                     }
                 }
-
-            },
+            }
 
             // This is an input port of an operator, or a scope output.
             // We want to walk back the edges leading to it.