yugan95 opened a new pull request, #3801:
URL: https://github.com/apache/celeborn/pull/3801
### What changes were proposed in this pull request?
When the C++ transport connection is already closed, `MessageDispatcher`
asserts instead of failing the request. Both send paths guard with a hard
`CELEBORN_CHECK(!closed_)`:
```cpp
folly::Future<std::unique_ptr<Message>> MessageDispatcher::operator()(...) {
CELEBORN_CHECK(!closed_); // aborts the process
...
this->pipeline_->write(std::move(toSendMsg));
CELEBORN_CHECK(!closed_); // aborts the process
return f;
}
```
and `TransportClientFactory::createClient` raises a non-retriable
`CELEBORN_FAIL` when a connection cannot be established.
This PR fails these conditions with a retriable `CelebornRuntimeError`
instead of aborting:
- `MessageDispatcher::operator()` and `sendFetchChunkRequest`: the leading
`CELEBORN_CHECK(!closed_)` becomes a fast path that returns a ready retriable
future; the trailing check now fulfils the just-registered promise with a
retriable exception (see the race note below) rather than asserting.
- `MessageDispatcher::cleanup()`: the two outstanding-request failures
switch from a plain `std::runtime_error` to the same retriable exception, so
every request failed on close carries consistent classification.
- `TransportClientFactory::createClient`: the connect failure throws a
retriable `CelebornRuntimeError` instead of `CELEBORN_FAIL`.
Three unit tests are added in `MessageDispatcherTest.cpp`: a send after
close, a fetch after close, and an in-flight request failed by `close()` — each
asserts the future is ready and failed with a retriable `CelebornException`.
### Why are the changes needed?
Sending on a closed connection is a normal recoverable condition — the peer
closed the socket because of a worker restart or an idle timeout — not an
invariant violation. Aborting the process (or raising a non-retriable error) on
that condition is wrong: a routine connection drop should surface to the caller
so its retry / replica-failover logic can recover, exactly as it does in the
Java client.
This aligns the C++ dispatcher's **connection-closed failure semantics**
with the Java client. In Java the request path never aborts on a dead channel:
- `TransportResponseHandler#channelInactive` calls
`failOutstandingRequests(new IOException("Connection from ... closed"))`,
invoking every outstanding fetch/rpc/push callback's `onFailure(cause)`. This
is the analogue of the C++ `cleanup()` and of the trailing-check path below.
- On the send path, `TransportClient#sendRpc` / `fetchChunk` / `pushData`
write unconditionally and let Netty's write-failure listener
(`StdChannelListener#operationComplete`) close the channel and call
`handleFailure` → the callback's `onFailure`.
In both cases Java fails the request rather than aborting, and
`CelebornInputStream#createReaderWithRetry` (and the push revive path) then
retry or fail over to the replica.
Two honest scope notes:
1. This aligns the *behavioral contract*, not the mechanism. The C++
`pipeline_->write` is fire-and-forget with no per-write failure listener, so
the fix uses explicit `closed_` checks rather than a Netty-style write
callback. Same observable outcome, C++-specific mechanism.
2. The `isRetriable=true` flag follows the existing C++ exception idiom and
encodes the intent that these conditions are recoverable. There is no retriable
flag on the Java exceptions, and no C++ path branches on `isRetriable()` yet —
both clients currently catch broadly and retry. The flag makes the
classification available for a future retriable-aware branch; it is not
consumed as a decision input today.
### Does the close-during-send race matter?
`close()` sets `closed_ = true` before `cleanup()` locks the registry. If
`cleanup()` runs between the leading fast-path check and the promise
registration, the promise would be registered into an already-drained registry
and never fulfilled — the future would hang until the request times out. The
trailing `closed_.load()` check re-inspects the registry after registration and
fails the promise with a retriable exception if it is still there, so the
caller sees a ready retriable error instead of a stall.
`closeFailsInFlightRequestsRetriably` covers the in-flight case.
### Does this PR resolve a correctness bug?
- [ ] Yes
Shuffle output is unaffected. This is a robustness / failure-handling fix: a
routine connection close no longer aborts the client (or, in the race window,
no longer hangs a request until timeout), and instead surfaces as a retriable
error the existing retry/failover paths can act on.
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
No config, API, or shuffle-behaviour change. A connection-closed condition
that previously aborted the process now fails the affected request retriably.
### How was this patch tested?
New unit tests in `cpp/celeborn/network/tests/MessageDispatcherTest.cpp`:
- `sendRpcRequestAfterCloseFailsRetriably` — a send issued after `close()`
returns a ready, retriable exception instead of tripping the assertion.
- `sendFetchChunkRequestAfterCloseFailsRetriably` — same for the fetch path.
- `closeFailsInFlightRequestsRetriably` — `close()` fails an
already-registered in-flight request retriably rather than leaving its future
pending.
Covered by the `Celeborn Cpp Integration Test` workflow (`Run Unittests of
Celeborn Cpp`).
--
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]
