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
frankmcsherry
pushed a commit
that referenced
this pull request
Aug 29, 2026
Port of prior work on a "server" for DDIR, in which various dataflows go through a create-run-delete lifecycle, concurrently. Rebased over master-next (#838-#849); the only conflict was Cargo.toml against #835's removal of the legacy server crates. Folds in three fixes that make the port work when scripted: - Reach the documented process exit on operator 'exit': execute_directly runs a trailing 'while has_dataflows { step_or_park }' loop, and the worker always still holds live dataflows (installed programs, generated sources, the diagnostics capture), so it parked forever and main's joins never reached their own process::exit. Exit where the shutdown decision lands, after a beat for session writers to flush the final 'ok bye'. - Skip '#' comment lines between commands, so annotated command scripts pipe straight to stdin. - README: mark auto-tick as demo furniture, not a design commitment (it makes observation advance time; a watcher-must-drive-time design may be better once the server has real tenants).
frankmcsherry
pushed a commit
that referenced
this pull request
Aug 29, 2026
Port of prior work on a "server" for DDIR, in which various dataflows go through a create-run-delete lifecycle, concurrently. Rebased over master-next (#838-#849); the only conflict was Cargo.toml against #835's removal of the legacy server crates. Folds in three fixes that make the port work when scripted: - Reach the documented process exit on operator 'exit': execute_directly runs a trailing 'while has_dataflows { step_or_park }' loop, and the worker always still holds live dataflows (installed programs, generated sources, the diagnostics capture), so it parked forever and main's joins never reached their own process::exit. Exit where the shutdown decision lands, after a beat for session writers to flush the final 'ok bye'. - Skip '#' comment lines between commands, so annotated command scripts pipe straight to stdin. - README: mark auto-tick as demo furniture, not a design commitment (it makes observation advance time; a watcher-must-drive-time design may be better once the server has real tenants).
frankmcsherry
pushed a commit
that referenced
this pull request
Aug 30, 2026
Port of prior work on a "server" for DDIR, in which various dataflows go through a create-run-delete lifecycle, concurrently. Rebased over master-next (#838-#849); the only conflict was Cargo.toml against #835's removal of the legacy server crates. Folds in three fixes that make the port work when scripted: - Reach the documented process exit on operator 'exit': execute_directly runs a trailing 'while has_dataflows { step_or_park }' loop, and the worker always still holds live dataflows (installed programs, generated sources, the diagnostics capture), so it parked forever and main's joins never reached their own process::exit. Exit where the shutdown decision lands, after a beat for session writers to flush the final 'ok bye'. - Skip '#' comment lines between commands, so annotated command scripts pipe straight to stdin. - README: mark auto-tick as demo furniture, not a design commitment (it makes observation advance time; a watcher-must-drive-time design may be better once the server has real tenants).
frankmcsherry
added a commit
that referenced
this pull request
Aug 30, 2026
* Port live DDIR server to current interpreter Port of prior work on a "server" for DDIR, in which various dataflows go through a create-run-delete lifecycle, concurrently. Rebased over master-next (#838-#849); the only conflict was Cargo.toml against #835's removal of the legacy server crates. Folds in three fixes that make the port work when scripted: - Reach the documented process exit on operator 'exit': execute_directly runs a trailing 'while has_dataflows { step_or_park }' loop, and the worker always still holds live dataflows (installed programs, generated sources, the diagnostics capture), so it parked forever and main's joins never reached their own process::exit. Exit where the shutdown decision lands, after a beat for session writers to flush the final 'ok bye'. - Skip '#' comment lines between commands, so annotated command scripts pipe straight to stdin. - README: mark auto-tick as demo furniture, not a design commitment (it makes observation advance time; a watcher-must-drive-time design may be better once the server has real tenants). * ddir-server: a write path -- feed, a size gate, and demos feed <prog> <in#> <key> [val=] [time=] [diff=] pushes one update into a loaded program's positional input, matching the ddir_server example's value grammar, parsed panic-safe on the session thread. The stance on contention: writes are open; policy lives in the dataflow. Cooperating clients include their id and an ordering epoch in the data, and programs resolve races over those facts deterministically. Identity is convention, not enforcement -- adversarial hardening is deliberately deferred. One intake gate: DDIR_MAX_PROGRAM_BYTES caps the load body before parsing (oversized bodies are swallowed and reported once; the parser stays in sync). Transport self-defense, not semantics. Demos: claims.txt (first-claim-wins as min over (epoch, id)); txn.txt (transaction processing in the data plane, after the 2025-04-27 blog post: intents as blind appends carrying the reads under which writes evaporate, a recursive non-monotone view resolving commits, and episodic OUTPUT-INVARIANT maintenance -- the demo retracts 13 of 15 intent rows and committed state is unchanged); two_sessions.py (TCP smoke: racing convention-stamped claims settle deterministically; the size gate fires; clean shutdown). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * ddir-server: bind -- export->input feedback, one epoch per tick bind <trace> <prog> <in#> delivers the trace's changes into the input at each tick, so the input mirrors the trace one epoch delayed. This is the write path for installed programs: a dataflow can act on the world -- or on itself -- with no client in the loop, one well-founded recursion step per tick. Rationale: the language is deliberately timeless (programs cannot observe the outer timestamp, and exports erase inner iteration coordinates), so all temporality lives in the server, and there are exactly two temporal operations: advance (tick) and unit delay (bind). Cross-program cycles cannot iterate within an epoch through the trace interface; the delay is also what keeps cycles among independently authored programs well-founded without joint monotonicity. Implemented in interactive::server::Server so the multi-worker example server gains bind/unbind too (each worker's tap sees its shard; the union delivers the delta exactly once). Lifecycle: a bound source holds an importer and cannot be dropped; nor can the bound target (unbind first); tick waits on tap probes before draining, so buffers hold every change through the just-closed epoch. The state-machine idiom (demo/counter.txt, tests/server_bind.rs): give a program a seed input and a feedback input, 'let state = seed + feedback', and bind 'f(state) + (seed | negate)' into the feedback input; then state(t) = f(state(t-1)) while later seed feeds inject as perturbations. Deliberately NOT for compaction/maintenance: a continuously-bound compactor oscillates (deleting a row un-derives its own deadness); maintenance is episodic peek+feed, per demo/txn.txt. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Frank McSherry <mcsherry@mini-00.tail4dbe6b.ts.net> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
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