Log each operator's frontier interest with its summary

mdbook/src/chapter_5/chapter_5_4.md

@@ -209,6 +209,8 @@ For example, a `map` operator has one input and one output with an identity summ
 
 This information is essential for understanding the progress guarantees of a dataflow. If you are trying to understand why a particular timestamp is not yet "complete" at some point in the graph, the operator summaries tell you which paths could still produce data at that timestamp.
 
+The `notify` field has one `FrontierInterest` per input port: whether the operator asks to be scheduled when that input's frontier changes (`Always`), only while it holds capabilities (`IfCapability`), or never (`Never`). A scope such as a region reports the strongest interest among its children, and at least `IfCapability`, so an `Always` child keeps its whole region scheduled as frontiers move. Joining `notify` with `OperatesEvent` names shows which operators those are.
+
 ## Additional Log Streams
 
 Beyond the main `"timely"` stream, there are typed log streams for deeper introspection:

timely/src/logging.rs

@@ -22,7 +22,7 @@ use serde::{Deserialize, Serialize};
 use crate::Container;
 use crate::container::CapacityContainerBuilder;
 use crate::dataflow::operators::capture::{Event, EventPusher};
-use crate::progress::operate::Connectivity;
+use crate::progress::operate::{Connectivity, FrontierInterest};
 
 /// Logs events as a timely stream, with progress statements.
 pub struct BatchLogger<P, C> where P: EventPusher<Duration, C> {
@@ -78,6 +78,8 @@ pub struct OperatesSummaryEvent<TS> {
     pub id: usize,
     /// Timestamp action summaries for (input, output) pairs.
     pub summary: Connectivity<TS>,
+    /// Per input, when the operator asks to be scheduled as that input's frontier changes.
+    pub notify: Vec<FrontierInterest>,
 }
 
 #[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]

timely/src/progress/operate.rs

@@ -64,7 +64,7 @@ pub trait Operate<T: Timestamp> {
 }
 
 /// The ways in which an operator can express interest in activation when an input frontier changes.
-#[derive(Ord, PartialOrd, Eq, PartialEq, Copy, Clone, Debug)]
+#[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Ord, PartialOrd, Eq, PartialEq, Copy, Clone, Debug)]
 pub enum FrontierInterest {
     /// Never interested in frontier changes, as for example the `map()` and `filter()` operators.
     Never,

timely/src/progress/subgraph.rs

@@ -683,6 +683,7 @@ impl<T: Timestamp> PerOperatorState<T> {
             l.log(crate::logging::OperatesSummaryEvent {
                 id: identifier,
                 summary: internal_summary.clone(),
+                notify: notify.clone(),
             })
         }
 

timely/tests/frontier_interest_log.rs

@@ -0,0 +1,62 @@
+//! Operator summaries report each input's frontier interest, so a log can audit who is woken by time.
+
+use std::cell::RefCell;
+use std::collections::HashMap;
+use std::rc::Rc;
+
+use timely::dataflow::operators::{Enter, Input, Inspect, Leave, Probe};
+use timely::dataflow::operators::vec::Map;
+use timely::logging::{TimelyEvent, TimelyEventBuilder, TimelySummaryEventBuilder};
+use timely::progress::operate::FrontierInterest;
+
+/// Builds `input -> region { map [-> inspect_core] } -> probe`, and returns each operator's name
+/// with the frontier interest its summary reports.
+fn interests(frontier: bool) -> Vec<(String, Vec<FrontierInterest>)> {
+    timely::execute_directly(move |worker| {
+        let names: Rc<RefCell<HashMap<usize, String>>> = Default::default();
+        let notify: Rc<RefCell<Vec<(usize, Vec<FrontierInterest>)>>> = Default::default();
+        let (names2, notify2) = (Rc::clone(&names), Rc::clone(&notify));
+        let mut registry = worker.log_register().unwrap();
+        registry.insert::<TimelyEventBuilder, _>("timely", move |_time, data| {
+            for (_, event) in data.iter().flat_map(|d| d.iter()) {
+                if let TimelyEvent::Operates(o) = event { names2.borrow_mut().insert(o.id, o.name.clone()); }
+            }
+        });
+        registry.insert::<TimelySummaryEventBuilder<u64>, _>("timely/summary/u64", move |_time, data| {
+            for (_, event) in data.iter().flat_map(|d| d.iter()) {
+                notify2.borrow_mut().push((event.id, event.notify.clone()));
+            }
+        });
+        drop(registry);
+
+        let (input, probe) = worker.dataflow::<u64, _, _>(|scope| {
+            let (input, stream) = scope.new_input::<Vec<u64>>();
+            let stream = scope.region(|inner| {
+                let stream = stream.enter(inner).map(|x| x + 1);
+                let stream = if frontier { stream.inspect_core(|_| {}) } else { stream };
+                stream.leave(scope)
+            });
+            (input, stream.probe().0)
+        });
+        drop(input);
+        while !probe.done() { worker.step(); }
+        worker.log_register().unwrap().remove("timely");
+        worker.log_register().unwrap().remove("timely/summary/u64");
+
+        let names = names.borrow();
+        let mut result: Vec<_> = notify.borrow().iter().map(|(id, n)| (names[id].clone(), n.clone())).collect();
+        result.sort();
+        result
+    })
+}
+
+#[test]
+fn summaries_report_frontier_interest() {
+    let quiet = interests(false);
+    assert!(quiet.contains(&("FlatMap".to_string(), vec![FrontierInterest::Never])), "{quiet:?}");
+    // A region is at least `IfCapability`: it may hold capabilities for messages it has yet to acknowledge.
+    assert!(quiet.contains(&("Region".to_string(), vec![FrontierInterest::IfCapability])), "{quiet:?}");
+    let awake = interests(true);
+    assert!(awake.contains(&("Inspect".to_string(), vec![FrontierInterest::Always])), "{awake:?}");
+    assert!(awake.contains(&("Region".to_string(), vec![FrontierInterest::Always])), "{awake:?}");
+}