From 7fbfd1e28e31da1062b95cbce6bd3b0c214531ca Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 8 Nov 2026 08:12:12 -0500 Subject: [PATCH 1/2] inspect does not observe frontiers `inspect` was built on `inspect_core`, which is scheduled whenever its input frontier changes. A region takes the strongest frontier interest of its children, so one `inspect` kept its whole region awake, and the region did progress work for every child each time any frontier moved. `inspect` now builds on `unary`, which declares no frontier interest. It is scheduled only when records arrive. `inspect_core` still observes frontiers, and its documentation says what that costs. A new test counts how often a region holding only an `inspect` is scheduled as an input advances without data. Co-Authored-By: Claude Opus 5.5 (1M context) --- timely/src/dataflow/operators/core/inspect.rs | 35 ++++++++--- timely/tests/inspect_interest.rs | 58 +++++++++++++++++++ 2 files changed, 84 insertions(+), 9 deletions(-) create mode 100644 timely/tests/inspect_interest.rs diff --git a/timely/src/dataflow/operators/core/inspect.rs b/timely/src/dataflow/operators/core/inspect.rs index 982193688..64dae12ac 100644 --- a/timely/src/dataflow/operators/core/inspect.rs +++ b/timely/src/dataflow/operators/core/inspect.rs @@ -16,6 +16,9 @@ pub trait Inspect: Sized { /// a singleton unless an upstream operator has stamped its messages with several times, or /// with none. /// + /// To observe frontiers the operator is scheduled whenever its input frontier changes, and + /// so is any region that contains it. Prefer `inspect` when frontiers are not needed. + /// /// # Examples /// ``` /// use timely::dataflow::operators::{ToStream, Inspect}; @@ -35,6 +38,9 @@ pub trait Inspect: Sized { /// Runs a supplied closure on each observed record. /// + /// Unlike `inspect_core`, the operator does not observe frontiers, and is scheduled only + /// when records arrive. + /// /// # Examples /// ``` /// use timely::dataflow::operators::{ToStream, Inspect}; @@ -45,17 +51,10 @@ pub trait Inspect: Sized { /// .inspect(|x| println!("seen: {:?}", x)); /// }); /// ``` - fn inspect(self, mut func: F) -> Self + fn inspect(self, func: F) -> Self where for<'a> &'a C: IntoIterator, - F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static, - { - self.inspect_core(move |event| { - if let Ok((_, data)) = event { - for datum in data.into_iter() { func(datum); } - } - }) - } + F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static; } impl Inspect for Stream<'_, T, C> { @@ -76,4 +75,22 @@ impl Inspect for Stream<'_, T, C> { }); }) } + + fn inspect(self, mut func: F) -> Self + where + for<'a> &'a C: IntoIterator, + F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static, + { + // `unary` rather than `unary_frontier`: an operator that never reads its frontier + // need not be scheduled when it changes, which lets an enclosing region sleep too. + self.unary(Pipeline, "Inspect", move |_,_| move |input, output| { + input.for_each_stamp(|cap, data| { + let mut session = output.session(&cap); + for container in data { + for datum in (&*container).into_iter() { func(datum); } + session.give_container(container); + } + }); + }) + } } diff --git a/timely/tests/inspect_interest.rs b/timely/tests/inspect_interest.rs new file mode 100644 index 000000000..59c1de81a --- /dev/null +++ b/timely/tests/inspect_interest.rs @@ -0,0 +1,58 @@ +//! `inspect` does not observe frontiers, so it does not wake an enclosing region when they move. + +use std::cell::RefCell; +use std::collections::HashMap; +use std::rc::Rc; + +use timely::dataflow::operators::{Enter, Input, Inspect, Leave, Probe}; +use timely::logging::{TimelyEvent, TimelyEventBuilder, StartStop}; + +/// Advances an input through `rounds` rounds without data, and returns how often the region +/// between the input and the probe was scheduled. `frontier` selects `inspect_core` over `inspect`. +fn region_schedules(frontier: bool, rounds: u64) -> usize { + timely::execute_directly(move |worker| { + let names: Rc>> = Default::default(); + let starts: Rc>> = Default::default(); + let (names2, starts2) = (Rc::clone(&names), Rc::clone(&starts)); + worker.log_register().unwrap().insert::("timely", move |_time, data| { + for (_, event) in data.iter().flat_map(|d| d.iter()) { + match event { + TimelyEvent::Operates(o) => { names2.borrow_mut().insert(o.id, o.name.clone()); } + TimelyEvent::Schedule(s) if s.start_stop == StartStop::Start => { *starts2.borrow_mut().entry(s.id).or_default() += 1; } + _ => {} + } + } + }); + + let (mut input, probe) = worker.dataflow::(|scope| { + let (input, stream) = scope.new_input::>(); + let stream = scope.region(|inner| { + let stream = stream.enter(inner); + let stream = if frontier { stream.inspect_core(|_| {}) } else { stream.inspect(|_| {}) }; + stream.leave(scope) + }); + (input, stream.probe().0) + }); + + for round in 1 ..= rounds { + input.advance_to(round); + while probe.less_than(&round) { worker.step(); } + } + drop(input); + while worker.step() { } + worker.log_register().unwrap().remove("timely"); + + let region = names.borrow().iter().find(|(_, n)| *n == "Region").map(|(id, _)| *id).expect("region logged"); + let count = starts.borrow().get(®ion).copied().unwrap_or(0); + count + }) +} + +#[test] +fn inspect_lets_region_sleep() { + let rounds = 100; + let sleepy = region_schedules(false, rounds); + let awake = region_schedules(true, rounds); + assert!(awake >= rounds as usize, "inspect_core should wake the region each round: {awake}"); + assert!(sleepy < 10, "inspect should not wake the region as frontiers move: {sleepy}"); +} From 3bac403956cddf71ccb145bd579a44b999f34b0b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 8 Nov 2026 08:45:49 -0500 Subject: [PATCH 2/2] inspect_stamped observes stamped containers without frontiers `inspect_stamped` runs a closure on each container and the stamp under which it travels. Like `inspect`, it does not observe frontiers, so it is scheduled only when containers arrive. Before this, the only way to see stamps was `inspect_core`, which also observes frontiers and is scheduled whenever they change. `inspect` is now built on `inspect_stamped`, so there is one data-only implementation. The test checks that a region holding either one sleeps while frontiers move. Co-Authored-By: Claude Opus 5.5 (1M context) --- timely/src/dataflow/operators/core/inspect.rs | 37 ++++++++++++++----- timely/tests/inspect_interest.rs | 22 ++++++++--- 2 files changed, 44 insertions(+), 15 deletions(-) diff --git a/timely/src/dataflow/operators/core/inspect.rs b/timely/src/dataflow/operators/core/inspect.rs index 64dae12ac..e9bef8a99 100644 --- a/timely/src/dataflow/operators/core/inspect.rs +++ b/timely/src/dataflow/operators/core/inspect.rs @@ -17,7 +17,8 @@ pub trait Inspect: Sized { /// with none. /// /// To observe frontiers the operator is scheduled whenever its input frontier changes, and - /// so is any region that contains it. Prefer `inspect` when frontiers are not needed. + /// so is any region that contains it. Prefer `inspect_stamped` or `inspect` when frontiers + /// are not needed. /// /// # Examples /// ``` @@ -36,6 +37,23 @@ pub trait Inspect: Sized { /// ``` fn inspect_core(self, func: F) -> Self where F: FnMut(Result<(&Stamp, &C), &[T]>)+'static; + /// Runs a supplied closure on each observed container and the stamp under which it travels. + /// + /// Unlike `inspect_core`, the operator does not observe frontiers, and is scheduled only + /// when containers arrive. + /// + /// # Examples + /// ``` + /// use timely::dataflow::operators::{ToStream, Inspect}; + /// + /// timely::example(|scope| { + /// (0..10).to_stream(scope) + /// .container::>() + /// .inspect_stamped(|stamp, data| println!("seen under {:?}: {:?} records", stamp, data.len())); + /// }); + /// ``` + fn inspect_stamped(self, func: F) -> Self where F: FnMut(&Stamp, &C)+'static; + /// Runs a supplied closure on each observed record. /// /// Unlike `inspect_core`, the operator does not observe frontiers, and is scheduled only @@ -51,10 +69,15 @@ pub trait Inspect: Sized { /// .inspect(|x| println!("seen: {:?}", x)); /// }); /// ``` - fn inspect(self, func: F) -> Self + fn inspect(self, mut func: F) -> Self where for<'a> &'a C: IntoIterator, - F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static; + F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static, + { + self.inspect_stamped(move |_, data| { + for datum in data.into_iter() { func(datum); } + }) + } } impl Inspect for Stream<'_, T, C> { @@ -76,18 +99,14 @@ impl Inspect for Stream<'_, T, C> { }) } - fn inspect(self, mut func: F) -> Self - where - for<'a> &'a C: IntoIterator, - F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static, - { + fn inspect_stamped(self, mut func: F) -> Self where F: FnMut(&Stamp, &C)+'static { // `unary` rather than `unary_frontier`: an operator that never reads its frontier // need not be scheduled when it changes, which lets an enclosing region sleep too. self.unary(Pipeline, "Inspect", move |_,_| move |input, output| { input.for_each_stamp(|cap, data| { let mut session = output.session(&cap); for container in data { - for datum in (&*container).into_iter() { func(datum); } + func(cap.stamp(), &*container); session.give_container(container); } }); diff --git a/timely/tests/inspect_interest.rs b/timely/tests/inspect_interest.rs index 59c1de81a..d0a7e3823 100644 --- a/timely/tests/inspect_interest.rs +++ b/timely/tests/inspect_interest.rs @@ -8,8 +8,8 @@ use timely::dataflow::operators::{Enter, Input, Inspect, Leave, Probe}; use timely::logging::{TimelyEvent, TimelyEventBuilder, StartStop}; /// Advances an input through `rounds` rounds without data, and returns how often the region -/// between the input and the probe was scheduled. `frontier` selects `inspect_core` over `inspect`. -fn region_schedules(frontier: bool, rounds: u64) -> usize { +/// between the input and the probe was scheduled, when the region holds the inspect `which`. +fn region_schedules(which: Which, rounds: u64) -> usize { timely::execute_directly(move |worker| { let names: Rc>> = Default::default(); let starts: Rc>> = Default::default(); @@ -28,7 +28,11 @@ fn region_schedules(frontier: bool, rounds: u64) -> usize { let (input, stream) = scope.new_input::>(); let stream = scope.region(|inner| { let stream = stream.enter(inner); - let stream = if frontier { stream.inspect_core(|_| {}) } else { stream.inspect(|_| {}) }; + let stream = match which { + Which::Core => stream.inspect_core(|_| {}), + Which::Stamped => stream.inspect_stamped(|_, _| {}), + Which::Records => stream.inspect(|_| {}), + }; stream.leave(scope) }); (input, stream.probe().0) @@ -48,11 +52,17 @@ fn region_schedules(frontier: bool, rounds: u64) -> usize { }) } +/// The inspect operator a region holds. +#[derive(Clone, Copy)] +enum Which { Core, Stamped, Records } + #[test] fn inspect_lets_region_sleep() { let rounds = 100; - let sleepy = region_schedules(false, rounds); - let awake = region_schedules(true, rounds); + let awake = region_schedules(Which::Core, rounds); assert!(awake >= rounds as usize, "inspect_core should wake the region each round: {awake}"); - assert!(sleepy < 10, "inspect should not wake the region as frontiers move: {sleepy}"); + let stamped = region_schedules(Which::Stamped, rounds); + assert!(stamped < 10, "inspect_stamped should not wake the region as frontiers move: {stamped}"); + let records = region_schedules(Which::Records, rounds); + assert!(records < 10, "inspect should not wake the region as frontiers move: {records}"); }