persist: deasync shard_source operators#37751
Draft
antiguru wants to merge 4 commits into
Draft
Conversation
Convert `shard_source_descs` and `shard_source_fetch` from `AsyncOperatorBuilder` to synchronous `OperatorBuilderRc` operators, each paired with a tokio task that owns the persist I/O. * `shard_source_descs` runs a listen task on the chosen worker that owns the reader, snapshot, listen loop, and stats filtering, and sends parts (split into `ExchangeableBatchPart` + `Lease`) plus progress over a channel. The former `shard_source_descs_return` operator is merged in as a disconnected `completed_fetches` input; the listen handle (the reader's SeqNo hold) is released via a oneshot once that frontier empties, preserving the old lease lifetime. * `shard_source_fetch` forwards descs to a fetch task and retains a per-flight capability pair keyed by time; results are matched by time and emitted at the data capability, with the completed-fetches capability dropped to release the lease only after the download completes. On a missing blob it reports through the `ErrorHandler` and freezes, so a later good result cannot advance the frontier past the missing part. * Both operators reproduce `builder_async`'s two-phase shutdown (`build_reschedule` + the coordinated button), so a local-only press cannot advance the downstream frontier past times other workers' instances still feed. * `filter_fn`, `listen_sleep`, and `start_signal` gain `Send` bounds; all callers already satisfy them. Preserves the two features that landed on the async operators after this work was first written: the fetch task runs up to `persist_source_fetch_concurrency` fetches at once via a `FuturesUnordered` of cloned `BatchFetcher`s (the time-keyed operator bookkeeping tolerates out-of-order completion), and the listen task coalesces hydration-time frontier downgrades under `persist_source_hydration_frontier_coalesce_bytes` while parts keep flowing per batch. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N1erbcNjoki3HsfBJCTFUd
The stress test parked its read workers indefinitely between steps, relying on the async persist source operators' listen retries as accidental wakeups. With the listen loop in a tokio task, a caught-up source with no retry timer produces no further activations, so the parks are now bounded. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N1erbcNjoki3HsfBJCTFUd
The lease-return operator is now a disconnected input on the merged `shard_source_descs` operator, so the introspection relation reports that name. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N1erbcNjoki3HsfBJCTFUd
The `shard_source_fetch` task drives its downloads with a `FuturesUnordered`, which only advances the futures inside it while it is being polled. The task forwarded each completed download with `blob_tx.send().await`, so once the bounded result channel filled the task blocked in `send` and stopped polling `in_flight`. That froze every other in-flight fetch, collapsing effective fetch concurrency far below `persist_source_fetch_concurrency` (the channel capacity became the real ceiling) and leaving persist reads slow while CPU and network sat idle. Park completed downloads in a `ready` queue and forward them with a reserved permit instead. When the channel is full the reserve arm stays pending rather than blocking the loop, so `in_flight` keeps being polled and the downloads overlap. Resident parts are bounded by `max_concurrency` (we stop starting new fetches once `in_flight.len() + ready.len()` reaches the cap), which replaces the channel capacity as the memory bound. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ALAjT4DRJhPf99SwXgpQdz
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.
Motivation
The persist source operators are built on the timely async bridge, which we are migrating away from (cf. #36810). This converts them to synchronous operators with no behavior change.
Rebase/re-derivation of #36910 on current main. Folds back in the two features that landed on the async operators after that PR was first written: concurrent part fetching (#37315) and hydration frontier coalescing (#37313).
Description
Convert
shard_source_descsandshard_source_fetchfromAsyncOperatorBuilderto synchronousOperatorBuilderRcoperators, each paired with a tokio task that owns the persist I/O.shard_source_descsruns a listen task on the chosen worker that owns the reader, snapshot, listen loop, and stats filtering, and sends parts (split intoExchangeableBatchPart+Lease) plus progress over a channel. The formershard_source_descs_returnoperator is merged in as a disconnectedcompleted_fetchesinput; the listen handle (the reader's SeqNo hold) is released via a oneshot once that frontier empties. The listen task coalesces hydration-time frontier downgrades underpersist_source_hydration_frontier_coalesce_byteswhile parts keep flowing per batch.shard_source_fetchforwards descs to a fetch task and retains a per-flight capability pair keyed by time; results are matched by time and emitted at the data capability, with the completed-fetches capability dropped to release the lease only after the download completes. The fetch task runs up topersist_source_fetch_concurrencydownloads at once via aFuturesUnorderedof clonedBatchFetchers. On a missing blob it reports through theErrorHandlerand freezes, so a later good result cannot advance the frontier past the missing part.builder_async's two-phase shutdown (build_reschedule+ the coordinated button), so a local-only press cannot advance the downstream frontier past times other workers' instances still feed.filter_fn,listen_sleep, andstart_signalgainSendbounds; all callers already satisfy them.🤖 Generated with Claude Code
https://claude.ai/code/session_01N1erbcNjoki3HsfBJCTFUd