From c62474077e6f355dbd97b59e2b25d249e491a366 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 07:31:12 -0400 Subject: [PATCH 01/14] Replace BTreeMap in PortConnectivity with a sorted vec; semi-naive summarize_outputs PortConnectivity now stores a sorted Vec<(usize, Antichain)> with lazy consolidation (appends set a dirty bit; reads expect consolidated data, with freeze points at Builder::add_node and PerOperatorState::new). The summarize_outputs fixed point is restructured into semi-naive rounds over log-structured sorted runs, removing its HashMap accumulator; the changed-bit insert methods it required are removed as dead. This excises all BTreeMap (and the one HashMap) instantiations from the progress subsystem: generated IR for the bfs example drops 9.1% (254,560 -> 231,310 llvm-lines; btree machinery 8,723 -> 0). Construction remains O(n log n) under any insertion order; shape_scaling timings at 10k/100k operators are unchanged or slightly improved. Co-Authored-By: Claude Fable 5 --- timely/src/progress/operate.rs | 93 ++++++++++++--- timely/src/progress/reachability.rs | 173 +++++++++++++++++++++------- timely/src/progress/subgraph.rs | 7 +- 3 files changed, 211 insertions(+), 62 deletions(-) diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 6c3f12955..708240787 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -77,49 +77,104 @@ pub enum FrontierInterest { /// Operator internal connectivity, from inputs to outputs. pub type Connectivity = Vec>; /// Internal connectivity from one port to any number of opposing ports. +/// +/// Represented as a list of `(port, antichain)` pairs. When `dirty` is unset the +/// list is sorted by port, ports are distinct, and no antichain is empty; reads +/// (`get`, `iter_ports`) require this canonical form. Mutations (`insert`, +/// `add_port`) append, and keep the canonical form when ports arrive in +/// non-decreasing order (the common case at all build sites); out-of-order +/// mutations set `dirty`, and a call to `consolidate` restores the canonical +/// form by sorting and merging duplicate ports (robustly `O(n log n)` for any +/// insertion order). #[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Debug, Clone, Eq, PartialEq)] pub struct PortConnectivity { - tree: std::collections::BTreeMap>, + /// Pairs of port and path summary antichain. + entries: Vec<(usize, Antichain)>, + /// Set when `entries` may be unsorted or contain duplicate ports. + dirty: bool, } impl Default for PortConnectivity { fn default() -> Self { - Self { tree: std::collections::BTreeMap::new() } + Self { entries: Vec::new(), dirty: false } } } impl PortConnectivity { - /// Inserts an element by reference, ensuring that the index exists. - pub fn insert(&mut self, index: usize, element: TS) -> bool where TS : crate::PartialOrder { - self.tree.entry(index).or_default().insert(element) - } - /// Inserts an element by reference, ensuring that the index exists. - pub fn insert_ref(&mut self, index: usize, element: &TS) -> bool where TS : crate::PartialOrder + Clone { - self.tree.entry(index).or_default().insert_ref(element) + /// Inserts a summary element for `index`, merging with any existing antichain at `index`. + pub fn insert(&mut self, index: usize, element: TS) where TS : crate::PartialOrder { + if !self.dirty { + match self.entries.last_mut() { + Some((port, antichain)) if *port == index => { antichain.insert(element); return; } + Some((port, _)) if *port > index => { self.dirty = true; } + _ => { } + } + } + self.entries.push((index, Antichain::from_elem(element))); } - /// Introduces a summary for `port`. Panics if a summary already exists. + /// Introduces a summary for `port`, which must not already have one. + /// + /// Panics if a summary already exists for `port`, when this can be cheaply detected + /// (ports added in non-decreasing order); otherwise duplicate additions are merged + /// by antichain insertion at the next `consolidate`. pub fn add_port(&mut self, port: usize, summary: Antichain) { - if !summary.is_empty() { - let prior = self.tree.insert(port, summary); - assert!(prior.is_none()); + if summary.is_empty() { return; } + if !self.dirty { + match self.entries.last() { + Some((last, _)) if *last == port => { panic!("add_port: summary already exists for port {}", port); } + Some((last, _)) if *last > port => { self.dirty = true; } + _ => { } + } } - else { - assert!(self.tree.remove(&port).is_none()); + self.entries.push((port, summary)); + } + /// Restores the canonical (sorted, distinct, non-empty) form after out-of-order mutations. + /// + /// Duplicate ports are merged by antichain insertion, whose result is independent of + /// the order in which elements were introduced. + pub fn consolidate(&mut self) where TS : crate::PartialOrder { + if self.dirty { + let mut entries = std::mem::take(&mut self.entries); + entries.sort_unstable_by_key(|(port, _)| *port); + for (port, summary) in entries { + match self.entries.last_mut() { + Some((last, antichain)) if *last == port => { + for element in summary { antichain.insert(element); } + } + _ => { self.entries.push((port, summary)); } + } + } + self.entries.retain(|(_, antichain)| !antichain.is_empty()); + self.dirty = false; } } /// Borrowing iterator of port identifiers and antichains. + /// + /// Requires the canonical form (see `consolidate`). pub fn iter_ports(&self) -> impl Iterator)> { - self.tree.iter().map(|(o,p)| (*o, p)) + debug_assert!(!self.dirty, "PortConnectivity read while unconsolidated"); + self.entries.iter().map(|(o,p)| (*o, p)) } /// Returns the associated path summary, if it exists. + /// + /// Requires the canonical form (see `consolidate`). pub fn get(&self, index: usize) -> Option<&Antichain> { - self.tree.get(&index) + debug_assert!(!self.dirty, "PortConnectivity read while unconsolidated"); + self.entries + .binary_search_by_key(&index, |(port, _)| *port) + .ok() + .map(|position| &self.entries[position].1) } } -impl FromIterator<(usize, Antichain)> for PortConnectivity { +impl FromIterator<(usize, Antichain)> for PortConnectivity { fn from_iter(iter: T) -> Self where T: IntoIterator)> { - Self { tree: iter.into_iter().filter(|(_,p)| !p.is_empty()).collect() } + let mut result = Self { + entries: iter.into_iter().filter(|(_,p)| !p.is_empty()).collect(), + dirty: true, + }; + result.consolidate(); + result } } diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 93e37668b..afacddfa3 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -74,7 +74,7 @@ //! assert_eq!(results[2], ((Location::new_target(2, 0), 17), -1)); //! ``` -use std::collections::{BinaryHeap, HashMap, VecDeque}; +use std::collections::{BinaryHeap, HashMap}; use std::cmp::Reverse; use columnar::{Vecs, Index as ColumnarIndex}; @@ -84,7 +84,7 @@ use crate::progress::{Source, Target}; use crate::progress::ChangeBatch; use crate::progress::{Location, Port}; use crate::progress::operate::{Connectivity, PortConnectivity}; -use crate::progress::frontier::MutableAntichain; +use crate::progress::frontier::{Antichain, MutableAntichain}; use crate::progress::timestamp::PathSummary; /// Build a `Vecs>>` from nested iterators. @@ -173,7 +173,12 @@ impl Builder { /// Add links internal to operators. /// /// This method overwrites any existing summary, instead of anything more sophisticated. - pub fn add_node(&mut self, index: usize, inputs: usize, outputs: usize, summary: Connectivity) { + pub fn add_node(&mut self, index: usize, inputs: usize, outputs: usize, mut summary: Connectivity) { + + // Restore canonical form for summaries built by out-of-order insertions. + for ports in summary.iter_mut() { + ports.consolidate(); + } // Assert that all summaries exist. debug_assert_eq!(inputs, summary.len()); @@ -777,93 +782,177 @@ impl Tracker { } } +/// Merges two sorted lists with disjoint keys into one sorted list. +fn merge_disjoint(a: Vec<(K, V)>, b: Vec<(K, V)>) -> Vec<(K, V)> { + let mut result = Vec::with_capacity(a.len() + b.len()); + let mut a = a.into_iter().peekable(); + let mut b = b.into_iter().peekable(); + loop { + match (a.peek(), b.peek()) { + (Some(x), Some(y)) => { + if x.0 < y.0 { result.push(a.next().unwrap()); } + else { result.push(b.next().unwrap()); } + } + (Some(_), None) => { result.push(a.next().unwrap()); } + (None, Some(_)) => { result.push(b.next().unwrap()); } + (None, None) => { break; } + } + } + result +} + /// Determines summaries from locations to scope outputs. /// /// Specifically, for each location whose node identifier is non-zero, we compile /// the summaries along which they can reach each output. /// /// Graph locations may be missing from the output, in which case they have no -/// paths to scope outputs. +/// paths to scope outputs. The result is sorted by location. fn summarize_outputs( nodes: &[Connectivity], edges: &[Vec>], - ) -> HashMap> + ) -> Vec<(Location, PortConnectivity)> { // A reverse edge map, to allow us to walk back up the dataflow graph. - let mut reverse = HashMap::new(); + // Sorted by target location; each target should have at most one source. + let mut reverse_edges = Vec::new(); for (node, outputs) in edges.iter().enumerate() { for (output, targets) in outputs.iter().enumerate() { for target in targets.iter() { - reverse.insert( + reverse_edges.push(( Location::from(*target), Location { node, port: Port::Source(output) } - ); + )); } } } + reverse_edges.sort_unstable(); + reverse_edges.dedup(); // A reverse map from operator outputs to inputs, along their internal summaries. - let mut reverse_internal: HashMap<_, Vec<_>> = HashMap::new(); + // Sorted by source location, so that the entries for a location are contiguous. + let mut reverse_internal = Vec::new(); for (node, connectivity) in nodes.iter().enumerate() { for (input, outputs) in connectivity.iter().enumerate() { for (output, summary) in outputs.iter_ports() { - reverse_internal - .entry(Location::new_source(node, output)) - .or_default() - .push((input, summary)); + reverse_internal.push((Location::new_source(node, output), input, summary)); } } } - - let mut results: HashMap> = HashMap::new(); - let mut worklist = VecDeque::<(Location, usize, T::Summary)>::new(); - - let outputs = + reverse_internal.sort_unstable_by(|x, y| (x.0, x.1).cmp(&(y.0, y.1))); + + // Accumulated summaries to scope outputs, as a sequence of sorted lists keyed by + // `(location, output)`. The lists hold disjoint keys and geometrically increasing + // sizes (towards the front): novel keys are introduced as a new list, and lists of + // comparable sizes are merged. This keeps key introduction amortized logarithmic + // and lookups `O(log^2 n)`, without ever rebuilding one large map per round. + let mut levels: Vec)>> = Vec::new(); + + // Round-based (semi-naive) fixed point. Each round walks reverse edges and reverse + // internal summaries from the triples that changed last round, and the proposals + // that improve the accumulated antichains form the next round's frontier. + // The scope may have no outputs, in which case we can do no work. + let mut frontier: Vec<(Location, usize, T::Summary)> = edges .iter() .flat_map(|x| x.iter()) .flat_map(|x| x.iter()) - .filter(|target| target.node == 0); + .filter(|target| target.node == 0) + .map(|target| (Location::from(*target), target.port, Default::default())) + .collect(); - // The scope may have no outputs, in which case we can do no work. - for output_target in outputs { - worklist.push_back((Location::from(*output_target), output_target.port, Default::default())); - } + let mut proposals: Vec<((Location, usize), T::Summary)> = Vec::new(); // Loop until we stop discovering novel reachability paths. - while let Some((location, output, summary)) = worklist.pop_front() { - match location.port { - - // This is an output port of an operator, or a scope input. - // We want to crawl up the operator, to its inputs. - Port::Source(_output_port) => { - if let Some(inputs) = reverse_internal.get(&location) { - for (input_port, operator_summary) in inputs.iter() { + while !frontier.is_empty() { + + // Collect proposed summaries from the triples changed last round. + for (location, output, summary) in frontier.drain(..) { + match location.port { + + // This is an output port of an operator, or a scope input. + // We want to crawl up the operator, to its inputs. + Port::Source(_output_port) => { + let start = reverse_internal.partition_point(|(source, _, _)| *source < location); + let inputs = reverse_internal[start..].iter().take_while(|(source, _, _)| *source == location); + for (_, input_port, operator_summary) in inputs { let new_location = Location::new_target(location.node, *input_port); for op_summary in operator_summary.elements().iter() { if let Some(combined) = op_summary.followed_by(&summary) { - if results.entry(new_location).or_default().insert_ref(output, &combined) { - worklist.push_back((new_location, output, combined)); - } + proposals.push(((new_location, output), combined)); } } } } + + // This is an input port of an operator, or a scope output. + // We want to walk back the (unique) edge leading to it. + Port::Target(_port) => { + if let Ok(index) = reverse_edges.binary_search_by_key(&location, |(target, _)| *target) { + proposals.push(((reverse_edges[index].1, output), summary)); + } + } } + } - // This is an input port of an operator, or a scope output. - // We want to walk back the edges leading to it. - Port::Target(_port) => { - // Each target should have (at most) one source. - if let Some(&source) = reverse.get(&location) { - if results.entry(source).or_default().insert_ref(output, &summary) { - worklist.push_back((source, output, summary)); + // Merge the batch of proposals into the accumulated summaries. Proposals which + // improve an antichain (in an order-independent sense) seed the next round. + proposals.sort_unstable_by(|x, y| x.0.cmp(&y.0)); + let mut fresh: Vec<((Location, usize), Antichain)> = Vec::new(); + for ((location, output), summary) in proposals.drain(..) { + // Find any existing antichain for this key: consecutive sorted proposals + // with a novel key accumulate in `fresh`, others live in some level. + let existing = + if fresh.last().map(|(key, _)| *key == (location, output)).unwrap_or(false) { + fresh.last_mut().map(|(_, antichain)| antichain) + } + else { + levels.iter_mut().find_map(|level| { + match level.binary_search_by_key(&(location, output), |(key, _)| *key) { + Ok(index) => Some(&mut level[index].1), + Err(_) => None, } + }) + }; + if let Some(antichain) = existing { + if antichain.insert_ref(&summary) { + frontier.push((location, output, summary)); } - }, + } + else { + frontier.push((location, output, summary.clone())); + fresh.push(((location, output), Antichain::from_elem(summary))); + } } + + // Introduce novel keys as a new level, restoring geometric level sizes. + if !fresh.is_empty() { + levels.push(fresh); + while levels.len() > 1 && levels[levels.len()-2].len() <= 2 * levels[levels.len()-1].len() { + let upper = levels.pop().unwrap(); + let lower = levels.pop().unwrap(); + levels.push(merge_disjoint(lower, upper)); + } + } + } + + // Merge all levels into one sorted list, and group it by location. + let mut merged = levels.pop().unwrap_or_default(); + while let Some(level) = levels.pop() { + merged = merge_disjoint(level, merged); } + let mut results: Vec<(Location, PortConnectivity)> = Vec::new(); + for ((location, output), antichain) in merged { + match results.last_mut() { + Some((last, connectivity)) if *last == location => { connectivity.add_port(output, antichain); } + _ => { + let mut connectivity = PortConnectivity::default(); + connectivity.add_port(output, antichain); + results.push((location, connectivity)); + } + } + } results } diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 393223f53..211e687a9 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -657,7 +657,12 @@ impl PerOperatorState { let outputs = scope.outputs(); let notify = scope.notify_me().to_vec(); - let (internal_summary, shared_progress, operator) = scope.initialize(); + let (mut internal_summary, shared_progress, operator) = scope.initialize(); + + // Restore canonical form for summaries built by out-of-order insertions. + for ports in internal_summary.iter_mut() { + ports.consolidate(); + } if let Some(l) = summary_logging { l.log(crate::logging::OperatesSummaryEvent { From 536fccc03bc7dc72a6b29147dda409bb481efc29 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 07:53:47 -0400 Subject: [PATCH 02/14] Replace is_acyclic's HashMap with dense per-location counts The cycle check already observed every location, so the map was dense in content; index in-degree counts by per-node location offsets instead. This removes the last std collection map from the progress subsystem. Co-Authored-By: Claude Fable 5 --- timely/src/progress/reachability.rs | 74 ++++++++++++++++++----------- 1 file changed, 45 insertions(+), 29 deletions(-) diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index afacddfa3..8e44fd085 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -74,7 +74,7 @@ //! assert_eq!(results[2], ((Location::new_target(2, 0), 17), -1)); //! ``` -use std::collections::{BinaryHeap, HashMap}; +use std::collections::BinaryHeap; use std::cmp::Reverse; use columnar::{Vecs, Index as ColumnarIndex}; @@ -287,31 +287,40 @@ impl Builder { /// ``` pub fn is_acyclic(&self) -> bool { - let locations = self.shape.iter().map(|(targets, sources)| targets + sources).sum(); - let mut in_degree = HashMap::with_capacity(locations); + // Dense per-location in-degree counts, with each node's targets and + // then sources laid out contiguously at a per-node offset. + let mut offsets = Vec::with_capacity(self.shape.len()); + let mut locations = 0; + for (targets, sources) in self.shape.iter() { + offsets.push(locations); + locations += targets + sources; + } + let index_of = |location: &Location| { + let (targets, _) = self.shape[location.node]; + match location.port { + Port::Target(port) => offsets[location.node] + port, + Port::Source(port) => offsets[location.node] + targets + port, + } + }; + let mut in_degree = vec![0usize; locations]; // Load edges as default summaries. - for (index, ports) in self.edges.iter().enumerate() { - for (output, targets) in ports.iter().enumerate() { - let source = Location::new_source(index, output); - in_degree.entry(source).or_insert(0); + for ports in self.edges.iter() { + for targets in ports.iter() { for &target in targets.iter() { - let target = Location::from(target); - *in_degree.entry(target).or_insert(0) += 1; + in_degree[index_of(&Location::from(target))] += 1; } } } // Load default intra-node summaries. for (index, summary) in self.nodes.iter().enumerate() { - for (input, outputs) in summary.iter().enumerate() { - let target = Location::new_target(index, input); - in_degree.entry(target).or_insert(0); + for outputs in summary.iter() { for (output, summaries) in outputs.iter_ports() { let source = Location::new_source(index, output); for summary in summaries.elements().iter() { if summary == &Default::default() { - *in_degree.entry(source).or_insert(0) += 1; + in_degree[index_of(&source)] += 1; } } } @@ -319,16 +328,21 @@ impl Builder { } // A worklist of nodes that cannot be reached from the whole graph. - // Initially this list contains observed locations with no incoming - // edges, but as the algorithm develops we add to it any locations - // that can only be reached by nodes that have been on this list. - let mut worklist = Vec::with_capacity(in_degree.len()); - for (key, val) in in_degree.iter() { - if *val == 0 { - worklist.push(*key); + // Initially this list contains locations with no incoming edges, but + // as the algorithm develops we add to it any locations that can only + // be reached by nodes that have been on this list. + let mut remaining = in_degree.iter().filter(|count| **count > 0).count(); + let mut worklist = Vec::with_capacity(locations); + for (node, &(targets, sources)) in self.shape.iter().enumerate() { + for port in 0 .. targets { + let location = Location::new_target(node, port); + if in_degree[index_of(&location)] == 0 { worklist.push(location); } + } + for port in 0 .. sources { + let location = Location::new_source(node, port); + if in_degree[index_of(&location)] == 0 { worklist.push(location); } } } - in_degree.retain(|_key, val| val != &0); // Repeatedly remove nodes and update adjacent in-edges. while let Some(Location { node, port }) = worklist.pop() { @@ -336,9 +350,10 @@ impl Builder { Port::Source(port) => { for target in self.edges[node][port].iter() { let target = Location::from(*target); - *in_degree.get_mut(&target).unwrap() -= 1; - if in_degree[&target] == 0 { - in_degree.remove(&target); + let index = index_of(&target); + in_degree[index] -= 1; + if in_degree[index] == 0 { + remaining -= 1; worklist.push(target); } } @@ -346,11 +361,12 @@ impl Builder { Port::Target(port) => { for (output, summaries) in self.nodes[node][port].iter_ports() { let source = Location::new_source(node, output); + let index = index_of(&source); for summary in summaries.elements().iter() { if summary == &Default::default() { - *in_degree.get_mut(&source).unwrap() -= 1; - if in_degree[&source] == 0 { - in_degree.remove(&source); + in_degree[index] -= 1; + if in_degree[index] == 0 { + remaining -= 1; worklist.push(source); } } @@ -360,8 +376,8 @@ impl Builder { } } - // Acyclic graphs should reduce to empty collections. - in_degree.is_empty() + // Acyclic graphs should drain every positive in-degree to zero. + remaining == 0 } } From b0e4566fddeb2ba9927ced5c27c40844c7bfe1f2 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 08:03:52 -0400 Subject: [PATCH 03/14] Unify add_port/insert as merging; canonical connectivity at initialize add_port now merges summaries at a port by antichain insertion, matching insert's semantics; it is the builder's responsibility to introduce multiple summaries for a port only when multiple paths exist. Operate::initialize is now documented to return connectivity in canonical (consolidated) form, and the in-tree implementations (builder_raw, Subgraph) establish that form before returning. The consolidation at the consumer in PerOperatorState::new remains as defense in depth against foreign Operate implementations. Co-Authored-By: Claude Fable 5 --- .../dataflow/operators/generic/builder_raw.rs | 8 +++++- timely/src/progress/operate.rs | 25 +++++++++++++------ timely/src/progress/subgraph.rs | 9 ++++++- 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/timely/src/dataflow/operators/generic/builder_raw.rs b/timely/src/dataflow/operators/generic/builder_raw.rs index 90a750be0..01be11125 100644 --- a/timely/src/dataflow/operators/generic/builder_raw.rs +++ b/timely/src/dataflow/operators/generic/builder_raw.rs @@ -241,7 +241,13 @@ where .iter_mut() .for_each(|output| output.update(T::minimum(), self.shape.peers as i64)); - (self.summary.clone(), Rc::clone(&self.shared_progress), self) + // Establish the canonical form required of `initialize` output. + let mut summary = self.summary.clone(); + for ports in summary.iter_mut() { + ports.consolidate(); + } + + (summary, Rc::clone(&self.shared_progress), self) } fn notify_me(&self) -> &[FrontierInterest] { &self.shape.notify } diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 708240787..6c7a30925 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -53,6 +53,12 @@ pub trait Operate { /// Importantly, it also indicates the initial internal capabilities for all of its outputs. /// This must happen at this moment, as it is the only moment where an operator is allowed to /// safely "create" capabilities without basing them on other, prior capabilities. + /// + /// The returned connectivity must be in canonical form: each `PortConnectivity` consolidated, + /// with ports sorted and distinct and antichains non-empty. This invariant is introduced here + /// and relied upon by all downstream consumers of the connectivity; implementations that + /// accumulate summaries out of port order should call `PortConnectivity::consolidate` before + /// returning. fn initialize(self: Box) -> (Connectivity, Rc>>, Box); /// Indicates for each input whether the operator should be invoked when that input's frontier changes. @@ -102,6 +108,8 @@ impl Default for PortConnectivity { impl PortConnectivity { /// Inserts a summary element for `index`, merging with any existing antichain at `index`. + /// + /// Equivalent to `add_port` with a single-element antichain. pub fn insert(&mut self, index: usize, element: TS) where TS : crate::PartialOrder { if !self.dirty { match self.entries.last_mut() { @@ -112,16 +120,19 @@ impl PortConnectivity { } self.entries.push((index, Antichain::from_elem(element))); } - /// Introduces a summary for `port`, which must not already have one. + /// Introduces a summary for `port`, merging with any summary already present. /// - /// Panics if a summary already exists for `port`, when this can be cheaply detected - /// (ports added in non-decreasing order); otherwise duplicate additions are merged - /// by antichain insertion at the next `consolidate`. - pub fn add_port(&mut self, port: usize, summary: Antichain) { + /// Summaries for the same port are merged by antichain insertion, and describe the + /// union of the claimed paths. It is the builder's responsibility to introduce + /// multiple summaries for a port only when multiple paths exist. + pub fn add_port(&mut self, port: usize, summary: Antichain) where TS : crate::PartialOrder { if summary.is_empty() { return; } if !self.dirty { - match self.entries.last() { - Some((last, _)) if *last == port => { panic!("add_port: summary already exists for port {}", port); } + match self.entries.last_mut() { + Some((last, antichain)) if *last == port => { + for element in summary { antichain.insert(element); } + return; + } Some((last, _)) if *last > port => { self.dirty = true; } _ => { } } diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 211e687a9..ccd10057a 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -574,6 +574,11 @@ where } } + // Establish the canonical form required of `initialize` output. + for ports in internal_summary.iter_mut() { + ports.consolidate(); + } + debug_assert_eq!( internal_summary.len(), self.inputs(), @@ -659,7 +664,9 @@ impl PerOperatorState { let (mut internal_summary, shared_progress, operator) = scope.initialize(); - // Restore canonical form for summaries built by out-of-order insertions. + // Defense in depth: `initialize` is required to produce canonical connectivity + // (see `Operate::initialize`); consolidating here guards against foreign + // implementations that have not upheld that obligation. for ports in internal_summary.iter_mut() { ports.consolidate(); } From 2ae63bc88027016b1e5f57d3fdc9854527e8e951 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 10:43:25 -0400 Subject: [PATCH 04/14] Express the canonical-form contract mechanically PortConnectivity gains is_consolidated(), and a Consolidate extension trait gives the Connectivity alias consolidate()/is_consolidated() verbs, so boundaries state the invariant as a predicate rather than prose. Operate::initialize's contract now reads 'must satisfy is_consolidated'; builder_raw consolidates its summary in place before cloning (leaving the retained copy canonical as well); the defense-in-depth site in PerOperatorState::new debug_asserts the contract before repairing. Co-Authored-By: Claude Fable 5 --- .../dataflow/operators/generic/builder_raw.rs | 10 ++--- timely/src/progress/operate.rs | 38 ++++++++++++++++--- timely/src/progress/reachability.rs | 8 ++-- timely/src/progress/subgraph.rs | 11 ++---- 4 files changed, 45 insertions(+), 22 deletions(-) diff --git a/timely/src/dataflow/operators/generic/builder_raw.rs b/timely/src/dataflow/operators/generic/builder_raw.rs index 01be11125..a4db14fdb 100644 --- a/timely/src/dataflow/operators/generic/builder_raw.rs +++ b/timely/src/dataflow/operators/generic/builder_raw.rs @@ -12,7 +12,7 @@ use crate::scheduling::{Schedule, Activations}; use crate::progress::{Source, Target}; use crate::progress::{Timestamp, Operate, operate::SharedProgress, Antichain}; -use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, Connectivity, Consolidate, PortConnectivity}; use crate::Container; use crate::dataflow::{Stream, Scope, OperatorSlot}; use crate::dataflow::channels::pushers::Tee; @@ -242,12 +242,10 @@ where .for_each(|output| output.update(T::minimum(), self.shape.peers as i64)); // Establish the canonical form required of `initialize` output. - let mut summary = self.summary.clone(); - for ports in summary.iter_mut() { - ports.consolidate(); - } + let mut this = self; + this.summary.consolidate(); - (summary, Rc::clone(&self.shared_progress), self) + (this.summary.clone(), Rc::clone(&this.shared_progress), this) } fn notify_me(&self) -> &[FrontierInterest] { &self.shape.notify } diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 6c7a30925..523ea0afc 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -54,11 +54,10 @@ pub trait Operate { /// This must happen at this moment, as it is the only moment where an operator is allowed to /// safely "create" capabilities without basing them on other, prior capabilities. /// - /// The returned connectivity must be in canonical form: each `PortConnectivity` consolidated, - /// with ports sorted and distinct and antichains non-empty. This invariant is introduced here - /// and relied upon by all downstream consumers of the connectivity; implementations that - /// accumulate summaries out of port order should call `PortConnectivity::consolidate` before - /// returning. + /// The returned connectivity must satisfy `Consolidate::is_consolidated`. This invariant is + /// introduced here and relied upon by all downstream consumers of the connectivity; + /// implementations that accumulate summaries out of port order should call + /// `Consolidate::consolidate` before returning. fn initialize(self: Box) -> (Connectivity, Rc>>, Box); /// Indicates for each input whether the operator should be invoked when that input's frontier changes. @@ -82,6 +81,29 @@ pub enum FrontierInterest { /// Operator internal connectivity, from inputs to outputs. pub type Connectivity = Vec>; + +/// Canonical-form maintenance for connectivity. +/// +/// `Connectivity` is an alias for `Vec>`, and this trait extends it +/// with the per-port `consolidate`/`is_consolidated` verbs so that boundaries which require +/// the canonical form (e.g. `Operate::initialize`) can state and establish it directly. +pub trait Consolidate { + /// Restores the canonical form (see `PortConnectivity::consolidate`). + fn consolidate(&mut self); + /// True when in canonical form (see `PortConnectivity::is_consolidated`). + fn is_consolidated(&self) -> bool; +} + +impl Consolidate for Connectivity { + fn consolidate(&mut self) { + for ports in self.iter_mut() { + ports.consolidate(); + } + } + fn is_consolidated(&self) -> bool { + self.iter().all(|ports| ports.is_consolidated()) + } +} /// Internal connectivity from one port to any number of opposing ports. /// /// Represented as a list of `(port, antichain)` pairs. When `dirty` is unset the @@ -139,6 +161,12 @@ impl PortConnectivity { } self.entries.push((port, summary)); } + /// True when in canonical form: ports sorted and distinct, antichains non-empty. + /// + /// Reads (`get`, `iter_ports`) require this; `consolidate` restores it. + pub fn is_consolidated(&self) -> bool { + !self.dirty + } /// Restores the canonical (sorted, distinct, non-empty) form after out-of-order mutations. /// /// Duplicate ports are merged by antichain insertion, whose result is independent of diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 8e44fd085..8941ca3a7 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -83,7 +83,7 @@ use crate::progress::Timestamp; use crate::progress::{Source, Target}; use crate::progress::ChangeBatch; use crate::progress::{Location, Port}; -use crate::progress::operate::{Connectivity, PortConnectivity}; +use crate::progress::operate::{Connectivity, Consolidate, PortConnectivity}; use crate::progress::frontier::{Antichain, MutableAntichain}; use crate::progress::timestamp::PathSummary; @@ -176,9 +176,9 @@ impl Builder { pub fn add_node(&mut self, index: usize, inputs: usize, outputs: usize, mut summary: Connectivity) { // Restore canonical form for summaries built by out-of-order insertions. - for ports in summary.iter_mut() { - ports.consolidate(); - } + // (`add_node` is a construction-time entry point, so unlike `initialize` + // it accepts unconsolidated input and canonicalizes it here.) + summary.consolidate(); // Assert that all summaries exist. debug_assert_eq!(inputs, summary.len()); diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index ccd10057a..1fa984732 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -19,7 +19,7 @@ use crate::scheduling::activate::Activations; use crate::progress::frontier::{MutableAntichain, MutableAntichainFilter}; use crate::progress::{Timestamp, Operate, operate::SharedProgress}; use crate::progress::{Location, Port, Source, Target}; -use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, Connectivity, Consolidate, PortConnectivity}; use crate::progress::ChangeBatch; use crate::progress::broadcast::Progcaster; use crate::progress::reachability; @@ -575,9 +575,7 @@ where } // Establish the canonical form required of `initialize` output. - for ports in internal_summary.iter_mut() { - ports.consolidate(); - } + internal_summary.consolidate(); debug_assert_eq!( internal_summary.len(), @@ -667,9 +665,8 @@ impl PerOperatorState { // Defense in depth: `initialize` is required to produce canonical connectivity // (see `Operate::initialize`); consolidating here guards against foreign // implementations that have not upheld that obligation. - for ports in internal_summary.iter_mut() { - ports.consolidate(); - } + debug_assert!(internal_summary.is_consolidated(), "`initialize` returned unconsolidated connectivity"); + internal_summary.consolidate(); if let Some(l) = summary_logging { l.log(crate::logging::OperatesSummaryEvent { From 1a3a862a383773f4130e4c1958ec86d6d89b1fc9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 10:59:03 -0400 Subject: [PATCH 05/14] Review feedback: trim docs, simplify insert, mut self in initialize insert now literally delegates to add_port (the in-place fast path was an unproven optimization); type and method docs state requirements without explaining the implementation; builder_raw's initialize takes mut self rather than rebinding. Co-Authored-By: Claude Fable 5 --- .../dataflow/operators/generic/builder_raw.rs | 7 +++-- timely/src/progress/operate.rs | 27 ++++--------------- 2 files changed, 8 insertions(+), 26 deletions(-) diff --git a/timely/src/dataflow/operators/generic/builder_raw.rs b/timely/src/dataflow/operators/generic/builder_raw.rs index a4db14fdb..a9d2e6bc2 100644 --- a/timely/src/dataflow/operators/generic/builder_raw.rs +++ b/timely/src/dataflow/operators/generic/builder_raw.rs @@ -229,7 +229,7 @@ where fn outputs(&self) -> usize { self.shape.outputs } // announce internal topology as fully connected, and hold all default capabilities. - fn initialize(self: Box) -> (Connectivity, Rc>>, Box) { + fn initialize(mut self: Box) -> (Connectivity, Rc>>, Box) { // Request the operator to be scheduled at least once. self.activations.borrow_mut().activate(&self.address[..]); @@ -242,10 +242,9 @@ where .for_each(|output| output.update(T::minimum(), self.shape.peers as i64)); // Establish the canonical form required of `initialize` output. - let mut this = self; - this.summary.consolidate(); + self.summary.consolidate(); - (this.summary.clone(), Rc::clone(&this.shared_progress), this) + (self.summary.clone(), Rc::clone(&self.shared_progress), self) } fn notify_me(&self) -> &[FrontierInterest] { &self.shape.notify } diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 523ea0afc..5af64449d 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -54,10 +54,7 @@ pub trait Operate { /// This must happen at this moment, as it is the only moment where an operator is allowed to /// safely "create" capabilities without basing them on other, prior capabilities. /// - /// The returned connectivity must satisfy `Consolidate::is_consolidated`. This invariant is - /// introduced here and relied upon by all downstream consumers of the connectivity; - /// implementations that accumulate summaries out of port order should call - /// `Consolidate::consolidate` before returning. + /// The returned connectivity must satisfy `Consolidate::is_consolidated`. fn initialize(self: Box) -> (Connectivity, Rc>>, Box); /// Indicates for each input whether the operator should be invoked when that input's frontier changes. @@ -106,14 +103,8 @@ impl Consolidate for Connectivity { } /// Internal connectivity from one port to any number of opposing ports. /// -/// Represented as a list of `(port, antichain)` pairs. When `dirty` is unset the -/// list is sorted by port, ports are distinct, and no antichain is empty; reads -/// (`get`, `iter_ports`) require this canonical form. Mutations (`insert`, -/// `add_port`) append, and keep the canonical form when ports arrive in -/// non-decreasing order (the common case at all build sites); out-of-order -/// mutations set `dirty`, and a call to `consolidate` restores the canonical -/// form by sorting and merging duplicate ports (robustly `O(n log n)` for any -/// insertion order). +/// Read methods (`get`, `iter_ports`) require a consolidated representation, +/// which the `consolidate` method ensures. #[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Debug, Clone, Eq, PartialEq)] pub struct PortConnectivity { /// Pairs of port and path summary antichain. @@ -133,20 +124,12 @@ impl PortConnectivity { /// /// Equivalent to `add_port` with a single-element antichain. pub fn insert(&mut self, index: usize, element: TS) where TS : crate::PartialOrder { - if !self.dirty { - match self.entries.last_mut() { - Some((port, antichain)) if *port == index => { antichain.insert(element); return; } - Some((port, _)) if *port > index => { self.dirty = true; } - _ => { } - } - } - self.entries.push((index, Antichain::from_elem(element))); + self.add_port(index, Antichain::from_elem(element)); } /// Introduces a summary for `port`, merging with any summary already present. /// /// Summaries for the same port are merged by antichain insertion, and describe the - /// union of the claimed paths. It is the builder's responsibility to introduce - /// multiple summaries for a port only when multiple paths exist. + /// union of the claimed paths. pub fn add_port(&mut self, port: usize, summary: Antichain) where TS : crate::PartialOrder { if summary.is_empty() { return; } if !self.dirty { From 7edd838592fd2487c8dbe8cb6f692a27030411a9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 11:09:42 -0400 Subject: [PATCH 06/14] Prototype two-type builder/frozen split of PortConnectivity Split PortConnectivity into an append-only PortConnectivityBuilder (insert/add_port/FromIterator, freeze) and a frozen, always-canonical PortConnectivity (get/iter_ports only). Deletes the dirty bit, the read-side debug_asserts, the Consolidate extension trait, the defense-in-depth consolidation in PerOperatorState::new and Builder::add_node, and the documented temporal contract on Operate::initialize, whose return type now carries the invariant. builder_rc's shared Rc> (also held by input handles and capabilities) stays frozen; new_output_connection re-freezes it in place via mem::take + into_builder + freeze. handles.rs and capability.rs are unchanged. Sizing notes in typed-split-sizing.md. cargo test -p timely passes. Co-Authored-By: Claude Fable 5 --- .../dataflow/operators/generic/builder_raw.rs | 16 +- .../dataflow/operators/generic/builder_rc.rs | 8 +- timely/src/progress/operate.rs | 141 ++++++++---------- timely/src/progress/reachability.rs | 15 +- timely/src/progress/subgraph.rs | 16 +- typed-split-sizing.md | 95 ++++++++++++ 6 files changed, 177 insertions(+), 114 deletions(-) create mode 100644 typed-split-sizing.md diff --git a/timely/src/dataflow/operators/generic/builder_raw.rs b/timely/src/dataflow/operators/generic/builder_raw.rs index a9d2e6bc2..50ebee4aa 100644 --- a/timely/src/dataflow/operators/generic/builder_raw.rs +++ b/timely/src/dataflow/operators/generic/builder_raw.rs @@ -12,7 +12,7 @@ use crate::scheduling::{Schedule, Activations}; use crate::progress::{Source, Target}; use crate::progress::{Timestamp, Operate, operate::SharedProgress, Antichain}; -use crate::progress::operate::{FrontierInterest, Connectivity, Consolidate, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivityBuilder}; use crate::Container; use crate::dataflow::{Stream, Scope, OperatorSlot}; use crate::dataflow::channels::pushers::Tee; @@ -55,7 +55,7 @@ pub struct OperatorBuilder<'scope, T: Timestamp> { slot: OperatorSlot<'scope, T>, address: Rc<[usize]>, // path to the operator (ending with index). shape: OperatorShape, - summary: Connectivity<::Summary>, + summary: Vec::Summary>>, } impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { @@ -113,8 +113,9 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { self.shape.inputs += 1; self.shape.notify.push(FrontierInterest::Always); - let connectivity: PortConnectivity<_> = connection.into_iter().collect(); - assert!(connectivity.iter_ports().all(|(o,_)| o < self.shape.outputs)); + let connectivity: PortConnectivityBuilder<_> = connection.into_iter() + .inspect(|(o,_)| assert!(*o < self.shape.outputs)) + .collect(); self.summary.push(connectivity); receiver @@ -182,7 +183,7 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { activations: self.scope.activations(), logic, shared_progress: Rc::new(RefCell::new(SharedProgress::new(inputs, outputs))), - summary: self.summary, + summary: self.summary.into_iter().map(|b| b.freeze()).collect(), }; self.slot.install(Box::new(operator)); @@ -229,7 +230,7 @@ where fn outputs(&self) -> usize { self.shape.outputs } // announce internal topology as fully connected, and hold all default capabilities. - fn initialize(mut self: Box) -> (Connectivity, Rc>>, Box) { + fn initialize(self: Box) -> (Connectivity, Rc>>, Box) { // Request the operator to be scheduled at least once. self.activations.borrow_mut().activate(&self.address[..]); @@ -241,9 +242,6 @@ where .iter_mut() .for_each(|output| output.update(T::minimum(), self.shape.peers as i64)); - // Establish the canonical form required of `initialize` output. - self.summary.consolidate(); - (self.summary.clone(), Rc::clone(&self.shared_progress), self) } diff --git a/timely/src/dataflow/operators/generic/builder_rc.rs b/timely/src/dataflow/operators/generic/builder_rc.rs index bdb4742e7..8b71d6e0a 100644 --- a/timely/src/dataflow/operators/generic/builder_rc.rs +++ b/timely/src/dataflow/operators/generic/builder_rc.rs @@ -115,7 +115,13 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { self.produced.push(Rc::clone(counter.produced())); for (input, entry) in connection { - self.summaries[input].borrow_mut().add_port(new_output, entry); + // The summaries are shared with input handles (and through them, input + // capabilities), so we re-freeze the shared value in place rather than + // replace it: take the frozen value, extend it, and put the result back. + let mut shared = self.summaries[input].borrow_mut(); + let mut builder = std::mem::take(&mut *shared).into_builder(); + builder.add_port(new_output, entry); + *shared = builder.freeze(); } (pushers::Output::new(counter, internal, new_output), stream) diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 5af64449d..044c69e1e 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -53,8 +53,6 @@ pub trait Operate { /// Importantly, it also indicates the initial internal capabilities for all of its outputs. /// This must happen at this moment, as it is the only moment where an operator is allowed to /// safely "create" capabilities without basing them on other, prior capabilities. - /// - /// The returned connectivity must satisfy `Consolidate::is_consolidated`. fn initialize(self: Box) -> (Connectivity, Rc>>, Box); /// Indicates for each input whether the operator should be invoked when that input's frontier changes. @@ -79,124 +77,103 @@ pub enum FrontierInterest { /// Operator internal connectivity, from inputs to outputs. pub type Connectivity = Vec>; -/// Canonical-form maintenance for connectivity. +/// Append-only accumulation of port summaries, prior to canonicalization. /// -/// `Connectivity` is an alias for `Vec>`, and this trait extends it -/// with the per-port `consolidate`/`is_consolidated` verbs so that boundaries which require -/// the canonical form (e.g. `Operate::initialize`) can state and establish it directly. -pub trait Consolidate { - /// Restores the canonical form (see `PortConnectivity::consolidate`). - fn consolidate(&mut self); - /// True when in canonical form (see `PortConnectivity::is_consolidated`). - fn is_consolidated(&self) -> bool; -} - -impl Consolidate for Connectivity { - fn consolidate(&mut self) { - for ports in self.iter_mut() { - ports.consolidate(); - } - } - fn is_consolidated(&self) -> bool { - self.iter().all(|ports| ports.is_consolidated()) - } -} -/// Internal connectivity from one port to any number of opposing ports. -/// -/// Read methods (`get`, `iter_ports`) require a consolidated representation, -/// which the `consolidate` method ensures. -#[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Debug, Clone, Eq, PartialEq)] -pub struct PortConnectivity { - /// Pairs of port and path summary antichain. +/// Summaries may be introduced in any order, and repeatedly for the same port. +/// The `freeze` method canonicalizes the accumulation into a `PortConnectivity`, +/// which is the only way to read the contents back out. +#[derive(Debug, Clone)] +pub struct PortConnectivityBuilder { + /// Pairs of port and path summary antichain, in insertion order. entries: Vec<(usize, Antichain)>, - /// Set when `entries` may be unsorted or contain duplicate ports. - dirty: bool, } -impl Default for PortConnectivity { +impl Default for PortConnectivityBuilder { fn default() -> Self { - Self { entries: Vec::new(), dirty: false } + Self { entries: Vec::new() } } } -impl PortConnectivity { - /// Inserts a summary element for `index`, merging with any existing antichain at `index`. +impl PortConnectivityBuilder { + /// Inserts a summary element for `index`. /// /// Equivalent to `add_port` with a single-element antichain. - pub fn insert(&mut self, index: usize, element: TS) where TS : crate::PartialOrder { + pub fn insert(&mut self, index: usize, element: TS) { self.add_port(index, Antichain::from_elem(element)); } - /// Introduces a summary for `port`, merging with any summary already present. + /// Introduces a summary for `port`, which `freeze` will merge with any other + /// summaries for the same port. /// /// Summaries for the same port are merged by antichain insertion, and describe the - /// union of the claimed paths. - pub fn add_port(&mut self, port: usize, summary: Antichain) where TS : crate::PartialOrder { - if summary.is_empty() { return; } - if !self.dirty { - match self.entries.last_mut() { - Some((last, antichain)) if *last == port => { - for element in summary { antichain.insert(element); } - return; - } - Some((last, _)) if *last > port => { self.dirty = true; } - _ => { } - } + /// union of the claimed paths. Empty summaries are discarded. + pub fn add_port(&mut self, port: usize, summary: Antichain) { + if !summary.is_empty() { + self.entries.push((port, summary)); } - self.entries.push((port, summary)); } - /// True when in canonical form: ports sorted and distinct, antichains non-empty. - /// - /// Reads (`get`, `iter_ports`) require this; `consolidate` restores it. - pub fn is_consolidated(&self) -> bool { - !self.dirty - } - /// Restores the canonical (sorted, distinct, non-empty) form after out-of-order mutations. + /// Canonicalizes the accumulated summaries into a readable `PortConnectivity`. /// /// Duplicate ports are merged by antichain insertion, whose result is independent of /// the order in which elements were introduced. - pub fn consolidate(&mut self) where TS : crate::PartialOrder { - if self.dirty { - let mut entries = std::mem::take(&mut self.entries); - entries.sort_unstable_by_key(|(port, _)| *port); - for (port, summary) in entries { - match self.entries.last_mut() { - Some((last, antichain)) if *last == port => { - for element in summary { antichain.insert(element); } - } - _ => { self.entries.push((port, summary)); } + pub fn freeze(mut self) -> PortConnectivity where TS : crate::PartialOrder { + self.entries.sort_unstable_by_key(|(port, _)| *port); + let mut entries: Vec<(usize, Antichain)> = Vec::with_capacity(self.entries.len()); + for (port, summary) in self.entries { + match entries.last_mut() { + Some((last, antichain)) if *last == port => { + for element in summary { antichain.insert(element); } } + _ => { entries.push((port, summary)); } } - self.entries.retain(|(_, antichain)| !antichain.is_empty()); - self.dirty = false; } + PortConnectivity { entries } } +} + +impl FromIterator<(usize, Antichain)> for PortConnectivityBuilder { + fn from_iter(iter: T) -> Self where T: IntoIterator)> { + Self { entries: iter.into_iter().filter(|(_,p)| !p.is_empty()).collect() } + } +} + +/// Internal connectivity from one port to any number of opposing ports. +/// +/// Always in canonical form: ports sorted and distinct, antichains non-empty. +/// Values are constructed by `PortConnectivityBuilder::freeze` (or collected from +/// an iterator), and offer no mutation. +#[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Debug, Clone, Eq, PartialEq)] +pub struct PortConnectivity { + /// Pairs of port and path summary antichain, sorted by distinct port. + entries: Vec<(usize, Antichain)>, +} + +impl Default for PortConnectivity { + fn default() -> Self { + Self { entries: Vec::new() } + } +} + +impl PortConnectivity { /// Borrowing iterator of port identifiers and antichains. - /// - /// Requires the canonical form (see `consolidate`). pub fn iter_ports(&self) -> impl Iterator)> { - debug_assert!(!self.dirty, "PortConnectivity read while unconsolidated"); self.entries.iter().map(|(o,p)| (*o, p)) } /// Returns the associated path summary, if it exists. - /// - /// Requires the canonical form (see `consolidate`). pub fn get(&self, index: usize) -> Option<&Antichain> { - debug_assert!(!self.dirty, "PortConnectivity read while unconsolidated"); self.entries .binary_search_by_key(&index, |(port, _)| *port) .ok() .map(|position| &self.entries[position].1) } + /// Recovers a builder, for further accumulation. + pub fn into_builder(self) -> PortConnectivityBuilder { + PortConnectivityBuilder { entries: self.entries } + } } impl FromIterator<(usize, Antichain)> for PortConnectivity { fn from_iter(iter: T) -> Self where T: IntoIterator)> { - let mut result = Self { - entries: iter.into_iter().filter(|(_,p)| !p.is_empty()).collect(), - dirty: true, - }; - result.consolidate(); - result + iter.into_iter().collect::>().freeze() } } diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 8941ca3a7..e0b84fc50 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -83,7 +83,7 @@ use crate::progress::Timestamp; use crate::progress::{Source, Target}; use crate::progress::ChangeBatch; use crate::progress::{Location, Port}; -use crate::progress::operate::{Connectivity, Consolidate, PortConnectivity}; +use crate::progress::operate::{Connectivity, PortConnectivity, PortConnectivityBuilder}; use crate::progress::frontier::{Antichain, MutableAntichain}; use crate::progress::timestamp::PathSummary; @@ -173,12 +173,7 @@ impl Builder { /// Add links internal to operators. /// /// This method overwrites any existing summary, instead of anything more sophisticated. - pub fn add_node(&mut self, index: usize, inputs: usize, outputs: usize, mut summary: Connectivity) { - - // Restore canonical form for summaries built by out-of-order insertions. - // (`add_node` is a construction-time entry point, so unlike `initialize` - // it accepts unconsolidated input and canonicalizes it here.) - summary.consolidate(); + pub fn add_node(&mut self, index: usize, inputs: usize, outputs: usize, summary: Connectivity) { // Assert that all summaries exist. debug_assert_eq!(inputs, summary.len()); @@ -958,18 +953,18 @@ fn summarize_outputs( merged = merge_disjoint(level, merged); } - let mut results: Vec<(Location, PortConnectivity)> = Vec::new(); + let mut results: Vec<(Location, PortConnectivityBuilder)> = Vec::new(); for ((location, output), antichain) in merged { match results.last_mut() { Some((last, connectivity)) if *last == location => { connectivity.add_port(output, antichain); } _ => { - let mut connectivity = PortConnectivity::default(); + let mut connectivity = PortConnectivityBuilder::default(); connectivity.add_port(output, antichain); results.push((location, connectivity)); } } } - results + results.into_iter().map(|(location, builder)| (location, builder.freeze())).collect() } /// Logging types for reachability tracking events. diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 1fa984732..d5d8ef81d 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -19,7 +19,7 @@ use crate::scheduling::activate::Activations; use crate::progress::frontier::{MutableAntichain, MutableAntichainFilter}; use crate::progress::{Timestamp, Operate, operate::SharedProgress}; use crate::progress::{Location, Port, Source, Target}; -use crate::progress::operate::{FrontierInterest, Connectivity, Consolidate, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivity, PortConnectivityBuilder}; use crate::progress::ChangeBatch; use crate::progress::broadcast::Progcaster; use crate::progress::reachability; @@ -565,7 +565,7 @@ where // Note that we need to have `self.inputs()` elements in the summary // with each element containing `self.outputs()` antichains regardless // of how long `self.scope_summary` is - let mut internal_summary = vec![PortConnectivity::default(); self.inputs()]; + let mut internal_summary = vec![PortConnectivityBuilder::default(); self.inputs()]; for (input_idx, input) in self.scope_summary.iter().enumerate() { for (output_idx, output) in input.iter_ports() { for outer in output.elements().iter().cloned().map(TInner::summarize) { @@ -573,9 +573,7 @@ where } } } - - // Establish the canonical form required of `initialize` output. - internal_summary.consolidate(); + let internal_summary: Connectivity<_> = internal_summary.into_iter().map(|b| b.freeze()).collect(); debug_assert_eq!( internal_summary.len(), @@ -660,13 +658,7 @@ impl PerOperatorState { let outputs = scope.outputs(); let notify = scope.notify_me().to_vec(); - let (mut internal_summary, shared_progress, operator) = scope.initialize(); - - // Defense in depth: `initialize` is required to produce canonical connectivity - // (see `Operate::initialize`); consolidating here guards against foreign - // implementations that have not upheld that obligation. - debug_assert!(internal_summary.is_consolidated(), "`initialize` returned unconsolidated connectivity"); - internal_summary.consolidate(); + let (internal_summary, shared_progress, operator) = scope.initialize(); if let Some(l) = summary_logging { l.log(crate::logging::OperatesSummaryEvent { diff --git a/typed-split-sizing.md b/typed-split-sizing.md new file mode 100644 index 000000000..e2a6e6450 --- /dev/null +++ b/typed-split-sizing.md @@ -0,0 +1,95 @@ +# Sizing: two-type (builder/frozen) split of `PortConnectivity` + +Branch `port_connectivity_typed`, built on `origin/port_connectivity_vec` (1a3a862a). +`cargo test -p timely` passes fully (unit, integration, doctests); the whole workspace +builds without warnings. + +## Diff stat vs. base + +``` + timely/src/dataflow/operators/generic/builder_raw.rs | 16 +-- + timely/src/dataflow/operators/generic/builder_rc.rs | 8 +- + timely/src/progress/operate.rs | 141 +++++++++---------- + timely/src/progress/reachability.rs | 15 +-- + timely/src/progress/subgraph.rs | 16 +-- + 5 files changed, 82 insertions(+), 114 deletions(-) +``` + +Net **−32 lines**. `handles.rs` and `capability.rs` are untouched (see below). + +## Deleted + +- The `dirty: bool` field, including its maintenance logic in `add_port` + (the "did this append stay sorted?" tracking) and its presence in the + serde/columnar-serialized and logged representation. +- Both `debug_assert!(!self.dirty, ...)` guards in `iter_ports`/`get`. +- `is_consolidated()` and `consolidate()` on `PortConnectivity` (the merge logic + survives, simplified, inside `PortConnectivityBuilder::freeze`, with no dirty + check and no `retain` pass). +- The entire `Consolidate` extension trait on `Connectivity` and its impl (~20 lines). +- The contract sentence in the `Operate::initialize` doc ("must satisfy + `Consolidate::is_consolidated`") — the return type now says it. +- The defense-in-depth block in `PerOperatorState::new` (subgraph.rs): + `debug_assert!(is_consolidated)` + `consolidate()` + 3-line comment. +- `summary.consolidate()` + comment in `reachability::Builder::add_node` + (it now accepts frozen input; non-canonical input is unrepresentable). +- `self.summary.consolidate()` in builder_raw's `initialize` (which also drops + back to `self: Box` from `mut self`), and `internal_summary.consolidate()` + in `Subgraph::initialize`. + +## Added + +- `PortConnectivityBuilder` (~55 lines with docs): `Default`, `insert`, + `add_port` (plain push, empty summaries discarded), `freeze(self) -> PortConnectivity` + (sort + merge), `FromIterator`. No reads; the only way out is `freeze`. +- `PortConnectivity::into_builder(self)` — needed for the builder_rc case. +- Freeze sites: **4** — builder_raw `build_typed` (freezes the accumulated + per-input summaries once, so `initialize` is read-only), `Subgraph::initialize` + (per-input builders frozen after summarizing), `summarize_outputs` in + reachability.rs (per-location builders frozen at the end), and builder_rc + `new_output_connection` (the re-freeze swap below). +- One behavioral nit in builder_raw `new_input_connection`: the port-bounds + `assert!` moved into an `.inspect()` before collection, so it now also fires + for out-of-range ports carrying *empty* antichains (previously filtered out + before the assert). Strictly stricter; no caller trips it. + +## The handles.rs shared-mutation case + +This turned out to be a non-event in the diff: `handles.rs` and `capability.rs` +keep `Rc>>` (frozen) unchanged. The timing +works out: `InputCapability` reads (`delayed`, `valid_for_output`) happen only at +operator runtime, after construction, while `new_output_connection` mutations +happen only during construction. Rather than store a builder in the `Rc` (which +would force the runtime read path through a non-canonical-tolerant API, +reintroducing the temporal invariant in worse form), builder_rc **re-freezes in +place** on each mutation: + +```rust +let mut shared = self.summaries[input].borrow_mut(); +let mut builder = std::mem::take(&mut *shared).into_builder(); +builder.add_port(new_output, entry); +*shared = builder.freeze(); +``` + +This is the "freeze-and-swap" option, applied per mutation rather than at +`build()` (per-build swap is impossible anyway: the `Rc`'s pointee type is fixed +when handles are created). Honest cost: 5 lines + comment where one line stood, +an `into_builder` round-trip, and an O(ports) re-sort per output addition — +construction-time only, on tiny nearly-sorted vectors. Honest benefit: the shared +value is canonical at *every* instant, not just at read time, so even a hostile +interleaving (reading a capability summary mid-construction) sees a coherent value. + +## Verdict + +**Net simpler.** The two-type version is −32 lines *and* deletes every piece of +non-local reasoning: the dirty bit, the "reads require consolidated form" doc +contract, the two read-side debug_asserts, the defense-in-depth consolidate at +the `initialize` boundary, and the `Consolidate` extension trait. What it adds — +one builder type and four `freeze` calls — is all locally checkable: each freeze +site is exactly where accumulation provably ends, and the compiler enforces it. +The serialized/logged form also sheds the dirty bit. The one place the design +pushed back (builder_rc's shared `Rc>`) cost five lines of +freeze-and-swap rather than any new invariant. The single-type-with-dirty-bit +design is fewer *nominal* types but more *temporal* obligations; this version +trades a documented invariant for a type, and comes out smaller even in raw +line count. From fd0c7a0ea7f65b61533a09c082debcadeac9cfc2 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 11:18:18 -0400 Subject: [PATCH 07/14] Share connectivity as Rc> The summaries shared with input handles are created at new_input, mutated only during construction, and read only at runtime through capabilities. The builder now accumulates them in PortConnectivityBuilders and freezes each into a shared OnceCell when the operator is built, so the lifecycle (write once at build, immutable thereafter) is expressed in the type and the re-freeze-in-place dance disappears along with RefCell borrows on the capability read path. Co-Authored-By: Claude Fable 5 --- timely/src/dataflow/operators/capability.rs | 10 +++--- .../dataflow/operators/generic/builder_rc.rs | 31 ++++++++++--------- .../src/dataflow/operators/generic/handles.rs | 6 ++-- 3 files changed, 25 insertions(+), 22 deletions(-) diff --git a/timely/src/dataflow/operators/capability.rs b/timely/src/dataflow/operators/capability.rs index f7a2b4b82..f4711761e 100644 --- a/timely/src/dataflow/operators/capability.rs +++ b/timely/src/dataflow/operators/capability.rs @@ -23,7 +23,7 @@ use std::{borrow, error::Error, fmt::Display, ops::Deref}; use std::rc::Rc; -use std::cell::RefCell; +use std::cell::{OnceCell, RefCell}; use std::fmt::{self, Debug}; use crate::order::PartialOrder; @@ -239,7 +239,7 @@ pub struct InputCapability { /// Output capability buffers, for use in minting capabilities. internal: CapabilityUpdates, /// Timestamp summaries for each output. - summaries: Rc>>, + summaries: Rc>>, /// A drop guard that updates the consumed capability this InputCapability refers to on drop consumed_guard: ConsumedGuard, } @@ -247,7 +247,7 @@ pub struct InputCapability { impl CapabilityTrait for InputCapability { fn time(&self) -> &T { self.time() } fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool { - let summaries_borrow = self.summaries.borrow(); + let summaries_borrow = self.summaries.get().expect("connectivity frozen at operator build"); let internal_borrow = self.internal.borrow(); // To be valid, the output buffer must match and the timestamp summary needs to be the default. Rc::ptr_eq(&internal_borrow[port], query_buffer) && @@ -258,7 +258,7 @@ impl CapabilityTrait for InputCapability { impl InputCapability { /// Creates a new capability reference at `time` while incrementing (and keeping a reference to) /// the provided [`ChangeBatch`]. - pub(crate) fn new(internal: CapabilityUpdates, summaries: Rc>>, guard: ConsumedGuard) -> Self { + pub(crate) fn new(internal: CapabilityUpdates, summaries: Rc>>, guard: ConsumedGuard) -> Self { InputCapability { internal, summaries, @@ -280,7 +280,7 @@ impl InputCapability { /// This method panics if `self.time` is not 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.borrow().get(output_port) { + 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)) { Capability::new(new_time.clone(), Rc::clone(&self.internal.borrow()[output_port])) } else { diff --git a/timely/src/dataflow/operators/generic/builder_rc.rs b/timely/src/dataflow/operators/generic/builder_rc.rs index 8b71d6e0a..245e6394d 100644 --- a/timely/src/dataflow/operators/generic/builder_rc.rs +++ b/timely/src/dataflow/operators/generic/builder_rc.rs @@ -1,7 +1,7 @@ //! Types to build operators with general shapes. use std::rc::Rc; -use std::cell::RefCell; +use std::cell::{OnceCell, RefCell}; use std::default::Default; use crate::progress::{ChangeBatch, Timestamp}; @@ -18,7 +18,7 @@ use crate::dataflow::operators::capability::Capability; use crate::dataflow::operators::generic::handles::{InputHandleCore, new_input_handle}; use crate::dataflow::operators::generic::operator_info::OperatorInfo; use crate::dataflow::operators::generic::builder_raw::OperatorShape; -use crate::progress::operate::{FrontierInterest, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, PortConnectivity, PortConnectivityBuilder}; use super::builder_raw::OperatorBuilder as OperatorBuilderRaw; @@ -29,8 +29,10 @@ pub struct OperatorBuilder<'scope, T: Timestamp> { frontier: Vec>, consumed: Vec>>>, internal: Rc>>>>>, - /// For each input, a shared list of summaries to each output. - summaries: Vec::Summary>>>>, + /// For each input, summaries to each output: a builder accumulating the + /// summaries during construction, and a shared cell that `build` freezes + /// them into for runtime readers (input handles and capabilities). + summaries: Vec<(Rc::Summary>>>, PortConnectivityBuilder<::Summary>)>, produced: Vec>>>, } @@ -81,8 +83,8 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { self.frontier.push(MutableAntichain::new()); self.consumed.push(Rc::clone(input.consumed())); - let shared_summary = Rc::new(RefCell::new(connection.into_iter().collect())); - self.summaries.push(Rc::clone(&shared_summary)); + let shared_summary = Rc::new(OnceCell::new()); + self.summaries.push((Rc::clone(&shared_summary), connection.into_iter().collect())); new_input_handle(input, Rc::clone(&self.internal), shared_summary) } @@ -115,13 +117,7 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { self.produced.push(Rc::clone(counter.produced())); for (input, entry) in connection { - // The summaries are shared with input handles (and through them, input - // capabilities), so we re-freeze the shared value in place rather than - // replace it: take the frozen value, extend it, and put the result back. - let mut shared = self.summaries[input].borrow_mut(); - let mut builder = std::mem::take(&mut *shared).into_builder(); - builder.add_port(new_output, entry); - *shared = builder.freeze(); + self.summaries[input].1.add_port(new_output, entry); } (pushers::Output::new(counter, internal, new_output), stream) @@ -174,11 +170,18 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { /// /// This method calls `build_typed` directly using a new closure, mirroring /// the variation in `L`, rather than forcing it to be reboxed via `build`. - pub fn build_reschedule_typed(self, constructor: B) + pub fn build_reschedule_typed(mut self, constructor: B) where B: FnOnce(Vec>) -> L, L: FnMut(&[MutableAntichain])->bool+'static { + // Freeze the per-input connectivity, now complete, for runtime readers. + for (cell, builder) in std::mem::take(&mut self.summaries) { + if cell.set(builder.freeze()).is_err() { + unreachable!("connectivity frozen before build"); + } + } + let mut logic = constructor(self.mint_capabilities()); let mut bookkeeping = ProgressBookkeeping { diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs index 4ecf53d07..9205e627e 100644 --- a/timely/src/dataflow/operators/generic/handles.rs +++ b/timely/src/dataflow/operators/generic/handles.rs @@ -4,7 +4,7 @@ //! the operator would with its input and output streams. use std::rc::Rc; -use std::cell::RefCell; +use std::cell::{OnceCell, RefCell}; use std::collections::VecDeque; use crate::progress::Timestamp; @@ -27,7 +27,7 @@ pub struct InputHandleCore>> { /// /// Each timestamp received through this input may only produce output timestamps /// greater or equal to the input timestamp subjected to at least one of these summaries. - summaries: Rc>>, + summaries: Rc>>, /// Staged capabilities and containers. staging: VecDeque<(InputCapability, C)>, staged: Vec, @@ -76,7 +76,7 @@ impl>> InputHandleCore>>( pull_counter: PullCounter, internal: Rc>>>>>, - summaries: Rc>>, + summaries: Rc>>, ) -> InputHandleCore { InputHandleCore { pull_counter, From 05827cdf595d5b068f0e7a6700962b21a1d3769d Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 11:37:00 -0400 Subject: [PATCH 08/14] Review feedback: remove dead into_builder, fix comment, expect, drop sizing doc into_builder's only caller was the re-freeze dance that the OnceCell sharing removed; freeze is one-way. The summaries field comment now matches the tuple order, set() failure uses expect, and the exploratory sizing document leaves the repository. Co-Authored-By: Claude Fable 5 --- .../dataflow/operators/generic/builder_rc.rs | 11 +-- timely/src/progress/operate.rs | 4 - typed-split-sizing.md | 95 ------------------- 3 files changed, 5 insertions(+), 105 deletions(-) delete mode 100644 typed-split-sizing.md diff --git a/timely/src/dataflow/operators/generic/builder_rc.rs b/timely/src/dataflow/operators/generic/builder_rc.rs index 245e6394d..1c9d809f2 100644 --- a/timely/src/dataflow/operators/generic/builder_rc.rs +++ b/timely/src/dataflow/operators/generic/builder_rc.rs @@ -29,9 +29,10 @@ pub struct OperatorBuilder<'scope, T: Timestamp> { frontier: Vec>, consumed: Vec>>>, internal: Rc>>>>>, - /// For each input, summaries to each output: a builder accumulating the - /// summaries during construction, and a shared cell that `build` freezes - /// them into for runtime readers (input handles and capabilities). + /// For each input, a shared cell from which input handles and capabilities + /// read the summaries to each output at runtime, and the builder in which + /// the summaries accumulate during construction. The cell is set once, from + /// the builder, when the operator is built. summaries: Vec<(Rc::Summary>>>, PortConnectivityBuilder<::Summary>)>, produced: Vec>>>, } @@ -177,9 +178,7 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { { // Freeze the per-input connectivity, now complete, for runtime readers. for (cell, builder) in std::mem::take(&mut self.summaries) { - if cell.set(builder.freeze()).is_err() { - unreachable!("connectivity frozen before build"); - } + cell.set(builder.freeze()).expect("connectivity already frozen"); } let mut logic = constructor(self.mint_capabilities()); diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 044c69e1e..c9453b49a 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -165,10 +165,6 @@ impl PortConnectivity { .ok() .map(|position| &self.entries[position].1) } - /// Recovers a builder, for further accumulation. - pub fn into_builder(self) -> PortConnectivityBuilder { - PortConnectivityBuilder { entries: self.entries } - } } impl FromIterator<(usize, Antichain)> for PortConnectivity { diff --git a/typed-split-sizing.md b/typed-split-sizing.md deleted file mode 100644 index e2a6e6450..000000000 --- a/typed-split-sizing.md +++ /dev/null @@ -1,95 +0,0 @@ -# Sizing: two-type (builder/frozen) split of `PortConnectivity` - -Branch `port_connectivity_typed`, built on `origin/port_connectivity_vec` (1a3a862a). -`cargo test -p timely` passes fully (unit, integration, doctests); the whole workspace -builds without warnings. - -## Diff stat vs. base - -``` - timely/src/dataflow/operators/generic/builder_raw.rs | 16 +-- - timely/src/dataflow/operators/generic/builder_rc.rs | 8 +- - timely/src/progress/operate.rs | 141 +++++++++---------- - timely/src/progress/reachability.rs | 15 +-- - timely/src/progress/subgraph.rs | 16 +-- - 5 files changed, 82 insertions(+), 114 deletions(-) -``` - -Net **−32 lines**. `handles.rs` and `capability.rs` are untouched (see below). - -## Deleted - -- The `dirty: bool` field, including its maintenance logic in `add_port` - (the "did this append stay sorted?" tracking) and its presence in the - serde/columnar-serialized and logged representation. -- Both `debug_assert!(!self.dirty, ...)` guards in `iter_ports`/`get`. -- `is_consolidated()` and `consolidate()` on `PortConnectivity` (the merge logic - survives, simplified, inside `PortConnectivityBuilder::freeze`, with no dirty - check and no `retain` pass). -- The entire `Consolidate` extension trait on `Connectivity` and its impl (~20 lines). -- The contract sentence in the `Operate::initialize` doc ("must satisfy - `Consolidate::is_consolidated`") — the return type now says it. -- The defense-in-depth block in `PerOperatorState::new` (subgraph.rs): - `debug_assert!(is_consolidated)` + `consolidate()` + 3-line comment. -- `summary.consolidate()` + comment in `reachability::Builder::add_node` - (it now accepts frozen input; non-canonical input is unrepresentable). -- `self.summary.consolidate()` in builder_raw's `initialize` (which also drops - back to `self: Box` from `mut self`), and `internal_summary.consolidate()` - in `Subgraph::initialize`. - -## Added - -- `PortConnectivityBuilder` (~55 lines with docs): `Default`, `insert`, - `add_port` (plain push, empty summaries discarded), `freeze(self) -> PortConnectivity` - (sort + merge), `FromIterator`. No reads; the only way out is `freeze`. -- `PortConnectivity::into_builder(self)` — needed for the builder_rc case. -- Freeze sites: **4** — builder_raw `build_typed` (freezes the accumulated - per-input summaries once, so `initialize` is read-only), `Subgraph::initialize` - (per-input builders frozen after summarizing), `summarize_outputs` in - reachability.rs (per-location builders frozen at the end), and builder_rc - `new_output_connection` (the re-freeze swap below). -- One behavioral nit in builder_raw `new_input_connection`: the port-bounds - `assert!` moved into an `.inspect()` before collection, so it now also fires - for out-of-range ports carrying *empty* antichains (previously filtered out - before the assert). Strictly stricter; no caller trips it. - -## The handles.rs shared-mutation case - -This turned out to be a non-event in the diff: `handles.rs` and `capability.rs` -keep `Rc>>` (frozen) unchanged. The timing -works out: `InputCapability` reads (`delayed`, `valid_for_output`) happen only at -operator runtime, after construction, while `new_output_connection` mutations -happen only during construction. Rather than store a builder in the `Rc` (which -would force the runtime read path through a non-canonical-tolerant API, -reintroducing the temporal invariant in worse form), builder_rc **re-freezes in -place** on each mutation: - -```rust -let mut shared = self.summaries[input].borrow_mut(); -let mut builder = std::mem::take(&mut *shared).into_builder(); -builder.add_port(new_output, entry); -*shared = builder.freeze(); -``` - -This is the "freeze-and-swap" option, applied per mutation rather than at -`build()` (per-build swap is impossible anyway: the `Rc`'s pointee type is fixed -when handles are created). Honest cost: 5 lines + comment where one line stood, -an `into_builder` round-trip, and an O(ports) re-sort per output addition — -construction-time only, on tiny nearly-sorted vectors. Honest benefit: the shared -value is canonical at *every* instant, not just at read time, so even a hostile -interleaving (reading a capability summary mid-construction) sees a coherent value. - -## Verdict - -**Net simpler.** The two-type version is −32 lines *and* deletes every piece of -non-local reasoning: the dirty bit, the "reads require consolidated form" doc -contract, the two read-side debug_asserts, the defense-in-depth consolidate at -the `initialize` boundary, and the `Consolidate` extension trait. What it adds — -one builder type and four `freeze` calls — is all locally checkable: each freeze -site is exactly where accumulation provably ends, and the compiler enforces it. -The serialized/logged form also sheds the dirty bit. The one place the design -pushed back (builder_rc's shared `Rc>`) cost five lines of -freeze-and-swap rather than any new invariant. The single-type-with-dirty-bit -design is fewer *nominal* types but more *temporal* obligations; this version -trades a documented invariant for a type, and comes out smaller even in raw -line count. From 4a211ef158b6eafd9bdb4758ab978e28f11ac9dd Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 12:04:51 -0400 Subject: [PATCH 09/14] Review feedback: idiomatic merge loop; rename frontier to todo merge_disjoint uses the while-let-both-Some then extend pattern, and summarize_outputs's per-round worklist avoids the name frontier, which means something else throughout this crate. Co-Authored-By: Claude Fable 5 --- timely/src/progress/reachability.rs | 27 +++++++++++---------------- 1 file changed, 11 insertions(+), 16 deletions(-) diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index e0b84fc50..282112d7b 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -798,17 +798,12 @@ fn merge_disjoint(a: Vec<(K, V)>, b: Vec<(K, V)>) -> Vec<(K, V)> { let mut result = Vec::with_capacity(a.len() + b.len()); let mut a = a.into_iter().peekable(); let mut b = b.into_iter().peekable(); - loop { - match (a.peek(), b.peek()) { - (Some(x), Some(y)) => { - if x.0 < y.0 { result.push(a.next().unwrap()); } - else { result.push(b.next().unwrap()); } - } - (Some(_), None) => { result.push(a.next().unwrap()); } - (None, Some(_)) => { result.push(b.next().unwrap()); } - (None, None) => { break; } - } + while let (Some(x), Some(y)) = (a.peek(), b.peek()) { + if x.0 < y.0 { result.push(a.next().unwrap()); } + else { result.push(b.next().unwrap()); } } + result.extend(a); + result.extend(b); result } @@ -861,9 +856,9 @@ fn summarize_outputs( // Round-based (semi-naive) fixed point. Each round walks reverse edges and reverse // internal summaries from the triples that changed last round, and the proposals - // that improve the accumulated antichains form the next round's frontier. + // that improve the accumulated antichains form the next round's work. // The scope may have no outputs, in which case we can do no work. - let mut frontier: Vec<(Location, usize, T::Summary)> = + let mut todo: Vec<(Location, usize, T::Summary)> = edges .iter() .flat_map(|x| x.iter()) @@ -875,10 +870,10 @@ fn summarize_outputs( let mut proposals: Vec<((Location, usize), T::Summary)> = Vec::new(); // Loop until we stop discovering novel reachability paths. - while !frontier.is_empty() { + while !todo.is_empty() { // Collect proposed summaries from the triples changed last round. - for (location, output, summary) in frontier.drain(..) { + for (location, output, summary) in todo.drain(..) { match location.port { // This is an output port of an operator, or a scope input. @@ -927,11 +922,11 @@ fn summarize_outputs( }; if let Some(antichain) = existing { if antichain.insert_ref(&summary) { - frontier.push((location, output, summary)); + todo.push((location, output, summary)); } } else { - frontier.push((location, output, summary.clone())); + todo.push((location, output, summary.clone())); fresh.push(((location, output), Antichain::from_elem(summary))); } } From 3a5884abbe19fd862fdb5f1fb794dcc5c5553d19 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 12:50:06 -0400 Subject: [PATCH 10/14] Inline the accumulated summaries as binary-sized sorted runs Replaces the Vec> levels in summarize_outputs with BinaryRuns: a single vector whose length's binary representation reveals its sorted runs, largest first. Batches of novel keys merge trailing runs as in binary addition (the smallest run is always the lowest set bit of the current length), keeping introduction amortized logarithmic and lookups to at most logarithmically many runs, in one allocation. Co-Authored-By: Claude Fable 5 --- timely/src/progress/reachability.rs | 104 +++++++++++++++++++++------- 1 file changed, 78 insertions(+), 26 deletions(-) diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 282112d7b..00eeb19b6 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -793,6 +793,77 @@ impl Tracker { } } +/// A sorted map maintained as a single vector of power-of-two sorted runs. +/// +/// The vector's length always reveals the run structure: the binary +/// representation of the length, read from the high bit down, gives the +/// sizes of the sorted runs in order. Adjacent runs may in fact be parts +/// of larger sorted runs, but we make no attempt to claim those wins. +/// +/// Keys are distinct across all runs. Batches of novel keys are introduced +/// by merging trailing runs as in binary addition, which makes introduction +/// amortized logarithmic per element, and lookups visit at most +/// logarithmically many runs. +struct BinaryRuns { entries: Vec<(K, V)> } + +impl Default for BinaryRuns { + fn default() -> Self { Self { entries: Vec::new() } } +} + +impl BinaryRuns { + /// A mutable reference to the value at `key`, if present. + fn get_mut(&mut self, key: &K) -> Option<&mut V> { + let mut position = None; + let mut offset = 0; + for bit in (0..usize::BITS).rev() { + let size = 1usize << bit; + if self.entries.len() & size != 0 { + let run = &self.entries[offset .. offset + size]; + if let Ok(index) = run.binary_search_by(|(k, _)| k.cmp(key)) { + position = Some(offset + index); + break; + } + offset += size; + } + } + position.map(|index| &mut self.entries[index].1) + } + + /// Introduces a sorted batch of keys distinct from each other and from those present. + fn insert_batch(&mut self, batch: Vec<(K, V)>) { + debug_assert!(batch.windows(2).all(|w| w[0].0 < w[1].0)); + if batch.is_empty() { return; } + let total = self.entries.len() + batch.len(); + // Runs at the common leading bits of the old and new lengths are unaffected. + let mut keep = 0; + for bit in (0..usize::BITS).rev() { + let size = 1usize << bit; + if (self.entries.len() & size) != (total & size) { break; } + keep += total & size; + } + // The remaining runs absorb the batch as in binary addition: the smallest + // run is always the lowest set bit of the current length. + let mut merged = batch; + while self.entries.len() > keep { + let size = 1usize << self.entries.len().trailing_zeros(); + let run = self.entries.split_off(self.entries.len() - size); + merged = merge_disjoint(run, merged); + } + self.entries.append(&mut merged); + } + + /// Merges all runs into one sorted vector. + fn into_sorted(mut self) -> Vec<(K, V)> { + let mut merged = Vec::new(); + while !self.entries.is_empty() { + let size = 1usize << self.entries.len().trailing_zeros(); + let run = self.entries.split_off(self.entries.len() - size); + merged = merge_disjoint(run, merged); + } + merged + } +} + /// Merges two sorted lists with disjoint keys into one sorted list. fn merge_disjoint(a: Vec<(K, V)>, b: Vec<(K, V)>) -> Vec<(K, V)> { let mut result = Vec::with_capacity(a.len() + b.len()); @@ -847,12 +918,8 @@ fn summarize_outputs( } reverse_internal.sort_unstable_by(|x, y| (x.0, x.1).cmp(&(y.0, y.1))); - // Accumulated summaries to scope outputs, as a sequence of sorted lists keyed by - // `(location, output)`. The lists hold disjoint keys and geometrically increasing - // sizes (towards the front): novel keys are introduced as a new list, and lists of - // comparable sizes are merged. This keeps key introduction amortized logarithmic - // and lookups `O(log^2 n)`, without ever rebuilding one large map per round. - let mut levels: Vec)>> = Vec::new(); + // Accumulated summaries to scope outputs, keyed by `(location, output)`. + let mut accumulated: BinaryRuns<(Location, usize), Antichain> = BinaryRuns::default(); // Round-based (semi-naive) fixed point. Each round walks reverse edges and reverse // internal summaries from the triples that changed last round, and the proposals @@ -913,12 +980,7 @@ fn summarize_outputs( fresh.last_mut().map(|(_, antichain)| antichain) } else { - levels.iter_mut().find_map(|level| { - match level.binary_search_by_key(&(location, output), |(key, _)| *key) { - Ok(index) => Some(&mut level[index].1), - Err(_) => None, - } - }) + accumulated.get_mut(&(location, output)) }; if let Some(antichain) = existing { if antichain.insert_ref(&summary) { @@ -931,22 +993,12 @@ fn summarize_outputs( } } - // Introduce novel keys as a new level, restoring geometric level sizes. - if !fresh.is_empty() { - levels.push(fresh); - while levels.len() > 1 && levels[levels.len()-2].len() <= 2 * levels[levels.len()-1].len() { - let upper = levels.pop().unwrap(); - let lower = levels.pop().unwrap(); - levels.push(merge_disjoint(lower, upper)); - } - } + // Introduce the novel keys. + accumulated.insert_batch(fresh); } - // Merge all levels into one sorted list, and group it by location. - let mut merged = levels.pop().unwrap_or_default(); - while let Some(level) = levels.pop() { - merged = merge_disjoint(level, merged); - } + // Merge all runs into one sorted list, and group it by location. + let merged = accumulated.into_sorted(); let mut results: Vec<(Location, PortConnectivityBuilder)> = Vec::new(); for ((location, output), antichain) in merged { From 8f00163f2bd4877fa15400a675041a383896550d Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 13:04:10 -0400 Subject: [PATCH 11/14] Simplify BinaryRuns: sort the unstable suffix rather than merge runs Introducing a batch now extends the vector and sort_unstables the suffix past the runs shared between the old and new lengths' decompositions. This removes merge_disjoint and the carry cascade entirely, costs a log factor more in comparisons (n log^2 n total), and runs in place, once per dataflow construction, on modest n. Co-Authored-By: Claude Fable 5 --- timely/src/progress/reachability.rs | 49 +++++++---------------------- 1 file changed, 12 insertions(+), 37 deletions(-) diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 00eeb19b6..d3d9dbd07 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -800,9 +800,11 @@ impl Tracker { /// sizes of the sorted runs in order. Adjacent runs may in fact be parts /// of larger sorted runs, but we make no attempt to claim those wins. /// -/// Keys are distinct across all runs. Batches of novel keys are introduced -/// by merging trailing runs as in binary addition, which makes introduction -/// amortized logarithmic per element, and lookups visit at most +/// Keys are distinct across all runs. Novel keys are introduced by +/// re-sorting the suffix whose run structure their addition changes, as +/// in binary addition. Each element is re-sorted at most logarithmically +/// often (so `O(n log^2 n)` comparisons in total, a log factor more than +/// merging would cost, for much less code), and lookups visit at most /// logarithmically many runs. struct BinaryRuns { entries: Vec<(K, V)> } @@ -829,53 +831,26 @@ impl BinaryRuns { position.map(|index| &mut self.entries[index].1) } - /// Introduces a sorted batch of keys distinct from each other and from those present. + /// Introduces a batch of keys distinct from each other and from those present. fn insert_batch(&mut self, batch: Vec<(K, V)>) { - debug_assert!(batch.windows(2).all(|w| w[0].0 < w[1].0)); if batch.is_empty() { return; } let total = self.entries.len() + batch.len(); // Runs at the common leading bits of the old and new lengths are unaffected. - let mut keep = 0; + let mut stable = 0; for bit in (0..usize::BITS).rev() { let size = 1usize << bit; if (self.entries.len() & size) != (total & size) { break; } - keep += total & size; + stable += total & size; } - // The remaining runs absorb the batch as in binary addition: the smallest - // run is always the lowest set bit of the current length. - let mut merged = batch; - while self.entries.len() > keep { - let size = 1usize << self.entries.len().trailing_zeros(); - let run = self.entries.split_off(self.entries.len() - size); - merged = merge_disjoint(run, merged); - } - self.entries.append(&mut merged); + self.entries.extend(batch); + self.entries[stable..].sort_unstable_by(|x, y| x.0.cmp(&y.0)); } /// Merges all runs into one sorted vector. fn into_sorted(mut self) -> Vec<(K, V)> { - let mut merged = Vec::new(); - while !self.entries.is_empty() { - let size = 1usize << self.entries.len().trailing_zeros(); - let run = self.entries.split_off(self.entries.len() - size); - merged = merge_disjoint(run, merged); - } - merged - } -} - -/// Merges two sorted lists with disjoint keys into one sorted list. -fn merge_disjoint(a: Vec<(K, V)>, b: Vec<(K, V)>) -> Vec<(K, V)> { - let mut result = Vec::with_capacity(a.len() + b.len()); - let mut a = a.into_iter().peekable(); - let mut b = b.into_iter().peekable(); - while let (Some(x), Some(y)) = (a.peek(), b.peek()) { - if x.0 < y.0 { result.push(a.next().unwrap()); } - else { result.push(b.next().unwrap()); } + self.entries.sort_unstable_by(|x, y| x.0.cmp(&y.0)); + self.entries } - result.extend(a); - result.extend(b); - result } /// Determines summaries from locations to scope outputs. From 7894e3d344d7fbd42b74fff2e9d5abcde14f56ac Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 13:12:26 -0400 Subject: [PATCH 12/14] Compute the stable prefix by masking at the highest differing bit The lengths' xor locates the highest bit on which they disagree; runs at the agreeing bits above it stay put, replacing the bit-scanning loop. Co-Authored-By: Claude Fable 5 --- timely/src/progress/reachability.rs | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index d3d9dbd07..8240eecb7 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -835,13 +835,9 @@ impl BinaryRuns { fn insert_batch(&mut self, batch: Vec<(K, V)>) { if batch.is_empty() { return; } let total = self.entries.len() + batch.len(); - // Runs at the common leading bits of the old and new lengths are unaffected. - let mut stable = 0; - for bit in (0..usize::BITS).rev() { - let size = 1usize << bit; - if (self.entries.len() & size) != (total & size) { break; } - stable += total & size; - } + // Runs at the leading bits on which the lengths agree are unaffected; mask + // away the highest differing bit (the xor is non-zero) and below. + let stable = total & !(usize::MAX >> (self.entries.len() ^ total).leading_zeros()); self.entries.extend(batch); self.entries[stable..].sort_unstable_by(|x, y| x.0.cmp(&y.0)); } From 6ada25551245dd1018ebbd96ff0a5b3a50d77b20 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 13:36:59 -0400 Subject: [PATCH 13/14] Collapse proposals per key before seeding the next round Same-key proposals within a round first collapse into an antichain, so only elements that survive both the round's own batch and insertion into the accumulated antichain are shipped as next-round work; previously every improving step shipped, including ones dominated within the same round. Also inlines the final into_sorted call. Co-Authored-By: Claude Fable 5 --- timely/src/progress/reachability.rs | 36 ++++++++++++++--------------- 1 file changed, 17 insertions(+), 19 deletions(-) diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 8240eecb7..340d372d0 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -939,28 +939,28 @@ fn summarize_outputs( } } - // Merge the batch of proposals into the accumulated summaries. Proposals which - // improve an antichain (in an order-independent sense) seed the next round. + // Merge the batch of proposals into the accumulated summaries. Proposals are + // first collapsed per key into an antichain, so that only elements novel to + // the accumulated antichain (in an order-independent sense) seed the next round. proposals.sort_unstable_by(|x, y| x.0.cmp(&y.0)); let mut fresh: Vec<((Location, usize), Antichain)> = Vec::new(); - for ((location, output), summary) in proposals.drain(..) { - // Find any existing antichain for this key: consecutive sorted proposals - // with a novel key accumulate in `fresh`, others live in some level. - let existing = - if fresh.last().map(|(key, _)| *key == (location, output)).unwrap_or(false) { - fresh.last_mut().map(|(_, antichain)| antichain) + let mut iter = proposals.drain(..).peekable(); + while let Some(((location, output), summary)) = iter.next() { + // Collapse this round's proposals for the key into one antichain. + let mut batch = Antichain::from_elem(summary); + while iter.peek().map(|(key, _)| *key == (location, output)).unwrap_or(false) { + batch.insert(iter.next().unwrap().1); } - else { - accumulated.get_mut(&(location, output)) - }; - if let Some(antichain) = existing { - if antichain.insert_ref(&summary) { - todo.push((location, output, summary)); + if let Some(antichain) = accumulated.get_mut(&(location, output)) { + for summary in batch { + if antichain.insert_ref(&summary) { + todo.push((location, output, summary)); + } } } else { - todo.push((location, output, summary.clone())); - fresh.push(((location, output), Antichain::from_elem(summary))); + todo.extend(batch.elements().iter().map(|summary| (location, output, summary.clone()))); + fresh.push(((location, output), batch)); } } @@ -969,10 +969,8 @@ fn summarize_outputs( } // Merge all runs into one sorted list, and group it by location. - let merged = accumulated.into_sorted(); - let mut results: Vec<(Location, PortConnectivityBuilder)> = Vec::new(); - for ((location, output), antichain) in merged { + for ((location, output), antichain) in accumulated.into_sorted() { match results.last_mut() { Some((last, connectivity)) if *last == location => { connectivity.add_port(output, antichain); } _ => { From 1024c32d510b4959560e4a15df834c2937cff557 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 12 Jun 2026 13:52:15 -0400 Subject: [PATCH 14/14] Add Antichain::drain; reuse the per-key proposal batch Antichain::drain yields owned elements while leaving the allocation for reuse, mirroring MutableAntichain::update_iter's smallvec::Drain. The proposal-collapsing batch in summarize_outputs is hoisted out of the key loop: existing keys drain it (no clones, allocation retained), and fresh keys surrender it via mem::take, which fresh must own anyway. Co-Authored-By: Claude Fable 5 --- timely/src/progress/frontier.rs | 3 +++ timely/src/progress/reachability.rs | 7 ++++--- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/timely/src/progress/frontier.rs b/timely/src/progress/frontier.rs index 15e3dfaf0..05e768766 100644 --- a/timely/src/progress/frontier.rs +++ b/timely/src/progress/frontier.rs @@ -238,6 +238,9 @@ impl Antichain { ///``` pub fn clear(&mut self) { self.elements.clear() } + /// Drains the elements, leaving the allocation for reuse. + pub fn drain(&mut self) -> smallvec::Drain<'_, [T; 1]> { self.elements.drain(..) } + /// Sorts the elements so that comparisons between antichains can be made. pub fn sort(&mut self) where T: Ord { self.elements.sort() } diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 340d372d0..8cea273ef 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -944,15 +944,16 @@ fn summarize_outputs( // the accumulated antichain (in an order-independent sense) seed the next round. proposals.sort_unstable_by(|x, y| x.0.cmp(&y.0)); let mut fresh: Vec<((Location, usize), Antichain)> = Vec::new(); + let mut batch: Antichain = Antichain::new(); let mut iter = proposals.drain(..).peekable(); while let Some(((location, output), summary)) = iter.next() { // Collapse this round's proposals for the key into one antichain. - let mut batch = Antichain::from_elem(summary); + batch.insert(summary); while iter.peek().map(|(key, _)| *key == (location, output)).unwrap_or(false) { batch.insert(iter.next().unwrap().1); } if let Some(antichain) = accumulated.get_mut(&(location, output)) { - for summary in batch { + for summary in batch.drain() { if antichain.insert_ref(&summary) { todo.push((location, output, summary)); } @@ -960,7 +961,7 @@ fn summarize_outputs( } else { todo.extend(batch.elements().iter().map(|summary| (location, output, summary.clone()))); - fresh.push(((location, output), batch)); + fresh.push(((location, output), std::mem::take(&mut batch))); } }