This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 9.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 3d3cd686050d2e478190cbec46cdf0a5765999a7 Author: opencode <[email protected]> AuthorDate: Sun Oct 11 16:10:01 2026 +0200 Do not poll or swallow interrupts when waiting for a pooled sender SenderQueue.getSender() parks on the queue monitor until a sender becomes available. Two problems made the wait misbehave as soon as maxWait was zero or negative: - The expiry test was "delta > timeout && timeout > 0", which can never succeed for a non-positive timeout, and the wait amount was then forced to 1 ms. The result was an endless loop waking every millisecond: continuous CPU churn while a blocked cluster send could not be woken by anything other than a sender eventually being returned. - The InterruptedException was ignored without restoring the interrupt status, so interrupting such a thread had no effect at all. It would only exit later, when the queue was closed, and then via an unchecked IllegalStateException escaping the caller. Wait once for the full remaining timeout when a positive timeout is configured, and wait without a timeout when it is not. No polling is needed: returnSender(), open() and close() all notify every waiting thread under the same monitor. On interruption, restore the interrupt status and return null, which the single caller, PooledParallelSender .sendMessage(), already reports as a ChannelException for the destination members. Document the non-positive maxWait semantics on getSender() and setMaxWait(); this preserves the observable behaviour, as a non-positive timeout already blocked indefinitely before. Since the wait no longer re-checks periodically, make the pool limit a notifier as well: setLimit() is now synchronized and wakes the waiting threads. Raising the limit can make room for a new sender, and previously only the 1 ms polling happened to notice it. Synchronizing the setter also removes an unsynchronized write of the limit, which getSender() and returnSender() read under the monitor. --- .../catalina/tribes/transport/PooledSender.java | 42 +++++++++++++++------- 1 file changed, 30 insertions(+), 12 deletions(-) diff --git a/java/org/apache/catalina/tribes/transport/PooledSender.java b/java/org/apache/catalina/tribes/transport/PooledSender.java index 80e1d961a2..c4b46a451a 100644 --- a/java/org/apache/catalina/tribes/transport/PooledSender.java +++ b/java/org/apache/catalina/tribes/transport/PooledSender.java @@ -52,8 +52,12 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS public abstract DataSender getNewDataSender(); /** - * Retrieves a sender from the pool, waiting up to maxWait milliseconds. - * @return a DataSender from the pool, or null if timeout expires + * Retrieves a sender from the pool, waiting up to maxWait milliseconds. If maxWait is zero or negative, the call + * waits indefinitely for a sender to become available. + * + * @return a DataSender from the pool, or null if the wait timed out or the calling thread was interrupted + * + * @throws IllegalStateException if the sender pool has been closed */ public DataSender getSender() { return queue.getSender(getMaxWait()); @@ -124,7 +128,9 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS } /** - * Sets the maximum time to wait for a sender from the pool, in milliseconds. + * Sets the maximum time to wait for a sender from the pool, in milliseconds. A value of zero or less means the + * wait for a sender is unbounded. + * * @param maxWait the maximum wait time */ public void setMaxWait(long maxWait) { @@ -172,15 +178,17 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS /** * @return Returns the limit. */ - public int getLimit() { + public synchronized int getLimit() { return limit; } /** * @param limit The limit to set. */ - public void setLimit(int limit) { + public synchronized void setLimit(int limit) { this.limit = limit; + // Raising the limit can make room for a new sender to be created + notifyAll(); } public synchronized int getInUsePoolSize() { @@ -217,15 +225,25 @@ public abstract class PooledSender extends AbstractSender implements MultiPointS inuse.add(sender); return sender; } - long delta = System.currentTimeMillis() - start; - if (delta > timeout && timeout > 0) { - return null; - } - long remaining = timeout > 0 ? Math.max(timeout - delta, 1) : 1; try { - wait(remaining); + if (timeout > 0) { + long delta = System.currentTimeMillis() - start; + if (delta >= timeout) { + return null; + } + // Everything that can make a sender available - returnSender(), + // open(), close() and setLimit() - notifies every waiting + // thread, so there is no need to poll. + wait(timeout - delta); + } else { + // No timeout configured: wait until a sender becomes + // available or the queue is closed. + wait(); + } } catch (InterruptedException x) { - // Ignore + // Restore the interrupt status and let the caller handle it + Thread.currentThread().interrupt(); + return null; } } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
