This is an automated email from the ASF dual-hosted git repository. SteNicholas pushed a commit to branch CELEBORN-2400 in repository https://gitbox.apache.org/repos/asf/celeborn.git
commit 0a31e9f9dba35c425cb479693cfe6ba9e234bb10 Author: Nicholas Jiang <[email protected]> AuthorDate: Mon Aug 3 16:11:31 2026 +0800 [CELEBORN-2400] Recreate the netty worker EventLoopGroup when a worker event loop thread dies This ports SPARK-58292 (apache/spark#57462) to Celeborn's network client stack, and extends it to invalidate clients that are already poisoned. A netty event-loop thread that dies (e.g. an uncaught error) is never replaced within a fixed-size EventLoopGroup, and the round-robin chooser keeps handing it out. Any channel pinned to the dead loop can no longer send or complete anything: writes and listener notifications route to the dead thread and are silently dropped, so a request on such a client can hang forever, and new connections registered on it fail with "event executor terminated". This makes the client recover from that state. Not reusing a poisoned client: - TransportClient.isActive() returns false once the channel's event loop is shutting down, so the pool evicts the client instead of reusing it. Unblocking whoever already holds one: - TransportClient.sendRpc() fails the callback up front rather than writing into a dead loop. Unlike pushes and fetches, an outstanding RPC has no timeout checker to fall back on, so an orphaned callback there hangs its owner forever. - TransportClientFactory, once a connection failure has revealed a dead loop, synchronously fails the outstanding requests of every pooled client still pinned to one. Marking a client inactive does not help an owner that keeps it for the lifetime of a stream -- e.g. Flink's CelebornBufferStream, which sends credits on the client it captured rather than reacquiring one -- because the dead loop delivers neither the write listener nor channelInactive(). The client cannot be force-closed either, since close() is itself submitted to the dead loop. The sweep runs outside the connection-pool lock, as failing a request invokes its callback on the calling thread and callbacks re-enter the factory. Restoring the worker group: - TransportClientFactory replaces its worker group when a connection fails with the terminated-executor rejection, then reconnects inline on the fresh group. The reconnect is deliberately not charged to celeborn.<module>.io.maxRetries, which may be as low as 1 and sleeps io.retryWait between attempts, and it also covers callers with no retry wrapper such as createUnmanagedClient. - The superseded group is not shut down eagerly, since its still-live threads may serve already-open channels. Its channels are tracked and the group is shut down once they drain, with close() as the backstop. It cannot be left to the GC: a netty thread keeps its executor, and the executor its parent group, strongly reachable. - Adds celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop (default true). It gates the group recreation and its channel bookkeeping only; refusing to reuse or write to a dead loop is unconditional, since such a request cannot succeed either way. Also logs the failure cause in CelebornBufferStream, which previously printed e.getCause() (null for these failures) or dropped the throwable entirely, leaving the recovery undiagnosable in exactly the scenario this change addresses. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --- .../plugin/flink/client/CelebornBufferStream.java | 14 +- .../common/network/client/TransportClient.java | 55 +++- .../network/client/TransportClientFactory.java | 354 ++++++++++++++++++--- .../network/client/TransportResponseHandler.java | 26 ++ .../common/network/util/TransportConf.java | 10 + .../org/apache/celeborn/common/CelebornConf.scala | 25 ++ .../network/SSLTransportClientFactorySuiteJ.java | 12 +- .../network/TransportClientFactorySuiteJ.java | 192 ++++++++++- .../network/client/TransportClientSuiteJ.java | 132 ++++++++ docs/configuration/network.md | 1 + 10 files changed, 761 insertions(+), 60 deletions(-) diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/client/CelebornBufferStream.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/client/CelebornBufferStream.java index 6588719f26..d6b59e8582 100644 --- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/client/CelebornBufferStream.java +++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/client/CelebornBufferStream.java @@ -115,10 +115,10 @@ public class CelebornBufferStream { @Override public void onFailure(Throwable e) { logger.error( - "Send PbReadAddCredit to {} failed, streamId {}, detail {}", + "Send PbReadAddCredit to {} failed, streamId {}", NettyUtils.getRemoteAddress(client.getChannel()), streamId, - e.getCause()); + e); messageConsumer.accept(new TransportableError(streamId, e)); } }); @@ -139,10 +139,10 @@ public class CelebornBufferStream { @Override public void onFailure(Throwable e) { logger.error( - "Send PbNotifyRequiredSegment to {} failed, streamId {}, detail {}", + "Send PbNotifyRequiredSegment to {} failed, streamId {}", NettyUtils.getRemoteAddress(client.getChannel()), streamId, - e.getCause()); + e); messageConsumer.accept(new TransportableError(streamId, e)); } }); @@ -358,7 +358,8 @@ public class CelebornBufferStream { "Open file {} stream for {} error from {}", fileName, shuffleKey, - NettyUtils.getRemoteAddress(client.getChannel())); + NettyUtils.getRemoteAddress(client.getChannel()), + e); messageConsumer.accept(new TransportableError(streamId, e)); } } @@ -369,7 +370,8 @@ public class CelebornBufferStream { "Open file {} stream for {} error from {}", fileName, shuffleKey, - NettyUtils.getRemoteAddress(client.getChannel())); + NettyUtils.getRemoteAddress(client.getChannel()), + e); messageConsumer.accept(new TransportableError(streamId, e)); } }; diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java index b0a61d054f..b6b84891d9 100644 --- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java +++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java @@ -36,6 +36,7 @@ import io.netty.util.concurrent.GenericFutureListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.celeborn.common.exception.CelebornIOException; import org.apache.celeborn.common.network.buffer.NioManagedBuffer; import org.apache.celeborn.common.network.protocol.OneWayMessage; import org.apache.celeborn.common.network.protocol.PushData; @@ -92,7 +93,53 @@ public class TransportClient implements Closeable { } 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() { + if (isEventLoopDead()) { + handler.failOutstandingRequestsOnDeadEventLoop(deadEventLoopException()); + } + } + + private CelebornIOException deadEventLoopException() { + return new CelebornIOException( + "Connection to " + + NettyUtils.getRemoteAddress(channel) + + " is pinned to a netty event loop that is no longer usable and cannot make progress"); } public SocketAddress getSocketAddress() { @@ -179,6 +226,12 @@ public class TransportClient implements Closeable { } long requestId = requestId(); + if (isEventLoopDead()) { + // Unlike pushes and fetches, an outstanding RPC has no timeout checker to fall back on, so a + // callback orphaned on a dead loop hangs its owner forever. Fail up front instead. + callback.onFailure(deadEventLoopException()); + return requestId; + } handler.addRpcRequest(requestId, callback); RpcChannelListener listener = new RpcChannelListener(requestId); diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java index b79383277a..189c309f18 100644 --- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java +++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java @@ -23,7 +23,10 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.List; import java.util.Random; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -96,7 +99,25 @@ public class TransportClientFactory implements Closeable { private final int sendBuf; private final Class<? extends Channel> socketChannelClass; - private EventLoopGroup workerGroup; + private final IOMode ioMode; + // The client worker EventLoopGroup new connections bind to. Replaced wholesale when one of its + // loops is found dead (see recreateWorkerGroup and TransportClient#isEventLoopDead); volatile so + // the swap is visible to concurrent callers. + private volatile EventLoopGroup workerGroup; + // Superseded worker groups, not shut down eagerly because their still-live threads may be + // serving already-open channels. See retireWorkerGroupIfDrained. + private final List<EventLoopGroup> supersededWorkerGroups = new CopyOnWriteArrayList<>(); + // Channels currently open on each worker group, keyed by group identity. + private final ConcurrentHashMap<EventLoopGroup, Set<Channel>> workerGroupChannels; + // Whether to recreate the worker group on a dead event loop (SPARK-58292); on by default. + private final boolean recreateWorkerGroupOnDeadEventLoop; + // Makes each recreated group's thread names distinct, and read outside the lock to detect that a + // recreation happened. Mutated under the recreateWorkerGroup lock. + private volatile int workerGroupRecreationCount; + // Set once close() has shut the worker group down, so the dead-event-loop path does not + // resurrect a closed factory by recreating a fresh (leaked) group. Guarded by the same lock as + // recreateWorkerGroup. + private boolean closed = false; protected ByteBufAllocator allocator; private final int maxClientConnectRetries; private final int maxClientConnectRetryWaitTimeMs; @@ -107,6 +128,7 @@ public class TransportClientFactory implements Closeable { TransportConf conf = context.getConf(); this.clientBootstraps = Lists.newArrayList(Preconditions.checkNotNull(clientBootstraps)); this.connectionPool = JavaUtils.newConcurrentHashMap(); + this.workerGroupChannels = JavaUtils.newConcurrentHashMap(); this.numConnectionsPerPeer = conf.numConnectionsPerPeer(); this.connectTimeoutMs = conf.connectTimeoutMs(); this.connectionTimeoutMs = conf.connectionTimeoutMs(); @@ -115,7 +137,7 @@ public class TransportClientFactory implements Closeable { this.sendBuf = conf.sendBuf(); this.rand = new Random(); - IOMode ioMode = IOMode.valueOf(conf.ioMode()); + this.ioMode = IOMode.valueOf(conf.ioMode()); this.socketChannelClass = NettyUtils.getClientChannelClass(ioMode); logger.info("Module {} mode {} threads {}", conf.getModuleName(), ioMode, conf.clientThreads()); this.workerGroup = @@ -124,6 +146,7 @@ public class TransportClientFactory implements Closeable { conf.clientThreads(), conf.conflictAvoidChooserEnable(), conf.getModuleName() + "-client"); + this.recreateWorkerGroupOnDeadEventLoop = conf.recreateWorkerGroupOnDeadEventLoop(); // Always disable thread-local cache when creating pooled ByteBuf allocator for TransportClients // because the ByteBufs are allocated by the event loop thread, but released by the executor // thread rather than the event loop thread. Those thread-local caches actually delay the @@ -134,6 +157,17 @@ public class TransportClientFactory implements Closeable { this.maxClientConnectRetryWaitTimeMs = conf.ioRetryWaitTimeMs(); } + @VisibleForTesting + public EventLoopGroup getWorkerGroup() { + return workerGroup; + } + + /** How many worker groups superseded after a dead event loop have not been retired yet. */ + @VisibleForTesting + public int supersededWorkerGroupCount() { + return supersededWorkerGroups.size(); + } + /** * Create a {@link TransportClient} connecting to the given remote host / port. * @@ -239,23 +273,31 @@ public class TransportClientFactory implements Closeable { "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); + return clientPool.clients[clientIndex]; } - clientPool.clients[clientIndex] = internalCreateClient(resolvedAddress, decoder); - return clientPool.clients[clientIndex]; + } finally { + // Runs once the pool lock above has been released: failing a client's outstanding requests + // invokes user callbacks, which may re-enter the factory and take another pool lock, so it + // must never happen while holding one. + failClientsOnDeadEventLoopsIfRecreated(recreationCountBefore); } } @@ -273,21 +315,41 @@ public class TransportClientFactory implements Closeable { public TransportClient createUnmanagedClient(String remoteHost, int remotePort) throws IOException, InterruptedException { final InetSocketAddress address = new InetSocketAddress(remoteHost, remotePort); - return internalCreateClient(address, NettyUtils.createFrameDecoder()); + final int recreationCountBefore = workerGroupRecreationCount; + try { + return internalCreateClient(address, NettyUtils.createFrameDecoder()); + } finally { + failClientsOnDeadEventLoopsIfRecreated(recreationCountBefore); + } + } + + private TransportClient internalCreateClient( + InetSocketAddress address, ChannelInboundHandlerAdapter decoder) + throws IOException, InterruptedException { + return internalCreateClient(address, decoder, true); } /** - * Create a completely new {@link TransportClient} to the given remote host / port. This - * connection is not pooled. + * Connect to the given address on the current worker group. * - * <p>As with {@link #createClient(String, int)}, this method is blocking. + * @param retryOnRecreatedWorkerGroup whether to immediately reconnect once if this attempt failed + * on a dead event loop and a fresh worker group was installed in response. That reconnect is + * the whole point of the recreation, so it must not be charged to the caller's I/O retry + * budget (celeborn.$module.io.maxRetries, which may be as low as 1, and which sleeps + * celeborn.$module.io.retryWait between attempts) and must also cover callers that have no + * retry wrapper at all, such as {@link #createUnmanagedClient}. */ private TransportClient internalCreateClient( - InetSocketAddress address, ChannelInboundHandlerAdapter decoder) + InetSocketAddress address, + ChannelInboundHandlerAdapter decoder, + boolean retryOnRecreatedWorkerGroup) throws IOException, InterruptedException { Bootstrap bootstrap = new Bootstrap(); + // Capture the group this connection uses, so that on a dead-event-loop failure we replace + // exactly this group (and not one a concurrent caller already swapped in). + final EventLoopGroup connectGroup = workerGroup; bootstrap - .group(workerGroup) + .group(connectGroup) .channel(socketChannelClass) // Disable Nagle's Algorithm since we don't want packets to wait .option(ChannelOption.TCP_NODELAY, true) @@ -317,29 +379,45 @@ public class TransportClientFactory implements Closeable { // 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); } - } 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()); + throw e; } + trackChannel(connectGroup, cf.channel()); if (context.sslEncryptionEnabled()) { final SslHandler sslHandler = cf.channel().pipeline().get(SslHandler.class); sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs); @@ -432,6 +510,181 @@ public class TransportClientFactory implements Closeable { } } + /** + * 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()) { + // Read without the pool lock: this is a best-effort sweep, and taking the lock here would + // invert the pool-then-factory lock order that createClient establishes. + for (TransportClient client : clientPool.clients) { + if (client == null || !client.isEventLoopDead()) { + continue; + } + client.invalidateIfEventLoopDead(); + // A dead loop never completes the close future, so untrack here or the group never retires. + Channel channel = client.getChannel(); + untrackChannel(channel.eventLoop().parent(), channel); + } + } + } + /** Close all connections in the connection pool, and shutdown the worker thread pool. */ @Override public void close() { @@ -447,10 +700,27 @@ public class TransportClientFactory implements Closeable { } connectionPool.clear(); + // Mark closed under the recreateWorkerGroup lock before shutting the group down, so a + // concurrent createClient() hitting the dead-event-loop path cannot recreate a fresh group + // after we have decided to close (which would leak threads). + synchronized (this) { + closed = true; + } + // SPARK-19147 if (workerGroup != null && !workerGroup.isShuttingDown()) { workerGroup.shutdownGracefully(); } + + // Backstop for worker groups superseded after a dead-event-loop recreation whose channels + // never drained, so retireWorkerGroupIfDrained could not shut them down earlier. + for (EventLoopGroup group : supersededWorkerGroups) { + if (!group.isShuttingDown()) { + group.shutdownGracefully(); + } + } + supersededWorkerGroups.clear(); + workerGroupChannels.clear(); } public TransportContext getContext() { diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java index 4ad8373c02..c3c99958a0 100644 --- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java +++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java @@ -263,6 +263,32 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> { } } + /** + * 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) { + pushCheckerScheduleFuture.cancel(false); + } + if (fetchCheckerScheduleFuture != null) { + fetchCheckerScheduleFuture.cancel(false); + } + } + @Override public void channelActive() {} diff --git a/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java b/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java index ce37ae9788..8cb809fa35 100644 --- a/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java +++ b/common/src/main/java/org/apache/celeborn/common/network/util/TransportConf.java @@ -92,6 +92,16 @@ public class TransportConf { return celebornConf.networkIoConflictAvoidChooserEnable(module); } + /** + * Whether to replace the client worker EventLoopGroup when a netty worker event loop is detected + * as dead (a connection is rejected with "event executor terminated"). A dead loop is never + * replaced within a fixed-size group and keeps being handed out by the round-robin chooser, so + * without this the degradation is permanent. On by default. See SPARK-58292. + */ + public boolean recreateWorkerGroupOnDeadEventLoop() { + return celebornConf.networkIoRecreateWorkerGroupOnDeadEventLoop(module); + } + /** * Receive buffer size (SO_RCVBUF). Note: the optimal size for receive buffer and send buffer * should be latency * network_bandwidth. Assuming latency = 1ms, network_bandwidth = 10Gbps diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala index 6dce12fec6..8f9bb59211 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -585,6 +585,10 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se getBoolean(key, NETWORK_IO_CLIENT_CONFLICT_AVOID_CHOOSER_ENABLE.defaultValue.get) } + def networkIoRecreateWorkerGroupOnDeadEventLoop(module: String): Boolean = { + getTransportConfBoolean(module, NETWORK_IO_RECREATE_WORKER_GROUP_ON_DEAD_EVENT_LOOP) + } + def networkIoReceiveBuf(module: String): Int = { getTransportConfSizeAsBytes(module, NETWORK_IO_RECEIVE_BUFFER).toInt } @@ -2296,6 +2300,27 @@ object CelebornConf extends Logging { .booleanConf .createWithDefault(false) + val NETWORK_IO_RECREATE_WORKER_GROUP_ON_DEAD_EVENT_LOOP: ConfigEntry[Boolean] = + buildConf("celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop") + .categories("network") + .version("1.0.0") + .doc("Whether to replace the netty client worker EventLoopGroup when one of its event-loop " + + "threads is detected as dead (a connection is rejected with \"event executor " + + "terminated\"). A dead event loop is never replaced within a fixed-size group and keeps " + + "being handed out by the round-robin chooser, permanently poisoning any channel pinned " + + "to it, so a request on it can hang forever. When enabled, such a failure recreates the " + + "worker group so subsequent connections bind to live threads and the retry can succeed. " + + s"If setting <module> to `${TransportModuleConstants.RPC_APP_MODULE}`, " + + s"works for shuffle client. " + + s"If setting <module> to `${TransportModuleConstants.RPC_SERVICE_MODULE}`, " + + s"works for master or worker. " + + s"If setting <module> to `${TransportModuleConstants.DATA_MODULE}`, " + + s"it works for shuffle client push and fetch data. " + + s"If setting <module> to `${TransportModuleConstants.REPLICATE_MODULE}`, " + + s"it works for replicate client of worker replicating data to peer worker.") + .booleanConf + .createWithDefault(true) + val NETWORK_IO_RECEIVE_BUFFER: ConfigEntry[Long] = buildConf("celeborn.<module>.io.receiveBuffer") .categories("network") diff --git a/common/src/test/java/org/apache/celeborn/common/network/SSLTransportClientFactorySuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/SSLTransportClientFactorySuiteJ.java index 8d60f68a90..90cfa5d2f2 100644 --- a/common/src/test/java/org/apache/celeborn/common/network/SSLTransportClientFactorySuiteJ.java +++ b/common/src/test/java/org/apache/celeborn/common/network/SSLTransportClientFactorySuiteJ.java @@ -28,13 +28,17 @@ import org.apache.celeborn.common.network.ssl.SslSampleConfigs; public class SSLTransportClientFactorySuiteJ extends TransportClientFactorySuiteJ { + /** Set up SSL for TEST_MODULE, for the suite's own servers and for any client a test builds. */ + @Override + protected CelebornConf newCelebornConf() { + return TestHelper.updateCelebornConfWithMap( + new CelebornConf(), SslSampleConfigs.createDefaultConfigMapForModule(TEST_MODULE)); + } + @Before @Override public void setUp() { - // set up SSL for TEST_MODULE - doSetup( - TestHelper.updateCelebornConfWithMap( - new CelebornConf(), SslSampleConfigs.createDefaultConfigMapForModule(TEST_MODULE))); + doSetup(newCelebornConf()); } @After diff --git a/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java index 3c51a175d3..923ba6b439 100644 --- a/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java +++ b/common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java @@ -25,9 +25,13 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import java.io.IOException; +import java.nio.ByteBuffer; import java.util.*; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import io.netty.channel.EventLoopGroup; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -35,6 +39,7 @@ import org.junit.Test; import org.mockito.Mockito; import org.apache.celeborn.common.CelebornConf; +import org.apache.celeborn.common.network.client.RpcResponseCallback; import org.apache.celeborn.common.network.client.TransportClient; import org.apache.celeborn.common.network.client.TransportClientFactory; import org.apache.celeborn.common.network.server.BaseMessageHandler; @@ -62,7 +67,17 @@ public class TransportClientFactorySuiteJ { @Before public void setUp() { - doSetup(new CelebornConf()); + doSetup(newCelebornConf()); + } + + /** + * A fresh conf carrying whatever this suite needs on top of the defaults. Tests that build their + * own TransportContext must start from this rather than a bare CelebornConf, or they would run a + * plain client against the SSL server1/server2 that {@link SSLTransportClientFactorySuiteJ} sets + * up, silently testing something other than what the subclass exists to cover. + */ + protected CelebornConf newCelebornConf() { + return new CelebornConf(); } // for validation in subclasses @@ -87,7 +102,7 @@ public class TransportClientFactorySuiteJ { private void testClientReuse(int maxConnections, boolean concurrent) throws IOException, InterruptedException { - CelebornConf _conf = new CelebornConf(); + CelebornConf _conf = newCelebornConf(); _conf.set("celeborn.shuffle.io.numConnectionsPerPeer", Integer.toString(maxConnections)); TransportConf conf = new TransportConf(TEST_MODULE, _conf); @@ -198,7 +213,7 @@ public class TransportClientFactorySuiteJ { @Test public void closeIdleConnectionForRequestTimeOut() throws IOException, InterruptedException { - CelebornConf _conf = new CelebornConf(); + CelebornConf _conf = newCelebornConf(); _conf.set("celeborn.shuffle.io.connectionTimeout", "1s"); TransportConf conf = new TransportConf(TEST_MODULE, _conf); TransportContext context = new TransportContext(conf, new BaseMessageHandler(), true); @@ -214,16 +229,21 @@ public class TransportClientFactorySuiteJ { context.close(); } - @Test(expected = IOException.class) - public void closeFactoryBeforeCreateClient() throws IOException, InterruptedException { + @Test + public void closeFactoryBeforeCreateClient() { TransportClientFactory factory = context.createClientFactory(); + EventLoopGroup groupBeforeClose = factory.getWorkerGroup(); factory.close(); - factory.createClient(getLocalHost(), server1.getPort()); + assertThrows(IOException.class, () -> factory.createClient(getLocalHost(), server1.getPort())); + // SPARK-58292: createClient on a closed factory fails with the terminated-executor cause, but + // the closed factory must NOT recreate a fresh worker group (which would leak threads). The + // group is left as-is, i.e. the shut-down one from before close(). + assertSame(groupBeforeClose, factory.getWorkerGroup()); } @Test public void unlimitedConnectionAndCreationTimeouts() throws IOException, InterruptedException { - CelebornConf _conf = new CelebornConf(); + CelebornConf _conf = newCelebornConf(); _conf.set("celeborn.shuffle.io.connectTimeout", "-1"); _conf.set("celeborn.shuffle.io.connectionTimeout", "-1"); TransportConf conf = new TransportConf(TEST_MODULE, _conf); @@ -248,6 +268,164 @@ public class TransportClientFactorySuiteJ { } } + /** + * A dead netty worker event loop is never replaced within a fixed-size group and permanently + * poisons connections (SPARK-58292). Simulate it by shutting the factory's whole worker group + * down: the next connection's channel registration is then rejected with "event executor + * terminated", exactly as it would be by a single loop whose thread has died. + */ + private static EventLoopGroup simulateDeadWorkerEventLoop(TransportClientFactory factory) + throws InterruptedException { + EventLoopGroup deadGroup = factory.getWorkerGroup(); + deadGroup.shutdownGracefully().sync(); + assertTrue(deadGroup.isShuttingDown()); + return deadGroup; + } + + private TransportContext newContext(CelebornConf celebornConf) { + return new TransportContext( + new TransportConf(TEST_MODULE, celebornConf), new BaseMessageHandler()); + } + + @Test + public void recreatesWorkerGroupWhenEventLoopIsDead() throws Exception { + CelebornConf _conf = newCelebornConf(); + _conf.set("celeborn.shuffle.io.retryWait", "100ms"); + TransportContext ctx = newContext(_conf); + try (TransportClientFactory factory = ctx.createClientFactory()) { + EventLoopGroup deadGroup = simulateDeadWorkerEventLoop(factory); + + // The first connect attempt fails because the (dead) group rejects the channel + // registration; the factory replaces the worker group and reconnects on it inline. + TransportClient client = factory.createClient(getLocalHost(), server1.getPort()); + assertTrue(client.isActive()); + + EventLoopGroup freshGroup = factory.getWorkerGroup(); + assertNotSame(deadGroup, freshGroup); + assertFalse(freshGroup.isShuttingDown()); + } finally { + ctx.close(); + } + } + + @Test + public void retiresSupersededWorkerGroupWithNoChannelsLeft() throws Exception { + // SPARK-58292: a superseded group must not be retained, along with its selector threads, until + // the factory closes. See TransportClientFactory#retireWorkerGroupIfDrained. + TransportContext ctx = newContext(newCelebornConf()); + try (TransportClientFactory factory = ctx.createClientFactory()) { + simulateDeadWorkerEventLoop(factory); + + assertTrue(factory.createClient(getLocalHost(), server1.getPort()).isActive()); + + // The superseded group had no channels left to serve, so it was retired at once rather than + // being parked on the retained list for close() to deal with. + assertEquals(0, factory.supersededWorkerGroupCount()); + } finally { + ctx.close(); + } + } + + @Test + public void recreatedWorkerGroupIsUsedWithoutConsumingTheRetryBudget() throws Exception { + // SPARK-58292: replacing the dead group only helps the triggering request if that request can + // actually use the replacement. celeborn.<module>.io.maxRetries may be as low as 1, leaving no + // retry to spend on the fresh group, so the reconnect must happen inline instead of being + // charged to retryCreateClient. + CelebornConf _conf = newCelebornConf(); + _conf.set("celeborn.shuffle.io.maxRetries", "1"); + TransportContext ctx = newContext(_conf); + try (TransportClientFactory factory = ctx.createClientFactory()) { + EventLoopGroup deadGroup = simulateDeadWorkerEventLoop(factory); + + TransportClient client = factory.createClient(getLocalHost(), server1.getPort()); + assertTrue(client.isActive()); + assertNotSame(deadGroup, factory.getWorkerGroup()); + } finally { + ctx.close(); + } + } + + @Test + public void createUnmanagedClientRecoversFromDeadEventLoop() throws Exception { + // createUnmanagedClient has no retry wrapper at all, so it can only recover from a dead event + // loop if the reconnect on the recreated group happens inline. + TransportContext ctx = newContext(newCelebornConf()); + try (TransportClientFactory factory = ctx.createClientFactory()) { + EventLoopGroup deadGroup = simulateDeadWorkerEventLoop(factory); + + TransportClient client = factory.createUnmanagedClient(getLocalHost(), server1.getPort()); + assertTrue(client.isActive()); + assertNotSame(deadGroup, factory.getWorkerGroup()); + } finally { + ctx.close(); + } + } + + @Test + public void failsOutstandingRequestsOfPooledClientsOnDeadEventLoops() throws Exception { + // SPARK-58292: the request that recovers the worker group must also unblock owners that are + // still holding a client pinned to the dead loop -- a dead loop delivers neither the write + // listener nor channelInactive(), so nothing else would ever fail their callbacks. + // + // shutdownGracefully's quiet period is the one window where a loop already reports + // isShuttingDown() while its channels are still open, i.e. exactly the state a dead loop leaves + // a pooled client in. The period is generous relative to this test body, and the assertion + // below fails loudly rather than silently testing the wrong thing if it ever elapses early. + TransportContext ctx = newContext(newCelebornConf()); + try (TransportClientFactory factory = ctx.createClientFactory()) { + TransportClient client = factory.createClient(getLocalHost(), server1.getPort()); + AtomicReference<Throwable> failure = new AtomicReference<>(); + client + .getHandler() + .addRpcRequest( + 1L, + new RpcResponseCallback() { + @Override + public void onSuccess(ByteBuffer response) {} + + @Override + public void onFailure(Throwable e) { + failure.set(e); + } + }); + + factory.getWorkerGroup().shutdownGracefully(30, 60, TimeUnit.SECONDS); + assertTrue( + "quiet period elapsed too early to exercise the sweep", client.getChannel().isOpen()); + assertFalse(client.isActive()); + assertTrue(client.getHandler().hasOutstandingRequests()); + + factory.failClientsOnDeadEventLoops(); + + assertNotNull("the pooled client's outstanding RPC must be failed", failure.get()); + assertFalse(client.getHandler().hasOutstandingRequests()); + } finally { + ctx.close(); + } + } + + @Test + public void doesNotRecreateWorkerGroupWhenDisabled() throws Exception { + // With the recreation disabled, a dead worker group is NOT recreated: createClient still + // fails, but the worker group is left unchanged. + CelebornConf _conf = newCelebornConf(); + _conf.set("celeborn.shuffle.io.recreateWorkerGroupOnDeadEventLoop", "false"); + _conf.set("celeborn.shuffle.io.retryWait", "100ms"); + TransportContext ctx = newContext(_conf); + try (TransportClientFactory factory = ctx.createClientFactory()) { + EventLoopGroup deadGroup = simulateDeadWorkerEventLoop(factory); + + assertThrows( + IOException.class, () -> factory.createClient(getLocalHost(), server1.getPort())); + assertSame(deadGroup, factory.getWorkerGroup()); + // Nothing was superseded, so none of the retirement bookkeeping kicked in either. + assertEquals(0, factory.supersededWorkerGroupCount()); + } finally { + ctx.close(); + } + } + @Test public void testRetryCreateClient() throws IOException, InterruptedException { TransportClientFactory factory = Mockito.spy(context.createClientFactory()); diff --git a/common/src/test/java/org/apache/celeborn/common/network/client/TransportClientSuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/client/TransportClientSuiteJ.java new file mode 100644 index 0000000000..dd04e3fbbb --- /dev/null +++ b/common/src/test/java/org/apache/celeborn/common/network/client/TransportClientSuiteJ.java @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.celeborn.common.network.client; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicReference; + +import io.netty.channel.Channel; +import io.netty.channel.EventLoop; +import org.junit.Before; +import org.junit.Test; + +import org.apache.celeborn.common.CelebornConf; +import org.apache.celeborn.common.network.util.TransportConf; + +public class TransportClientSuiteJ { + + private Channel channel; + private EventLoop eventLoop; + private TransportResponseHandler handler; + private TransportClient client; + + @Before + public void setUp() { + TransportConf conf = new TransportConf("shuffle", new CelebornConf()); + channel = mock(Channel.class); + eventLoop = mock(EventLoop.class); + when(channel.eventLoop()).thenReturn(eventLoop); + handler = new TransportResponseHandler(conf, channel); + client = new TransportClient(channel, handler); + } + + @Test + public void isActiveFalseWhenEventLoopIsShuttingDown() { + // SPARK-58292: a client whose netty event loop has terminated must not be treated as active, + // even though the TCP channel still reports open/active (nothing can close it -- closing runs + // on the now-dead loop). + when(channel.isOpen()).thenReturn(true); + when(channel.isActive()).thenReturn(true); + + // Live event loop -> active. + when(eventLoop.isShuttingDown()).thenReturn(false); + assertTrue(client.isActive()); + + // Terminated event loop -> NOT active, so it will be evicted from the pool and not reused. + when(eventLoop.isShuttingDown()).thenReturn(true); + assertFalse(client.isActive()); + } + + @Test + public void sendRpcFailsFastWhenEventLoopIsDead() { + // SPARK-58292: a request written to a dead loop orphans, and unlike pushes and fetches an + // outstanding RPC has no timeout checker to fall back on. + when(eventLoop.isShuttingDown()).thenReturn(true); + + AtomicReference<Throwable> failure = new AtomicReference<>(); + client.sendRpc(ByteBuffer.allocate(8), new CapturingCallback(failure)); + + assertNotNull("callback must be failed, not left outstanding", failure.get()); + assertFalse(handler.hasOutstandingRequests()); + // The request was never handed to the dead loop. + verify(channel, never()).writeAndFlush(any()); + } + + @Test + public void invalidatesClientWhenEventLoopIsDead() { + // SPARK-58292: owners that hold a client for the lifetime of a stream (e.g. Flink's + // CelebornBufferStream) never reacquire it from the factory, so marking it inactive is not + // enough -- their in-flight callbacks must be failed synchronously or they wait forever. + when(eventLoop.isShuttingDown()).thenReturn(false); + + // An RPC issued while the loop was still alive stays outstanding. + AtomicReference<Throwable> failure = new AtomicReference<>(); + handler.addRpcRequest(1L, new CapturingCallback(failure)); + assertTrue(handler.hasOutstandingRequests()); + + // A healthy client is left alone. + client.invalidateIfEventLoopDead(); + assertTrue(handler.hasOutstandingRequests()); + assertNull(failure.get()); + + when(eventLoop.isShuttingDown()).thenReturn(true); + client.invalidateIfEventLoopDead(); + assertNotNull(failure.get()); + assertFalse(handler.hasOutstandingRequests()); + + // Idempotent: a second sweep over an already drained handler is a no-op. + client.invalidateIfEventLoopDead(); + assertFalse(handler.hasOutstandingRequests()); + } + + private static class CapturingCallback implements RpcResponseCallback { + private final AtomicReference<Throwable> failure; + + CapturingCallback(AtomicReference<Throwable> failure) { + this.failure = failure; + } + + @Override + public void onSuccess(ByteBuffer response) {} + + @Override + public void onFailure(Throwable e) { + failure.set(e); + } + } +} diff --git a/docs/configuration/network.md b/docs/configuration/network.md index 636bca1bcb..9ee360e11a 100644 --- a/docs/configuration/network.md +++ b/docs/configuration/network.md @@ -33,6 +33,7 @@ license: | | celeborn.<module>.io.numConnectionsPerPeer | 1 | false | Number of concurrent connections between two nodes. If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `replicate`, it works for replicate client of worker replicating data to peer worker. | | | | celeborn.<module>.io.preferDirectBufs | true | false | If true, we will prefer allocating off-heap byte buffers within Netty. If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `push`, it works for worker receiving push data. If setting <module> to `replicate`, it works for replicate server or client of w [...] | celeborn.<module>.io.receiveBuffer | 0b | false | Receive buffer size (SO_RCVBUF). Note: the optimal size for receive buffer and send buffer should be latency * network_bandwidth. Assuming latency = 1ms, network_bandwidth = 10Gbps buffer size should be ~ 1.25MB. If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <mod [...] +| celeborn.<module>.io.recreateWorkerGroupOnDeadEventLoop | true | false | Whether to replace the netty client worker EventLoopGroup when one of its event-loop threads is detected as dead (a connection is rejected with "event executor terminated"). A dead event loop is never replaced within a fixed-size group and keeps being handed out by the round-robin chooser, permanently poisoning any channel pinned to it, so a request on it can hang forever. When enabled, such a failure recrea [...] | celeborn.<module>.io.retryWait | 5s | false | Time that we will wait in order to perform a retry after an IOException. Only relevant if maxIORetries > 0. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `replicate`, it works for replicate client of worker replicating data to peer worker. If setting <module> to `push`, it works for Flink shuffle client push data. | 0.2.0 | | | celeborn.<module>.io.saslTimeout | 30s | false | Timeout for a single round trip of auth message exchange, in milliseconds. | 0.5.0 | | | celeborn.<module>.io.sendBuffer | 0b | false | Send buffer size (SO_SNDBUF). If setting <module> to `rpc_app`, works for shuffle client. If setting <module> to `rpc_service`, works for master or worker. If setting <module> to `data`, it works for shuffle client push and fetch data. If setting <module> to `push`, it works for worker receiving push data. If setting <module> to `replicate`, it works for replicate server or client of worker replicating data to peer worker. If setting [...]
