haiyangsun-db commented on PR #56702: URL: https://github.com/apache/spark/pull/56702#issuecomment-5106945232
@Yicong-Huang , Thanks for the thorough pass — this was really helpful. I've pushed a series of commits addressing everything; summary below, with per-thread replies inline. Would you please help take another look? Some summary of the changes: ▎ HL-2 (Cancel-after-init-error via Cancelling, not early Failed): done — the init-error path no longer settles Failed in the callback; it records the error, and doInit (where requestObserver is published) sends the Cancel and drains the CancelResponse, so the terminal is Cancelled. Transitions.failed was removed entirely. ▎ ▎ HL-1 (init-handshake complexity): addressed by cleanup rather than the full handler-split — details in the inline reply on that thread. ▎ ▎ I also took the interrupt discussion (#5) further into a dedicated Interrupted termination — see that thread. --- HL-1 — "init logic still feels too complex / isolate init as a one-shot" ▎ You're right that the signaling was the main source of noise, and it turned out to be mostly accidental. onTerminalSettled is now the single place a settled terminal wakes a blocked init(), so every branch that settles a terminal (the responseObserver failure/terminator cases, the timeout paths, close()) no longer signals the init waiter itself — it just settles. That removed 7 redundant signalWithoutValue() calls; the only direct init signals left are the three genuinely non-terminal paths (INIT-ok, INIT-error, pre-init ERROR), which must wake init() without settling a terminal so the session stays Initializing and process() can still run. That's the irreducible core of the handshake. ▎ On fully isolating init behind a one-shot handler and swapping to a data-phase handler once it resolves: I looked at this and decided against it. gRPC gives us a single StreamObserver for the stream's whole life, so "switching" means either a swappable @volatile var handler or an if (initResolved) … at the top of onNext. The former adds a mutable concurrency surface and a swap-vs-reentrant-delivery race — exactly the trickiest part of this class — and the latter just relocates the two initResolved checks that already exist. Neither removes the essential constraint (reentrant directExecutor delivery forcing the deferred Cancel), so I don't think the extra indirection pays for itself. Happy to revisit if you feel strongly. --- #1 (id 3593697357) — "settling failed before sendCancelInternal means Cancel is never written; contradicts proto" ▎ Fixed. The INIT-error branch no longer settles a terminal in the callback — it records the error (executionError) and completes the init waiter, leaving the session in Initializing. doInit then runs failInitWithError, which sends the Cancel (where requestObserver is guaranteed published) and drains the CancelResponse, so the terminal settles Cancelled and the proto's "Cancel after init error" holds on both the directExecutor and cross-thread paths. --- #2 (id 3593697365) — "awaitTerminal() is a no-op; drain comment is stale" ▎ Fixed by the same change as #1. Since the terminal is no longer settled early, awaitTerminal() in failInitWithError now genuinely blocks for the CancelResponse before throwing — the drain is real, and the comment is accurate again. --- #3 (id 3593697374) — "doClose null-path returns a fresh Cancelled even when a terminal already settled" ▎ Fixed. The null-path now only settles Cancelled if (!currentState.isTerminal), then returns settledTermination — so an already-settled terminal (e.g. the channel-shutdown TransportFailed) is returned as-is rather than a bare Cancelled that disagrees with the state. --- #4 (id 3593856689) — "can TransportFailed be merged to failed?" ▎ This one's moot now: Transitions.failed was removed (the init-error path settles Cancelled via the drain, not Failed). The failure terminals are deliberately distinct — TransportFailed (unsalvageable, unknown worker state) vs. the new Interrupted (salvageable, engine-side event) — so I'd keep them separate rather than merge. See #5. --- #5 (id 3593874187) — "who can interrupt this thread? cancel vs failed?" ▎ Good question — it prompted a dedicated outcome. An interrupt here is an engine-side event (cancelled query / killed task), not a worker fault, so it's now neither Cancelled nor Failed: it settles a new Termination.Interrupted. handleInterrupt sends a best-effort Cancel and waits a short, configurable interruptCancelTimeoutMs (default 2s) for the ack — if the worker responds in time the session settles a clean, salvageable Cancelled; otherwise it falls back to Interrupted (also salvageable, since the worker isn't at fault). It's applied at every blocking point (init, terminal drain, close, result poll), not just init. There's a TODO [SPARK-57640] to back the optimistic "salvageable" assumption with a worker liveness check. --- #6 (id 3593884935) — "make await raise a timeout error instead of returning a Boolean" ▎ Done. OneShotValue.await now returns Unit and throws TimeoutException on expiry, so doInit handles the timeout on the exception path instead of inspecting a return value. --- #7 (id 3593892046) — "complete the value before transit the state?" ▎ I kept the order as transition-then-complete here, deliberately. Transitions.initAccepted() is a CAS from Initializing that a raced terminal (e.g. a transport error) must be able to lose against; completing the value first would publish an InitResponse that a subsequently-winning terminal contradicts. Completing after the CAS preserves settle/transition-before-release: init() wakes to a state consistent with the value it reads. Added a comment noting this. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
