From f367ed0270cbd09bf56eb642e1e0a7cf82c421fc Mon Sep 17 00:00:00 2001 From: Bryan O'Sullivan <23374+bos@users.noreply.github.com> Date: Sat, 24 Oct 2026 13:05:36 -0700 Subject: [PATCH] Tracker visits locations by strongly connected component `Tracker::propagate_all` visits locations by strongly connected component, in a topological order of the components computed when the tracker is built, rather than by time and location. A location outside any cycle is visited once per call, after every change that can reach it, applies all its changes together and pushes its net changes directly. The locations of a component with a cycle are visited in order of time, as before. Pending changes are kept per location, so consolidating them sorts each location's few times rather than all changes by location and time. `Tracker::pushed` now returns a `Vec` of net changes, consolidated per location and in no particular order. --- timely/src/progress/reachability.rs | 560 ++++++++++++++++++++++------ timely/src/progress/subgraph.rs | 2 +- 2 files changed, 450 insertions(+), 112 deletions(-) diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 4ef1b0c32..d371a5988 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -39,7 +39,7 @@ //! tracker //! .pushed() //! .0 -//! .drain() +//! .drain(..) //! .filter(|((location, time), delta)| location.is_target()) //! .collect::>(); //! @@ -62,7 +62,7 @@ //! tracker //! .pushed() //! .0 -//! .drain() +//! .drain(..) //! .filter(|((location, time), delta)| location.is_target()) //! .collect::>(); //! @@ -410,14 +410,16 @@ pub struct Tracker { /// Source and target changes are buffered, which allows us to delay processing until propagation, /// and so consolidate updates, but to leap directly to those frontiers that may have changed. - target_changes: ChangeBatch<(Target, T)>, - source_changes: ChangeBatch<(Source, T)>, + pending_changes: LocationChanges, - /// Worklist of updates to perform, ordered by increasing timestamp and target. - worklist: BinaryHeap>, + /// Worklist of changes to implications to perform, by location, with + /// locations visited in an order consistent with the graph's edges. + worklist: Worklist, - /// Buffer of consequent changes. - pushed_changes: ChangeBatch<(Location, T)>, + /// Buffer of consequent changes: the net change to the frontier at each location. + pushed_changes: Vec<((Location, T), i64)>, + /// The changes to the frontiers in the cyclic component being visited, to consolidate. + cycle_changes: LocationChanges, /// Compiled summaries from each internal location (not scope inputs) to each scope output. output_changes: Vec>, @@ -509,18 +511,17 @@ impl Tracker { /// Updates the count for a time at a target (operator input, scope output). #[inline] pub fn update_target(&mut self, target: Target, time: T, value: i64) { - self.target_changes.update((target, time), value); + self.pending_changes.update(Location::from(target), time, value); } /// Updates the count for a time at a source (operator output, scope input). #[inline] pub fn update_source(&mut self, source: Source, time: T, value: i64) { - self.source_changes.update((source, time), value); + self.pending_changes.update(Location::from(source), time, value); } /// Indicates if any pointstamps have positive count. pub fn tracking_anything(&mut self) -> bool { - !self.source_changes.is_empty() || - !self.target_changes.is_empty() || + !self.pending_changes.is_empty() || self.total_counts > 0 } @@ -540,6 +541,9 @@ impl Tracker { .map(|&(inputs, outputs)| PerOperator::new(inputs, outputs)) .collect::>(); + // The order in which to visit locations when propagating changes. + let worklist = Worklist::new(&builder.shape, &builder.nodes, &builder.edges); + // Summary of scope inputs to scope outputs. let mut builder_summary = vec![PortConnectivity::default(); builder.shape[0].1]; @@ -618,10 +622,10 @@ impl Tracker { target_summaries, source_summaries, per_operator, - target_changes: ChangeBatch::new(), - source_changes: ChangeBatch::new(), - worklist: BinaryHeap::new(), - pushed_changes: ChangeBatch::new(), + pending_changes: LocationChanges::new(&builder.shape), + worklist, + pushed_changes: Vec::new(), + cycle_changes: LocationChanges::new(&builder.shape), output_changes, total_counts: 0, logger, @@ -640,128 +644,109 @@ impl Tracker { if let Some(logger) = &mut self.logger { let target_changes = - self.target_changes + self.pending_changes .iter() - .map(|((target, time), diff)| (target.node, target.port, time, *diff)); + .filter_map(|(location, time, diff)| match location.port { + Port::Target(port) => Some((location.node, port, time, diff)), + Port::Source(_) => None, + }); logger.log_target_updates(target_changes); let source_changes = - self.source_changes + self.pending_changes .iter() - .map(|((source, time), diff)| (source.node, source.port, time, *diff)); + .filter_map(|(location, time, diff)| match location.port { + Port::Source(port) => Some((location.node, port, time, diff)), + Port::Target(_) => None, + }); logger.log_source_updates(source_changes); } - // Step 1: Drain `self.input_changes` and determine actual frontier changes. + // Step 1: Drain `self.pending_changes` and determine actual frontier changes. // - // Not all changes in `self.input_changes` may alter the frontier at a location. + // Not all pending changes may alter the frontier at a location. // By filtering the changes through `self.pointstamps` we react only to discrete // changes in the frontier, rather than changes in the pointstamp counts that // witness that frontier. - use itertools::Itertools; - let mut target_changes = self.target_changes.drain().peekable(); - while let Some(((target, _), _)) = target_changes.peek() { - - let target = *target; - let operator = &mut self.per_operator[target.node].targets[target.port]; - let target_updates = target_changes.peeking_take_while(|((t, _),_)| t == &target).map(|((_,time),diff)| (time,diff)); - let changes = operator.pointstamps.update_iter(target_updates); - - for (time, diff) in changes { - self.total_counts += diff; - for &(output, ref summary) in (&self.target_summaries).get(target.node).get(target.port).into_index_iter() { - if let Some(out_time) = summary.results_in(&time) { - self.output_changes[output].update(out_time, diff); + for (location, updates) in self.pending_changes.drain_locations() { + let operator = &mut self.per_operator[location.node]; + match location.port { + Port::Target(port) => { + let changes = operator.targets[port].pointstamps.update_iter(updates); + for (time, diff) in changes { + self.total_counts += diff; + for &(output, ref summary) in (&self.target_summaries).get(location.node).get(port).into_index_iter() { + if let Some(out_time) = summary.results_in(&time) { + self.output_changes[output].update(out_time, diff); + } + } + self.worklist.update(location, time, diff); } } - self.worklist.push(Reverse((time, Location::from(target), diff))); - } - } - - let mut source_changes = self.source_changes.drain().peekable(); - while let Some(((source, _), _)) = source_changes.peek() { - - let source = *source; - let operator = &mut self.per_operator[source.node]; - let op_source = &mut operator.sources[source.port]; - let source_updates = source_changes.peeking_take_while(|((s, _),_)| s == &source).map(|((_,time),diff)| (time,diff)); - let changes = op_source.pointstamps.update_iter(source_updates); - - for (time, diff) in changes { - self.total_counts += diff; - operator.cap_counts += diff; - for &(output, ref summary) in (&self.source_summaries).get(source.node).get(source.port).into_index_iter() { - if let Some(out_time) = summary.results_in(&time) { - self.output_changes[output].update(out_time, diff); + Port::Source(port) => { + let changes = operator.sources[port].pointstamps.update_iter(updates); + for (time, diff) in changes { + self.total_counts += diff; + operator.cap_counts += diff; + for &(output, ref summary) in (&self.source_summaries).get(location.node).get(port).into_index_iter() { + if let Some(out_time) = summary.results_in(&time) { + self.output_changes[output].update(out_time, diff); + } + } + self.worklist.update(location, time, diff); } } - self.worklist.push(Reverse((time, Location::from(source), diff))); } } // Step 2: Circulate implications of changes to `self.pointstamps`. // - // TODO: The argument that this always terminates is subtle, and should be made. - // The intent is that that by moving forward in layers through `time`, we - // will discover zero-change times when we first visit them, as no further - // changes can be made to them once we complete them. - while let Some(Reverse((time, location, mut diff))) = self.worklist.pop() { - - // Drain and accumulate all updates that have the same time and location. - while self.worklist.peek().map(|x| ((x.0).0 == time) && ((x.0).1 == location)).unwrap_or(false) { - diff += (self.worklist.pop().unwrap().0).2; + // Locations are visited by strongly connected component, in a + // topological order of the components, so that every change that + // can reach a location has reached it before it is visited. A + // location outside any cycle is then visited once, and applies all + // its changes, at whatever times, together: its frontier moves once, + // directly from the old to the new, and its successors receive the + // net change. Within a component with a cycle, changes are applied + // in order of increasing time, as they reach each location. + // + // The order affects only the work done, not the result: each + // location's implications accumulate the same changes, and the + // changes pushed for it are consolidated, so they are the net change + // to its frontier. + while let Some(component) = self.worklist.pop_component() { + if !self.worklist.cyclic(component) { + let location = self.worklist.members(component)[0]; + let updates = self.worklist.take(location); + let implications = match location.port { + Port::Target(port) => &mut self.per_operator[location.node].targets[port].implications, + Port::Source(port) => &mut self.per_operator[location.node].sources[port].implications, + }; + // The location is visited once, so these are its net changes. + for (time, diff) in implications.update_iter(updates) { + push_implication(&self.nodes, &self.edges, &mut self.worklist, location, &time, diff, None); + self.pushed_changes.push(((location, time), diff)); + } + continue; } - // Only act if there is a net change, positive or negative. - if diff != 0 { - - match location.port { - // Update to an operator input. - // Propagate any changes forward across the operator. - Port::Target(port_index) => { - - let changes = - self.per_operator[location.node] - .targets[port_index] - .implications - .update_iter(Some((time, diff))); - - for (time, diff) in changes { - for &(output_port, ref summary) in (&self.nodes).get(location.node).get(port_index).into_index_iter() { - if let Some(new_time) = summary.results_in(&time) { - let source = Location { node: location.node, port: Port::Source(output_port) }; - self.worklist.push(Reverse((new_time, source, diff))); - } - } - self.pushed_changes.update((location, time), diff); - } - } - // Update to an operator output. - // Propagate any changes forward along outgoing edges. - Port::Source(port_index) => { - - let changes = - self.per_operator[location.node] - .sources[port_index] - .implications - .update_iter(Some((time, diff))); - - for (time, diff) in changes { - for new_target in (&self.edges).get(location.node).get(port_index).into_index_iter() { - self.worklist.push(Reverse(( - time.clone(), - Location::from(*new_target), - diff, - ))); - } - self.pushed_changes.update((location, time), diff); - } - }, + // A component with a cycle: its changes in order of increasing time. + self.worklist.load_cycle(component); + while let Some((at, location, change)) = self.worklist.pop_cycle() { + let implications = match location.port { + Port::Target(port) => &mut self.per_operator[location.node].targets[port].implications, + Port::Source(port) => &mut self.per_operator[location.node].sources[port].implications, }; + for (time, diff) in implications.update_iter(Some((at, change))) { + push_implication(&self.nodes, &self.edges, &mut self.worklist, location, &time, diff, Some(component)); + self.cycle_changes.update(location, time, diff); + } } + self.pushed_changes.extend(self.cycle_changes.drain()); } + self.worklist.clear(); } /// Implications of maintained capabilities projected to each output. @@ -770,7 +755,7 @@ impl Tracker { } /// A mutable reference to the pushed results of changes. - pub fn pushed(&mut self) -> (&mut ChangeBatch<(Location, T)>, &[PerOperator]) { + pub fn pushed(&mut self) -> (&mut Vec<((Location, T), i64)>, &[PerOperator]) { (&mut self.pushed_changes, &self.per_operator) } @@ -793,6 +778,359 @@ impl Tracker { } } +/// Pushes the implications of a change of `diff` to `time` in the frontier +/// at `location` to its successors: across the operator, from an input, or +/// along the edges, from an output. A successor in the cyclic `component` +/// being visited receives it in time order; any other, when its component is +/// visited. +fn push_implication( + nodes: &Vecs>>, + edges: &Vecs>>, + worklist: &mut Worklist, + location: Location, + time: &T, + diff: i64, + component: Option, +) { + match location.port { + // Propagate the change forward across the operator. + Port::Target(port) => { + for &(output, ref summary) in nodes.get(location.node).get(port).into_index_iter() { + if let Some(new_time) = summary.results_in(time) { + worklist.push(Location::new_source(location.node, output), new_time, diff, component); + } + } + } + // Propagate the change forward along outgoing edges. + Port::Source(port) => { + for target in edges.get(location.node).get(port).into_index_iter() { + worklist.push(Location::from(*target), time.clone(), diff, component); + } + } + } +} + +/// Changes to the implications at locations, waiting to be applied. +/// +/// The locations are grouped into the strongly connected components of the +/// graph whose edges lead from each operator input to the outputs it may +/// reach, and from each output to the inputs it is connected to. The +/// components are numbered in a topological order, so a change at a location +/// can reach only locations in the same component or a later one. +struct Worklist { + /// The changes at each location. + changes: LocationChanges, + /// The component of each location, indexed as `changes` indexes them. + component: Vec, + /// The locations of each component, those of component `c` being + /// `members[starts[c] .. starts[c + 1]]`. + members: Vec, + /// The start in `members` of each component, and the number of locations. + starts: Vec, + /// The components with changes, possibly more than once each. + components: BinaryHeap>, + /// The changes in the cyclic component being visited, by time and location. + cycle: BinaryHeap>, +} + +impl Worklist { + /// A worklist for a graph with nodes of the `(inputs, outputs)` of + /// `shape`, the internal connectivity of `nodes`, and the edges of `edges`. + fn new(shape: &[(usize, usize)], nodes: &[Connectivity], edges: &[Vec>]) -> Self { + let changes = LocationChanges::new(shape); + let locations = changes.slot.len(); + + // Each location's successors, by index. + let mut successors = vec![Vec::new(); locations]; + for (node, inputs) in nodes.iter().enumerate() { + for (input, outputs) in inputs.iter().enumerate() { + let from = changes.index(Location::new_target(node, input)); + for (output, summaries) in outputs.iter_ports() { + if !summaries.elements().is_empty() { + successors[from].push(changes.index(Location::new_source(node, output))); + } + } + } + } + for (node, outputs) in edges.iter().enumerate() { + for (output, targets) in outputs.iter().enumerate() { + let from = changes.index(Location::new_source(node, output)); + for &target in targets { + successors[from].push(changes.index(Location::from(target))); + } + } + } + + // Each location by index. + let mut location = Vec::with_capacity(locations); + for (node, &(inputs, outputs)) in shape.iter().enumerate() { + location.extend((0 .. inputs).map(|port| Location::new_target(node, port))); + location.extend((0 .. outputs).map(|port| Location::new_source(node, port))); + } + + // Tarjan's algorithm finds the components in reverse topological order. + let found = strongly_connected_components(&successors); + let count = found.len(); + let mut component = vec![0; locations]; + let mut members = Vec::with_capacity(locations); + let mut starts = Vec::with_capacity(count + 1); + for (number, scc) in found.iter().rev().enumerate() { + starts.push(members.len()); + for &index in scc { + component[index] = number; + members.push(location[index]); + } + } + starts.push(members.len()); + + Worklist { changes, component, members, starts, components: BinaryHeap::new(), cycle: BinaryHeap::new() } + } + + /// The locations of `component`. + #[inline] + fn members(&self, component: usize) -> &[Location] { + &self.members[self.starts[component] .. self.starts[component + 1]] + } + + /// Indicates that `component` has a cycle: more than one location, as + /// no location is its own successor. + #[inline] + fn cyclic(&self, component: usize) -> bool { + self.starts[component + 1] - self.starts[component] > 1 + } + + /// Adds a change of `diff` to `time` at `location`, to be applied in + /// time order if `location` is in the cyclic `visiting` component, and + /// otherwise when its component is visited. + #[inline] + fn push(&mut self, location: Location, time: T, diff: i64, visiting: Option) { + let component = self.component[self.changes.index(location)]; + if Some(component) == visiting { + self.cycle.push(Reverse((time, location, diff))); + } + else { + self.update(location, time, diff); + } + } + + /// Adds a change of `diff` to `time` at `location`, to be applied when + /// its component is visited. + #[inline] + fn update(&mut self, location: Location, time: T, diff: i64) { + if self.changes.update(location, time, diff) { + self.components.push(Reverse(self.component[self.changes.index(location)])); + } + } + + /// The first component with changes, in topological order. + fn pop_component(&mut self) -> Option { + let Reverse(component) = self.components.pop()?; + while self.components.peek() == Some(&Reverse(component)) { + self.components.pop(); + } + Some(component) + } + + /// Removes the changes at `location`, consolidated. + fn take(&mut self, location: Location) -> smallvec::IntoIter<[(T, i64); 4]> { + self.changes.take(location).into_inner().into_iter() + } + + /// Moves the changes at the locations of `component` into the time-ordered `cycle`. + fn load_cycle(&mut self, component: usize) { + for index in self.starts[component] .. self.starts[component + 1] { + let location = self.members[index]; + for (time, diff) in self.changes.take(location).into_inner() { + self.cycle.push(Reverse((time, location, diff))); + } + } + } + + /// The net change at the earliest time and location in the cyclic + /// component being visited, skipping those with none. + fn pop_cycle(&mut self) -> Option<(T, Location, i64)> { + while let Some(Reverse((time, location, mut diff))) = self.cycle.pop() { + // Drain and accumulate all updates that have the same time and location. + while self.cycle.peek().is_some_and(|x| (x.0).0 == time && (x.0).1 == location) { + diff += (self.cycle.pop().unwrap().0).2; + } + if diff != 0 { + return Some((time, location, diff)); + } + } + None + } + + /// Forgets the locations that had changes, once all are applied. + fn clear(&mut self) { + self.changes.changes.clear(); + } +} + +/// The strongly connected components of the graph with the `successors` of +/// each vertex, in reverse topological order: a component is listed before +/// any component with an edge to it. Tarjan's algorithm, without recursion. +fn strongly_connected_components(successors: &[Vec]) -> Vec> { + const UNVISITED: usize = usize::MAX; + let count = successors.len(); + let mut index = vec![UNVISITED; count]; + let mut lowlink = vec![0; count]; + let mut on_stack = vec![false; count]; + let mut stack = Vec::new(); + let mut components = Vec::new(); + let mut next = 0; + // The depth-first path: each vertex, and how many of its successors it has visited. + let mut path: Vec<(usize, usize)> = Vec::new(); + for root in 0 .. count { + if index[root] != UNVISITED { + continue; + } + path.push((root, 0)); + index[root] = next; + lowlink[root] = next; + next += 1; + stack.push(root); + on_stack[root] = true; + while let Some(&mut (vertex, ref mut visited)) = path.last_mut() { + if let Some(&successor) = <[usize]>::get(&successors[vertex], *visited) { + *visited += 1; + if index[successor] == UNVISITED { + index[successor] = next; + lowlink[successor] = next; + next += 1; + stack.push(successor); + on_stack[successor] = true; + path.push((successor, 0)); + } + else if on_stack[successor] { + lowlink[vertex] = lowlink[vertex].min(index[successor]); + } + continue; + } + path.pop(); + if let Some(&(parent, _)) = <[(usize, usize)]>::last(&path) { + lowlink[parent] = lowlink[parent].min(lowlink[vertex]); + } + if lowlink[vertex] == index[vertex] { + let mut component = Vec::new(); + loop { + let member = stack.pop().unwrap(); + on_stack[member] = false; + component.push(member); + if member == vertex { + break; + } + } + components.push(component); + } + } + } + components +} + +/// Changes to the counts of times at the locations of a graph. +/// +/// The changes are kept per location, each location's in its own +/// `ChangeBatch`, with the locations in the order they first changed. +/// Consolidating them then sorts only each location's few times, rather +/// than all the changes by location and time together; their consumers +/// need each location's changes consolidated, in no particular order. +struct LocationChanges { + /// For each node, the index in `slot` of its first input, its outputs + /// following its inputs, and its number of inputs. + offsets: Vec<(usize, usize)>, + /// The index in `changes` of each location's changes, valid only while + /// `changes` holds that location at that index. + slot: Vec, + /// The locations that have changes, in the order first changed, and their changes. + changes: Vec<(Location, ChangeBatch)>, +} + +impl LocationChanges { + /// Changes for a graph whose nodes have the `(inputs, outputs)` of `shape`. + fn new(shape: &[(usize, usize)]) -> Self { + let mut offsets = Vec::with_capacity(shape.len()); + let mut ports = 0; + for &(inputs, outputs) in shape { + offsets.push((ports, inputs)); + ports += inputs + outputs; + } + LocationChanges { offsets, slot: vec![usize::MAX; ports], changes: Vec::new() } + } + + /// The index in `slot` of `location`. + #[inline] + fn index(&self, location: Location) -> usize { + let (first, inputs) = self.offsets[location.node]; + match location.port { + Port::Target(port) => first + port, + Port::Source(port) => first + inputs + port, + } + } + + /// Adds `diff` to the count of `time` at `location`, and indicates + /// whether `location` had no changes before. + #[inline] + fn update(&mut self, location: Location, time: T, diff: i64) -> bool { + if diff == 0 { + return false; + } + let index = self.index(location); + let slot = self.slot[index]; + match self.changes.get_mut(slot) { + Some((at, batch)) if *at == location => { + let first = batch.unstable_internal_updates().is_empty(); + batch.update(time, diff); + first + } + _ => { + self.slot[index] = self.changes.len(); + self.changes.push((location, ChangeBatch::new_from(time, diff))); + true + } + } + } + + /// Removes the changes at `location`. + #[inline] + fn take(&mut self, location: Location) -> ChangeBatch { + let slot = self.slot[self.index(location)]; + match self.changes.get_mut(slot) { + Some((at, batch)) if *at == location => std::mem::take(batch), + _ => ChangeBatch::new(), + } + } + + /// Indicates that no location has a net change. + fn is_empty(&mut self) -> bool { + self.changes.iter_mut().all(|(_, batch)| batch.is_empty()) + } + + /// The net changes, by location, time and change. + fn iter(&mut self) -> impl Iterator + '_ { + self.changes.iter_mut().flat_map(|(location, batch)| { + let location = *location; + batch.iter().map(move |(time, diff)| (location, time, *diff)) + }) + } + + /// Removes the changes, yielding each location with a net change and + /// its net changes, consolidated. + fn drain_locations(&mut self) -> impl Iterator)> + '_ { + self.changes.drain(..).filter_map(|(location, batch)| { + let updates = batch.into_inner(); + (!updates.is_empty()).then(|| (location, updates.into_iter())) + }) + } + + /// Removes the changes, yielding each net change at each location. + fn drain(&mut self) -> impl Iterator + '_ { + self.drain_locations().flat_map(|(location, updates)| { + updates.map(move |(time, diff)| ((location, time), diff)) + }) + } +} + /// A sorted map maintained as a single vector of power-of-two sorted runs. /// /// The vector's length always reveals the run structure: the binary diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index d863132fb..2bffcd3e5 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -475,7 +475,7 @@ where // Drain propagated information into shared progress structure. let (pushed, operators) = self.pointstamp_tracker.pushed(); - for ((location, time), diff) in pushed.drain() { + for ((location, time), diff) in pushed.drain(..) { self.maybe_shutdown.push(location.node); // Targets are actionable, sources are not. if let crate::progress::Port::Target(port) = location.port {