inspect and inspect_stamped do not observe frontiers

timely/src/dataflow/operators/core/inspect.rs

@@ -16,6 +16,10 @@ pub trait Inspect<T: Timestamp, C>: 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<T: Timestamp, C>: Sized {
     /// ```
     fn inspect_core<F>(self, func: F) -> Self where F: FnMut(Result<(&Stamp<T>, &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::<Vec<_>>()
+    ///            .inspect_stamped(|stamp, data| println!("seen under {:?}: {:?} records", stamp, data.len()));
+    /// });
+    /// ```
+    fn inspect_stamped<F>(self, func: F) -> Self where F: FnMut(&Stamp<T>, &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<T: Timestamp, C>: 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<T: Timestamp, C: Container> Inspect<T, C> for Stream<'_, T, C> {
             });
         })
     }
+
+    fn inspect_stamped<F>(self, mut func: F) -> Self where F: FnMut(&Stamp<T>, &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);
+                }
+            });
+        })
+    }
 }

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<RefCell<HashMap<usize, String>>> = Default::default();
+        let starts: Rc<RefCell<HashMap<usize, usize>>> = Default::default();
+        let (names2, starts2) = (Rc::clone(&names), Rc::clone(&starts));
+        worker.log_register().unwrap().insert::<TimelyEventBuilder, _>("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::<u64, _, _>(|scope| {
+            let (input, stream) = scope.new_input::<Vec<u64>>();
+            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(&region).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}");
+}