From 7d8b5f08a0a358cec3d55b01ba014f328e01b4c6 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 8 Nov 2026 08:58:31 -0500 Subject: [PATCH] Log each operator's frontier interest with its summary `OperatesSummaryEvent` gains `notify`, one `FrontierInterest` per input, taken from `notify_me()` where the summary is already logged. A region's entry is the strongest interest among its children, and at least `IfCapability`. Joined with `OperatesEvent` names, this shows which operators are scheduled as frontiers move, and which child keeps a region awake. `FrontierInterest` derives `Serialize`, `Deserialize` and `Columnar` so it can be logged. Co-Authored-By: Claude Opus 5.5 (1M context) --- mdbook/src/chapter_5/chapter_5_4.md | 2 + timely/src/logging.rs | 4 +- timely/src/progress/operate.rs | 2 +- timely/src/progress/subgraph.rs | 1 + timely/tests/frontier_interest_log.rs | 62 +++++++++++++++++++++++++++ 5 files changed, 69 insertions(+), 2 deletions(-) create mode 100644 timely/tests/frontier_interest_log.rs diff --git a/mdbook/src/chapter_5/chapter_5_4.md b/mdbook/src/chapter_5/chapter_5_4.md index 17656487d..0816a16cf 100644 --- a/mdbook/src/chapter_5/chapter_5_4.md +++ b/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: diff --git a/timely/src/logging.rs b/timely/src/logging.rs index 8abfb0202..d594a5c5e 100644 --- a/timely/src/logging.rs +++ b/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 where P: EventPusher { @@ -78,6 +78,8 @@ pub struct OperatesSummaryEvent { pub id: usize, /// Timestamp action summaries for (input, output) pairs. pub summary: Connectivity, + /// Per input, when the operator asks to be scheduled as that input's frontier changes. + pub notify: Vec, } #[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)] diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 088bd8940..ca352ff56 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -64,7 +64,7 @@ pub trait Operate { } /// 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, diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index ef923f426..ee014452d 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -683,6 +683,7 @@ impl PerOperatorState { l.log(crate::logging::OperatesSummaryEvent { id: identifier, summary: internal_summary.clone(), + notify: notify.clone(), }) } diff --git a/timely/tests/frontier_interest_log.rs b/timely/tests/frontier_interest_log.rs new file mode 100644 index 000000000..5caedad2a --- /dev/null +++ b/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)> { + timely::execute_directly(move |worker| { + let names: Rc>> = Default::default(); + let notify: Rc)>>> = Default::default(); + let (names2, notify2) = (Rc::clone(&names), Rc::clone(¬ify)); + let mut registry = worker.log_register().unwrap(); + registry.insert::("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::, _>("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::(|scope| { + let (input, stream) = scope.new_input::>(); + 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:?}"); +}