timely 0.29.0

//! Manages pointstamp reachability within a timely dataflow graph.
//!
//! Timely dataflow is concerned with understanding and communicating the potential
//! for capabilities to reach nodes in a directed graph, by following paths through
//! the graph (along edges and through nodes). This module contains one abstraction
//! for managing this information.
//!
//! # Examples
//!
//! ```rust
//! use timely::progress::{Location, Port};
//! use timely::progress::frontier::Antichain;
//! use timely::progress::{Source, Target};
//! use timely::progress::reachability::{Builder, Tracker};
//!
//! // allocate a new empty topology builder.
//! let mut builder = Builder::<usize>::new();
//!
//! // Each node with one input connected to one output.
//! builder.add_node(0, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
//! builder.add_node(1, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
//! builder.add_node(2, 1, 1, vec![[(0, Antichain::from_elem(1))].into_iter().collect()]);
//!
//! // Connect nodes in sequence, looping around to the first from the last.
//! builder.add_edge(Source::new(0, 0), Target::new(1, 0));
//! builder.add_edge(Source::new(1, 0), Target::new(2, 0));
//! builder.add_edge(Source::new(2, 0), Target::new(0, 0));
//!
//! // Construct a reachability tracker.
//! let (mut tracker, _) = builder.build(None);
//!
//! // Introduce a pointstamp at the output of the first node.
//! tracker.update_source(Source::new(0, 0), 17, 1);
//!
//! // Propagate changes; until this call updates are simply buffered.
//! tracker.propagate_all();
//!
//! let mut results =
//! tracker
//!     .pushed()
//!     .0
//!     .drain()
//!     .filter(|((location, time), delta)| location.is_target())
//!     .collect::<Vec<_>>();
//!
//! results.sort();
//!
//! println!("{:?}", results);
//!
//! assert_eq!(results.len(), 3);
//! assert_eq!(results[0], ((Location::new_target(0, 0), 18), 1));
//! assert_eq!(results[1], ((Location::new_target(1, 0), 17), 1));
//! assert_eq!(results[2], ((Location::new_target(2, 0), 17), 1));
//!
//! // Introduce a pointstamp at the output of the first node.
//! tracker.update_source(Source::new(0, 0), 17, -1);
//!
//! // Propagate changes; until this call updates are simply buffered.
//! tracker.propagate_all();
//!
//! let mut results =
//! tracker
//!     .pushed()
//!     .0
//!     .drain()
//!     .filter(|((location, time), delta)| location.is_target())
//!     .collect::<Vec<_>>();
//!
//! results.sort();
//!
//! assert_eq!(results.len(), 3);
//! assert_eq!(results[0], ((Location::new_target(0, 0), 18), -1));
//! assert_eq!(results[1], ((Location::new_target(1, 0), 17), -1));
//! assert_eq!(results[2], ((Location::new_target(2, 0), 17), -1));
//! ```

use std::collections::{BinaryHeap, HashMap, VecDeque};
use std::cmp::Reverse;

use columnar::{Vecs, Index as ColumnarIndex};

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::timestamp::PathSummary;

/// Build a `Vecs<Vecs<Vec<S>>>` from nested iterators.
///
/// The outer iterator yields nodes, each node yields ports, each port yields data items.
fn build_nested_vecs<S>(nodes: impl Iterator<Item = impl Iterator<Item = impl Iterator<Item = S>>>) -> Vecs<Vecs<Vec<S>>> {
    let mut result: Vecs<Vecs<Vec<S>>> = Default::default();
    for node in nodes {
        for port in node {
            result.values.push_iter(port);
        }
        result.bounds.push(result.values.bounds.len() as u64);
    }
    result
}

