sunchao commented on code in PR #3778:
URL: https://github.com/apache/celeborn/pull/3778#discussion_r3759741619
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java:
##########
@@ -263,6 +263,32 @@ private void failOutstandingRequests(Throwable cause) {
}
}
+ /**
+ * Fail all outstanding requests because the netty event loop this channel
is pinned to has died
+ * (see {@link TransportClient#isEventLoopDead()}). Unlike {@link
#channelInactive()} this runs on
+ * an arbitrary caller thread, because the dead loop will never deliver
channelInactive() itself.
+ *
+ * <p>Idempotent: the outstanding maps are drained by {@code remove}, so a
second call is a no-op.
+ *
+ * @param cause the failure handed to every outstanding callback.
+ */
+ void failOutstandingRequestsOnDeadEventLoop(Throwable cause) {
+ if (hasOutstandingRequests()) {
+ logger.error(
+ "Failing {} outstanding requests to {}: the netty event loop this
channel is pinned to "
+ + "is no longer usable, so they can never complete",
+ numOutstandingRequests(),
+ NettyUtils.getRemoteAddress(channel));
+ failOutstandingRequests(cause);
+ }
+ if (pushCheckerScheduleFuture != null) {
Review Comment:
[P1] Keep timeout checkers alive until new push/fetch requests are rejected.
Dead-loop invalidation cancels the per-client push/fetch timeout checkers,
but `TransportClient.fetchChunk()`, `pushData()`, and `pushMergedData()` still
register and write new requests without a dead-loop guard. For example,
`WorkerPartitionReader` can observe its client active, another thread can then
invalidate the now-dead client and cancel this checker, and the reader can
resume into `fetchChunk()`. The write listener never runs and the timeout
backstop has been removed, so `next()` polls forever. I reproduced an expired
fetch remaining outstanding after cancellation while an otherwise identical
uncanceled checker fails the request normally. Please keep the checkers running
until actual channel closure, or atomically mark the handler terminal and
immediately fail every later push/fetch registration.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -239,23 +273,31 @@ public TransportClient createClient(
"DNS resolution {} for {} took {} ms", resolveMsg, resolvedAddress,
hostResolveTimeMs);
}
- synchronized (clientPool.locks[clientIndex]) {
- cachedClient = clientPool.clients[clientIndex];
-
- if (cachedClient != null) {
- if (cachedClient.isActive()) {
- logger.debug(
- "Returning cached connection from {} to {}: {}",
- cachedClient.getChannel().localAddress(),
- resolvedAddress,
- cachedClient);
- return cachedClient;
- } else {
- logger.info("Found inactive connection to {}, creating a new one.",
resolvedAddress);
+ final int recreationCountBefore = workerGroupRecreationCount;
+ try {
+ synchronized (clientPool.locks[clientIndex]) {
+ cachedClient = clientPool.clients[clientIndex];
+
+ if (cachedClient != null) {
+ if (cachedClient.isActive()) {
+ logger.debug(
+ "Returning cached connection from {} to {}: {}",
+ cachedClient.getChannel().localAddress(),
+ resolvedAddress,
+ cachedClient);
+ return cachedClient;
+ } else {
+ logger.info("Found inactive connection to {}, creating a new
one.", resolvedAddress);
+ }
}
+ clientPool.clients[clientIndex] =
internalCreateClient(resolvedAddress, decoder);
Review Comment:
[P1] Preserve the displaced client until its outstanding work is failed.
With the default one connection per peer, reconnecting to the same peer
replaces the dead `cachedClient` here before the `finally` block runs
`failClientsOnDeadEventLoopsIfRecreated()`. That sweep walks only the current
pool slots, so the displaced client is already unreachable: its outstanding RPC
never completes, and its dead channel remains tracked. I reproduced the actual
recovery path on this head: the replacement was active while the old callback
was never failed and its RPC remained outstanding. The new regression test
calls `failClientsOnDeadEventLoops()` directly while the old client is still
pooled, so it misses this case. Please retain the evicted client and
invalidate/untrack it after releasing the pool lock, including when
reconnection succeeds on another live loop without recreating the group.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -92,7 +93,53 @@ public Channel getChannel() {
}
public boolean isActive() {
- return !timedOut && (channel.isOpen() || channel.isActive());
+ // A client on a dead event loop can neither send nor complete anything,
so it must not be
+ // reused. See isEventLoopDead().
+ return !timedOut && !isEventLoopDead() && (channel.isOpen() ||
channel.isActive());
+ }
+
+ /**
+ * Whether this channel's netty event loop can no longer be relied on, i.e.
it has terminated or
+ * is shutting down. The motivating case is SPARK-58292: a channel is pinned
to one loop for its
+ * lifetime, and netty neither replaces a loop whose thread has died within
a fixed-size group nor
+ * stops handing it out.
+ *
+ * <p>Everything submitted to a terminated loop is silently dropped - the
write never happens, and
+ * the listener that would have failed the callback never runs either, since
netty's {@code
+ * safeExecute} only logs the rejection. So a request issued here orphans
rather than failing,
+ * {@code channelInactive()} is never delivered, and even {@code close()}
cannot take effect
+ * because it is itself submitted to the dead loop. Nothing but an explicit
sweep recovers such a
+ * client.
+ *
+ * <p>Deliberately keyed on {@code isShuttingDown()} rather than the exact
{@code isShutdown()}
+ * that netty's "event executor terminated" rejection uses: a loop in {@code
shutdownGracefully}'s
+ * quiet period would still drain its queue, so this is conservative. The
one visible consequence
+ * is that a best-effort message guarded by {@link #isActive()} - e.g. the
BUFFER_STREAM_END a
+ * reader sends on close - is skipped while the owning factory is closing.
The server reclaims
+ * those streams when the connection drops, and refusing new work on a group
that is going away is
+ * what we want anyway.
+ */
+ public boolean isEventLoopDead() {
+ return channel.eventLoop().isShuttingDown();
+ }
+
+ /**
+ * Invalidate this client if its event loop has died, by failing every
outstanding request so that
+ * owners holding the client directly - rather than reacquiring it from
{@link
+ * TransportClientFactory} - are notified instead of waiting for a
completion that can never come.
+ * No-op for a healthy client, and idempotent.
+ */
+ public void invalidateIfEventLoopDead() {
Review Comment:
[P1] Notify established inbound streams when invalidating a dead client.
An established Flink credit stream normally has no outstanding RPC:
`OPEN_STREAM` and subsequent credit/segment updates have already been
acknowledged. Its ownership instead lives in `ReadClientHandler.streamClients`,
and `ReadClientHandler.channelInactive()` is what sends `TransportableError` to
the corresponding stream readers. This method only drains the response handler
and never invokes the inbound request handler, while a genuinely dead loop
cannot deliver `channelInactive()` itself. Consequently, the recovery sweep
notifies nobody and an already-open Flink reader remains stalled indefinitely.
I reproduced that dead-loop invalidation leaves an installed inbound handler
unnotified, whereas the normal `TransportChannelHandler.channelInactive()` path
notifies it. Please propagate dead-client invalidation to the request/inbound
stream handler exactly once.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -179,6 +226,12 @@ public long sendRpc(ByteBuffer message,
RpcResponseCallback callback) {
}
long requestId = requestId();
+ if (isEventLoopDead()) {
Review Comment:
[P1] Close the race between the dead-loop check and RPC registration.
`sendRpc()` checks the loop before adding its callback to `outstandingRpcs`.
A sender can observe a live loop, pause, and let another thread observe the
loop die, recreate the group, and finish the entire invalidation sweep while
this callback is not yet in the map. The sender then registers its callback and
writes to the dead loop; neither the write listener nor `channelInactive()`
runs, RPCs have no timeout checker, and no further sweep occurs unless another
recreation happens. A controlled interleaving on this head leaves the RPC
outstanding after the sweep with its callback never invoked. Please register
before rechecking/invalidation, or synchronize registration with terminal
invalidation so neither operation can miss the other.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -317,29 +379,45 @@ 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());
+ }
+ } catch (IOException e) {
+ // Registration may have been rejected because the loop it landed on is
dead, which degrades
+ // the whole group permanently. Replace it, then reconnect straight away
so the request that
+ // triggered the recovery is the first to benefit from it rather than
the one that pays.
+ if (recreateWorkerGroupIfEventLoopDead(connectGroup, cf.cause())
+ && retryOnRecreatedWorkerGroup) {
+ logger.warn("Retrying the connection to {} on a fresh worker group",
address, e);
+ // Reusing `decoder` is safe here: the dead loop rejected the channel
registration, so the
+ // ChannelInitializer above never ran and the decoder was never added
to a pipeline.
+ return internalCreateClient(address, decoder, false);
Review Comment:
[P2] Create a fresh decoder for the replacement connection.
The comment assumes `"event executor terminated"` means the original
`ChannelInitializer` never ran, but Netty 4.2.10 can initialize/register the
channel and install this decoder before `Bootstrap.doConnect()` submits the
actual connect task to the event loop. If the loop dies between those steps,
the same terminated-executor rejection triggers this retry with a decoder that
is already present in the old pipeline. Both `TransportFrameDecoder` and the
Flink decoder are non-`@Sharable`; I reproduced Netty rejecting the second
installation with `ChannelPipelineException`. Recovery therefore fails despite
creating a healthy replacement, including for unmanaged clients and
`maxRetries=1`. Please pass a decoder supplier through the retry and allocate a
fresh handler for the second channel.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -432,6 +510,181 @@ 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. Without this the degradation is permanent - see {@link
+ * TransportClient#isEventLoopDead()}.
+ *
+ * @return whether the caller may now retry: either this call replaced the
group, or a concurrent
+ * caller already did and the current group is therefore a fresh one.
+ */
+ private boolean recreateWorkerGroupIfEventLoopDead(EventLoopGroup
connectGroup, Throwable cause) {
+ if (!recreateWorkerGroupOnDeadEventLoop) {
+ return false;
+ }
+ 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;
+ }
+ }
+ return 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 already-open channels, so it is retired by {@link
#retireWorkerGroupIfDrained}
+ * instead. Synchronized and identity-guarded so concurrent callers that all
hit the same dead
+ * group replace it exactly once rather than spawning many groups.
+ *
+ * @return whether a fresh group is now installed and the caller may retry
on it.
+ */
+ private synchronized boolean 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 false;
+ }
+ // A concurrent caller that hit the same dead group already swapped it
out. Nothing to do, but
+ // the current group is a fresh one, so the caller can still retry on it.
+ if (workerGroup != connectGroup) {
+ return true;
+ }
+ // Tag the thread names so a dead-event-loop recovery is obvious in a
thread dump, and so
+ // successive recreations stay distinguishable if a loop dies more than
once.
+ workerGroupRecreationCount++;
+ TransportConf conf = context.getConf();
+ String threadPrefix = conf.getModuleName() + "-client-recreated-" +
workerGroupRecreationCount;
+ workerGroup =
+ NettyUtils.createEventLoop(
+ ioMode, conf.clientThreads(), conf.conflictAvoidChooserEnable(),
threadPrefix);
+ supersededWorkerGroups.add(connectGroup);
+ logger.warn(
+ "Detected a dead netty event loop in the {} client worker group;
replaced it with {}. "
+ + "The superseded group keeps serving its {} already-open channels
until they drain "
+ + "(SPARK-58292).",
+ conf.getModuleName(),
+ threadPrefix,
+ channelCount(connectGroup));
+ // If it has no channels left, nothing will ever untrack one on its
behalf. Check once, here.
+ retireWorkerGroupIfDrained(connectGroup);
Review Comment:
[P2] Include in-flight registrations when deciding whether the old group
drained.
`trackChannel()` is called only after a connect completes, but this
immediate retirement treats an empty tracked-channel set as proof that the old
group has no users. With multiple loops, a concurrent connection can already be
registered on a still-healthy loop in that group while its connect is pending
and its channel has not yet reached `trackChannel()`. Recovery on another dead
loop then shuts down the entire old group and closes the otherwise valid
in-flight connection. I reproduced this with a real registered
`NioSocketChannel`: recreation observed no tracked channels, shut down the old
group, and closed the registered channel. Please track pending registrations
under the same retirement coordination, or register channels before testing
whether the group has drained.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -432,6 +510,181 @@ 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. Without this the degradation is permanent - see {@link
+ * TransportClient#isEventLoopDead()}.
+ *
+ * @return whether the caller may now retry: either this call replaced the
group, or a concurrent
+ * caller already did and the current group is therefore a fresh one.
+ */
+ private boolean recreateWorkerGroupIfEventLoopDead(EventLoopGroup
connectGroup, Throwable cause) {
+ if (!recreateWorkerGroupOnDeadEventLoop) {
+ return false;
+ }
+ 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;
+ }
+ }
+ return 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 already-open channels, so it is retired by {@link
#retireWorkerGroupIfDrained}
+ * instead. Synchronized and identity-guarded so concurrent callers that all
hit the same dead
+ * group replace it exactly once rather than spawning many groups.
+ *
+ * @return whether a fresh group is now installed and the caller may retry
on it.
+ */
+ private synchronized boolean 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 false;
+ }
+ // A concurrent caller that hit the same dead group already swapped it
out. Nothing to do, but
+ // the current group is a fresh one, so the caller can still retry on it.
+ if (workerGroup != connectGroup) {
+ return true;
+ }
+ // Tag the thread names so a dead-event-loop recovery is obvious in a
thread dump, and so
+ // successive recreations stay distinguishable if a loop dies more than
once.
+ workerGroupRecreationCount++;
+ TransportConf conf = context.getConf();
+ String threadPrefix = conf.getModuleName() + "-client-recreated-" +
workerGroupRecreationCount;
+ workerGroup =
+ NettyUtils.createEventLoop(
+ ioMode, conf.clientThreads(), conf.conflictAvoidChooserEnable(),
threadPrefix);
+ supersededWorkerGroups.add(connectGroup);
+ logger.warn(
+ "Detected a dead netty event loop in the {} client worker group;
replaced it with {}. "
+ + "The superseded group keeps serving its {} already-open channels
until they drain "
+ + "(SPARK-58292).",
+ conf.getModuleName(),
+ threadPrefix,
+ channelCount(connectGroup));
+ // If it has no channels left, nothing will ever untrack one on its
behalf. Check once, here.
+ retireWorkerGroupIfDrained(connectGroup);
+ return true;
+ }
+
+ /**
+ * Register a newly connected channel against the worker group it is pinned
to. Tracking exists
+ * solely so a superseded group can be retired, which cannot happen unless
recreation is enabled,
+ * so skip the bookkeeping entirely when it is off.
+ */
+ private void trackChannel(EventLoopGroup group, Channel channel) {
+ if (!recreateWorkerGroupOnDeadEventLoop) {
+ return;
+ }
+ workerGroupChannels
+ .computeIfAbsent(group, unused -> ConcurrentHashMap.newKeySet())
+ .add(channel);
+ channel.closeFuture().addListener(future -> untrackChannel(group,
channel));
+ }
+
+ /** Called from the channel's close future, and from {@link
#failClientsOnDeadEventLoops()}. */
+ private void untrackChannel(EventLoopGroup group, Channel channel) {
+ Set<Channel> channels = workerGroupChannels.get(group);
+ if (channels != null) {
+ channels.remove(channel);
+ }
+ retireWorkerGroupIfDrained(group);
+ }
+
+ /**
+ * Shut down a superseded worker group once it has no channels left to
serve. It cannot simply be
+ * dropped and left to the GC: a netty thread keeps its executor, and the
executor its parent
+ * group, strongly reachable. So without this, one dead event loop would
cost the process the
+ * group's other clientThreads() - 1 selector threads for the lifetime of
the factory, and
+ * repeated recoveries would accumulate them.
+ *
+ * <p>Best-effort: a connection that captured this group before it was
superseded may still
+ * register a channel afterwards, but such a connection is failing anyway,
and {@link #close()}
+ * remains the backstop for any group that never drains.
+ */
+ private synchronized void retireWorkerGroupIfDrained(EventLoopGroup group) {
+ if (closed || group == workerGroup) {
+ return;
+ }
+ Set<Channel> channels = workerGroupChannels.get(group);
+ if (channels != null && !channels.isEmpty()) {
+ return;
+ }
+ workerGroupChannels.remove(group);
+ if (supersededWorkerGroups.remove(group) && !group.isShuttingDown()) {
+ logger.info(
+ "A superseded {} client worker group has drained; shutting it down.
{} superseded "
+ + "group(s) still retained.",
+ context.getConf().getModuleName(),
+ supersededWorkerGroups.size());
+ group.shutdownGracefully();
+ }
+ }
+
+ /** Number of channels currently tracked as open on the given worker group.
*/
+ private int channelCount(EventLoopGroup group) {
+ Set<Channel> channels = workerGroupChannels.get(group);
+ return channels == null ? 0 : channels.size();
+ }
+
+ /** Sweep only if the worker group has been recreated since {@code
recreationCountBefore}. */
+ private void failClientsOnDeadEventLoopsIfRecreated(int
recreationCountBefore) {
+ if (workerGroupRecreationCount == recreationCountBefore) {
+ return;
+ }
+ try {
+ failClientsOnDeadEventLoops();
+ } catch (Throwable t) {
+ // Never let this mask the outcome of the createClient call it is
attached to.
+ logger.warn("Error while invalidating clients pinned to a dead netty
event loop", t);
+ }
+ }
+
+ /**
+ * Synchronously fail the outstanding requests of every pooled client pinned
to a dead event loop.
+ * Marking such a client inactive stops the factory handing it out again,
but does nothing for
+ * whoever already holds it: an owner that keeps a client for the lifetime
of a stream - e.g.
+ * Flink's {@code CelebornBufferStream}, which sends credits on the client
it captured rather than
+ * reacquiring one - would otherwise wait forever. Failing its callbacks is
the only way it can
+ * notice; the client cannot be force-closed. See {@link
TransportClient#isEventLoopDead()}.
+ *
+ * <p>MUST be called with no pool lock held: failing a request invokes its
callback on this
+ * thread, and callbacks re-enter the factory.
+ *
+ * <p>Only pooled clients are reachable from here, so a client handed out by
{@link
+ * #createUnmanagedClient} is left to its owner to invalidate via {@link
+ * TransportClient#invalidateIfEventLoopDead()}.
+ */
+ @VisibleForTesting
+ public void failClientsOnDeadEventLoops() {
+ for (ClientPool clientPool : connectionPool.values()) {
Review Comment:
[P2] Untrack dead unmanaged channels before retiring superseded groups.
`createUnmanagedClient()` still registers its channel in
`workerGroupChannels`, but this sweep visits only `connectionPool`. If an
unmanaged channel is pinned to the dead loop, its `closeFuture()` never
completes, so it is never untracked; even calling
`TransportClient.invalidateIfEventLoopDead()` only fails response callbacks and
does not inform the factory. The superseded group therefore appears permanently
nonempty, keeping its remaining selector threads alive until the whole factory
closes. I reproduced a tracked dead unmanaged channel surviving both the pool
sweep and owner-side invalidation while `supersededWorkerGroupCount()` remains
one. Please sweep the tracked channels themselves and explicitly remove
dead-loop channels regardless of whether their clients were pooled.
--
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]
