[bos]

Problem

Tracker::propagate_all drives a single worklist ordered by (time, location). A location whose frontier moves from t to t' is visited at t and again at t', and each visit propagates downstream, so the frontier typically steps through an intermediate state (often empty) and every downstream location repeats the work. Pending source and target changes, and the pushed changes, are ChangeBatch<(Location, T)>s, so each round also sorts all its changes by location and time. In programs that run many small rounds, this sorting and the repeated visits dominate.

Change

• When the tracker is built, compute the strongly connected components of the location graph (input to the outputs it can reach, output to the inputs it is connected to), with an iterative Tarjan, and number them in topological order.
• propagate_all visits components in that order. A location outside any cycle is visited once, after every change that can reach it: it applies all its changes to its implications in one update_iter and pushes the net frontier change directly. The locations of a component with a cycle are visited in order of time, within the component, as before.
• Pending changes are kept per location (LocationChanges, a ChangeBatch<T, 4> per location), so consolidation sorts a location's few times, never the round's changes by location and time.
• Tracker::pushed now returns &mut Vec<((Location, T), i64)>: net changes, consolidated per location, in no particular order. This is a public signature change; the in-tree caller (Subgraph) and the module docs are updated to drain(..).

Why results and progress semantics are unchanged

A location's implications is a MutableAntichain, which accumulates the same updates to the same counts in any order, so its frontier after propagate_all is the same. Because components are visited in topological order, every change that can reach a location outside a cycle has arrived before it is visited, so visiting it once is enough. Within a cyclic component the existing time-ordered argument applies unchanged: changes leaving the component are buffered for later components and cannot return. The pushed changes are consolidated to the net change of each location's frontier, as the ChangeBatch made them before. Subgraph applies them to each input's frontier, activates the operators whose inputs changed and considers the same shutdowns, and none of these depends on order. output_changes, total_counts and cap_counts are updated exactly as before in step 1. Only the work in between differs. The order of the tracker's logged source and target updates changes, but not their content.

Measurements

A new example, progress_rounds (below, not part of this PR), runs one input through a dataflow for many rounds, one record and advance_to(round + 1) per round, stepping until the probe passes. One worker, release build, pinned with taskset, perf stat on a c7a.8xlarge, medians of 3 runs:

Workload	master	this PR	with #827 as well
chain 100 20000 (region of 100 maps)	19.33 G instr, 1.71 s	16.88 G (−12.7%), 1.47 s (−15%)	16.17 G (−16.4%), 1.39 s
chain 10 100000	11.72 G, 1.23 s	11.29 G (−3.6%), 1.13 s (−8%)	11.03 G (−5.9%), 1.09 s
loop 20 20000 (iterate, 20 maps in the body)	18.30 G, 1.70 s	17.87 G (−2.3%), 1.68 s	17.43 G (−4.7%), 1.64 s

Run-to-run spread was under 0.05% in instructions and about 2% in time. The "with #827" times come from a second batch, where this PR alone measured 1.48 s, 1.11 s and 1.68 s. The loop gains little because most of its locations share one cycle, which is visited as before.

This came out of FlowLog, a Datalog engine on differential dataflow, where progress tracking was 76% of the samples and sorting ChangeBatch<(Location, T)> a third of them in a workload committing once a second (279 locations in the root, 29 in one cycle). With this change, #827 and a small change to ChangeBatch::compact (not proposed, as it costs instructions on the workloads above), engine CPU fell 32% there (instructions −31%), 15% on another, and about 1% where rounds are rare, with identical outputs.

Testing

cargo test --workspace --all-targets and cargo test --doc pass (234 tests), alone and stacked on #827. cargo clippy --workspace --all-targets adds no warnings (104, down from 108).

timely/examples/progress_rounds.rs

//! Many small rounds through one dataflow, to time progress tracking.
//!
//! `progress_rounds chain <length> <rounds>`: a region holding a chain of
//! `length` maps, with no cycle.
//! `progress_rounds loop <length> <rounds>`: an iterative scope whose loop
//! body is a chain of `length` maps; each record goes around three times.