/// A topology builder, which can summarize reachability along paths.
///
/// A `Builder` takes descriptions of the nodes and edges in a graph, and compiles
/// a static summary of the minimal actions a timestamp must endure going from any
/// input or output port to a destination input port.
///
/// A graph is provides as (i) several indexed nodes, each with some number of input
/// and output ports, and each with a summary of the internal paths connecting each
/// input to each output, and (ii) a set of edges connecting output ports to input
/// ports. Edges do not adjust timestamps; only nodes do this.
///
/// The resulting summary describes, for each origin port in the graph and destination
/// input port, a set of incomparable path summaries, each describing what happens to
/// a timestamp as it moves along the path. There may be multiple summaries for each
/// part of origin and destination due to the fact that the actions on timestamps may
/// not be totally ordered (e.g., "increment the timestamp" and "take the maximum of
/// the timestamp and seven").
///
/// # Examples
///
/// ```rust
/// use timely::progress::frontier::Antichain;
/// use timely::progress::{Source, Target};
/// use timely::progress::reachability::Builder;
///
/// // allocate a new empty topology builder.
/// let mut builder = Builder::<usize>::new();
///
/// // Each node with one input connected to one output.
/// builder.add_node(0, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
/// builder.add_node(1, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
/// builder.add_node(2, 1, 1, vec![[(0, Antichain::from_elem(1))].into_iter().collect()]);
///
/// // Connect nodes in sequence, looping around to the first from the last.
/// builder.add_edge(Source::new(0, 0), Target::new(1, 0));
/// builder.add_edge(Source::new(1, 0), Target::new(2, 0));
/// builder.add_edge(Source::new(2, 0), Target::new(0, 0));
///
/// // Summarize reachability information.
/// let (tracker, _) = builder.build(None);
/// ```
#[derive(Clone, Debug)]
pub struct Builder<T: Timestamp> {
    /// Internal connections within hosted operators.
    ///
    /// Indexed by operator index, then input port, then output port. This is the
    /// same format returned by `initialize`, as if we simply appended
    /// all of the summaries for the hosted nodes.
    pub nodes: Vec<Connectivity<T::Summary>>,
    /// Direct connections from sources to targets.
    ///
    /// Edges do not affect timestamps, so we only need to know the connectivity.
    /// Indexed by operator index then output port.
    pub edges: Vec<Vec<Vec<Target>>>,
    /// Numbers of inputs and outputs for each node.
    pub shape: Vec<(usize, usize)>,
}

impl<T: Timestamp> Builder<T> {

    /// Create a new empty topology builder.
    pub fn new() -> Self {
        Builder {
            nodes: Vec::new(),
            edges: Vec::new(),
            shape: Vec::new(),
        }
    }

    /// Add links internal to operators.
    ///
    /// This method overwrites any existing summary, instead of anything more sophisticated.
    pub fn add_node(&mut self, index: usize, inputs: usize, outputs: usize, summary: Connectivity<T::Summary>) {

        // Assert that all summaries exist.
        debug_assert_eq!(inputs, summary.len());
        debug_assert!(summary.iter().all(|os| os.iter_ports().all(|(o,_)| o < outputs)));

        while self.nodes.len() <= index {
            self.nodes.push(Vec::new());
            self.edges.push(Vec::new());
            self.shape.push((0, 0));
        }

        self.nodes[index] = summary;
        if self.edges[index].len() != outputs {
            self.edges[index] = vec![Vec::new(); outputs];
        }
        self.shape[index] = (inputs, outputs);
    }

    /// Add links between operators.
    ///
    /// This method does not check that the associated nodes and ports exist. References to
    /// missing nodes or ports are discovered in `build`.
    pub fn add_edge(&mut self, source: Source, target: Target) {

        // Assert that the edge is between existing ports.
        debug_assert!(source.port < self.shape[source.node].1);
        debug_assert!(target.port < self.shape[target.node].0);

        self.edges[source.node][source.port].push(target);
    }

    /// Compiles the current nodes and edges into immutable path summaries.
    ///
    /// This method has the opportunity to perform some error checking that the path summaries
    /// are valid, including references to undefined nodes and ports, as well as self-loops with
    /// default summaries (a serious liveness issue).
    ///
    /// The optional logger information is baked into the resulting tracker.
    pub fn build(self, logger: Option<logging::TrackerLogger<T>>) -> (Tracker<T>, Connectivity<T::Summary>) {

        if !self.is_acyclic() {
            println!("Cycle detected without timestamp increment");
            println!("{:?}", self);
        }

        Tracker::allocate_from(self, logger)
    }

