This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit be0483d6a743b8a1b102a0142ecb34cf87483a6b Author: opencode <[email protected]> AuthorDate: Sun Oct 11 15:37:52 2026 +0200 Wake up pool waiters on stop and never leak receive tasks RxTaskPool.stop() never notified the monitor that threads parked in getRxTask() wait on when the pool is exhausted, so such a thread would never observe that running had become false and would hang forever. stop() now closes and clears the idle tasks and calls notifyAll() under the pool monitor. Relatedly, NioReplicationTask.run() only returned its worker to the pool as the last statement of the method. A java.lang.Error escaping the catch(Exception) block, or the defensive early return when the key is null, permanently removed one worker from the receive task pool. Once all workers were lost, the single NioReceiver selector thread would block forever in getRxTask() and no further cluster message could ever be received. The key reset and the returnWorker() call now happen in a finally block so a worker can never leak. Also update the NioReplicationTask class javadoc, which documented the old early return without pool return, and drop the now unused Iterator import from RxTaskPool. --- .../catalina/tribes/transport/RxTaskPool.java | 12 +-- .../tribes/transport/nio/NioReplicationTask.java | 102 +++++++++++---------- 2 files changed, 59 insertions(+), 55 deletions(-) diff --git a/java/org/apache/catalina/tribes/transport/RxTaskPool.java b/java/org/apache/catalina/tribes/transport/RxTaskPool.java index c2169f7f96..07a872a8d0 100644 --- a/java/org/apache/catalina/tribes/transport/RxTaskPool.java +++ b/java/org/apache/catalina/tribes/transport/RxTaskPool.java @@ -17,7 +17,6 @@ package org.apache.catalina.tribes.transport; import java.util.ArrayList; -import java.util.Iterator; import java.util.List; /** @@ -151,12 +150,13 @@ public class RxTaskPool { public void stop() { running = false; synchronized (mutex) { - Iterator<AbstractRxTask> i = idle.iterator(); - while (i.hasNext()) { - AbstractRxTask worker = i.next(); - returnWorker(worker); - i.remove(); + for (AbstractRxTask worker : idle) { + worker.close(); } + idle.clear(); + // Wake up any thread parked in getRxTask() so that it sees that + // running is false and returns rather than waiting forever + mutex.notifyAll(); } } diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java index 94dfa766c0..e7eb825570 100644 --- a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java +++ b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java @@ -44,8 +44,7 @@ import org.apache.juli.logging.LogFactory; /** * A worker thread class which drains a channel and processes received messages. * Each instance services a single SelectionKey per invocation. After completion, - * the worker returns itself to its parent pool for reuse, unless the key is null - * in which case it returns early without doing so. + * the worker returns itself to its parent pool for reuse. */ public class NioReplicationTask extends AbstractRxTask { @@ -89,61 +88,66 @@ public class NioReplicationTask extends AbstractRxTask { // loop forever waiting for work to do @Override public synchronized void run() { - if (buffer == null) { - int size = getRxBufSize(); - if (key.channel() instanceof DatagramChannel) { - size = ChannelReceiver.MAX_UDP_SIZE; - } - if ((getOptions() & OPTION_DIRECT_BUFFER) == OPTION_DIRECT_BUFFER) { - buffer = ByteBuffer.allocateDirect(size); - } else { - buffer = ByteBuffer.allocate(size); - } - } else { - buffer.clear(); - } - if (key == null) { - return; // just in case - } - if (log.isTraceEnabled()) { - log.trace("Servicing key:" + key); - } - try { - ObjectReader reader = (ObjectReader) key.attachment(); - if (reader == null) { - if (log.isTraceEnabled()) { - log.trace("No object reader, cancelling:" + key); + if (buffer == null) { + int size = getRxBufSize(); + if (key.channel() instanceof DatagramChannel) { + size = ChannelReceiver.MAX_UDP_SIZE; } - cancelKey(key); - } else { - if (log.isTraceEnabled()) { - log.trace("Draining channel:" + key); + if ((getOptions() & OPTION_DIRECT_BUFFER) == OPTION_DIRECT_BUFFER) { + buffer = ByteBuffer.allocateDirect(size); + } else { + buffer = ByteBuffer.allocate(size); } - - drainChannel(key, reader); + } else { + buffer.clear(); } - } catch (Exception e) { - // this is common, since the sockets on the other - // end expire after a certain time. - if (e instanceof CancelledKeyException) { - // do nothing - } else if (e instanceof IOException) { - // don't spew out stack traces for IO exceptions unless debug is enabled. - if (log.isDebugEnabled()) { - log.debug(sm.getString("nioReplicationTask.unable.drainChannel.ioe", e.getMessage()), e); + if (key == null) { + return; // just in case + } + if (log.isTraceEnabled()) { + log.trace("Servicing key:" + key); + } + + try { + ObjectReader reader = (ObjectReader) key.attachment(); + if (reader == null) { + if (log.isTraceEnabled()) { + log.trace("No object reader, cancelling:" + key); + } + cancelKey(key); } else { - log.warn(sm.getString("nioReplicationTask.unable.drainChannel.ioe", e.getMessage())); + if (log.isTraceEnabled()) { + log.trace("Draining channel:" + key); + } + + drainChannel(key, reader); } - } else if (log.isErrorEnabled()) { - // this is a real error, log it. - log.error(sm.getString("nioReplicationTask.exception.drainChannel"), e); + } catch (Exception e) { + // this is common, since the sockets on the other + // end expire after a certain time. + if (e instanceof CancelledKeyException) { + // do nothing + } else if (e instanceof IOException) { + // don't spew out stack traces for IO exceptions unless debug is enabled. + if (log.isDebugEnabled()) { + log.debug(sm.getString("nioReplicationTask.unable.drainChannel.ioe", e.getMessage()), e); + } else { + log.warn(sm.getString("nioReplicationTask.unable.drainChannel.ioe", e.getMessage())); + } + } else if (log.isErrorEnabled()) { + // this is a real error, log it. + log.error(sm.getString("nioReplicationTask.exception.drainChannel"), e); + } + cancelKey(key); } - cancelKey(key); + } finally { + key = null; + // done, ready for more, return to pool -- done from a finally + // block so the worker can never leak out of the pool, not even + // when an Error escapes the catch block above + getTaskPool().returnWorker(this); } - key = null; - // done, ready for more, return to pool - getTaskPool().returnWorker(this); } /** --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
