Introduce tests; catch further issues

Author: frankmcsherry

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

@@ -124,7 +124,6 @@ impl<G: Scope> OperatorBuilder<G> {
         stream.connect_to(target, sender, channel_id);
 
         self.shape.inputs += 1;
-        assert_eq!(self.shape.outputs, connection.len());
         self.summary.push(connection);
 
         receiver

timely/src/progress/reachability.rs

@@ -754,6 +754,19 @@ fn summarize_outputs<T: Timestamp>(
         }
     }
 
+    // A reverse map from operator outputs to inputs, along their internal summaries.
+    let mut reverse_internal = HashMap::new();
+    for (node, connectivity) in nodes.iter().enumerate() {
+        for (input, outputs) in connectivity.iter().enumerate() {
+            for (output, summary) in outputs.iter() {
+                reverse_internal
+                    .entry(Location::new_source(node, *output))
+                    .or_insert_with(|| Vec::default())
+                    .push((input, summary));
+            }
+        }
+    }
+
     let mut results: HashMap<Location, PortConnectivity<T::Summary>> = HashMap::new();
     let mut worklist = VecDeque::<(Location, usize, T::Summary)>::new();
 
@@ -776,20 +789,18 @@ 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_insert_with(|| PortConnectivity::default());
-
-                    // Combine each operator-internal summary to the output with `summary`.
-                    if let Some(port_connectivity) = summaries.get(&output_port) {
-                        for operator_summary in port_connectivity.elements().iter() {
-                            if let Some(combined) = operator_summary.followed_by(&summary) {
-                                let update = antichains.entry(output).or_insert_with(|| Antichain::default()).insert_ref(&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_insert_with(|| PortConnectivity::default())
+                                    .entry(output)
+                                    .or_insert_with(|| Antichain::default())
+                                    .insert_ref(&combined);
                                 if update {
                                     worklist.push_back((location, output, combined));
                                 }

timely/src/progress/subgraph.rs

@@ -572,8 +572,8 @@ where
             "the internal summary should have as many elements as there are inputs",
         );
         debug_assert!(
-            internal_summary.iter().all(|summary| summary.len() == self.outputs()),
-            "each element of the internal summary should have as many elements as there are outputs",
+            internal_summary.iter().all(|summary| summary.keys().all(|output| output < &self.outputs())),
+            "each element of the internal summary should only reference recognized outputs",
         );
 
         // Each child has expressed initial capabilities (their `shared_progress.internals`).
@@ -673,7 +673,7 @@ impl<T: Timestamp> PerOperatorState<T> {
             inputs,
         );
         assert!(
-            !internal_summary.iter().any(|x| x.len() != outputs),
+            !internal_summary.iter().any(|x| x.keys().any(|k| k >= &outputs)),
             "operator summary had too few outputs",
         );
 

timely/tests/shape_scaling.rs

@@ -0,0 +1,79 @@
+use timely::dataflow::channels::pact::Pipeline;
+use timely::dataflow::operators::Input;
+use timely::dataflow::InputHandle;
+use timely::Config;
+
+#[test] fn operator_scaling_1() { operator_scaling(1); }
+#[test] fn operator_scaling_10() { operator_scaling(10); }
+#[test] fn operator_scaling_100() { operator_scaling(100); }
+#[test] fn operator_scaling_1000() { operator_scaling(1000); }
+#[test] fn operator_scaling_10000() { operator_scaling(10000); }
+#[test] fn operator_scaling_100000() { operator_scaling(100000); }
+
+fn operator_scaling(scale: u64) {
+    timely::execute(Config::thread(), move |worker| {
+        let mut input = InputHandle::new();
+        worker.dataflow::<u64, _, _>(|scope| {
+            use timely::dataflow::operators::Partition;
+            let parts =
+            scope
+                .input_from(&mut input)
+                .partition(scale, |()| (0, ()));
+
+            use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
+            let mut builder = OperatorBuilder::new("OpScaling".to_owned(), scope.clone());
+            let mut handles = Vec::with_capacity(parts.len());
+            let mut outputs = Vec::with_capacity(parts.len());
+            for (index, part) in parts.into_iter().enumerate() {
+                use timely::container::CapacityContainerBuilder;
+                let (output, stream) = builder.new_output_connection::<CapacityContainerBuilder<Vec<()>>>(Default::default());
+                use timely::progress::Antichain;
+                let connectivity = [(index, Antichain::from_elem(Default::default()))].into();
+                handles.push((builder.new_input_connection(&part, Pipeline, connectivity), output));
+                outputs.push(stream);
+            }
+
+            builder.build(move |_| {
+                move |_frontiers| {
+                    for (input, output) in handles.iter_mut() {
+                        let mut output = output.activate();
+                        input.for_each(|time, data| {
+                            let mut output = output.session_with_builder(&time);
+                            for datum in data.drain(..) {
+                                output.give(datum);
+                            }
+                        });
+                    }
+                }
+            });
+        });
+    })
+    .unwrap();
+}
+
+#[test] fn subgraph_scaling_1() { subgraph_scaling(1); }
+#[test] fn subgraph_scaling_10() { subgraph_scaling(10); }
+#[test] fn subgraph_scaling_100() { subgraph_scaling(100); }
+#[test] fn subgraph_scaling_1000() { subgraph_scaling(1000); }
+#[test] fn subgraph_scaling_10000() { subgraph_scaling(10000); }
+#[test] fn subgraph_scaling_100000() { subgraph_scaling(100000); }
+
+fn subgraph_scaling(scale: u64) {
+    timely::execute(Config::thread(), move |worker| {
+        let mut input = InputHandle::new();
+        worker.dataflow::<u64, _, _>(|scope| {
+            use timely::dataflow::operators::Partition;
+            let parts =
+            scope
+                .input_from(&mut input)
+                .partition(scale, |()| (0, ()));
+
+            use timely::dataflow::Scope;
+            let _outputs = scope.region(|inner| {
+                use timely::dataflow::operators::{Enter, Leave};
+                parts.into_iter().map(|part| part.enter(inner).leave()).collect::<Vec<_>>()
+            });
+        });
+    })
+    .unwrap();
+}