From 73f35b6b2523e05caa848b62ddf66600e4374966 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 18 Aug 2026 20:51:30 -0400 Subject: [PATCH 1/2] Messages carry a Stamp of timestamps Replaces the single timestamp on each message with a Stamp: a multiset of timestamps affixed to the message, of which the message's contents 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. Messages are accounted in progress tracking once per stamp element, on both the produce and consume sides, so multiplicities are significant; element order is not, and is maintained sorted so that equal stamps are structurally equal. The progress tracker itself is unchanged, as it already consumes multisets of pointstamp updates. This commit is a representation change only: no public interface constructs a stamp with other than exactly one element, every reachable path produces singletons, and singleton stamps follow the same operations as the prior single timestamp. Stamp stores a single element inline, mirroring Antichain's storage, so the common case allocates nothing; the most message-overhead-bound microbenchmark (pingpong, one-element messages, one worker) measures within ~5%, and progress-only benchmarks (barrier) are unchanged. Stamp transformations come in two shapes with distinct obligations, marked by construction: map_pointwise preserves multiplicities and is 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 and must agree element-wise; map_into restores minimality and is reserved for operators that account for their own messages. Breaking changes: Message.time becomes Message.stamp; Message::push_at takes a Stamp; CapabilityTrait::time() becomes stamp() -> Stamp; Distributor implementations receive a &Stamp and must reproduce it on every produced sub-message; Event::Messages carries a Stamp, so captured event streams from prior versions are not readable; the bincode wire layout of Message changes accordingly. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh --- timely/src/dataflow/channels/mod.rs | 29 +-- .../src/dataflow/channels/pullers/counter.rs | 17 +- .../src/dataflow/channels/pushers/counter.rs | 5 +- .../src/dataflow/channels/pushers/exchange.rs | 45 +++-- .../src/dataflow/channels/pushers/progress.rs | 2 +- timely/src/dataflow/operators/capability.rs | 50 +++-- .../operators/core/capture/capture.rs | 4 +- .../dataflow/operators/core/capture/event.rs | 6 +- .../operators/core/capture/extract.rs | 5 +- .../src/dataflow/operators/core/enterleave.rs | 6 +- timely/src/dataflow/operators/core/input.rs | 4 +- timely/src/dataflow/operators/core/probe.rs | 2 +- .../src/dataflow/operators/generic/handles.rs | 4 +- timely/src/logging.rs | 2 +- timely/src/progress/mod.rs | 2 + timely/src/progress/stamp.rs | 191 ++++++++++++++++++ 16 files changed, 301 insertions(+), 73 deletions(-) create mode 100644 timely/src/progress/stamp.rs 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..ac3bdb2b9 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,14 @@ 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) } + } impl Deref for InputCapability { @@ -332,7 +350,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) } 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/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() + } +} From c46d67a644abfef4697ea585b4fc69bb812271a9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 18 Aug 2026 20:51:30 -0400 Subject: [PATCH 2/2] Multi- and zero-capability messages via CapabilitySet sessions Allows an output session to be opened with a CapabilitySet, stamping its messages with the set's timestamps: a batch of updates can ship whole, stamped with the antichain that justifies its contents, rather than shredded into one message per capability. An empty set stamps messages with no timestamps at all: such messages make no progress claims, are invisible to progress tracking, and sit outside the frontier-ordered delivery guarantee, including best-effort delivery near dataflow teardown. Non-singleton stamps arise only downstream of an explicit CapabilitySet session; no existing program contains one, and unaware code is unaffected. On the receive side, InputCapability::stamp() exposes the stamp and retain_stamp() mints a capability for each element; interfaces that insist on a single timestamp per message (InputCapability::time(), retain(), and capture's Extract) panic with a message directing callers to the stamp interfaces, and can only be provoked by an opt-in upstream in the same dataflow. Feedback advances each stamp element through its summary, discarding elements that cannot traverse and restoring minimality, which is sound there because operators account for their own consumed and produced messages. Tests cover a multi-capability message traversing channels, retain_stamp, and delayed minting; zero-capability delivery after the sender drops all capabilities; multi-stamp messages crossing a data exchange; and a regression test that a stamp leaving a scope remains accounted at one outer pointstamp per element when projection makes elements comparable. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh --- timely/src/dataflow/operators/capability.rs | 51 ++++ .../src/dataflow/operators/core/feedback.rs | 15 +- timely/tests/stamps.rs | 245 ++++++++++++++++++ 3 files changed, 308 insertions(+), 3 deletions(-) create mode 100644 timely/tests/stamps.rs diff --git a/timely/src/dataflow/operators/capability.rs b/timely/src/dataflow/operators/capability.rs index ac3bdb2b9..32e589093 100644 --- a/timely/src/dataflow/operators/capability.rs +++ b/timely/src/dataflow/operators/capability.rs @@ -322,6 +322,18 @@ impl InputCapability { 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 { @@ -535,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/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/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]); +}