SteNicholas commented on code in PR #3778:
URL: https://github.com/apache/celeborn/pull/3778#discussion_r3755898715
##########
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:
Correct on both counts, and the `WeakReference` was worse than useless -- it
bought nothing while making the retention look bounded. A running netty thread
keeps its `SingleThreadEventExecutor` reachable, and
`AbstractEventExecutor.parent` keeps the group reachable, so the referent was
never going to be cleared. The javadoc claiming the group was "not yet
collected" described something that could not happen.
Changed to a strong `supersededWorkerGroups` list plus per-group channel
tracking:
- `trackChannel` registers each newly connected channel against the group it
is pinned to, and a `closeFuture` listener untracks it.
- `retireWorkerGroupIfDrained` shuts the group down once it is superseded
and has no channels left. `close()` remains the backstop for any group that
never drains.
- A channel on the dead loop never completes its `closeFuture`, so
`failClientsOnDeadEventLoops` untracks it explicitly -- otherwise that one
channel would pin the group forever and the drain condition could never be met.
- Tracking is skipped entirely when `recreateWorkerGroupOnDeadEventLoop` is
off, since retirement cannot happen then and the bookkeeping would be pure
overhead.
The log line now carries the module, the new thread prefix, and how many
already-open channels the superseded group is still serving, so the retention
is visible rather than inferred.
One residual I want to be explicit about: retiring the group reclaims the
*other* `clientThreads() - 1` selector threads, but the socket pinned to the
dead loop still cannot be closed, for the same reason force-close does not work
in your other comment.
Covered by
`TransportClientFactorySuiteJ.retiresSupersededWorkerGroupWithNoChannelsLeft`.
I could not build a deterministic test for the complementary case (a superseded
group that still has open channels, retired later once they drain) --
constructing it needs one loop of a multi-loop group to die while the others
stay live, which I could not do without reaching into netty internals.
`close()` covers it as a backstop.
--
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]