    /// Tests whether the graph includes a cycle of default path summaries.
    ///
    /// Graphs containing cycles of default path summaries will most likely
    /// not work well with progress tracking, as a timestamp can result in
    /// itself. Such computations can still *run*, but one should not block
    /// on frontier information before yielding results, as you many never
    /// unblock.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use timely::progress::frontier::Antichain;
    /// use timely::progress::{Source, Target};
    /// use timely::progress::reachability::Builder;
    ///
    /// // allocate a new empty topology builder.
    /// let mut builder = Builder::<usize>::new();
    ///
    /// // Each node with one input connected to one output.
    /// builder.add_node(0, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
    /// builder.add_node(1, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
    /// builder.add_node(2, 1, 1, vec![[(0, Antichain::from_elem(0))].into_iter().collect()]);
    ///
    /// // Connect nodes in sequence, looping around to the first from the last.
    /// builder.add_edge(Source::new(0, 0), Target::new(1, 0));
    /// builder.add_edge(Source::new(1, 0), Target::new(2, 0));
    ///
    /// assert!(builder.is_acyclic());
    ///
    /// builder.add_edge(Source::new(2, 0), Target::new(0, 0));
    ///
    /// assert!(!builder.is_acyclic());
    /// ```
    ///
    /// This test exists because it is possible to describe dataflow graphs that
    /// do not contain non-incrementing cycles, but without feedback nodes that
    /// strictly increment timestamps. For example,
    ///
    /// ```rust
    /// use timely::progress::frontier::Antichain;
    /// use timely::progress::{Source, Target};
    /// use timely::progress::reachability::Builder;
    ///
    /// // allocate a new empty topology builder.
    /// let mut builder = Builder::<usize>::new();
    ///
    /// // Two inputs and outputs, only one of which advances.
    /// builder.add_node(0, 2, 2, vec![
    ///     [(0,Antichain::from_elem(0)),(1,Antichain::new())].into_iter().collect(),
    ///     [(0,Antichain::new()),(1,Antichain::from_elem(1))].into_iter().collect(),
    /// ]);
    ///
    /// // Connect each output to the opposite input.
    /// builder.add_edge(Source::new(0, 0), Target::new(0, 1));
    /// builder.add_edge(Source::new(0, 1), Target::new(0, 0));
    ///
    /// assert!(builder.is_acyclic());
    /// ```
    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);

        // 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 &target in targets.iter() {
                    let target = Location::from(target);
                    *in_degree.entry(target).or_insert(0) += 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 (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;
                        }
                    }
                }
            }
        }

        // 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);
            }
        }
        in_degree.retain(|_key, val| val != &0);

        // Repeatedly remove nodes and update adjacent in-edges.
        while let Some(Location { node, port }) = worklist.pop() {
            match port {
                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);
                            worklist.push(target);
                        }
                    }
                },
                Port::Target(port) => {
                    for (output, summaries) in self.nodes[node][port].iter_ports() {
                        let source = Location::new_source(node, output);
                        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);
                                    worklist.push(source);
                                }
                            }
                        }
                    }
                },
            }
        }

        // Acyclic graphs should reduce to empty collections.
        in_degree.is_empty()
    }
}

impl<T: Timestamp> Default for Builder<T> {
    fn default() -> Self {
        Self::new()
    }
}

/// An interactive tracker of propagated reachability information.
///
/// A `Tracker` tracks, for a fixed graph topology, the implications of
/// pointstamp changes at various node input and output ports. These changes may
/// alter the potential pointstamps that could arrive at downstream input ports.
pub struct Tracker<T:Timestamp> {

    /// Internal operator connectivity, columnar form of `Vec<Vec<PortConnectivity<T::Summary>>>`.
    /// Indexed by `(node, input_port)` to yield `(output_port, summary)` pairs.
    nodes: Vecs<Vecs<Vec<(usize, T::Summary)>>>,
    /// Edge connectivity, columnar form of `Vec<Vec<Vec<Target>>>`.
    /// Indexed by `(node, output_port)` to yield target slices.
    edges: Vecs<Vecs<Vec<Target>>>,

    /// Summaries from each target (operator input) to scope outputs.
    /// Indexed by `(node, target_port)` to yield `(scope_output, summary)` pairs.
    target_summaries: Vecs<Vecs<Vec<(usize, T::Summary)>>>,
    /// Summaries from each source (operator output) to scope outputs.
    /// Indexed by `(node, source_port)` to yield `(scope_output, summary)` pairs.
    source_summaries: Vecs<Vecs<Vec<(usize, T::Summary)>>>,

    /// Each source and target has a mutable antichain to ensure that we track their discrete frontiers,
    /// rather than their multiplicities. We separately track the frontiers resulting from propagated
    /// frontiers, to protect them from transient negativity in inbound target updates.
    per_operator: Vec<PerOperator<T>>,

    /// 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)>,

    /// Worklist of updates to perform, ordered by increasing timestamp and target.
    worklist: BinaryHeap<Reverse<(T, Location, i64)>>,

    /// Buffer of consequent changes.
    pushed_changes: ChangeBatch<(Location, T)>,

    /// Compiled summaries from each internal location (not scope inputs) to each scope output.
    output_changes: Vec<ChangeBatch<T>>,

    /// A non-negative sum of post-filtration input changes.
    ///
    /// This sum should be zero exactly when the accumulated input changes are zero,
    /// indicating that the progress tracker is currently tracking nothing. It should
    /// always be exactly equal to the sum across all operators of the frontier sizes
    /// of the target and source `pointstamps` member.
    total_counts: i64,

    /// Optionally, a unique logging identifier and logging for tracking events.
    logger: Option<logging::TrackerLogger<T>>,
}

/// Target and source information for each operator.
pub struct PerOperator<T: Timestamp> {
    /// Port information for each target.
    pub targets: Vec<PortInformation<T>>,
    /// Port information for each source.
    pub sources: Vec<PortInformation<T>>,
    /// Sum across outputs of capabilities.
    pub cap_counts: i64,
}

impl<T: Timestamp> PerOperator<T> {
    /// A new PerOperator bundle from numbers of input and output ports.
    pub fn new(inputs: usize, outputs: usize) -> Self {
        PerOperator {
            targets: vec![PortInformation::new(); inputs],
            sources: vec![PortInformation::new(); outputs],
            cap_counts: 0,
        }
    }
}

/// Per-port progress-tracking information.
#[derive(Clone)]
pub struct PortInformation<T: Timestamp> {
    /// Current counts of active pointstamps.
    pub pointstamps: MutableAntichain<T>,
    /// Current implications of active pointstamps across the dataflow.
    pub implications: MutableAntichain<T>,
}

impl<T: Timestamp> PortInformation<T> {
    /// Creates empty port information.
    pub fn new() -> Self {
        PortInformation {
            pointstamps: MutableAntichain::new(),
            implications: MutableAntichain::new(),
        }
    }

    /// Returns `true` if updates at this pointstamp uniquely block progress.
    ///
    /// This method returns `true` if the currently maintained pointstamp
    /// counts are such that zeroing out outstanding updates at *this*
    /// pointstamp would change the frontiers at this operator. When the
    /// method returns `false` it means that, temporarily at least, there
    /// are outstanding pointstamp updates that are strictly less than
    /// this pointstamp.
    #[inline]
    pub fn is_global(&self, time: &T) -> bool {
        let dominated = self.implications.frontier().iter().any(|t| t.less_than(time));
        let redundant = self.implications.count_for(time) > 1;
        !dominated && !redundant
    }
}

impl<T: Timestamp> Default for PortInformation<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T:Timestamp> Tracker<T> {

