Linear connectivity (#651)

* Connectivity as a type alias

* Introduce PortConnectivity as BTreeMap

* Introduce tests; catch further issues

* Respond to review feedback

Author: frankmcsherry

timely/src/dataflow/operators/capability.rs

@@ -27,9 +27,9 @@ use std::cell::RefCell;
 use std::fmt::{self, Debug};
 
 use crate::order::PartialOrder;
-use crate::progress::Antichain;
 use crate::progress::Timestamp;
 use crate::progress::ChangeBatch;
+use crate::progress::operate::PortConnectivity;
 use crate::scheduling::Activations;
 use crate::dataflow::channels::pullers::counter::ConsumedGuard;
 
@@ -238,26 +238,26 @@ pub struct InputCapability<T: Timestamp> {
     /// Output capability buffers, for use in minting capabilities.
     internal: CapabilityUpdates<T>,
     /// Timestamp summaries for each output.
-    summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>,
+    summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,
     /// A drop guard that updates the consumed capability this InputCapability refers to on drop
     consumed_guard: ConsumedGuard<T>,
 }
 
 impl<T: Timestamp> CapabilityTrait<T> for InputCapability<T> {
     fn time(&self) -> &T { self.time() }
     fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
-        let borrow = self.summaries.borrow();
-        self.internal.borrow().iter().enumerate().any(|(index, rc)| {
+        let internal_borrow = self.internal.borrow();
+        self.summaries.borrow().iter().any(|(output, summary)| {
             // To be valid, the output buffer must match and the timestamp summary needs to be the default.
-            Rc::ptr_eq(rc, query_buffer) && borrow[index].len() == 1 && borrow[index][0] == Default::default()
+            Rc::ptr_eq(&internal_borrow[*output], query_buffer) && summary.len() == 1 && summary[0] == Default::default()
         })
     }
 }
 
 impl<T: Timestamp> InputCapability<T> {
     /// Creates a new capability reference at `time` while incrementing (and keeping a reference to)
     /// the provided [`ChangeBatch`].
-    pub(crate) fn new(internal: CapabilityUpdates<T>, summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>, guard: ConsumedGuard<T>) -> Self {
+    pub(crate) fn new(internal: CapabilityUpdates<T>, summaries: Rc<RefCell<PortConnectivity<T::Summary>>>, guard: ConsumedGuard<T>) -> Self {
         InputCapability {
             internal,
             summaries,
@@ -281,10 +281,10 @@ impl<T: Timestamp> InputCapability<T> {
     /// Delays capability for a specific output port.
     pub fn delayed_for_output(&self, new_time: &T, output_port: usize) -> Capability<T> {
         use crate::progress::timestamp::PathSummary;
-        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) {
+        if self.summaries.borrow()[&output_port].iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) {
             Capability::new(new_time.clone(), self.internal.borrow()[output_port].clone())
         } else {
-            panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", new_time, self.summaries.borrow()[output_port], self.time());
+            panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", new_time, self.summaries.borrow()[&output_port], self.time());
         }
     }
 
@@ -305,11 +305,11 @@ impl<T: Timestamp> InputCapability<T> {
     pub fn retain_for_output(self, output_port: usize) -> Capability<T> {
         use crate::progress::timestamp::PathSummary;
         let self_time = self.time().clone();
-        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(&self_time)).any(|time| time.less_equal(&self_time)) {
+        if self.summaries.borrow()[&output_port].iter().flat_map(|summary| summary.results_in(&self_time)).any(|time| time.less_equal(&self_time)) {
             Capability::new(self_time, self.internal.borrow()[output_port].clone())
         }
         else {
-            panic!("Attempted to retain a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", self_time, self.summaries.borrow()[output_port], self_time);
+            panic!("Attempted to retain a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", self_time, self.summaries.borrow()[&output_port], self_time);
         }
     }
 }

timely/src/dataflow/operators/core/feedback.rs

@@ -113,7 +113,8 @@ impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
         let summary = handle.summary;
         let mut output = handle.output;
 
-        let mut input = builder.new_input_connection(self, Pipeline, vec![Antichain::from_elem(summary.clone())]);
+        let connection = Some((0, Antichain::from_elem(summary.clone()))).into_iter().collect();
+        let mut input = builder.new_input_connection(self, Pipeline, connection);
 
         builder.build(move |_capability| move |_frontier| {
             let mut output = output.activate();

timely/src/dataflow/operators/core/input.rs

@@ -7,9 +7,9 @@ use crate::container::{CapacityContainerBuilder, ContainerBuilder, PushInto};
 
 use crate::scheduling::{Schedule, Activator};
 
-use crate::progress::frontier::Antichain;
 use crate::progress::{Operate, operate::SharedProgress, Timestamp, ChangeBatch};
 use crate::progress::Source;
+use crate::progress::operate::Connectivity;
 
 use crate::{Container, Data};
 use crate::communication::Push;
@@ -205,7 +205,7 @@ impl<T:Timestamp> Operate<T> for Operator<T> {
     fn inputs(&self) -> usize { 0 }
     fn outputs(&self) -> usize { 1 }
 
-    fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<<T as Timestamp>::Summary>>>, Rc<RefCell<SharedProgress<T>>>) {
+    fn get_internal_summary(&mut self) -> (Connectivity<T::Summary>, Rc<RefCell<SharedProgress<T>>>) {
         self.shared_progress.borrow_mut().internals[0].update(T::minimum(), self.copies as i64);
         (Vec::new(), self.shared_progress.clone())
     }

timely/src/dataflow/operators/core/unordered_input.rs

@@ -7,10 +7,10 @@ use crate::container::{ContainerBuilder, CapacityContainerBuilder};
 
 use crate::scheduling::{Schedule, ActivateOnDrop};
 
-use crate::progress::frontier::Antichain;
 use crate::progress::{Operate, operate::SharedProgress, Timestamp};
 use crate::progress::Source;
 use crate::progress::ChangeBatch;
+use crate::progress::operate::Connectivity;
 
 use crate::dataflow::channels::pushers::{Counter, Tee};
 use crate::dataflow::channels::pushers::buffer::{Buffer as PushBuffer, AutoflushSession};
@@ -134,7 +134,7 @@ impl<T:Timestamp> Operate<T> for UnorderedOperator<T> {
     fn inputs(&self) -> usize { 0 }
     fn outputs(&self) -> usize { 1 }
 
-    fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<<T as Timestamp>::Summary>>>, Rc<RefCell<SharedProgress<T>>>) {
+    fn get_internal_summary(&mut self) -> (Connectivity<T::Summary>, Rc<RefCell<SharedProgress<T>>>) {
         let mut borrow = self.internal.borrow_mut();
         for (time, count) in borrow.drain() {
             self.shared_progress.borrow_mut().internals[0].update(time, count * (self.peers as i64));

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

@@ -12,6 +12,7 @@ use crate::scheduling::{Schedule, Activations};
 
 use crate::progress::{Source, Target};
 use crate::progress::{Timestamp, Operate, operate::SharedProgress, Antichain};
+use crate::progress::operate::{Connectivity, PortConnectivity};
 
 use crate::Container;
 use crate::dataflow::{StreamCore, Scope};
@@ -60,7 +61,7 @@ pub struct OperatorBuilder<G: Scope> {
     global: usize,
     address: Rc<[usize]>,    // path to the operator (ending with index).
     shape: OperatorShape,
-    summary: Vec<Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>>,
+    summary: Connectivity<<G::Timestamp as Timestamp>::Summary>,
 }
 
 impl<G: Scope> OperatorBuilder<G> {
@@ -107,12 +108,12 @@ impl<G: Scope> OperatorBuilder<G> {
     pub fn new_input<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P) -> P::Puller
         where
             P: ParallelizationContract<G::Timestamp, C> {
-        let connection = vec![Antichain::from_elem(Default::default()); self.shape.outputs];
+        let connection = (0 .. self.shape.outputs).map(|o| (o, Antichain::from_elem(Default::default()))).collect();
         self.new_input_connection(stream, pact, connection)
     }
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
-    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> P::Puller
+    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>) -> P::Puller
     where
         P: ParallelizationContract<G::Timestamp, C> {
 
@@ -123,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
@@ -132,21 +132,20 @@ impl<G: Scope> OperatorBuilder<G> {
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
     pub fn new_output<C: Container>(&mut self) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
 
-        let connection = vec![Antichain::from_elem(Default::default()); self.shape.inputs];
+        let connection = (0 .. self.shape.inputs).map(|i| (i, Antichain::from_elem(Default::default()))).collect();
         self.new_output_connection(connection)
     }
 
     /// 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>) {
+    pub fn new_output_connection<C: Container>(&mut self, connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
 
         let (targets, registrar) = Tee::<G::Timestamp,C>::new();
         let source = Source::new(self.index, self.shape.outputs);
         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.push(entry);
+        for (input, entry) in connection.into_iter() {
+            self.summary[input].insert(self.shape.outputs - 1, entry);
         }
 
         (targets, stream)
@@ -188,7 +187,7 @@ where
     logic: L,
     shared_progress: Rc<RefCell<SharedProgress<T>>>,
     activations: Rc<RefCell<Activations>>,
-    summary: Vec<Vec<Antichain<T::Summary>>>,
+    summary: Connectivity<T::Summary>,
 }
 
 impl<T, L> Schedule for OperatorCore<T, L>
@@ -213,7 +212,7 @@ where
     fn outputs(&self) -> usize { self.shape.outputs }
 
     // announce internal topology as fully connected, and hold all default capabilities.
-    fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<T::Summary>>>, Rc<RefCell<SharedProgress<T>>>) {
+    fn get_internal_summary(&mut self) -> (Connectivity<T::Summary>, Rc<RefCell<SharedProgress<T>>>) {
 
         // Request the operator to be scheduled at least once.
         self.activations.borrow_mut().activate(&self.address[..]);

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

@@ -20,7 +20,7 @@ use crate::dataflow::operators::capability::Capability;
 use crate::dataflow::operators::generic::handles::{InputHandleCore, new_input_handle, OutputWrapper};
 use crate::dataflow::operators::generic::operator_info::OperatorInfo;
 use crate::dataflow::operators::generic::builder_raw::OperatorShape;
-
+use crate::progress::operate::PortConnectivity;
 use crate::logging::TimelyLogger as Logger;
 
 use super::builder_raw::OperatorBuilder as OperatorBuilderRaw;
@@ -33,7 +33,7 @@ pub struct OperatorBuilder<G: Scope> {
     consumed: Vec<Rc<RefCell<ChangeBatch<G::Timestamp>>>>,
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<G::Timestamp>>>>>>,
     /// For each input, a shared list of summaries to each output.
-    summaries: Vec<Rc<RefCell<Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>>>>,
+    summaries: Vec<Rc<RefCell<PortConnectivity<<G::Timestamp as Timestamp>::Summary>>>>,
     produced: Vec<Rc<RefCell<ChangeBatch<G::Timestamp>>>>,
     logging: Option<Logger>,
 }
@@ -64,7 +64,7 @@ impl<G: Scope> OperatorBuilder<G> {
     where
         P: ParallelizationContract<G::Timestamp, C> {
 
-        let connection = (0..self.builder.shape().outputs()).map(|_| Antichain::from_elem(Default::default())).collect();
+        let connection = (0..self.builder.shape().outputs()).map(|o| (o, Antichain::from_elem(Default::default()))).collect();
         self.new_input_connection(stream, pact, connection)
     }
 
@@ -76,7 +76,7 @@ impl<G: Scope> OperatorBuilder<G> {
     ///
     /// Commonly the connections are either the unit summary, indicating the same timestamp might be produced as output, or an empty
     /// antichain indicating that there is no connection from the input to the output.
-    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> InputHandleCore<G::Timestamp, C, P::Puller>
+    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>) -> InputHandleCore<G::Timestamp, C, P::Puller>
         where
             P: ParallelizationContract<G::Timestamp, C> {
 
@@ -94,7 +94,7 @@ impl<G: Scope> OperatorBuilder<G> {
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
     pub fn new_output<CB: ContainerBuilder>(&mut self) -> (OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>, StreamCore<G, CB::Container>) {
-        let connection = (0..self.builder.shape().inputs()).map(|_| Antichain::from_elem(Default::default())).collect();
+        let connection = (0..self.builder.shape().inputs()).map(|i| (i, Antichain::from_elem(Default::default()))).collect();
         self.new_output_connection(connection)
     }
 
@@ -108,7 +108,7 @@ impl<G: Scope> OperatorBuilder<G> {
     /// antichain indicating that there is no connection from the input to the output.
     pub fn new_output_connection<CB: ContainerBuilder>(
         &mut self,
-        connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>
+        connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>
     ) -> (
         OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
         StreamCore<G, CB::Container>
@@ -122,8 +122,9 @@ impl<G: Scope> OperatorBuilder<G> {
         let mut buffer = PushBuffer::new(PushCounter::new(tee));
         self.produced.push(buffer.inner().produced().clone());
 
-        for (summary, connection) in self.summaries.iter().zip(connection.into_iter()) {
-            summary.borrow_mut().push(connection.clone());
+        for (input, summary) in connection.into_iter() {
+            let mut borrow = self.summaries[input].borrow_mut();
+            borrow.insert(self.shape().outputs() - 1, summary);
         }
 
         (OutputWrapper::new(buffer, internal), stream)

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

@@ -6,10 +6,10 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 
-use crate::progress::Antichain;
 use crate::progress::Timestamp;
 use crate::progress::ChangeBatch;
 use crate::progress::frontier::MutableAntichain;
+use crate::progress::operate::PortConnectivity;
 use crate::dataflow::channels::pullers::Counter as PullCounter;
 use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::{Buffer, Session};
@@ -30,7 +30,7 @@ pub struct InputHandleCore<T: Timestamp, C: Container, P: Pull<Message<T, C>>> {
     ///
     /// Each timestamp received through this input may only produce output timestamps
     /// greater or equal to the input timestamp subjected to at least one of these summaries.
-    summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>, 
+    summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,
     logging: Option<Logger>,
 }
 
@@ -148,8 +148,8 @@ pub fn _access_pull_counter<T: Timestamp, C: Container, P: Pull<Message<T, C>>>(
 /// Declared separately so that it can be kept private when `InputHandle` is re-exported.
 pub fn new_input_handle<T: Timestamp, C: Container, P: Pull<Message<T, C>>>(
     pull_counter: PullCounter<T, C, P>, 
-    internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>, 
-    summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>, 
+    internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>,
+    summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,
     logging: Option<Logger>
 ) -> InputHandleCore<T, C, P> {
     InputHandleCore {

timely/src/logging.rs

@@ -22,6 +22,7 @@ use serde::{Deserialize, Serialize};
 use crate::Container;
 use crate::container::CapacityContainerBuilder;
 use crate::dataflow::operators::capture::{Event, EventPusher};
+use crate::progress::operate::Connectivity;
 
 /// Logs events as a timely stream, with progress statements.
 pub struct BatchLogger<P, C> where P: EventPusher<Duration, C> {
@@ -76,7 +77,7 @@ pub struct OperatesSummaryEvent<TS> {
     /// Worker-unique identifier for the operator.
     pub id: usize,
     /// Timestamp action summaries for (input, output) pairs.
-    pub summary: Vec<Vec<crate::progress::Antichain<TS>>>,
+    pub summary: Connectivity<TS>,
 }
 
 #[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]

timely/src/progress/operate.rs

@@ -44,7 +44,7 @@ pub trait Operate<T: Timestamp> : Schedule {
     ///
     /// The default behavior is to indicate that timestamps on any input can emerge unchanged on
     /// any output, and no initial capabilities are held.
-    fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<T::Summary>>>, Rc<RefCell<SharedProgress<T>>>);
+    fn get_internal_summary(&mut self) -> (Connectivity<T::Summary>, Rc<RefCell<SharedProgress<T>>>);
 
     /// Signals that external frontiers have been set.
     ///
@@ -58,6 +58,11 @@ pub trait Operate<T: Timestamp> : Schedule {
     fn notify_me(&self) -> bool { true }
 }
 
+/// 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> = std::collections::BTreeMap<usize, Antichain<TS>>;
+
 /// Progress information shared between parent and child.
 #[derive(Debug)]
 pub struct SharedProgress<T: Timestamp> {