DEBUG (do not merge): instrument shard_source fetch/descs depth#37765
Draft
antiguru wants to merge 5 commits into
Draft
DEBUG (do not merge): instrument shard_source fetch/descs depth#37765antiguru wants to merge 5 commits into
antiguru wants to merge 5 commits into
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
Temporary instrumentation to locate the ~3-part in-flight ceiling during hydration reads. Emits `INSTR` info logs: * fetch task: once/sec, in_flight/ready/outstanding/max_concurrency plus cumulative pushed/completed. The tick fires even while the loop is blocked in `desc_rx.recv()`, so a starved loop (low in_flight while outstanding < max_concurrency) is visible. * fetch operator: descs forwarded to the fetch task per activation and cumulatively. * descs operator: parts emitted per `ListenMessage::Parts`. Together these show whether desc delivery meters the pipeline at depth ~3 or the fetch loop is the throttle. 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.
Temporary instrumentation to locate the ~3-part in-flight ceiling during persist hydration reads. Not for merge; built to deploy to staging.
Stacked on the
shard_source_fetchresult-channel fix (mh/persist-deasync-v2). AddsINSTRinfo logs:in_flight/ready/outstanding/max_concurrency+ cumulative pushed/completed. The tick fires even while blocked indesc_rx.recv(), so a desc-starved loop is visible.ListenMessage::Parts.Together these show whether desc delivery meters the pipeline at depth ~3 or the fetch loop is the throttle.
🤖 Generated with Claude Code