    /// Updates the count for a time at a location.
    #[inline]
    pub fn update(&mut self, location: Location, time: T, value: i64) {
        match location.port {
            Port::Target(port) => self.update_target(Target::new(location.node, port), time, value),
            Port::Source(port) => self.update_source(Source::new(location.node, port), time, value),
        };
    }

    /// 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);
    }
    /// 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);
    }

    /// 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.total_counts > 0
    }

    /// Allocate a new `Tracker` using the shape from `summaries`.
    ///
    /// The result is a pair of tracker, and the summaries from each input port to each
    /// output port.
    ///
    /// If the optional logger is provided, it will be used to log various tracker events.
    pub fn allocate_from(builder: Builder<T>, logger: Option<logging::TrackerLogger<T>>) -> (Self, Connectivity<T::Summary>) {

        // Allocate buffer space for each input and input port.
        let per_operator =
        builder
            .shape
            .iter()
            .map(|&(inputs, outputs)| PerOperator::new(inputs, outputs))
            .collect::<Vec<_>>();

        // Summary of scope inputs to scope outputs.
        let mut builder_summary = vec![PortConnectivity::default(); builder.shape[0].1];

        // Compile summaries from each location to each scope output.
        // Collect into per-node, per-port buckets for flattening.
        let output_summaries = summarize_outputs::<T>(&builder.nodes, &builder.edges);

        // Temporary storage: target_sum[node][port] and source_sum[node][port].
        let mut target_sum: Vec<Vec<PortConnectivity<T::Summary>>> = builder.shape.iter()
            .map(|&(inputs, _)| vec![PortConnectivity::default(); inputs])
            .collect();
        let mut source_sum: Vec<Vec<PortConnectivity<T::Summary>>> = builder.shape.iter()
            .map(|&(_, outputs)| vec![PortConnectivity::default(); outputs])
            .collect();

        for (location, summaries) in output_summaries.into_iter() {
            // Summaries from scope inputs are useful in summarizing the scope.
            if location.node == 0 {
                if let Port::Source(port) = location.port {
                    builder_summary[port] = summaries;
                }
                else {
                    // Ignore (ideally trivial) output to output summaries.
                }
            }
            // Summaries from internal nodes are important for projecting capabilities.
            else {
                match location.port {
                    Port::Target(port) => {
                        target_sum[location.node][port] = summaries;
                    },
                    Port::Source(port) => {
                        source_sum[location.node][port] = summaries;
                    },
                }
            }
        }

        // Build columnar nodes: Vecs<Vecs<Vec<(usize, T::Summary)>>>.
        let nodes = build_nested_vecs(builder.nodes.iter().map(|connectivity| {
            connectivity.iter().map(|port_conn| {
                port_conn.iter_ports().flat_map(|(port, antichain)| {
                    antichain.elements().iter().map(move |s| (port, s.clone()))
                })
            })
        }));

        // Build columnar edges: Vecs<Vecs<Vec<Target>>>.
        let edges = build_nested_vecs(builder.edges.iter().map(|node_edges| {
            node_edges.iter().map(|port_edges| port_edges.iter().cloned())
        }));

        // Build columnar target and source summaries.
        let target_summaries = build_nested_vecs(target_sum.iter().map(|ports| {
            ports.iter().map(|port_conn| {
                port_conn.iter_ports().flat_map(|(port, antichain)| {
                    antichain.elements().iter().map(move |s| (port, s.clone()))
                })
            })
        }));
        let source_summaries = build_nested_vecs(source_sum.iter().map(|ports| {
            ports.iter().map(|port_conn| {
                port_conn.iter_ports().flat_map(|(port, antichain)| {
                    antichain.elements().iter().map(move |s| (port, s.clone()))
                })
            })
        }));

        let scope_outputs = builder.shape[0].0;
        let output_changes = vec![ChangeBatch::new(); scope_outputs];

        let tracker =
        Tracker {
            nodes,
            edges,
            target_summaries,
            source_summaries,
            per_operator,
            target_changes: ChangeBatch::new(),
            source_changes: ChangeBatch::new(),
            worklist: BinaryHeap::new(),
            pushed_changes: ChangeBatch::new(),
            output_changes,
            total_counts: 0,
            logger,
        };

        (tracker, builder_summary)
    }

    /// Propagates all pending updates.
    ///
    /// The method drains `self.input_changes` and circulates their implications
    /// until we cease deriving new implications.
    pub fn propagate_all(&mut self) {

        // Step 0: If logging is enabled, construct and log inbound changes.
        if let Some(logger) = &mut self.logger {

            let target_changes =
            self.target_changes
                .iter()
                .map(|((target, time), diff)| (target.node, target.port, time, *diff));

            logger.log_target_updates(target_changes);

            let source_changes =
            self.source_changes
                .iter()
                .map(|((source, time), diff)| (source.node, source.port, time, *diff));

            logger.log_source_updates(source_changes);
        }

        // Step 1: Drain `self.input_changes` and determine actual frontier changes.
        //
        // Not all changes in `self.input_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);
                    }
                }
                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);
                    }
                }
                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;
            }

            // 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);
                        }
                    },
                };
            }
        }
    }

    /// Implications of maintained capabilities projected to each output.
    pub fn pushed_output(&mut self) -> &mut [ChangeBatch<T>] {
        &mut self.output_changes[..]
    }

    /// A mutable reference to the pushed results of changes.
    pub fn pushed(&mut self) -> (&mut ChangeBatch<(Location, T)>, &[PerOperator<T>]) {
        (&mut self.pushed_changes, &self.per_operator)
    }

    /// Reveals per-operator frontier state.
    pub fn node_state(&self, index: usize) -> &PerOperator<T> {
        &self.per_operator[index]
    }

    /// Indicates if pointstamp is in the scope-wide frontier.
    ///
    /// Such a pointstamp would, if removed from `self.pointstamps`, cause a change
    /// to `self.implications`, which is what we track for per operator input frontiers.
    /// If the above do not hold, then its removal either 1. shouldn't be possible,
    /// or 2. will not affect the output of `self.implications`.
    pub fn is_global(&self, location: Location, time: &T) -> bool {
        match location.port {
            Port::Target(port) => self.per_operator[location.node].targets[port].is_global(time),
            Port::Source(port) => self.per_operator[location.node].sources[port].is_global(time),
        }
    }
}

