sunchao commented on code in PR #3778:
URL: https://github.com/apache/celeborn/pull/3778#discussion_r3730347505
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -317,28 +349,39 @@ public void initChannel(SocketChannel ch) {
// Connect to the remote server
long preConnect = System.nanoTime();
ChannelFuture cf = bootstrap.connect(address);
- if (connectTimeoutMs <= 0) {
- awaitWithChannelCleanup(
- () -> {
- cf.await();
- return true;
- },
- cf);
- assert cf.isDone();
- if (cf.isCancelled()) {
+ try {
+ if (connectTimeoutMs <= 0) {
+ awaitWithChannelCleanup(
+ () -> {
+ cf.await();
+ return true;
+ },
+ cf);
+ assert cf.isDone();
+ if (cf.isCancelled()) {
+ closeChannel(cf);
+ throw new IOException(String.format("Connecting to %s cancelled",
address));
+ } else if (!cf.isSuccess()) {
+ closeChannel(cf);
+ throw new IOException(String.format("Failed to connect to %s",
address), cf.cause());
+ }
+ } else if (!awaitWithChannelCleanup(() -> cf.await(connectTimeoutMs),
cf)) {
closeChannel(cf);
- throw new IOException(String.format("Connecting to %s cancelled",
address));
- } else if (!cf.isSuccess()) {
+ throw new CelebornIOException(
+ String.format("Connecting to %s timed out (%s ms)", address,
connectTimeoutMs));
+ } else if (cf.cause() != null) {
closeChannel(cf);
- throw new IOException(String.format("Failed to connect to %s",
address), cf.cause());
+ throw new CelebornIOException(
+ String.format("Failed to connect to %s", address), cf.cause());
}
- } else if (!awaitWithChannelCleanup(() -> cf.await(connectTimeoutMs), cf))
{
- closeChannel(cf);
- throw new CelebornIOException(
- String.format("Connecting to %s timed out (%s ms)", address,
connectTimeoutMs));
- } else if (cf.cause() != null) {
- closeChannel(cf);
- throw new CelebornIOException(String.format("Failed to connect to %s",
address), cf.cause());
+ } catch (IOException e) {
+ // If the connection failed because the channel could not be registered
on its netty event
+ // loop (the loop's thread has died and netty rejects new tasks with
"event executor
+ // terminated"), the worker group is permanently degraded: that dead
loop is never replaced
+ // and keeps being handed out by the round-robin chooser. Replace the
group so retries bind
+ // to fresh live threads, then rethrow so the caller (e.g.
retryCreateClient) retries.
+ recreateWorkerGroupIfEventLoopDead(connectGroup, cf.cause());
Review Comment:
[P2] Retry the replacement group independently of the ordinary retry budget.
This path successfully installs a fresh worker group but immediately
rethrows the original connection failure. `retryCreateClient()` counts that
failure against `celeborn.<module>.io.maxRetries`, so with the supported value
`1` the triggering request fails without ever trying the healthy replacement
group. `createUnmanagedClient()` has no retry wrapper and likewise always fails
on the recovery-triggering attempt. Even with the default retry count, recovery
incurs the configured five-second retry wait. Please retry once immediately
after replacing the dead group without consuming the normal I/O retry budget,
or propagate a distinct recovery signal that the caller retries unconditionally.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -92,7 +92,15 @@ public Channel getChannel() {
}
public boolean isActive() {
- return !timedOut && (channel.isOpen() || channel.isActive());
+ // A channel is pinned to one netty event-loop thread for its lifetime. If
that event loop has
+ // terminated (e.g. an uncaught error killed the thread; netty does not
replace it in a
+ // fixed-size EventLoopGroup), the channel can no longer send or complete
anything: writes and
+ // listener notifications route to the dead loop and are silently dropped.
Such a client must
+ // not be treated as active/reused, otherwise a request on it can orphan
and hang. See
+ // SPARK-58292.
+ return !timedOut
+ && !channel.eventLoop().isShuttingDown()
Review Comment:
[P1] Fail outstanding work before abandoning a dead-loop client.
`CelebornBufferStream` keeps its `TransportClient` for the lifetime of a
partition, and `addCredit()` / `notifyRequiredSegment()` continue calling
`sendRpc()` on that instance rather than reacquiring through the factory. Once
this check marks the dead-loop client inactive, the factory can overwrite its
pooled entry without failing outstanding RPCs, while the existing stream still
holds the original client. The terminated event loop never runs write listeners
or `channelInactive()`, so the Flink reader can hang forever even after another
request recreates the worker group; `closeStream()` also skips
`BUFFER_STREAM_END`. I reproduced with Netty 4.2.10 that the channel remains
open, the write listener never executes, and an ordinary `channel.close()`
cannot close it. Please synchronously fail outstanding callbacks and
force-close or otherwise invalidate dead-loop clients so existing stream owners
can recover.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -432,6 +475,71 @@ static boolean awaitWithChannelCleanup(
}
}
+ /**
+ * If the given connection-failure cause was a rejection by a dead netty
event loop (its worker
+ * thread terminated and netty rejects new registrations with a {@link
+ * RejectedExecutionException}), replace the worker group so subsequent
connections bind to fresh,
+ * live threads. A dead loop is never replaced within a fixed-size group and
keeps being selected
+ * by the round-robin chooser, so without this the degradation is permanent.
See SPARK-58292.
+ */
+ private void recreateWorkerGroupIfEventLoopDead(EventLoopGroup connectGroup,
Throwable cause) {
+ if (!recreateWorkerGroupOnDeadEventLoop) {
+ return;
+ }
+ boolean eventLoopDead = false;
+ for (Throwable t = cause; t != null; t = t.getCause()) {
+ // Match ONLY the terminated-loop rejection, not a transient
task-queue-full rejection.
+ // netty's SingleThreadEventExecutor.reject() throws exactly this
message when isShutdown();
+ // the queue-full handler path throws a RejectedExecutionException with
no message.
+ if (t instanceof RejectedExecutionException
+ && "event executor terminated".equals(t.getMessage())) {
+ eventLoopDead = true;
+ break;
+ }
+ }
+ if (eventLoopDead) {
+ recreateWorkerGroup(connectGroup);
+ }
+ }
+
+ /**
+ * Replace the worker group with a fresh one, if it is still the group the
failed connection used
+ * ({@code connectGroup}). The superseded group is not shut down here: its
still-live threads may
+ * be serving channels that are already open. We keep a weak reference and
shut it down
+ * best-effort at {@link #close()}; its threads are daemon, so a
not-yet-collected group cannot
+ * block JVM shutdown. Synchronized and identity-guarded so concurrent
callers that all hit the
+ * same dead group replace it exactly once rather than spawning many groups.
+ */
+ private synchronized void recreateWorkerGroup(EventLoopGroup connectGroup) {
+ // The factory is closed (or closing): its worker group was shut down by
close(), so a
+ // createClient() racing or following close() must not recreate a fresh
group and resurrect a
+ // closed factory (which would leak threads that close() will never reap
again).
+ if (closed) {
+ return;
+ }
+ // A concurrent caller that hit the same dead group already swapped it
out; nothing to do.
+ if (workerGroup != connectGroup) {
+ return;
+ }
+ // Use a distinct thread-name prefix so the recreated group is
self-identifying in thread
+ // dumps and logs. Netty's DefaultThreadFactory also appends an
incrementing pool id, so the
+ // names would not collide even with the same prefix, but tagging it
"-recreated-<n>" makes the
+ // dead-event-loop recovery obvious to anyone inspecting the process, and
the <n> distinguishes
+ // successive recreations if a loop dies more than once.
+ workerGroupRecreationCount++;
+ TransportConf conf = context.getConf();
+ workerGroup =
+ NettyUtils.createEventLoop(
+ ioMode,
+ conf.clientThreads(),
+ conf.conflictAvoidChooserEnable(),
+ conf.getModuleName() + "-client-recreated-" +
workerGroupRecreationCount);
+ deprecatedWorkerGroups.add(new WeakReference<>(connectGroup));
Review Comment:
[P2] Retire superseded worker groups after their channels drain.
A `WeakReference` does not make the old group collectible: each surviving
Netty event-loop thread retains its executor, and that executor strongly
retains its parent group. There is no shutdown or retirement when the old
group's channels drain; the group is only shut down when the entire factory
closes. With the default client-thread count of `2 * cores`, one partial worker
failure can therefore retain up to `2 * cores - 1` obsolete threads and their
selectors for the lifetime of a long-lived factory, and repeated recoveries
accumulate them. I reproduced this with a two-thread Netty 4.2.10 group: after
terminating one child and dropping application references, the weak reference
remained live and the other thread kept running. Please track the superseded
group's active channels and shut it down once they have drained.
--
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]
