[dbsp] Publish accumulated outputs atomically across workers#6589
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
Elegant, minimally intrusive fix. A Mutex-backed per-host cohort barrier avoids busy-waiting and only lets the completing worker publish, so there is no thundering herd on the mailboxes. Design and tests look good; a few small notes inline.
| _marker: PhantomData<T>, | ||
| } | ||
|
|
||
| unsafe impl<T> Sync for CohortId<T> {} |
There was a problem hiding this comment.
Nit: unsafe impl Sync deserves a // SAFETY: note — same fix would apply to the existing OutputId<T> above. Something like: // SAFETY: CohortId only stores a usize and a PhantomData<T>; it never owns or exposes a T, so it is Sync regardless of T.
There was a problem hiding this comment.
I think that this unsafe impl could be eliminated by declaring _marker as:
_marker: PhantomData<fn() -> T>
based on https://doc.rust-lang.org/nomicon/phantom-data.html#table-of-phantomdata-patterns
|
|
||
| // The upstream accumulator emits exactly once per transaction per | ||
| // worker, so the slot must be free. | ||
| debug_assert!(pending[worker].is_none()); |
There was a problem hiding this comment.
Consider assert! instead of debug_assert!. If the upstream accumulator's "once per transaction per worker" invariant is ever violated (say by a future refactor, a transaction abort/rollback leaving a stale slot behind, or a checkpoint-restore race), a release build would silently overwrite the previous value, publish a mixed cohort, and hand connectors like Delta Lake a corrupted transaction. Cheap enough to keep on in release. (Same spirit as NASA Power of Ten rule 5 for shared-state invariants.)
| /// | ||
| /// Slot `w` stores worker `w`'s output until every worker on this host has | ||
| /// produced its output for the transaction. | ||
| struct CohortSlots<T>(Arc<Mutex<Vec<Option<T>>>>); |
There was a problem hiding this comment.
Related to the debug_assert above: what happens to a partially-filled CohortSlots if a transaction is aborted / rolled back before every worker has emitted, or if the runtime restarts mid-commit and a checkpoint is restored? The next transaction's put on the stale worker would fire the debug_assert (release: silent overwrite). If the accumulator machinery guarantees that can't happen, a one-liner in the CohortSlots doc noting the assumption would help future readers; if it can happen, we probably want a reset hook (on abort / on restore).
| /// host and publishes the whole cohort to the mailboxes when the last | ||
| /// worker's output arrives, so a reader sees either all of a transaction's | ||
| /// outputs or none. | ||
| pub struct AccumulateOutput<B> |
There was a problem hiding this comment.
The docstring is clear that this is a per-host barrier. Worth explicitly noting the multi-host implication: with N hosts, each host publishes its own cohort into its own local mailboxes, and cross-host atomicity for a transaction still depends on the controller aggregating per-host results at commit time (which is presumably where the "alternative approach" you mentioned in the PR description would kick in). Just so the next reader doesn't assume this fix already covers the distributed case.
There was a problem hiding this comment.
@ryzhyk Fred prompted me to consider the multihost case. I don't understand how this works in that case. The CohortSlots object is necessarily per-host, but each instance of it has a slot for every worker overall. I think only the slots for the current host will ever get filled, so that this will hang on multihost.
Maybe I misunderstand, or maybe you have tested with multiple hosts, so I will approve this.
There was a problem hiding this comment.
Thanks for spotting this. Should be fixed now.
| /// commit; `DBSPHandle::transaction` would only return after the commit | ||
| /// completed, when the cohort is trivially complete. | ||
| #[test] | ||
| fn test_accumulate_output() { |
There was a problem hiding this comment.
Nice that the test drives the manual start_transaction / start_commit_transaction / step API — that's the only way to catch a partial-cohort leak. One thing this doesn't exercise is workers emitting in different commit steps (which is the entire scenario the fix is aimed at); with a small add_input_zset it likely all completes in a single step. A follow-up test that forces multi-step commits with staggered per-worker emission (via a slow operator or an artificial streaming-exchange delay) would give the barrier a real workout. Not a blocker.
| _marker: PhantomData<T>, | ||
| } | ||
|
|
||
| unsafe impl<T> Sync for CohortId<T> {} |
There was a problem hiding this comment.
I think that this unsafe impl could be eliminated by declaring _marker as:
_marker: PhantomData<fn() -> T>
based on https://doc.rust-lang.org/nomicon/phantom-data.html#table-of-phantomdata-patterns
| /// host and publishes the whole cohort to the mailboxes when the last | ||
| /// worker's output arrives, so a reader sees either all of a transaction's | ||
| /// outputs or none. | ||
| pub struct AccumulateOutput<B> |
There was a problem hiding this comment.
@ryzhyk Fred prompted me to consider the multihost case. I don't understand how this works in that case. The CohortSlots object is necessarily per-host, but each instance of it has a slot for every worker overall. I think only the slots for the current host will ever get filled, so that this will hang on multihost.
Maybe I misunderstand, or maybe you have tested with multiple hosts, so I will approve this.
|
I removed it from the queue because I wanted thoughts on #6589 (comment) before queuing. |
The AccumulateOutput operator produces a single output per transaction per worker. The idea was that this would generate a single output batch per transaction for each output connector, giving the connector an opportunity to commit these changes in a single transaction to the data sink (e.g., the Delta Lake connector does that). However different workers can produce their per-transaction outputs in different steps, resulting in multiple non-empty outputs from the circuit. This was unlikely to happen before the introduction of streaming exchange, since the upstream `shard` operator contains a barrier that synchronizes transaction completion across all threads. This commit introduces a local (per-host) barrier that synchronizes transaction output across all workers. There is an alternative approach: allow workers to produce outputs at any step, but only handle them in the controller when the transaction commits. This may be an even cleaner solution. One advantage of my fix is that it allows views that are computed early to produce outputs before all other views are fully evaluated in the transaction. Some users rely on this behavior. We may want to revise this in the future, but for now I wanted to implement the least intrusive fix. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
c1b0532 to
5c4db8e
Compare
The AccumulateOutput operator produces a single output per transaction per worker. The idea was that this would generate a single output batch per transaction for each output connector, giving the connector an opportunity to commit these changes in a single transaction to the data sink (e.g., the Delta Lake connector does that). However different workers can produce their per-transaction outputs in different steps, resulting in multiple non-empty outputs from the circuit. This was unlikely to happen before the introduction of streaming exchange, since the upstream
shardoperator contains a barrier that synchronizes transaction completion across all threads.This commit introduces a local (per-host) barrier that synchronizes transaction output across all workers.
There is an alternative approach: allow workers to produce outputs at any step, but only handle them in the controller when the transaction commits. This may be an even cleaner solution. One advantage of my fix is that it allows views that are computed early to produce outputs before all other views are fully evaluated in the transaction. Some users rely on this behavior. We may want to revise this in the future, but for now I wanted to implement the least intrusive fix.
Describe Manual Test Plan
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes