haiyangsun-db commented on PR #56702: URL: https://github.com/apache/spark/pull/56702#issuecomment-5107090751
@Yicong-Huang Thanks for the thorough review -- this was really helpful. I've pushed a series of commits addressing everything; recap below, with the design discussion on high-level point 1 at the end. **High-level point 2 (Cancel-after-init-error should go through Cancelling, not settle Failed early):** done. The init-error path no longer settles `Failed` in the response callback; it records the error and leaves the session `Initializing`, and `doInit` (where `requestObserver` is guaranteed published) sends the `Cancel` and drains the `CancelResponse`, so the terminal settles `Cancelled`. `Transitions.failed` was removed entirely. **Inline comments:** all addressed in the threads (recap so it's in one place): - INIT-error no longer settles a terminal early -> the wire `Cancel` is now actually written and its `CancelResponse` drained. - `awaitTerminal()` genuinely blocks for the `CancelResponse` again (it was a no-op only because of the early-settle bug above). - `doClose`'s no-stream path only settles `Cancelled` when not already terminal, then returns `settledTermination` (no more fresh `Cancelled` disagreeing with a settled state). - `OneShotValue.await` now returns `Unit` and throws `TimeoutException` instead of returning a boolean. - Interrupt handling is now a first-class `Interrupted` termination (see that thread): a best-effort `Cancel` with a short bounded wait for the ack -- a clean, salvageable `Cancelled` if the worker responds in time, otherwise `Interrupted` (also salvageable, since the interrupt is an engine-side event, not a worker fault). **High-level point 1 (init-handshake complexity):** 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 switching 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. -- 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]
