From db5e38cd17f98e4ad22f06a5542399cb61bc56c9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 25 Oct 2026 11:10:33 -0400 Subject: [PATCH 1/2] Avoid rebuilding frontiers when adding counts at existing minima --- timely/examples/frontier_counts.rs | 28 +++++++++++++ timely/src/progress/frontier.rs | 64 ++++++++++++++++++++++-------- 2 files changed, 76 insertions(+), 16 deletions(-) create mode 100644 timely/examples/frontier_counts.rs diff --git a/timely/examples/frontier_counts.rs b/timely/examples/frontier_counts.rs new file mode 100644 index 000000000..9c3595d0c --- /dev/null +++ b/timely/examples/frontier_counts.rs @@ -0,0 +1,28 @@ +//! Add counts at an existing frontier while retaining many later timestamps. +//! +//! For example: `cargo run --release -p timely --example frontier_counts -- 16000 16000`. +//! The first argument is the retained timestamp count; the second is the number +//! of separately applied additions. Setup and final validation are not timed. +use std::hint::black_box; +use std::time::Instant; +use timely::progress::frontier::MutableAntichain; + +fn main() { + let mut args = std::env::args().skip(1); + let retained: usize = args.next().expect("retained timestamps").parse().unwrap(); + let additions: usize = args.next().expect("additions").parse().unwrap(); + assert!(retained > 1); + let additions_i64 = i64::try_from(additions).unwrap(); + assert!(additions_i64 < i64::MAX); + + let mut frontier = MutableAntichain::::new(); + frontier.update_iter((0..retained).map(|time| (time, 1))); + let start = Instant::now(); + for _ in 0..additions { + assert_eq!(black_box(&mut frontier).update_iter([(0, 1)]).count(), 0); + } + let elapsed = start.elapsed(); + assert_eq!(frontier.count_for(&0), additions_i64 + 1); + assert_eq!(frontier.update_iter([(0, -additions_i64 - 1)]).collect::>(), vec![(0, -1), (1, 1)]); + println!("{retained},{additions},{}", elapsed.as_nanos()); +} diff --git a/timely/src/progress/frontier.rs b/timely/src/progress/frontier.rs index 05e768766..175d9a70b 100644 --- a/timely/src/progress/frontier.rs +++ b/timely/src/progress/frontier.rs @@ -562,23 +562,15 @@ impl MutableAntichain { where T: PartialOrder, { - // Single-pass `for` loop (instead of two `Iterator::any` calls) avoids - // monomorphizing `slice::Iter::any` over per-call-site closure types and - // traverses `self.frontier` at most once. - let mut beyond_frontier = false; - let mut before_frontier = true; - for f in &self.frontier { - if !beyond_frontier && f.less_than(time) { - beyond_frontier = true; - } - if before_frontier && f.less_equal(time) { - before_frontier = false; - } - if beyond_frontier && !before_frontier { - break; - } + if delta > 0 { + // Adding a count at or beyond the frontier cannot change it. + !self.frontier.iter().any(|f| f.less_equal(time)) + } + else { + // Removing a count can only expose new frontier elements if it + // removes an existing frontier element. Zero changes do nothing. + delta < 0 && self.frontier.contains(time) } - !(beyond_frontier || (delta < 0 && before_frontier)) } /// Rebuilds `self.frontier` from `self.updates`. @@ -866,4 +858,44 @@ mod tests { } assert!(mutable.updates.unstable_internal_updates().len() <= 32); } + + #[test] + fn mutable_matches_counted_set() { + // Include negative counts, zero changes, frontier retreat, and batches + // that remove and replace a frontier element before they are applied. + fn check(domain: Vec) { + let mut counts = std::collections::BTreeMap::::new(); + let mut mutable = MutableAntichain::new(); + let mut seed = 1u64; + for _ in 0..2000 { + let before = mutable.frontier().to_vec(); + let mut updates = Vec::new(); + for _ in 0..1 + (seed as usize % 5) { + seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1); + let time = domain[((seed >> 32) as usize) % domain.len()].clone(); + let diff = ((seed >> 48) % 5) as i64 - 2; + *counts.entry(time.clone()).or_default() += diff; + updates.push((time, diff)); + } + let actual: Vec<_> = mutable.update_iter(updates).collect(); + let expected: Antichain<_> = counts.iter() + .filter(|(_, count)| **count > 0) + .map(|(time, _)| time.clone()) + .collect(); + assert_eq!(mutable.frontier(), expected.borrow()); + + let mut changes = std::collections::BTreeMap::::new(); + for time in before { *changes.entry(time).or_default() -= 1; } + for time in expected.elements() { *changes.entry(time.clone()).or_default() += 1; } + let expected_changes: Vec<_> = changes.into_iter().filter(|(_, d)| *d != 0).collect(); + assert_eq!(actual, expected_changes); + for time in &domain { + assert_eq!(mutable.count_for(time), *counts.get(time).unwrap_or(&0)); + } + } + } + check((0..16usize).collect()); + check((0..4).flat_map(|outer| (0..4).map(move |inner| + crate::order::Product::new(outer, inner))).collect()); + } } From 1216de83509aa5e40aba58b3b4e5180c3f1d2041 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 25 Oct 2026 11:53:27 -0400 Subject: [PATCH 2/2] Keep the isolated frontier benchmark outside maintained examples --- timely/examples/frontier_counts.rs | 28 ---------------------------- 1 file changed, 28 deletions(-) delete mode 100644 timely/examples/frontier_counts.rs diff --git a/timely/examples/frontier_counts.rs b/timely/examples/frontier_counts.rs deleted file mode 100644 index 9c3595d0c..000000000 --- a/timely/examples/frontier_counts.rs +++ /dev/null @@ -1,28 +0,0 @@ -//! Add counts at an existing frontier while retaining many later timestamps. -//! -//! For example: `cargo run --release -p timely --example frontier_counts -- 16000 16000`. -//! The first argument is the retained timestamp count; the second is the number -//! of separately applied additions. Setup and final validation are not timed. -use std::hint::black_box; -use std::time::Instant; -use timely::progress::frontier::MutableAntichain; - -fn main() { - let mut args = std::env::args().skip(1); - let retained: usize = args.next().expect("retained timestamps").parse().unwrap(); - let additions: usize = args.next().expect("additions").parse().unwrap(); - assert!(retained > 1); - let additions_i64 = i64::try_from(additions).unwrap(); - assert!(additions_i64 < i64::MAX); - - let mut frontier = MutableAntichain::::new(); - frontier.update_iter((0..retained).map(|time| (time, 1))); - let start = Instant::now(); - for _ in 0..additions { - assert_eq!(black_box(&mut frontier).update_iter([(0, 1)]).count(), 0); - } - let elapsed = start.elapsed(); - assert_eq!(frontier.count_for(&0), additions_i64 + 1); - assert_eq!(frontier.update_iter([(0, -additions_i64 - 1)]).collect::>(), vec![(0, -1), (1, 1)]); - println!("{retained},{additions},{}", elapsed.as_nanos()); -}