/// 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.
fn summarize_outputs<T: Timestamp>(
    nodes: &[Connectivity<T::Summary>],
    edges: &[Vec<Vec<Target>>],
    ) -> HashMap<Location, PortConnectivity<T::Summary>>
{
    // A reverse edge map, to allow us to walk back up the dataflow graph.
    let mut reverse = HashMap::new();
    for (node, outputs) in edges.iter().enumerate() {
        for (output, targets) in outputs.iter().enumerate() {
            for target in targets.iter() {
                reverse.insert(
                    Location::from(*target),
                    Location { node, port: Port::Source(output) }
                );
            }
        }
    }

    // A reverse map from operator outputs to inputs, along their internal summaries.
    let mut reverse_internal: HashMap<_, Vec<_>> = HashMap::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));
            }
        }
    }

    let mut results: HashMap<Location, PortConnectivity<T::Summary>> = HashMap::new();
    let mut worklist = VecDeque::<(Location, usize, T::Summary)>::new();

    let outputs =
    edges
        .iter()
        .flat_map(|x| x.iter())
        .flat_map(|x| x.iter())
        .filter(|target| target.node == 0);

    // 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()));
    }

    // 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() {
                        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));
                                }
                            }
                        }
                    }
                }
            }

            // 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));
                    }
                }
            },
        }
    }

    results
}

/// Logging types for reachability tracking events.
pub mod logging {
    use std::time::Duration;

    use timely_container::CapacityContainerBuilder;
    use timely_logging::TypedLogger;
    use crate::logging_core::Logger;

    /// A container builder for tracker events.
    pub type TrackerEventBuilder<T> = CapacityContainerBuilder<Vec<(Duration, TrackerEvent<T>)>>;

