diff --git a/timely/src/dataflow/operators/core/inspect.rs b/timely/src/dataflow/operators/core/inspect.rs index 982193688..e9bef8a99 100644 --- a/timely/src/dataflow/operators/core/inspect.rs +++ b/timely/src/dataflow/operators/core/inspect.rs @@ -16,6 +16,10 @@ 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_stamped` or `inspect` when frontiers + /// are not needed. + /// /// # Examples /// ``` /// use timely::dataflow::operators::{ToStream, Inspect}; @@ -33,8 +37,28 @@ 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 + /// when records arrive. + /// /// # Examples /// ``` /// use timely::dataflow::operators::{ToStream, Inspect}; @@ -50,10 +74,8 @@ pub trait Inspect: Sized { 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); } - } + self.inspect_stamped(move |_, data| { + for datum in data.into_iter() { func(datum); } }) } } @@ -76,4 +98,18 @@ impl Inspect for Stream<'_, T, C> { }); }) } + + 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 { + func(cap.stamp(), &*container); + session.give_container(container); + } + }); + }) + } } diff --git a/timely/tests/inspect_interest.rs b/timely/tests/inspect_interest.rs new file mode 100644 index 000000000..d0a7e3823 --- /dev/null +++ b/timely/tests/inspect_interest.rs @@ -0,0 +1,68 @@ +//! `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, 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(); + 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 = match which { + Which::Core => stream.inspect_core(|_| {}), + Which::Stamped => stream.inspect_stamped(|_, _| {}), + Which::Records => 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 + }) +} + +/// The inspect operator a region holds. +#[derive(Clone, Copy)] +enum Which { Core, Stamped, Records } + +#[test] +fn inspect_lets_region_sleep() { + let rounds = 100; + let awake = region_schedules(Which::Core, rounds); + assert!(awake >= rounds as usize, "inspect_core should wake the region each round: {awake}"); + 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}"); +}