use timely::dataflow::operators::{Input, Probe, Enter, Leave, Feedback, ConnectLoop, Concat};
use timely::dataflow::operators::vec::{Map, Filter};
use timely::order::Product;

fn main() {
    let mut args = std::env::args();
    args.next();
    let shape = args.next().unwrap();
    let length: usize = args.next().unwrap().parse().unwrap();
    let rounds: usize = args.next().unwrap().parse().unwrap();

    timely::execute_from_args(args, move |worker| {
        let timer = std::time::Instant::now();
        let (mut input, probe) = worker.dataflow::<usize, _, _>(|scope| {
            let (input, stream) = scope.new_input::<Vec<u64>>();
            let stream = match shape.as_str() {
                "chain" => scope.region(|inner| {
                    let mut stream = stream.enter(inner);
                    for _ in 0 .. length {
                        stream = stream.map(|x| x);
                    }
                    stream.leave(scope)
                }),
                "loop" => scope.iterative::<u64, _, _>(|inner| {
                    let (handle, cycle) = inner.feedback(Product::new(0, 1));
                    let mut body = stream.enter(inner).concat(cycle);
                    for _ in 0 .. length {
                        body = body.map(|x| x);
                    }
                    body.clone().filter(|x| *x > 0).map(|x| x - 1).connect_loop(handle);
                    body.leave(scope)
                }),
                other => panic!("unknown shape {other}"),
            };
            let (probe, _stream) = stream.probe();
            (input, probe)
        });

        for round in 0 .. rounds {
            input.send(3);
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
        println!("{:?}\t{} rounds of {} x {}", timer.elapsed(), rounds, shape, length);
    }).unwrap();
}

[frankmcsherry]

Have had a read, and want to share some thoughts.

1. Seems smart. At least, the point is right that this can and should be done by SCC.
2. There are a bunch of other things to do around here, and it's the locus of complexity in the project, so a bit sensitive. It will be a process to adopt; seems worth doing, but will take a bit of free time.
3. From experience, it's less common to have multiple non-trivial SCCs in a graph. I'm not sure what FlowLog does here, but e.g. with Datalog and other iterative computations, it's often good for that layer to form the SCCs as regions, which will each iterate to convergence and be scheduled in topological order. There's often (not always) some value in letting the earlier components reach fixed point before starting the work that depends on them. Certainly with non-monotonic iteration this is the case (skip the intermediate states that may be retracted), but there is also some potential value for monotonic iteration as well (coalescing work, mostly; each produced fact is valid, it's just faster to do things together than in a trickle).

So a bit of TBD thinking for me to do about whether this is the layer to tidy up graphs that others have built. We've concluded "no" in the past, but if it's a thing that cannot be worked around then it becomes timely's responsibility. I'm not clear on the whole stack you have, and where it is ok to interpose, but that might be helpful to unstick you. For example, DD specifically handles fine grained update timestamps (the 20,000 and 100,000 I think?) as distinct from coarser timestamps it hands to timely, for the reason that the progress tracking pays in terms of timestamp granularity. It sounds like this could be a fix at the FlowLog layer, unless something is forcing a large number of distinct timely timestamps.

[bos-voxel]

Thanks for reading it closely. FlowLog builds against a vendored copy of timely with this patch applied, so there's no urgency on our side; take whatever time you need to think it through. If it would make review easier, I can split the PR into two:

1. Per-location pending changes, along with the pushed signature change.
2. Visiting locations in SCC order.

You asked whether this could be fixed in FlowLog instead. I don't think so. I'll share some context, then results of a quick experiment.

What the benchmark measures

In progress_rounds, the arguments 20,000 and 100,000 are the number of rounds. Each round sends one record and advances the input by one, so the run time is mostly the tracker's cost per round. It isn't meant to show how a program should choose timestamps. The chain-of-100-maps case (−15%) uses only timely, with no FlowLog code involved.

How FlowLog structures its dataflows

FlowLog already gives each recursive stratum its own iterative scope. The incident program's root scope has 279 locations, and 29 of them form a cycle. That cycle isn't a recursive Datalog stratum. It comes from a relation the program reads one event instant prior, implemented as a Variable whose feedback summary advances event time by one instant.

Why the root scope sees so many rounds

FlowLog doesn't create extra distinct timestamps. At 1 s commits, each commit takes about 3.1 rounds of progress tracking. What costs time is the work in each round: the tracker walks the root's worklist in (time, location) order, and sorts every pending (Location, T) change together.

Committing less often would reduce the number of rounds. At 60 s commits, this PR's gain on our workloads falls to about 1%, which agrees with your point about timestamp granularity. But 60 s is almost two orders of magnitude worse latency than 1 s. Our workloads are soft real-time, so latency is an operational problem for us. Longer commit intervals also cause other problems, such as memory pressure from the updates that build up between commits.

Putting the cycle in a region

To test your suggestion, I moved exactly that cycle (the variable, the three rules that close it, and their unions) into a region, with enter_region and leave_region at its two inputs and its one output. That ran on stock timely 0.31.0. The table shows commit CPU against stock timely, as medians over three build paths × three runs, pinned to cores, with byte-identical outputs:

	1 s commits	tick clock
cycle in a region	1.027	1.030
this PR	0.690	0.675

The region doesn't help, because most of the tracker's work is in the roughly 250 root locations outside the cycle. A region removes the cycle from the root's tracker, but those 250 locations remain, and they're the expensive part.

[frankmcsherry]

Thanks for the context.

One confusing bit, that Claude articulated better than I was able to, is

The FlowLog numbers don't fit that, unless the engine is mostly idle. They describe 279 locations, about 3.1 rounds per commit, and one commit per second. That is on the order of 1,000–2,000 location visits per second. Even with a generous per-visit cost, it is around a millisecond of CPU per second. For that to be 76% of samples, or for a constant-factor cut to remove 31% of engine instructions, the engine must be doing almost nothing else.

Our read is that these numbers seem like they might be about optimizing a largely idle system. When there is little data to put in the data plane, the fixed costs of the control plane dominate. This will also get worse with more workers, but .. from experience .. it vanishes (relatively) as you fill up the data plane. It's meant to be a fixed tax, not unlike spinning up a thread; useful, but only if you have enough work to pay against the overhead.

If the system is largely idle, it's expected that progress tracking will be the dominant cost. I think the other work we've discussed on avoiding scheduling operators and regions that hold no capabilities may be more impactful, in that case. That is the step that gives the order of magnitude reduction when parts of the dataflow are idle; whole blocks aren't even investigated, until data actually arrive.

Let me know if that is roughly accurate, that is whether the FlowLog times you are seeing spent in progress tracking are on a loaded system, or whether it's largely not busy. E.g. how many absolute milliseconds per second of non-progress work are you seeing each second (and, if large the absolute milliseconds per second of progress work). Optimizing the low utilization CPU is always good, but .. I'd lean into that other technique instead (working on it! :D).

[bos]

Fair question, and you’re mostly right.

Some context: this is a prototype running in a moderately realistic test environment. Those numbers come from replaying recorded input as fast as it’ll go, so “1 s commits” means one commit per second of event time, not wall time. The run I quoted is about 22,500 commits in 1.86 s of CPU.

Per commit that’s about 83 µs on stock timely: roughly 63 µs in progress tracking and 20 µs for everything else, with only a few records per round. With this PR it’s about 56 µs. So yes, the data plane is nearly empty each round, and this cuts the fixed tax. Run live at one commit a second, it would be close to idle.

A heavier program does about 4.7 ms of work per commit and still drops 15% at 1 s commits, falling to 1% at 60 s. Same picture: the tax amortises as rounds fill up.

Small rounds at 1 s latency are the regime we expect to run in, so this tax is what we hit. If that’s not a regime you want timely tuned for, fair enough. I’m happy to try the capability-skipping work when it’s ready and see how much of this it recovers.

One question about it: here every round advances the input frontier, so every location downstream of it moves. Does skipping regions that hold no capabilities avoid propagating those changes too, or only the operator schedules?

[bos]

One other bit of context: even though the main engine runs in real time with a low load, having the full-rate replays be as fast as possible is very beneficial, because replays are used for debugging, crash recovery, counterfactual analysis, and sundry other fun purposes where throughput becomes a bottleneck.