Track timely master: un-shred arrange with multi-capability stamps - #838
Merged
Conversation
Points the timely dependency at timely master, which stamps each message with a multiset of timestamps rather than exactly one. The adaptations are mechanical: Distributor implementations receive the stamp in place of a time and reproduce it on each produced sub-message, and capture events carry a stamp. All stamps remain singletons; behavior is unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh
frankmcsherry
force-pushed
the
stamps
branch
2 times, most recently
from
August 19, 2026 13:37
7a36999 to
f6b7ab6
Compare
The arrange operator retired capabilities one at a time, carving the batcher into per-capability tiles because each message could carry only one capability (its comment: 'Until timely dataflow supports multiple capabilities on messages, at least'). It now seals one batch per frontier advance and ships it under a CapabilitySet of the retiring capabilities; for totally ordered times the capability antichain has at most one element and behavior is unchanged. TraceReplayInstruction's capability hint becomes a Stamp (empty exactly for empty batches), and trace import replays batches under capability sets minted with delayed_stamp. Consumers accept multi-stamp batches: join retains the stamp and lower-bounds its unit's consolidation meet by the lattice meet of the stamp's elements (it must be the meet — any single element would be unsound for the history advance); reduce, count, threshold, and arrange's own input retain each stamp element. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh
The tactic contract required output batches to tile [lower, upper), one per held capability time, because each message could carry only one capability. With stamped messages the tiling is unnecessary: retire now returns a single optional batch spanning the interval, which the driver ships stamped with the held times not in advance of the upper limit — precisely the times that justify its contents. Everything that existed per tile becomes single: the cursor tactics keep one output buffer and one builder in place of one per held time, and their thinkers append to a plain vector rather than routing each update through a reverse search for its covering capability; the proxy tactic accumulates one delta buffer per window rather than one per held time, and the tile_descriptions machinery (descriptions, held-time tags, and the tile_of routing map) is deleted. The ProxyReduceBackend lifecycle keeps its shape — begin, then windows of corrections and emissions, then finish — but begin now receives its one description owned rather than borrowing a list it must clone, emit loses its tile index, and finish yields the batch rather than a vector of them. On scc_bench with one worker (spreads under 1%): 100 nodes / 200 edges / 2,000 concurrently open rounds runs in 0.88s against 4.06s before (4.6x); 20 / 40 / 20,000 rounds in 14.2s against 20.6s (1.45x). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Tracks timely master, whose messages are stamped by multisets of timestamps (TimelyDataflow/timely-dataflow#813), and removes the shredding of arranged and reduced batches into per-capability tiles.
Three commits:
master-next) plus mechanical adaptations:Distributorimplementations receive the stamp in place of a time and reproduce it on each produced sub-message, and capture events carry a stamp. All stamps remain singletons; behavior is unchanged.batcher.seal(frontier)shipped under aCapabilitySetof the retiring capabilities. For totally ordered times the capability antichain has at most one element and behavior is identical.TraceReplayInstruction's capability hint becomes aStamp(empty exactly for empty batches); trace import replays underdelayed_stamp. Consumers accept multi-stamp batches: join lower-bounds each unit's consolidation meet by the lattice meet of the stamp's elements (it must be the meet — any single element would be unsound for the history advance); reduce, count, threshold, and arrange's own input retain each stamp element.retirereturns a single optional batch spanning[lower, upper), shipped stamped with the held times not in advance of the upper limit, which justify its contents. Everything per-tile becomes single-step: one output buffer and builder in the cursor tactics, no per-update reverse routing search in the thinkers, one delta buffer per window in the proxy tactic, and thetile_descriptionsmachinery is deleted.ProxyReduceBackendkeeps itsbegin/emit/finishlifecycle, butbegintakes its one description owned rather than borrowing a list it must clone,emitloses its tile index, andfinishyields the batch rather than a vector of them.Measured on an SCC benchmark with all update rounds concurrently open (single worker, spreads under 1%): 100 nodes / 200 edges / 2,000 rounds runs in 0.88s against 4.06s before the reduce fusion (4.6x); 20 / 40 / 20,000 rounds in 14.2s against 20.6s (1.45x). A separate benchmark probed whether mixed stamps — say {(0, 100), (1, 0)}, whose lattice meet (0, 0) forgoes the consolidation a batch rooted at (0, 100) would seed — cost join anything: no difference against per-capability tiles, to within noise, at tens of millions of updates; the thinker's in-sweep per-key meet advancement recovers the consolidation regardless of the unit's initial meet.
🤖 Generated with Claude Code
https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh