diff --git a/timely/src/dataflow/channels/mod.rs b/timely/src/dataflow/channels/mod.rs index ce109363c..d8d618117 100644 --- a/timely/src/dataflow/channels/mod.rs +++ b/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 { - /// 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, /// The data in the message. pub data: C, /// The source worker. @@ -35,17 +40,17 @@ impl Message { /// 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, 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>>(buffer: &mut C, time: T, pusher: &mut P) where C: Default { + pub fn push_at>>(buffer: &mut C, stamp: Stamp, 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::().unwrap().try_into().unwrap(); let seq: usize = slice.read_u64::().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 = ::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::(self.from.try_into().unwrap()).unwrap(); writer.write_u64::(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); diff --git a/timely/src/dataflow/channels/pullers/counter.rs b/timely/src/dataflow/channels/pullers/counter.rs index ab5e30e1a..8ce055fc4 100644 --- a/timely/src/dataflow/channels/pullers/counter.rs +++ b/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 { /// A guard type that updates the change batch counts on drop pub struct ConsumedGuard { consumed: Rc>>, - time: Option, + stamp: Option>, record_count: i64, } impl ConsumedGuard { #[inline] - pub(crate) fn time(&self) -> &T { - self.time.as_ref().unwrap() + pub(crate) fn stamp(&self) -> &Stamp { + self.stamp.as_ref().unwrap() } } impl Drop for ConsumedGuard { 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>> Counter Push> for Counter wher #[inline] fn push(&mut self, message: &mut Option>) { 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) diff --git a/timely/src/dataflow/channels/pushers/exchange.rs b/timely/src/dataflow/channels/pushers/exchange.rs index e57bcf8fe..ea8a3f17b 100644 --- a/timely/src/dataflow/channels/pushers/exchange.rs +++ b/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 { - /// Partition the contents of `container` at `time` into the `pushers`. - fn partition>>(&mut self, container: &mut C, time: &T, pushers: &mut [P]); - /// Flush any remaining contents into the `pushers` at time `time`. - fn flush>>(&mut self, time: &T, pushers: &mut [P]); + /// Partition the contents of `container` at `stamp` into the `pushers`. + fn partition>>(&mut self, container: &mut C, stamp: &Stamp, pushers: &mut [P]); + /// Flush any remaining contents into the `pushers` at stamp `stamp`. + fn flush>>(&mut self, stamp: &Stamp, pushers: &mut [P]); /// Optionally release resources, such as memory. fn relax(&mut self) { } } @@ -46,7 +47,7 @@ where CB: ContainerBuilder + for<'a> PushInto<::Item<'a>>, for<'a> H: FnMut(&::Item<'a>) -> u64, { - fn partition>>(&mut self, container: &mut CB::Container, time: &T, pushers: &mut [P]) { + fn partition>>(&mut self, container: &mut CB::Container, stamp: &Stamp, 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; @@ -54,7 +55,7 @@ where 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>>(&mut self, time: &T, pushers: &mut [P]) { + fn flush>>(&mut self, stamp: &Stamp, 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 { pushers: Vec

, - current: Option, + current: Option>, 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() { diff --git a/timely/src/dataflow/channels/pushers/progress.rs b/timely/src/dataflow/channels/pushers/progress.rs index c4ff4d9cd..89251106c 100644 --- a/timely/src/dataflow/channels/pushers/progress.rs +++ b/timely/src/dataflow/channels/pushers/progress.rs @@ -23,7 +23,7 @@ impl Progress { /// On return, the container may hold undefined contents and should be cleared before it is reused. #[inline] pub fn give>(&mut self, capability: &CT, container: &mut C) where P: Push> { 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> { diff --git a/timely/src/dataflow/operators/capability.rs b/timely/src/dataflow/operators/capability.rs index e2cf53043..32e589093 100644 --- a/timely/src/dataflow/operators/capability.rs +++ b/timely/src/dataflow/operators/capability.rs @@ -28,27 +28,32 @@ use std::fmt::{self, Debug}; use crate::order::PartialOrder; use crate::progress::Timestamp; -use crate::progress::ChangeBatch; +use crate::progress::{ChangeBatch, Stamp}; use crate::progress::operate::PortConnectivity; use crate::scheduling::Activations; use crate::dataflow::channels::pullers::counter::ConsumedGuard; -/// An internal trait expressing the capability to send messages with a given timestamp. +/// An internal trait expressing the capability to send messages with given timestamps. pub trait CapabilityTrait { - /// The timestamp associated with the capability. - fn time(&self) -> &T; + /// The stamp of timestamps to attach to messages sent with this capability. + /// + /// Messages may only result in downstream work at times greater or equal to + /// some element of the stamp. An empty stamp makes no progress claims, and + /// such messages may be delivered after downstream frontiers have advanced + /// past all times in their contents. + fn stamp(&self) -> Stamp; /// Validates that the capability is valid for a specific internal buffer and output port. fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool; } impl> CapabilityTrait for &C { - fn time(&self) -> &T { (**self).time() } + fn stamp(&self) -> Stamp { (**self).stamp() } fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool { (**self).valid_for_output(query_buffer, port) } } impl> CapabilityTrait for &mut C { - fn time(&self) -> &T { (**self).time() } + fn stamp(&self) -> Stamp { (**self).stamp() } fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool { (**self).valid_for_output(query_buffer, port) } @@ -66,7 +71,7 @@ pub struct Capability { } impl CapabilityTrait for Capability { - fn time(&self) -> &T { &self.time } + fn stamp(&self) -> Stamp { Stamp::from_elem(self.time.clone()) } fn valid_for_output(&self, query_buffer: &Rc>>, _port: usize) -> bool { Rc::ptr_eq(&self.internal, query_buffer) } @@ -246,7 +251,7 @@ pub struct InputCapability { } impl CapabilityTrait for InputCapability { - fn time(&self) -> &T { self.time() } + fn stamp(&self) -> Stamp { self.stamp().clone() } fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool { let summaries_borrow = self.summaries.get().expect("connectivity frozen at operator build"); let internal_borrow = self.internal.borrow(); @@ -267,25 +272,35 @@ impl InputCapability { } } + /// The stamp of timestamps associated with the received message. + #[inline] + pub fn stamp(&self) -> &Stamp { + self.consumed_guard.stamp() + } + /// The timestamp associated with this capability. + /// + /// This method panics if the message's stamp is not a singleton, as is the case + /// when an upstream operator sends messages stamped by multiple timestamps, or by + /// none at all. Such messages must be accessed through [`InputCapability::stamp`]. #[inline] pub fn time(&self) -> &T { - self.consumed_guard.time() + self.stamp().expect_singleton() } /// Delays capability for a specific output port. /// - /// Makes a new capability for a timestamp `new_time` greater or equal to the timestamp of - /// the source capability (`self`). + /// Makes a new capability for a timestamp `new_time` greater or equal to some element + /// of the stamp of the source capability (`self`). /// - /// This method panics if `self.time` is not less or equal to `new_time`. + /// This method panics if no element of `self.stamp()` is less or equal to `new_time`. pub fn delayed(&self, new_time: &T, output_port: usize) -> Capability { use crate::progress::timestamp::PathSummary; if let Some(path) = self.summaries.get().expect("connectivity frozen at operator build").get(output_port) { - if path.iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) { + if self.stamp().iter().flat_map(|elem| path.iter().flat_map(move |summary| summary.results_in(elem))).any(|time| time.less_equal(new_time)) { Capability::new(new_time.clone(), Rc::clone(&self.internal.borrow()[output_port])) } 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, path, self.time()); + panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to any element of the capability's stamp ({:?})", new_time, path, self.stamp()); } } else { @@ -299,11 +314,26 @@ impl InputCapability { /// capability. Users should take care that these capabilities are only stored for /// as long as they are required, as failing to drop them may result in livelock. /// - /// This method panics if the timestamp summary to `output_port` strictly advances the time. + /// This method panics if the message's stamp is not a singleton, or if the timestamp + /// summary to `output_port` strictly advances the time. Stamps with zero or multiple + /// elements must be retained with [`InputCapability::retain_stamp`]. #[inline] pub fn retain(&self, output_port: usize) -> Capability { self.delayed(self.time(), output_port) } + + /// Transforms to an owned capability set for a specific output port, with one + /// capability for each element of the message's stamp. + /// + /// An empty stamp produces an empty capability set. Data sent through a session + /// keyed by an empty capability set makes no progress claims, and may be delivered + /// after downstream frontiers have advanced past all times in its contents. + /// + /// This method panics if the timestamp summary to `output_port` strictly advances + /// any element of the stamp. + pub fn retain_stamp(&self, output_port: usize) -> CapabilitySet { + self.stamp().iter().map(|time| self.delayed(time, output_port)).collect() + } } impl Deref for InputCapability { @@ -332,7 +362,7 @@ pub struct ActivateCapability { } impl CapabilityTrait for ActivateCapability { - fn time(&self) -> &T { self.capability.time() } + fn stamp(&self) -> Stamp { self.capability.stamp() } fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool { self.capability.valid_for_output(query_buffer, port) } @@ -517,6 +547,45 @@ impl CapabilitySet { } } +impl CapabilityTrait for CapabilitySet { + /// The stamp of the set's timestamps. + /// + /// An empty set stamps messages with an empty stamp: such messages make no + /// progress claims, are invisible to progress tracking, and may be delivered + /// after downstream frontiers have advanced past all times in their contents. + fn stamp(&self) -> Stamp { + self.elements.iter().map(|capability| capability.time().clone()).collect() + } + /// Valid iff every member capability is valid for the output. + /// + /// An empty set is vacuously valid for any output: constructing an empty set + /// and opening a session with it is the deliberate mechanism for sending + /// messages that make no progress claims. + fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool { + self.elements.iter().all(|capability| capability.valid_for_output(query_buffer, port)) + } +} + +impl CapabilitySet { + /// Creates a new capability set to send data at each element of `stamp`. + /// + /// This method panics if any element of `stamp` is without a capability in + /// `self.elements` less or equal to it. + pub fn delayed_stamp(&self, stamp: &Stamp) -> CapabilitySet { + stamp.iter().map(|time| self.delayed(time)).collect() + } +} + +impl FromIterator> for CapabilitySet { + fn from_iter>>(iter: I) -> Self { + let mut result = Self::new(); + for capability in iter { + result.insert(capability); + } + result + } +} + impl From>> for CapabilitySet where T: Timestamp, diff --git a/timely/src/dataflow/operators/core/capture/capture.rs b/timely/src/dataflow/operators/core/capture/capture.rs index 1a484581c..e793d8030 100644 --- a/timely/src/dataflow/operators/core/capture/capture.rs +++ b/timely/src/dataflow/operators/core/capture/capture.rs @@ -138,10 +138,10 @@ impl Capture for Stream<'_, T, C> { // turn each received message into an event. while let Some(message) = input.next() { - let time = &message.time; + let stamp = &message.stamp; let data = &mut message.data; let vector = std::mem::take(data); - event_pusher.push(Event::Messages(time.clone(), vector)); + event_pusher.push(Event::Messages(stamp.clone(), vector)); } input.consumed().borrow_mut().drain_into(&mut progress.consumeds[0]); false diff --git a/timely/src/dataflow/operators/core/capture/event.rs b/timely/src/dataflow/operators/core/capture/event.rs index d43a1e51c..920af4405 100644 --- a/timely/src/dataflow/operators/core/capture/event.rs +++ b/timely/src/dataflow/operators/core/capture/event.rs @@ -7,13 +7,15 @@ use columnar::Columnar; use serde::{Deserialize, Serialize}; +use crate::progress::Stamp; + /// Data and progress events of the captured stream. #[derive(Debug, Clone, Hash, Ord, PartialOrd, Eq, PartialEq, Deserialize, Serialize, Columnar)] pub enum Event { /// Progress received via `push_external_progress`. Progress(Vec<(T, i64)>), - /// Messages received via the data stream. - Messages(T, C), + /// Messages received via the data stream, stamped by an antichain of timestamps. + Messages(Stamp, C), } /// Iterates over contained `Event`. diff --git a/timely/src/dataflow/operators/core/capture/extract.rs b/timely/src/dataflow/operators/core/capture/extract.rs index e6ed7b9cc..01321df20 100644 --- a/timely/src/dataflow/operators/core/capture/extract.rs +++ b/timely/src/dataflow/operators/core/capture/extract.rs @@ -57,8 +57,9 @@ where fn extract(self) -> Vec<(T, C)> { let mut staged = std::collections::BTreeMap::new(); for event in self { - if let Event::Messages(time, data) = event { - staged.entry(time) + if let Event::Messages(stamp, data) = event { + // This testing convenience insists on singleton stamps. + staged.entry({ let mut elements = stamp.into_elements(); assert!(elements.len() == 1, "Extract insists on singleton stamps"); elements.pop().unwrap() }) .or_insert_with(Vec::new) .push(data); } diff --git a/timely/src/dataflow/operators/core/enterleave.rs b/timely/src/dataflow/operators/core/enterleave.rs index cbcd5a7ad..35f9ffac3 100644 --- a/timely/src/dataflow/operators/core/enterleave.rs +++ b/timely/src/dataflow/operators/core/enterleave.rs @@ -168,7 +168,8 @@ impl, TContainer: Container fn push(&mut self, element: &mut Option>) { if let Some(outer_message) = element { let data = ::std::mem::take(&mut outer_message.data); - let mut inner_message = Some(Message::new(TInner::to_inner(outer_message.time.clone()), data)); + let stamp = outer_message.stamp.map_pointwise(|time| TInner::to_inner(time.clone())); + let mut inner_message = Some(Message::new(stamp, data)); self.targets.push(&mut inner_message); if let Some(inner_message) = inner_message { outer_message.data = inner_message.data; @@ -196,7 +197,8 @@ where TOuter: Timestamp, TInner: Timestamp+Refines, { fn push(&mut self, message: &mut Option>) { if let Some(inner_message) = message { let data = ::std::mem::take(&mut inner_message.data); - let mut outer_message = Some(Message::new(inner_message.time.clone().to_outer(), data)); + let stamp = inner_message.stamp.map_pointwise(|time| time.clone().to_outer()); + let mut outer_message = Some(Message::new(stamp, data)); self.targets.push(&mut outer_message); if let Some(outer_message) = outer_message { inner_message.data = outer_message.data; diff --git a/timely/src/dataflow/operators/core/feedback.rs b/timely/src/dataflow/operators/core/feedback.rs index 4c0686713..14e1643d6 100644 --- a/timely/src/dataflow/operators/core/feedback.rs +++ b/timely/src/dataflow/operators/core/feedback.rs @@ -122,9 +122,18 @@ impl<'scope, T: Timestamp, C: Container> ConnectLoop<'scope, T, C> for Stream<'s builder.build(move |_capability| move |_frontier| { let mut output = output.activate(); input.for_each(|cap, data| { - if let Some(new_time) = summary.results_in(cap.time()) { - let new_cap = cap.delayed(&new_time, output.output_index()); - output.give(&new_cap, data); + // Advance each stamp element by the summary, discarding elements that + // cannot traverse the feedback edge and restoring minimality. Contents + // at times supported only by discarded elements cannot be sent + // downstream, just as a message with a singleton stamp is discarded + // entirely when its element cannot traverse. + let new_caps = cap.stamp() + .map_into(|time| summary.results_in(time)) + .iter() + .map(|time| cap.delayed(time, output.output_index())) + .collect::>(); + if !new_caps.is_empty() || cap.stamp().is_empty() { + output.give(&new_caps, data); } }); }); diff --git a/timely/src/dataflow/operators/core/input.rs b/timely/src/dataflow/operators/core/input.rs index 94ed474d2..20a1413aa 100644 --- a/timely/src/dataflow/operators/core/input.rs +++ b/timely/src/dataflow/operators/core/input.rs @@ -390,10 +390,10 @@ impl> Handle { for index in 0 .. pushers.len() { if index < pushers.len() - 1 { buffer.clone_from(container); - Message::push_at(buffer, now_at.clone(), &mut pushers[index]); + Message::push_at(buffer, crate::progress::Stamp::from_elem(now_at.clone()), &mut pushers[index]); } else { - Message::push_at(container, now_at.clone(), &mut pushers[index]); + Message::push_at(container, crate::progress::Stamp::from_elem(now_at.clone()), &mut pushers[index]); } } } diff --git a/timely/src/dataflow/operators/core/probe.rs b/timely/src/dataflow/operators/core/probe.rs index 91b73e629..a824e7042 100644 --- a/timely/src/dataflow/operators/core/probe.rs +++ b/timely/src/dataflow/operators/core/probe.rs @@ -125,7 +125,7 @@ impl Probe for Stream<'_, T, C> { } while let Some(message) = input.next() { - Message::push_at(&mut message.data, message.time.clone(), &mut output); + Message::push_at(&mut message.data, message.stamp.clone(), &mut output); } use timely_communication::Push; output.done(); diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs index 9205e627e..715f2d2c8 100644 --- a/timely/src/dataflow/operators/generic/handles.rs +++ b/timely/src/dataflow/operators/generic/handles.rs @@ -58,11 +58,11 @@ impl>> InputHandleCore BatchLogger where P: EventPusher, C: Container { /// Publishes a batch of logged events and advances the capability. pub fn publish_batch(&mut self, &time: &Duration, data: &mut Option) { if let Some(data) = data { - self.event_pusher.push(Event::Messages(self.time, std::mem::take(data))); + self.event_pusher.push(Event::Messages(crate::progress::Stamp::from_elem(self.time), std::mem::take(data))); } if self.time < time { let new_frontier = time; diff --git a/timely/src/progress/mod.rs b/timely/src/progress/mod.rs index 1ff95a977..4fb2cc1d9 100644 --- a/timely/src/progress/mod.rs +++ b/timely/src/progress/mod.rs @@ -7,8 +7,10 @@ pub use self::subgraph::{Subgraph, SubgraphBuilder}; pub use self::timestamp::{Timestamp, PathSummary}; pub use self::change_batch::ChangeBatch; pub use self::frontier::Antichain; +pub use self::stamp::Stamp; pub mod change_batch; +pub mod stamp; pub mod frontier; pub mod timestamp; pub mod operate; diff --git a/timely/src/progress/stamp.rs b/timely/src/progress/stamp.rs new file mode 100644 index 000000000..e3deb37f8 --- /dev/null +++ b/timely/src/progress/stamp.rs @@ -0,0 +1,191 @@ +//! The stamp of a message: the multiset of timestamps affixed to it. +//! +//! Each message in timely dataflow carries a stamp: a multiset of timestamps, +//! such that the message may only result in downstream work at times greater +//! or equal to some element. Like postage, the stamp records the capabilities +//! under which the message travels. Multiplicities are significant, as the +//! message is accounted once per element in progress tracking; the order of +//! elements is not, and is maintained sorted so that equal stamps are +//! structurally equal. The common case is a singleton stamp, +//! corresponding to the classic "one capability per message" design, and is +//! represented inline without allocation. Stamps may contain multiple elements +//! (e.g. a batch of updates stamped with its lower antichain, as in differential +//! dataflow) or no elements at all (data that makes no progress claims, and which +//! may be delivered after the frontier has passed all times in its payload). +//! +//! `Stamp` maintains its elements sorted by `Ord`, so that equal stamps are +//! structurally equal and stamps can be used as grouping and sorting keys. +//! Stamps built by insertion are minimal (an antichain), but stamps that have +//! crossed a scope boundary may contain comparable or duplicate elements: the +//! stamp is the multiset of pointstamps at which the message is accounted, and +//! boundary maps must preserve counts element-wise (see [`Stamp::map_pointwise`]). + +use serde::{Deserialize, Serialize}; +use smallvec::SmallVec; + +use crate::order::PartialOrder; + +/// A multiset of timestamps affixed to a message, stored sorted. +/// +/// The elements are stored in increasing order under `Ord` so that `Eq`, `Ord`, +/// and `Hash` are structural. Stamps built by insertion are minimal antichains; +/// stamps mapped across scope boundaries may contain comparable or duplicate +/// elements, whose multiplicities are significant for progress accounting. +/// +/// As with [`Antichain`](crate::progress::Antichain), the storage holds a single +/// element inline, so the overwhelmingly common zero- and one-element stamps +/// require no allocation. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, columnar::Columnar)] +pub struct Stamp { + elements: SmallVec<[T; 1]>, +} + +impl Stamp { + /// An empty stamp, making no progress claims. + pub fn new() -> Self { + Stamp { elements: SmallVec::new() } + } + + /// A stamp containing a single element. + pub fn from_elem(element: T) -> Self { + Stamp { elements: SmallVec::from_buf([element]) } + } + + /// The elements of the stamp, in increasing `Ord` order. + #[inline] + pub fn elements(&self) -> &[T] { + &self.elements[..] + } + + /// An iterator over the elements of the stamp. + #[inline] + pub fn iter(&self) -> std::slice::Iter<'_, T> { + self.elements.iter() + } + + /// True iff the stamp contains no elements. + #[inline] + pub fn is_empty(&self) -> bool { + self.elements.is_empty() + } + + /// The number of elements in the stamp. + #[inline] + pub fn len(&self) -> usize { + self.elements.len() + } + + /// The elements of the stamp, by value, in increasing `Ord` order. + #[inline] + pub fn into_elements(self) -> Vec { + self.elements.into_vec() + } + + /// The sole element of the stamp, if the stamp is a singleton. + #[inline] + pub fn as_singleton(&self) -> Option<&T> { + if self.elements.len() == 1 { Some(&self.elements[0]) } else { None } + } + + /// The sole element of the stamp; panics if the stamp is not a singleton. + /// + /// This supports pre-stamp interfaces that insist on a single timestamp per + /// message; such interfaces cannot be used with multi- or zero-element stamps. + #[inline] + pub fn expect_singleton(&self) -> &T where T: std::fmt::Debug { + self.as_singleton().unwrap_or_else(|| { + panic!("expected a singleton stamp; found {:?} elements: {:?}", self.elements.len(), self.elements()) + }) + } +} + +impl Stamp { + /// Inserts `element` unless it is redundant, removing elements it dominates. + /// + /// Returns true iff the element was inserted. + pub fn insert(&mut self, element: T) -> bool { + if self.elements.iter().any(|x| x.less_equal(&element)) { + false + } else { + self.elements.retain(|x| !element.less_equal(x)); + let position = self.elements.partition_point(|x| x < &element); + self.elements.insert(position, element); + true + } + } + + /// True iff some element of the stamp is less or equal to `time`. + #[inline] + pub fn less_equal(&self, time: &T) -> bool { + self.elements.iter().any(|x| x.less_equal(time)) + } + + /// Maps each element through `logic`, discarding `None` results and + /// restoring minimality and canonical order. + /// + /// This is the shape required when timestamps traverse path summaries + /// (feedback edges), where elements may fail to traverse, or may become + /// comparable after mapping. It may only be used by operators that account + /// for their own consumed and produced messages, as it changes the number + /// of pointstamps a message is accounted at. + pub fn map_into(&self, mut logic: impl FnMut(&T) -> Option) -> Stamp { + let mut result = Stamp::new(); + for element in self.elements.iter() { + if let Some(mapped) = logic(element) { + result.insert(mapped); + } + } + result + } + + /// Maps each element through `logic`, preserving the number of elements and + /// restoring sorted order, but *not* minimality. + /// + /// This is the shape required at scope boundaries (enter and leave), whose + /// produced and consumed accounting is inferred independently at either end + /// of the channel from the stamp itself: the counts must agree element-wise, + /// so the map must not collapse elements that become comparable or equal + /// after mapping (as when leaving a scope projects away a timestamp + /// coordinate). The resulting stamp may contain comparable or duplicate + /// elements; this weakens no guarantee, as the stamp promises only that + /// message contents are greater or equal to *some* element. + pub fn map_pointwise(&self, logic: impl FnMut(&T) -> T2) -> Stamp { + let mut elements: SmallVec<[T2; 1]> = self.elements.iter().map(logic).collect(); + elements.sort(); + Stamp { elements } + } +} + +impl Default for Stamp { + fn default() -> Self { Self::new() } +} + +impl FromIterator for Stamp { + fn from_iter>(iter: I) -> Self { + let mut result = Stamp::new(); + for element in iter { + result.insert(element); + } + result + } +} + +impl<'a, T> IntoIterator for &'a Stamp { + type Item = &'a T; + type IntoIter = std::slice::Iter<'a, T>; + fn into_iter(self) -> Self::IntoIter { self.iter() } +} + +impl From> for Stamp { + fn from(antichain: crate::progress::Antichain) -> Self { + let mut elements: SmallVec<[T; 1]> = antichain.into(); + elements.sort(); + Stamp { elements } + } +} + +impl From> for crate::progress::Antichain { + fn from(stamp: Stamp) -> Self { + stamp.into_elements().into() + } +} diff --git a/timely/tests/stamps.rs b/timely/tests/stamps.rs new file mode 100644 index 000000000..750f0cdfe --- /dev/null +++ b/timely/tests/stamps.rs @@ -0,0 +1,245 @@ +//! Tests for messages stamped by multiple, or zero, capabilities. + +use std::cell::RefCell; +use std::rc::Rc; + +use timely::dataflow::channels::pact::{Exchange, Pipeline}; +use timely::dataflow::operators::CapabilitySet; +use timely::dataflow::operators::generic::builder_rc::OperatorBuilder; +use timely::order::Product; +use timely::progress::Stamp; + +/// A message stamped by two incomparable timestamps traverses a channel whole, +/// is observed with both stamp elements, can be re-sent via `retain_stamp`, +/// and the computation drains (the progress books balance). +#[test] +fn multi_capability_stamps() { + let seen = timely::execute_directly(move |worker| { + let seen = Rc::new(RefCell::new(Vec::new())); + let seen2 = Rc::clone(&seen); + worker.dataflow::(move |scope| { + scope.iterative::(move |inner| { + // A source holding two incomparable capabilities, sending one + // message stamped by both. + let mut builder = OperatorBuilder::new("source".to_owned(), inner.clone()); + let (mut output, stream) = builder.new_output::>(); + builder.build(move |mut init_caps| { + let cap = init_caps.pop().unwrap(); + let mut caps = Some(CapabilitySet::from(vec![ + cap.delayed(&Product::new(0, 1)), + cap.delayed(&Product::new(1, 0)), + ])); + move |_frontiers| { + if let Some(caps) = caps.take() { + let mut data = vec![1u64, 2, 3]; + output.activate().give(&caps, &mut data); + } + } + }); + + // A relay that receives the stamped message and forwards it under + // the capability set minted from the message's stamp. + let mut builder = OperatorBuilder::new("relay".to_owned(), inner.clone()); + let (mut output, forwarded) = builder.new_output::>(); + let mut input = builder.new_input(stream, Pipeline); + builder.build(move |_init_caps| { + move |_frontiers| { + let mut output = output.activate(); + input.for_each(|cap, data| { + let caps = cap.retain_stamp(0); + // A stamp element must justify times beyond it. + let _upper = cap.delayed(&Product::new(1, 1), 0); + output.give(&caps, data); + }); + } + }); + + // A sink recording the stamps and data it observes. + let mut builder = OperatorBuilder::new("sink".to_owned(), inner.clone()); + let mut input = builder.new_input(forwarded, Pipeline); + builder.build(move |_init_caps| { + move |_frontiers| { + input.for_each(|cap, data| { + seen2.borrow_mut().push((cap.stamp().clone(), std::mem::take(data))); + }); + } + }); + }); + }); + while worker.step() { } + Rc::try_unwrap(seen).unwrap().into_inner() + }); + + let expected_stamp: Stamp> = + vec![Product::new(0, 1), Product::new(1, 0)].into_iter().collect(); + assert_eq!(seen, vec![(expected_stamp, vec![1, 2, 3])]); +} + +/// A message sent under an empty capability set is delivered with an empty +/// stamp, despite the sender holding no capabilities at all, and despite the +/// message making no progress claims. +#[test] +fn zero_capability_stamps() { + let seen = timely::execute_directly(move |worker| { + let seen = Rc::new(RefCell::new(Vec::new())); + let seen2 = Rc::clone(&seen); + worker.dataflow::(move |scope| { + let mut builder = OperatorBuilder::new("source".to_owned(), scope.clone()); + let (mut output, stream) = builder.new_output::>(); + builder.build(move |init_caps| { + let mut once = Some(init_caps); + move |_frontiers| { + if let Some(init_caps) = once.take() { + // Drop all capabilities, then send anyway. + drop(init_caps); + let empty = CapabilitySet::::new(); + let mut data = vec![4u64, 5, 6]; + output.activate().give(&empty, &mut data); + } + } + }); + + let mut builder = OperatorBuilder::new("sink".to_owned(), scope.clone()); + let mut input = builder.new_input(stream, Pipeline); + builder.build(move |_init_caps| { + move |_frontiers| { + input.for_each(|cap, data| { + seen2.borrow_mut().push((cap.stamp().clone(), std::mem::take(data))); + }); + } + }); + }); + while worker.step() { } + Rc::try_unwrap(seen).unwrap().into_inner() + }); + + assert_eq!(seen, vec![(Stamp::new(), vec![4, 5, 6])]); +} + +/// A multi-stamp message crossing a data exchange arrives at each worker with +/// the stamp intact, partitioned by the exchange function. +#[test] +fn multi_stamps_exchange() { + let guards = timely::execute(timely::Config::process(2), move |worker| { + let index = worker.index(); + let seen = Rc::new(RefCell::new(Vec::new())); + let seen2 = Rc::clone(&seen); + worker.dataflow::(move |scope| { + scope.iterative::(move |inner| { + let mut builder = OperatorBuilder::new("source".to_owned(), inner.clone()); + let (mut output, stream) = builder.new_output::>(); + builder.build(move |mut init_caps| { + let cap = init_caps.pop().unwrap(); + let mut caps = (index == 0).then(|| CapabilitySet::from(vec![ + cap.delayed(&Product::new(0, 1)), + cap.delayed(&Product::new(1, 0)), + ])); + move |_frontiers| { + if let Some(caps) = caps.take() { + let mut data = (0..10u64).collect::>(); + output.activate().give(&caps, &mut data); + } + } + }); + + let mut builder = OperatorBuilder::new("sink".to_owned(), inner.clone()); + let mut input = builder.new_input(stream, Exchange::new(|x: &u64| *x)); + builder.build(move |_init_caps| { + move |_frontiers| { + input.for_each(|cap, data| { + seen2.borrow_mut().push((cap.stamp().clone(), std::mem::take(data))); + }); + } + }); + }); + }); + while worker.step() { } + Rc::try_unwrap(seen).unwrap().into_inner() + }).unwrap(); + + let expected_stamp: Stamp> = + vec![Product::new(0, 1), Product::new(1, 0)].into_iter().collect(); + let results = guards.join().into_iter().map(|r| r.unwrap()).collect::>(); + assert_eq!(results.len(), 2); + for (worker, seen) in results.iter().enumerate() { + let mut received = Vec::new(); + for (stamp, data) in seen.iter() { + assert_eq!(stamp, &expected_stamp); + received.extend(data.iter().copied()); + } + received.sort(); + let expected = (0..10u64).filter(|x| (*x as usize) % 2 == worker).collect::>(); + assert_eq!(received, expected); + } +} + +/// Stamps restore minimality and canonical order under insertion and mapping. +#[test] +fn frame_canonical_form() { + let mut stamp = Stamp::new(); + assert!(stamp.insert(Product::new(1u64, 0u64))); + assert!(stamp.insert(Product::new(0, 1))); + // Dominated by (0, 1). + assert!(!stamp.insert(Product::new(1, 1))); + assert_eq!(stamp.elements(), &[Product::new(0, 1), Product::new(1, 0)]); + + // Projecting away the inner coordinate collapses the antichain. + let outer = stamp.map_into(|time| Some(time.outer)); + assert_eq!(outer.elements(), &[0u64]); + + // Mapping may discard elements entirely. + let filtered = stamp.map_into(|time| if time.outer == 0 { None } else { Some(time.clone()) }); + assert_eq!(filtered.elements(), &[Product::new(1, 0)]); +} + +/// A multi-stamp message leaving a scope must remain accounted at one outer +/// pointstamp per inner stamp element, even when projecting away the inner +/// timestamp coordinate makes elements comparable. Collapsing the stamp here +/// would strand produced counts at the parent and wedge the computation. +#[test] +fn multi_stamp_leave_collapse() { + use timely::dataflow::operators::Leave; + let seen = timely::execute_directly(move |worker| { + let seen = Rc::new(RefCell::new(Vec::new())); + let seen2 = Rc::clone(&seen); + worker.dataflow::(move |scope| { + let outer = scope.clone(); + let stream = scope.iterative::(move |inner| { + let mut builder = OperatorBuilder::new("source".to_owned(), inner.clone()); + let (mut output, stream) = builder.new_output::>(); + builder.build(move |mut init_caps| { + let cap = init_caps.pop().unwrap(); + let mut caps = Some(CapabilitySet::from(vec![ + cap.delayed(&Product::new(0, 1)), + cap.delayed(&Product::new(1, 0)), + ])); + move |_frontiers| { + if let Some(caps) = caps.take() { + let mut data = vec![7u64]; + output.activate().give(&caps, &mut data); + } + } + }); + stream.leave(outer) + }); + + let mut builder = OperatorBuilder::new("sink".to_owned(), stream.scope()); + let mut input = builder.new_input(stream, Pipeline); + builder.build(move |_init_caps| { + move |_frontiers| { + input.for_each(|cap, data| { + seen2.borrow_mut().push((cap.stamp().clone(), std::mem::take(data))); + }); + } + }); + }); + while worker.step() { } + Rc::try_unwrap(seen).unwrap().into_inner() + }); + + // The outer stamp retains both elements, unminimized: the message remains + // accounted at both outer times 0 and 1. + assert_eq!(seen.len(), 1); + assert_eq!(seen[0].0.elements(), &[0u64, 1]); + assert_eq!(seen[0].1, vec![7u64]); +}