Messages stamped by sets of capabilities

timely/src/dataflow/channels/mod.rs

@@ -10,11 +10,16 @@ pub mod pullers;
 /// Parallelization contracts, describing how data must be exchanged between operators.
 pub mod pact;
 
+pub use crate::progress::Stamp;
+
 /// A serializable representation of timestamped data.
 #[derive(Clone)]
 pub struct Message<T, C> {
-    /// The timestamp associated with the message.
-    pub time: T,
+    /// The multiset of timestamps affixed to the message.
+    ///
+    /// The message may only result in downstream work at times greater or equal
+    /// to some element of the stamp. An empty stamp makes no progress claims.
+    pub stamp: Stamp<T>,
     /// The data in the message.
     pub data: C,
     /// The source worker.
@@ -35,17 +40,17 @@ impl<T, C> Message<T, C> {
     /// Creates a new message instance from arguments.
     ///
     /// Zero values are installed for `from` and `seq`, and are meant to be populated by `LogPusher`.
-    pub fn new(time: T, data: C) -> Self {
-        Message { time, data, from: 0, seq: 0 }
+    pub fn new(stamp: Stamp<T>, data: C) -> Self {
+        Message { stamp, data, from: 0, seq: 0 }
     }
 
     /// Forms a message from borrowed parts, and replaces `buffer` with what is left by the `push` call.
     /// If the pusher returns nothing, then `buffer` is set to the default for the container.
     #[inline]
-    pub fn push_at<P: Push<Message<T, C>>>(buffer: &mut C, time: T, pusher: &mut P) where C: Default {
+    pub fn push_at<P: Push<Message<T, C>>>(buffer: &mut C, stamp: Stamp<T>, pusher: &mut P) where C: Default {
 
         let data = ::std::mem::take(buffer);
-        let message = Message::new(time, data);
+        let message = Message::new(stamp, data);
         let mut bundle = Some(message);
 
         pusher.push(&mut bundle);
@@ -70,17 +75,17 @@ where
         let mut slice = &bytes[..];
         let from: usize = slice.read_u64::<byteorder::LittleEndian>().unwrap().try_into().unwrap();
         let seq: usize = slice.read_u64::<byteorder::LittleEndian>().unwrap().try_into().unwrap();
-        let time: T = ::bincode::deserialize_from(&mut slice).expect("bincode::deserialize() failed");
-        let time_size = ::bincode::serialized_size(&time).expect("bincode::serialized_size() failed") as usize;
+        let stamp: Stamp<T> = ::bincode::deserialize_from(&mut slice).expect("bincode::deserialize() failed");
+        let time_size = ::bincode::serialized_size(&stamp).expect("bincode::serialized_size() failed") as usize;
         // We expect to find the `data` payload at `8 + 8 + round_up(time_size)`;
         let bytes_read = 8 + 8 + ((time_size + 7) & !7);
         bytes.extract_to(bytes_read);
         let data: C = ContainerBytes::from_bytes(bytes);
-        Self { time, data, from, seq }
+        Self { stamp, data, from, seq }
     }
 
     fn length_in_bytes(&self) -> usize {
-        let time_size = ::bincode::serialized_size(&self.time).expect("bincode::serialized_size() failed") as usize;
+        let time_size = ::bincode::serialized_size(&self.stamp).expect("bincode::serialized_size() failed") as usize;
         // 16 comes from the two `u64` fields: `from` and `seq`.
         16 + ((time_size + 7) & !7) + self.data.length_in_bytes()
     }
@@ -89,8 +94,8 @@ where
         use byteorder::WriteBytesExt;
         writer.write_u64::<byteorder::LittleEndian>(self.from.try_into().unwrap()).unwrap();
         writer.write_u64::<byteorder::LittleEndian>(self.seq.try_into().unwrap()).unwrap();
-        ::bincode::serialize_into(&mut *writer, &self.time).expect("bincode::serialize_into() failed");
-        let time_size = ::bincode::serialized_size(&self.time).expect("bincode::serialized_size() failed") as usize;
+        ::bincode::serialize_into(&mut *writer, &self.stamp).expect("bincode::serialize_into() failed");
+        let time_size = ::bincode::serialized_size(&self.stamp).expect("bincode::serialized_size() failed") as usize;
         let time_slop = ((time_size + 7) & !7) - time_size;
         writer.write_all(&[0u8; 8][..time_slop]).unwrap();
         self.data.into_bytes(&mut *writer);

timely/src/dataflow/channels/pullers/counter.rs

@@ -4,7 +4,7 @@ use std::rc::Rc;
 use std::cell::RefCell;
 
 use crate::dataflow::channels::Message;
-use crate::progress::ChangeBatch;
+use crate::progress::{ChangeBatch, Stamp};
 use crate::communication::Pull;
 use crate::Accountable;
 
@@ -18,22 +18,25 @@ pub struct Counter<T, C, P> {
 /// A guard type that updates the change batch counts on drop
 pub struct ConsumedGuard<T: Ord + Clone + 'static> {
     consumed: Rc<RefCell<ChangeBatch<T>>>,
-    time: Option<T>,
+    stamp: Option<Stamp<T>>,
     record_count: i64,
 }
 
 impl<T:Ord+Clone+'static> ConsumedGuard<T> {
     #[inline]
-    pub(crate) fn time(&self) -> &T {
-        self.time.as_ref().unwrap()
+    pub(crate) fn stamp(&self) -> &Stamp<T> {
+        self.stamp.as_ref().unwrap()
     }
 }
 
 impl<T:Ord+Clone+'static> Drop for ConsumedGuard<T> {
     fn drop(&mut self) {
         // SAFETY: we're in a Drop impl, so this runs at most once
-        let time = self.time.take().unwrap();
-        self.consumed.borrow_mut().update(time, self.record_count);
+        let stamp = self.stamp.take().unwrap();
+        let mut consumed = self.consumed.borrow_mut();
+        for time in stamp.iter() {
+            consumed.update(time.clone(), self.record_count);
+        }
     }
 }
 
@@ -49,7 +52,7 @@ impl<T:Ord+Clone+'static, C: Accountable, P: Pull<Message<T, C>>> Counter<T, C,
         if let Some(message) = self.pullable.pull() {
             let guard = ConsumedGuard {
                 consumed: Rc::clone(&self.consumed),
-                time: Some(message.time.clone()),
+                stamp: Some(message.stamp.clone()),
                 record_count: message.data.record_count(),
             };
             Some((guard, message))

timely/src/dataflow/channels/pushers/counter.rs

@@ -19,7 +19,10 @@ impl<T: Clone+Ord, C: Accountable, P> Push<Message<T, C>> for Counter<T, P> wher
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, C>>) {
         if let Some(message) = message {
-            self.produced.borrow_mut().update(message.time.clone(), message.data.record_count());
+            let mut produced = self.produced.borrow_mut();
+            for time in message.stamp.iter() {
+                produced.update(time.clone(), message.data.record_count());
+            }
         }
 
         // only propagate `None` if dirty (indicates flush)

timely/src/dataflow/channels/pushers/exchange.rs

@@ -4,6 +4,7 @@ use crate::ContainerBuilder;
 use crate::communication::Push;
 use crate::container::{DrainContainer, PushInto};
 use crate::dataflow::channels::Message;
+use crate::progress::Stamp;
 
 /// Distribute containers to several pushers.
 ///
@@ -15,10 +16,10 @@ use crate::dataflow::channels::Message;
 /// must be preserved across the output containers, from the first call to `partition` until the
 /// call to `flush` for a specific time stamp.
 pub trait Distributor<C> {
-    /// Partition the contents of `container` at `time` into the `pushers`.
-    fn partition<T: Clone, P: Push<Message<T, C>>>(&mut self, container: &mut C, time: &T, pushers: &mut [P]);
-    /// Flush any remaining contents into the `pushers` at time `time`.
-    fn flush<T: Clone, P: Push<Message<T, C>>>(&mut self, time: &T, pushers: &mut [P]);
+    /// Partition the contents of `container` at `stamp` into the `pushers`.
+    fn partition<T: Clone, P: Push<Message<T, C>>>(&mut self, container: &mut C, stamp: &Stamp<T>, pushers: &mut [P]);
+    /// Flush any remaining contents into the `pushers` at stamp `stamp`.
+    fn flush<T: Clone, P: Push<Message<T, C>>>(&mut self, stamp: &Stamp<T>, pushers: &mut [P]);
     /// Optionally release resources, such as memory.
     fn relax(&mut self) { }
 }
@@ -46,15 +47,15 @@ where
     CB: ContainerBuilder<Container: DrainContainer> + for<'a> PushInto<<CB::Container as DrainContainer>::Item<'a>>,
     for<'a> H: FnMut(&<CB::Container as DrainContainer>::Item<'a>) -> u64,
 {
-    fn partition<T: Clone, P: Push<Message<T, CB::Container>>>(&mut self, container: &mut CB::Container, time: &T, pushers: &mut [P]) {
+    fn partition<T: Clone, P: Push<Message<T, CB::Container>>>(&mut self, container: &mut CB::Container, stamp: &Stamp<T>, pushers: &mut [P]) {
         debug_assert_eq!(self.builders.len(), pushers.len());
         if pushers.len().is_power_of_two() {
             let mask = (pushers.len() - 1) as u64;
             for datum in container.drain() {
                 let index = ((self.hash_func)(&datum) & mask) as usize;
                 self.builders[index].push_into(datum);
                 while let Some(produced) = self.builders[index].extract() {
-                    Message::push_at(produced, time.clone(), &mut pushers[index]);
+                    Message::push_at(produced, stamp.clone(), &mut pushers[index]);
                 }
             }
         }
@@ -64,16 +65,16 @@ where
                 let index = ((self.hash_func)(&datum) % num_pushers) as usize;
                 self.builders[index].push_into(datum);
                 while let Some(produced) = self.builders[index].extract() {
-                    Message::push_at(produced, time.clone(), &mut pushers[index]);
+                    Message::push_at(produced, stamp.clone(), &mut pushers[index]);
                 }
             }
         }
     }
 
-    fn flush<T: Clone, P: Push<Message<T, CB::Container>>>(&mut self, time: &T, pushers: &mut [P]) {
+    fn flush<T: Clone, P: Push<Message<T, CB::Container>>>(&mut self, stamp: &Stamp<T>, pushers: &mut [P]) {
         for (builder, pusher) in self.builders.iter_mut().zip(pushers.iter_mut()) {
             while let Some(container) = builder.finish() {
-                Message::push_at(container, time.clone(), pusher);
+                Message::push_at(container, stamp.clone(), pusher);
             }
         }
     }
@@ -89,7 +90,7 @@ where
 /// Distributes records among target pushees according to a distributor.
 pub struct Exchange<T, P, D> {
     pushers: Vec<P>,
-    current: Option<T>,
+    current: Option<Stamp<T>>,
     distributor: D,
 }
 
@@ -117,28 +118,28 @@ where
         }
         else if let Some(message) = message {
 
-            let time = &message.time;
+            let stamp = &message.stamp;
             let data = &mut message.data;
 
-            // if the time isn't right, flush everything.
+            // if the stamp isn't right, flush everything.
             match self.current.as_ref() {
-                // We have a current time, and it is different from the new time.
-                Some(current_time) if current_time != time => {
-                    self.distributor.flush(current_time, &mut self.pushers);
-                    self.current = Some(time.clone());
+                // We have a current stamp, and it is different from the new stamp.
+                Some(current_stamp) if current_stamp != stamp => {
+                    self.distributor.flush(current_stamp, &mut self.pushers);
+                    self.current = Some(stamp.clone());
                 }
-                // We had no time before, or flushed.
-                None => self.current = Some(time.clone()),
-                // Time didn't change since last call.
+                // We had no stamp before, or flushed.
+                None => self.current = Some(stamp.clone()),
+                // Stamp didn't change since last call.
                 _ => {}
             }
 
-            self.distributor.partition(data, time, &mut self.pushers);
+            self.distributor.partition(data, stamp, &mut self.pushers);
         }
         else {
             // flush
-            if let Some(time) = self.current.take() {
-                self.distributor.flush(&time, &mut self.pushers);
+            if let Some(stamp) = self.current.take() {
+                self.distributor.flush(&stamp, &mut self.pushers);
             }
             self.distributor.relax();
             for index in 0..self.pushers.len() {

timely/src/dataflow/channels/pushers/progress.rs

@@ -23,7 +23,7 @@ impl<T: Timestamp, P> Progress<T, P> {
     /// On return, the container may hold undefined contents and should be cleared before it is reused.
     #[inline] pub fn give<C: Container, CT: CapabilityTrait<T>>(&mut self, capability: &CT, container: &mut C) where P: Push<Message<T, C>> {
         debug_assert!(self.valid(capability), "Attempted to open output session with invalid capability");
-        if !container.is_empty() { Message::push_at(container, capability.time().clone(), &mut self.pushee); }
+        if !container.is_empty() { Message::push_at(container, capability.stamp(), &mut self.pushee); }
     }
     /// Activates a `Progress` into a `ProgressSession` which will flush when dropped.
     pub fn activate<'a, C>(&'a mut self) -> ProgressSession<'a, T, C, P> where P: Push<Message<T, C>> {