    /// A logger with additional identifying information about the tracker.
    pub struct TrackerLogger<T: Clone + 'static> {
        identifier: usize,
        logger: TypedLogger<TrackerEventBuilder<T>, TrackerEvent<T>>,
    }

    impl<T: Clone + 'static> TrackerLogger<T> {
        /// Create a new tracker logger from its fields.
        pub fn new(identifier: usize, logger: Logger<TrackerEventBuilder<T>>) -> Self {
            Self { identifier, logger: logger.into() }
        }

        /// Log source update events with additional identifying information.
        pub fn log_source_updates<'a, I>(&mut self, updates: I)
        where
            I: IntoIterator<Item = (usize, usize, &'a T, i64)>
        {
            let updates: Vec<_> = updates.into_iter().map(|(a,b,c,d)| (a,b,c.clone(),d)).collect();
            if !updates.is_empty() {
                self.logger.log({
                    SourceUpdate {
                        tracker_id: self.identifier,
                        updates
                    }
                });
            }
        }
        /// Log target update events with additional identifying information.
        pub fn log_target_updates<'a, I>(&mut self, updates: I)
        where
            I: IntoIterator<Item = (usize, usize, &'a T, i64)>
        {
            let updates: Vec<_> = updates.into_iter().map(|(a,b,c,d)| (a,b,c.clone(),d)).collect();
            if !updates.is_empty() {
                self.logger.log({
                    TargetUpdate {
                        tracker_id: self.identifier,
                        updates
                    }
                });
            }
        }
    }

    /// Events that the tracker may record.
    #[derive(Debug, Clone)]
    pub enum TrackerEvent<T> {
        /// Updates made at a source of data.
        SourceUpdate(SourceUpdate<T>),
        /// Updates made at a target of data.
        TargetUpdate(TargetUpdate<T>),
    }

    /// An update made at a source of data.
    #[derive(Debug, Clone)]
    pub struct SourceUpdate<T> {
        /// An identifier for the tracker.
        pub tracker_id: usize,
        /// Updates themselves, as `(node, port, time, diff)`.
        pub updates: Vec<(usize, usize, T, i64)>,
    }

    /// An update made at a target of data.
    #[derive(Debug, Clone)]
    pub struct TargetUpdate<T> {
        /// An identifier for the tracker.
        pub tracker_id: usize,
        /// Updates themselves, as `(node, port, time, diff)`.
        pub updates: Vec<(usize, usize, T, i64)>,
    }

    impl<T> From<SourceUpdate<T>> for TrackerEvent<T> {
        fn from(v: SourceUpdate<T>) -> TrackerEvent<T> { TrackerEvent::SourceUpdate(v) }
    }

    impl<T> From<TargetUpdate<T>> for TrackerEvent<T> {
        fn from(v: TargetUpdate<T>) -> TrackerEvent<T> { TrackerEvent::TargetUpdate(v) }
    }
}

// The Drop implementation for `Tracker` makes sure that reachability logging is correct for
// prematurely dropped dataflows. At the moment, this is only possible through `drop_dataflow`,
// because in all other cases the tracker stays alive while it has outstanding work, leaving no
// remaining work for this Drop implementation.
impl<T: Timestamp> Drop for Tracker<T> {
    fn drop(&mut self) {
        let logger = if let Some(logger) = &mut self.logger {
            logger
        } else {
            // No cleanup necessary when there is no logger.
            return;
        };

        // Retract pending data that `propagate_all` would normally log.
        for (index, per_operator) in self.per_operator.iter_mut().enumerate() {
            let target_changes = per_operator.targets
                .iter_mut()
                .enumerate()
                .flat_map(|(port, target)| {
                    target.pointstamps
                        .updates()
                        .map(move |(time, diff)| (index, port, time, -diff))
                });

            logger.log_target_updates(target_changes);

            let source_changes = per_operator.sources
                .iter_mut()
                .enumerate()
                .flat_map(|(port, source)| {
                    source.pointstamps
                        .updates()
                        .map(move |(time, diff)| (index, port, time, -diff))
                });

            logger.log_source_updates(source_changes);
        }
    }
}