diff --git a/timely/src/dataflow/operators/capability.rs b/timely/src/dataflow/operators/capability.rs index f7a2b4b82..f4711761e 100644 --- a/timely/src/dataflow/operators/capability.rs +++ b/timely/src/dataflow/operators/capability.rs @@ -23,7 +23,7 @@ use std::{borrow, error::Error, fmt::Display, ops::Deref}; use std::rc::Rc; -use std::cell::RefCell; +use std::cell::{OnceCell, RefCell}; use std::fmt::{self, Debug}; use crate::order::PartialOrder; @@ -239,7 +239,7 @@ pub struct InputCapability { /// Output capability buffers, for use in minting capabilities. internal: CapabilityUpdates, /// Timestamp summaries for each output. - summaries: Rc>>, + summaries: Rc>>, /// A drop guard that updates the consumed capability this InputCapability refers to on drop consumed_guard: ConsumedGuard, } @@ -247,7 +247,7 @@ pub struct InputCapability { impl CapabilityTrait for InputCapability { fn time(&self) -> &T { self.time() } fn valid_for_output(&self, query_buffer: &Rc>>, port: usize) -> bool { - let summaries_borrow = self.summaries.borrow(); + let summaries_borrow = self.summaries.get().expect("connectivity frozen at operator build"); let internal_borrow = self.internal.borrow(); // To be valid, the output buffer must match and the timestamp summary needs to be the default. Rc::ptr_eq(&internal_borrow[port], query_buffer) && @@ -258,7 +258,7 @@ impl CapabilityTrait for InputCapability { impl InputCapability { /// Creates a new capability reference at `time` while incrementing (and keeping a reference to) /// the provided [`ChangeBatch`]. - pub(crate) fn new(internal: CapabilityUpdates, summaries: Rc>>, guard: ConsumedGuard) -> Self { + pub(crate) fn new(internal: CapabilityUpdates, summaries: Rc>>, guard: ConsumedGuard) -> Self { InputCapability { internal, summaries, @@ -280,7 +280,7 @@ impl InputCapability { /// This method panics if `self.time` is not less or equal to `new_time`. pub fn delayed(&self, new_time: &T, output_port: usize) -> Capability { use crate::progress::timestamp::PathSummary; - if let Some(path) = self.summaries.borrow().get(output_port) { + if let Some(path) = self.summaries.get().expect("connectivity frozen at operator build").get(output_port) { if path.iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) { Capability::new(new_time.clone(), Rc::clone(&self.internal.borrow()[output_port])) } else { diff --git a/timely/src/dataflow/operators/generic/builder_raw.rs b/timely/src/dataflow/operators/generic/builder_raw.rs index 90a750be0..50ebee4aa 100644 --- a/timely/src/dataflow/operators/generic/builder_raw.rs +++ b/timely/src/dataflow/operators/generic/builder_raw.rs @@ -12,7 +12,7 @@ use crate::scheduling::{Schedule, Activations}; use crate::progress::{Source, Target}; use crate::progress::{Timestamp, Operate, operate::SharedProgress, Antichain}; -use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivityBuilder}; use crate::Container; use crate::dataflow::{Stream, Scope, OperatorSlot}; use crate::dataflow::channels::pushers::Tee; @@ -55,7 +55,7 @@ pub struct OperatorBuilder<'scope, T: Timestamp> { slot: OperatorSlot<'scope, T>, address: Rc<[usize]>, // path to the operator (ending with index). shape: OperatorShape, - summary: Connectivity<::Summary>, + summary: Vec::Summary>>, } impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { @@ -113,8 +113,9 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { self.shape.inputs += 1; self.shape.notify.push(FrontierInterest::Always); - let connectivity: PortConnectivity<_> = connection.into_iter().collect(); - assert!(connectivity.iter_ports().all(|(o,_)| o < self.shape.outputs)); + let connectivity: PortConnectivityBuilder<_> = connection.into_iter() + .inspect(|(o,_)| assert!(*o < self.shape.outputs)) + .collect(); self.summary.push(connectivity); receiver @@ -182,7 +183,7 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { activations: self.scope.activations(), logic, shared_progress: Rc::new(RefCell::new(SharedProgress::new(inputs, outputs))), - summary: self.summary, + summary: self.summary.into_iter().map(|b| b.freeze()).collect(), }; self.slot.install(Box::new(operator)); diff --git a/timely/src/dataflow/operators/generic/builder_rc.rs b/timely/src/dataflow/operators/generic/builder_rc.rs index bdb4742e7..1c9d809f2 100644 --- a/timely/src/dataflow/operators/generic/builder_rc.rs +++ b/timely/src/dataflow/operators/generic/builder_rc.rs @@ -1,7 +1,7 @@ //! Types to build operators with general shapes. use std::rc::Rc; -use std::cell::RefCell; +use std::cell::{OnceCell, RefCell}; use std::default::Default; use crate::progress::{ChangeBatch, Timestamp}; @@ -18,7 +18,7 @@ use crate::dataflow::operators::capability::Capability; use crate::dataflow::operators::generic::handles::{InputHandleCore, new_input_handle}; use crate::dataflow::operators::generic::operator_info::OperatorInfo; use crate::dataflow::operators::generic::builder_raw::OperatorShape; -use crate::progress::operate::{FrontierInterest, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, PortConnectivity, PortConnectivityBuilder}; use super::builder_raw::OperatorBuilder as OperatorBuilderRaw; @@ -29,8 +29,11 @@ pub struct OperatorBuilder<'scope, T: Timestamp> { frontier: Vec>, consumed: Vec>>>, internal: Rc>>>>>, - /// For each input, a shared list of summaries to each output. - summaries: Vec::Summary>>>>, + /// For each input, a shared cell from which input handles and capabilities + /// read the summaries to each output at runtime, and the builder in which + /// the summaries accumulate during construction. The cell is set once, from + /// the builder, when the operator is built. + summaries: Vec<(Rc::Summary>>>, PortConnectivityBuilder<::Summary>)>, produced: Vec>>>, } @@ -81,8 +84,8 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { self.frontier.push(MutableAntichain::new()); self.consumed.push(Rc::clone(input.consumed())); - let shared_summary = Rc::new(RefCell::new(connection.into_iter().collect())); - self.summaries.push(Rc::clone(&shared_summary)); + let shared_summary = Rc::new(OnceCell::new()); + self.summaries.push((Rc::clone(&shared_summary), connection.into_iter().collect())); new_input_handle(input, Rc::clone(&self.internal), shared_summary) } @@ -115,7 +118,7 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { self.produced.push(Rc::clone(counter.produced())); for (input, entry) in connection { - self.summaries[input].borrow_mut().add_port(new_output, entry); + self.summaries[input].1.add_port(new_output, entry); } (pushers::Output::new(counter, internal, new_output), stream) @@ -168,11 +171,16 @@ impl<'scope, T: Timestamp> OperatorBuilder<'scope, T> { /// /// This method calls `build_typed` directly using a new closure, mirroring /// the variation in `L`, rather than forcing it to be reboxed via `build`. - pub fn build_reschedule_typed(self, constructor: B) + pub fn build_reschedule_typed(mut self, constructor: B) where B: FnOnce(Vec>) -> L, L: FnMut(&[MutableAntichain])->bool+'static { + // Freeze the per-input connectivity, now complete, for runtime readers. + for (cell, builder) in std::mem::take(&mut self.summaries) { + cell.set(builder.freeze()).expect("connectivity already frozen"); + } + let mut logic = constructor(self.mint_capabilities()); let mut bookkeeping = ProgressBookkeeping { diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs index 4ecf53d07..9205e627e 100644 --- a/timely/src/dataflow/operators/generic/handles.rs +++ b/timely/src/dataflow/operators/generic/handles.rs @@ -4,7 +4,7 @@ //! the operator would with its input and output streams. use std::rc::Rc; -use std::cell::RefCell; +use std::cell::{OnceCell, RefCell}; use std::collections::VecDeque; use crate::progress::Timestamp; @@ -27,7 +27,7 @@ pub struct InputHandleCore>> { /// /// Each timestamp received through this input may only produce output timestamps /// greater or equal to the input timestamp subjected to at least one of these summaries. - summaries: Rc>>, + summaries: Rc>>, /// Staged capabilities and containers. staging: VecDeque<(InputCapability, C)>, staged: Vec, @@ -76,7 +76,7 @@ impl>> InputHandleCore>>( pull_counter: PullCounter, internal: Rc>>>>>, - summaries: Rc>>, + summaries: Rc>>, ) -> InputHandleCore { InputHandleCore { pull_counter, diff --git a/timely/src/progress/frontier.rs b/timely/src/progress/frontier.rs index 15e3dfaf0..05e768766 100644 --- a/timely/src/progress/frontier.rs +++ b/timely/src/progress/frontier.rs @@ -238,6 +238,9 @@ impl Antichain { ///``` pub fn clear(&mut self) { self.elements.clear() } + /// Drains the elements, leaving the allocation for reuse. + pub fn drain(&mut self) -> smallvec::Drain<'_, [T; 1]> { self.elements.drain(..) } + /// Sorts the elements so that comparisons between antichains can be made. pub fn sort(&mut self) where T: Ord { self.elements.sort() } diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs index 6c3f12955..c9453b49a 100644 --- a/timely/src/progress/operate.rs +++ b/timely/src/progress/operate.rs @@ -76,50 +76,100 @@ pub enum FrontierInterest { /// Operator internal connectivity, from inputs to outputs. pub type Connectivity = Vec>; -/// Internal connectivity from one port to any number of opposing ports. -#[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Debug, Clone, Eq, PartialEq)] -pub struct PortConnectivity { - tree: std::collections::BTreeMap>, + +/// Append-only accumulation of port summaries, prior to canonicalization. +/// +/// Summaries may be introduced in any order, and repeatedly for the same port. +/// The `freeze` method canonicalizes the accumulation into a `PortConnectivity`, +/// which is the only way to read the contents back out. +#[derive(Debug, Clone)] +pub struct PortConnectivityBuilder { + /// Pairs of port and path summary antichain, in insertion order. + entries: Vec<(usize, Antichain)>, } -impl Default for PortConnectivity { +impl Default for PortConnectivityBuilder { fn default() -> Self { - Self { tree: std::collections::BTreeMap::new() } + Self { entries: Vec::new() } } } -impl PortConnectivity { - /// Inserts an element by reference, ensuring that the index exists. - pub fn insert(&mut self, index: usize, element: TS) -> bool where TS : crate::PartialOrder { - self.tree.entry(index).or_default().insert(element) - } - /// Inserts an element by reference, ensuring that the index exists. - pub fn insert_ref(&mut self, index: usize, element: &TS) -> bool where TS : crate::PartialOrder + Clone { - self.tree.entry(index).or_default().insert_ref(element) +impl PortConnectivityBuilder { + /// Inserts a summary element for `index`. + /// + /// Equivalent to `add_port` with a single-element antichain. + pub fn insert(&mut self, index: usize, element: TS) { + self.add_port(index, Antichain::from_elem(element)); } - /// Introduces a summary for `port`. Panics if a summary already exists. + /// Introduces a summary for `port`, which `freeze` will merge with any other + /// summaries for the same port. + /// + /// Summaries for the same port are merged by antichain insertion, and describe the + /// union of the claimed paths. Empty summaries are discarded. pub fn add_port(&mut self, port: usize, summary: Antichain) { if !summary.is_empty() { - let prior = self.tree.insert(port, summary); - assert!(prior.is_none()); + self.entries.push((port, summary)); } - else { - assert!(self.tree.remove(&port).is_none()); + } + /// Canonicalizes the accumulated summaries into a readable `PortConnectivity`. + /// + /// Duplicate ports are merged by antichain insertion, whose result is independent of + /// the order in which elements were introduced. + pub fn freeze(mut self) -> PortConnectivity where TS : crate::PartialOrder { + self.entries.sort_unstable_by_key(|(port, _)| *port); + let mut entries: Vec<(usize, Antichain)> = Vec::with_capacity(self.entries.len()); + for (port, summary) in self.entries { + match entries.last_mut() { + Some((last, antichain)) if *last == port => { + for element in summary { antichain.insert(element); } + } + _ => { entries.push((port, summary)); } + } } + PortConnectivity { entries } + } +} + +impl FromIterator<(usize, Antichain)> for PortConnectivityBuilder { + fn from_iter(iter: T) -> Self where T: IntoIterator)> { + Self { entries: iter.into_iter().filter(|(_,p)| !p.is_empty()).collect() } } +} + +/// Internal connectivity from one port to any number of opposing ports. +/// +/// Always in canonical form: ports sorted and distinct, antichains non-empty. +/// Values are constructed by `PortConnectivityBuilder::freeze` (or collected from +/// an iterator), and offer no mutation. +#[derive(serde::Serialize, serde::Deserialize, columnar::Columnar, Debug, Clone, Eq, PartialEq)] +pub struct PortConnectivity { + /// Pairs of port and path summary antichain, sorted by distinct port. + entries: Vec<(usize, Antichain)>, +} + +impl Default for PortConnectivity { + fn default() -> Self { + Self { entries: Vec::new() } + } +} + +impl PortConnectivity { /// Borrowing iterator of port identifiers and antichains. pub fn iter_ports(&self) -> impl Iterator)> { - self.tree.iter().map(|(o,p)| (*o, p)) + self.entries.iter().map(|(o,p)| (*o, p)) } /// Returns the associated path summary, if it exists. pub fn get(&self, index: usize) -> Option<&Antichain> { - self.tree.get(&index) + self.entries + .binary_search_by_key(&index, |(port, _)| *port) + .ok() + .map(|position| &self.entries[position].1) } } -impl FromIterator<(usize, Antichain)> for PortConnectivity { +impl FromIterator<(usize, Antichain)> for PortConnectivity { fn from_iter(iter: T) -> Self where T: IntoIterator)> { - Self { tree: iter.into_iter().filter(|(_,p)| !p.is_empty()).collect() } + iter.into_iter().collect::>().freeze() } } diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs index 93e37668b..8cea273ef 100644 --- a/timely/src/progress/reachability.rs +++ b/timely/src/progress/reachability.rs @@ -74,7 +74,7 @@ //! assert_eq!(results[2], ((Location::new_target(2, 0), 17), -1)); //! ``` -use std::collections::{BinaryHeap, HashMap, VecDeque}; +use std::collections::BinaryHeap; use std::cmp::Reverse; use columnar::{Vecs, Index as ColumnarIndex}; @@ -83,8 +83,8 @@ use crate::progress::Timestamp; use crate::progress::{Source, Target}; use crate::progress::ChangeBatch; use crate::progress::{Location, Port}; -use crate::progress::operate::{Connectivity, PortConnectivity}; -use crate::progress::frontier::MutableAntichain; +use crate::progress::operate::{Connectivity, PortConnectivity, PortConnectivityBuilder}; +use crate::progress::frontier::{Antichain, MutableAntichain}; use crate::progress::timestamp::PathSummary; /// Build a `Vecs>>` from nested iterators. @@ -282,31 +282,40 @@ impl Builder { /// ``` pub fn is_acyclic(&self) -> bool { - let locations = self.shape.iter().map(|(targets, sources)| targets + sources).sum(); - let mut in_degree = HashMap::with_capacity(locations); + // Dense per-location in-degree counts, with each node's targets and + // then sources laid out contiguously at a per-node offset. + let mut offsets = Vec::with_capacity(self.shape.len()); + let mut locations = 0; + for (targets, sources) in self.shape.iter() { + offsets.push(locations); + locations += targets + sources; + } + let index_of = |location: &Location| { + let (targets, _) = self.shape[location.node]; + match location.port { + Port::Target(port) => offsets[location.node] + port, + Port::Source(port) => offsets[location.node] + targets + port, + } + }; + let mut in_degree = vec![0usize; locations]; // Load edges as default summaries. - for (index, ports) in self.edges.iter().enumerate() { - for (output, targets) in ports.iter().enumerate() { - let source = Location::new_source(index, output); - in_degree.entry(source).or_insert(0); + for ports in self.edges.iter() { + for targets in ports.iter() { for &target in targets.iter() { - let target = Location::from(target); - *in_degree.entry(target).or_insert(0) += 1; + in_degree[index_of(&Location::from(target))] += 1; } } } // Load default intra-node summaries. for (index, summary) in self.nodes.iter().enumerate() { - for (input, outputs) in summary.iter().enumerate() { - let target = Location::new_target(index, input); - in_degree.entry(target).or_insert(0); + for outputs in summary.iter() { for (output, summaries) in outputs.iter_ports() { let source = Location::new_source(index, output); for summary in summaries.elements().iter() { if summary == &Default::default() { - *in_degree.entry(source).or_insert(0) += 1; + in_degree[index_of(&source)] += 1; } } } @@ -314,16 +323,21 @@ impl Builder { } // A worklist of nodes that cannot be reached from the whole graph. - // Initially this list contains observed locations with no incoming - // edges, but as the algorithm develops we add to it any locations - // that can only be reached by nodes that have been on this list. - let mut worklist = Vec::with_capacity(in_degree.len()); - for (key, val) in in_degree.iter() { - if *val == 0 { - worklist.push(*key); + // Initially this list contains locations with no incoming edges, but + // as the algorithm develops we add to it any locations that can only + // be reached by nodes that have been on this list. + let mut remaining = in_degree.iter().filter(|count| **count > 0).count(); + let mut worklist = Vec::with_capacity(locations); + for (node, &(targets, sources)) in self.shape.iter().enumerate() { + for port in 0 .. targets { + let location = Location::new_target(node, port); + if in_degree[index_of(&location)] == 0 { worklist.push(location); } + } + for port in 0 .. sources { + let location = Location::new_source(node, port); + if in_degree[index_of(&location)] == 0 { worklist.push(location); } } } - in_degree.retain(|_key, val| val != &0); // Repeatedly remove nodes and update adjacent in-edges. while let Some(Location { node, port }) = worklist.pop() { @@ -331,9 +345,10 @@ impl Builder { Port::Source(port) => { for target in self.edges[node][port].iter() { let target = Location::from(*target); - *in_degree.get_mut(&target).unwrap() -= 1; - if in_degree[&target] == 0 { - in_degree.remove(&target); + let index = index_of(&target); + in_degree[index] -= 1; + if in_degree[index] == 0 { + remaining -= 1; worklist.push(target); } } @@ -341,11 +356,12 @@ impl Builder { Port::Target(port) => { for (output, summaries) in self.nodes[node][port].iter_ports() { let source = Location::new_source(node, output); + let index = index_of(&source); for summary in summaries.elements().iter() { if summary == &Default::default() { - *in_degree.get_mut(&source).unwrap() -= 1; - if in_degree[&source] == 0 { - in_degree.remove(&source); + in_degree[index] -= 1; + if in_degree[index] == 0 { + remaining -= 1; worklist.push(source); } } @@ -355,8 +371,8 @@ impl Builder { } } - // Acyclic graphs should reduce to empty collections. - in_degree.is_empty() + // Acyclic graphs should drain every positive in-degree to zero. + remaining == 0 } } @@ -777,94 +793,195 @@ impl Tracker { } } +/// 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 +/// representation of the length, read from the high bit down, gives the +/// sizes of the sorted runs in order. Adjacent runs may in fact be parts +/// of larger sorted runs, but we make no attempt to claim those wins. +/// +/// Keys are distinct across all runs. Novel keys are introduced by +/// re-sorting the suffix whose run structure their addition changes, as +/// in binary addition. Each element is re-sorted at most logarithmically +/// often (so `O(n log^2 n)` comparisons in total, a log factor more than +/// merging would cost, for much less code), and lookups visit at most +/// logarithmically many runs. +struct BinaryRuns { entries: Vec<(K, V)> } + +impl Default for BinaryRuns { + fn default() -> Self { Self { entries: Vec::new() } } +} + +impl BinaryRuns { + /// A mutable reference to the value at `key`, if present. + fn get_mut(&mut self, key: &K) -> Option<&mut V> { + let mut position = None; + let mut offset = 0; + for bit in (0..usize::BITS).rev() { + let size = 1usize << bit; + if self.entries.len() & size != 0 { + let run = &self.entries[offset .. offset + size]; + if let Ok(index) = run.binary_search_by(|(k, _)| k.cmp(key)) { + position = Some(offset + index); + break; + } + offset += size; + } + } + position.map(|index| &mut self.entries[index].1) + } + + /// Introduces a batch of keys distinct from each other and from those present. + fn insert_batch(&mut self, batch: Vec<(K, V)>) { + if batch.is_empty() { return; } + let total = self.entries.len() + batch.len(); + // Runs at the leading bits on which the lengths agree are unaffected; mask + // away the highest differing bit (the xor is non-zero) and below. + let stable = total & !(usize::MAX >> (self.entries.len() ^ total).leading_zeros()); + self.entries.extend(batch); + self.entries[stable..].sort_unstable_by(|x, y| x.0.cmp(&y.0)); + } + + /// Merges all runs into one sorted vector. + fn into_sorted(mut self) -> Vec<(K, V)> { + self.entries.sort_unstable_by(|x, y| x.0.cmp(&y.0)); + self.entries + } +} + /// Determines summaries from locations to scope outputs. /// /// Specifically, for each location whose node identifier is non-zero, we compile /// the summaries along which they can reach each output. /// /// Graph locations may be missing from the output, in which case they have no -/// paths to scope outputs. +/// paths to scope outputs. The result is sorted by location. fn summarize_outputs( nodes: &[Connectivity], edges: &[Vec>], - ) -> HashMap> + ) -> Vec<(Location, PortConnectivity)> { // A reverse edge map, to allow us to walk back up the dataflow graph. - let mut reverse = HashMap::new(); + // Sorted by target location; each target should have at most one source. + let mut reverse_edges = Vec::new(); for (node, outputs) in edges.iter().enumerate() { for (output, targets) in outputs.iter().enumerate() { for target in targets.iter() { - reverse.insert( + reverse_edges.push(( Location::from(*target), Location { node, port: Port::Source(output) } - ); + )); } } } + reverse_edges.sort_unstable(); + reverse_edges.dedup(); // A reverse map from operator outputs to inputs, along their internal summaries. - let mut reverse_internal: HashMap<_, Vec<_>> = HashMap::new(); + // Sorted by source location, so that the entries for a location are contiguous. + let mut reverse_internal = Vec::new(); for (node, connectivity) in nodes.iter().enumerate() { for (input, outputs) in connectivity.iter().enumerate() { for (output, summary) in outputs.iter_ports() { - reverse_internal - .entry(Location::new_source(node, output)) - .or_default() - .push((input, summary)); + reverse_internal.push((Location::new_source(node, output), input, summary)); } } } + reverse_internal.sort_unstable_by(|x, y| (x.0, x.1).cmp(&(y.0, y.1))); - let mut results: HashMap> = HashMap::new(); - let mut worklist = VecDeque::<(Location, usize, T::Summary)>::new(); + // Accumulated summaries to scope outputs, keyed by `(location, output)`. + let mut accumulated: BinaryRuns<(Location, usize), Antichain> = BinaryRuns::default(); - let outputs = + // Round-based (semi-naive) fixed point. Each round walks reverse edges and reverse + // internal summaries from the triples that changed last round, and the proposals + // that improve the accumulated antichains form the next round's work. + // The scope may have no outputs, in which case we can do no work. + let mut todo: Vec<(Location, usize, T::Summary)> = edges .iter() .flat_map(|x| x.iter()) .flat_map(|x| x.iter()) - .filter(|target| target.node == 0); + .filter(|target| target.node == 0) + .map(|target| (Location::from(*target), target.port, Default::default())) + .collect(); - // The scope may have no outputs, in which case we can do no work. - for output_target in outputs { - worklist.push_back((Location::from(*output_target), output_target.port, Default::default())); - } + let mut proposals: Vec<((Location, usize), T::Summary)> = Vec::new(); // Loop until we stop discovering novel reachability paths. - while let Some((location, output, summary)) = worklist.pop_front() { - match location.port { - - // This is an output port of an operator, or a scope input. - // We want to crawl up the operator, to its inputs. - Port::Source(_output_port) => { - if let Some(inputs) = reverse_internal.get(&location) { - for (input_port, operator_summary) in inputs.iter() { + while !todo.is_empty() { + + // Collect proposed summaries from the triples changed last round. + for (location, output, summary) in todo.drain(..) { + match location.port { + + // This is an output port of an operator, or a scope input. + // We want to crawl up the operator, to its inputs. + Port::Source(_output_port) => { + let start = reverse_internal.partition_point(|(source, _, _)| *source < location); + let inputs = reverse_internal[start..].iter().take_while(|(source, _, _)| *source == location); + for (_, input_port, operator_summary) in inputs { let new_location = Location::new_target(location.node, *input_port); for op_summary in operator_summary.elements().iter() { if let Some(combined) = op_summary.followed_by(&summary) { - if results.entry(new_location).or_default().insert_ref(output, &combined) { - worklist.push_back((new_location, output, combined)); - } + proposals.push(((new_location, output), combined)); } } } } + + // This is an input port of an operator, or a scope output. + // We want to walk back the (unique) edge leading to it. + Port::Target(_port) => { + if let Ok(index) = reverse_edges.binary_search_by_key(&location, |(target, _)| *target) { + proposals.push(((reverse_edges[index].1, output), summary)); + } + } } + } - // This is an input port of an operator, or a scope output. - // We want to walk back the edges leading to it. - Port::Target(_port) => { - // Each target should have (at most) one source. - if let Some(&source) = reverse.get(&location) { - if results.entry(source).or_default().insert_ref(output, &summary) { - worklist.push_back((source, output, summary)); + // Merge the batch of proposals into the accumulated summaries. Proposals are + // first collapsed per key into an antichain, so that only elements novel to + // the accumulated antichain (in an order-independent sense) seed the next round. + proposals.sort_unstable_by(|x, y| x.0.cmp(&y.0)); + let mut fresh: Vec<((Location, usize), Antichain)> = Vec::new(); + let mut batch: Antichain = Antichain::new(); + let mut iter = proposals.drain(..).peekable(); + while let Some(((location, output), summary)) = iter.next() { + // Collapse this round's proposals for the key into one antichain. + batch.insert(summary); + while iter.peek().map(|(key, _)| *key == (location, output)).unwrap_or(false) { + batch.insert(iter.next().unwrap().1); + } + if let Some(antichain) = accumulated.get_mut(&(location, output)) { + for summary in batch.drain() { + if antichain.insert_ref(&summary) { + todo.push((location, output, summary)); } } - }, + } + else { + todo.extend(batch.elements().iter().map(|summary| (location, output, summary.clone()))); + fresh.push(((location, output), std::mem::take(&mut batch))); + } } + + // Introduce the novel keys. + accumulated.insert_batch(fresh); } - results + // Merge all runs into one sorted list, and group it by location. + let mut results: Vec<(Location, PortConnectivityBuilder)> = Vec::new(); + for ((location, output), antichain) in accumulated.into_sorted() { + match results.last_mut() { + Some((last, connectivity)) if *last == location => { connectivity.add_port(output, antichain); } + _ => { + let mut connectivity = PortConnectivityBuilder::default(); + connectivity.add_port(output, antichain); + results.push((location, connectivity)); + } + } + } + results.into_iter().map(|(location, builder)| (location, builder.freeze())).collect() } /// Logging types for reachability tracking events. diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs index 393223f53..d5d8ef81d 100644 --- a/timely/src/progress/subgraph.rs +++ b/timely/src/progress/subgraph.rs @@ -19,7 +19,7 @@ use crate::scheduling::activate::Activations; use crate::progress::frontier::{MutableAntichain, MutableAntichainFilter}; use crate::progress::{Timestamp, Operate, operate::SharedProgress}; use crate::progress::{Location, Port, Source, Target}; -use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivity}; +use crate::progress::operate::{FrontierInterest, Connectivity, PortConnectivity, PortConnectivityBuilder}; use crate::progress::ChangeBatch; use crate::progress::broadcast::Progcaster; use crate::progress::reachability; @@ -565,7 +565,7 @@ where // Note that we need to have `self.inputs()` elements in the summary // with each element containing `self.outputs()` antichains regardless // of how long `self.scope_summary` is - let mut internal_summary = vec![PortConnectivity::default(); self.inputs()]; + let mut internal_summary = vec![PortConnectivityBuilder::default(); self.inputs()]; for (input_idx, input) in self.scope_summary.iter().enumerate() { for (output_idx, output) in input.iter_ports() { for outer in output.elements().iter().cloned().map(TInner::summarize) { @@ -573,6 +573,7 @@ where } } } + let internal_summary: Connectivity<_> = internal_summary.into_iter().map(|b| b.freeze()).collect(); debug_assert_eq!( internal_summary.len(),