This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/11.0.x by this push:
new 08b97c1ce1 Review "Wake up pool waiters on stop and never leak receive
tasks"
08b97c1ce1 is described below
commit 08b97c1ce199b048e003f716167b4183720f4247
Author: Mark Thomas <[email protected]>
AuthorDate: Fri Oct 23 17:55:44 2026 +0100
Review "Wake up pool waiters on stop and never leak receive tasks"
- Ensure task pool is stopped on receiver shutdown
- Additional error handling
- Fix possible busy spin
Co-authored by: CoPilot / GCP-5.6 Sol
Co-authored by: Claude / Sonnet 5.5
---
.../catalina/tribes/transport/RxTaskPool.java | 28 ++--
.../catalina/tribes/transport/nio/NioReceiver.java | 23 ++-
.../tribes/transport/nio/NioReplicationTask.java | 34 ++++-
.../catalina/tribes/transport/TestRxTaskPool.java | 159 +++++++++++++++++++++
.../tribes/transport/nio/TestNioReceiver.java | 40 ++++++
5 files changed, 260 insertions(+), 24 deletions(-)
diff --git a/java/org/apache/catalina/tribes/transport/RxTaskPool.java
b/java/org/apache/catalina/tribes/transport/RxTaskPool.java
index 518ba0d753..9608d82f06 100644
--- a/java/org/apache/catalina/tribes/transport/RxTaskPool.java
+++ b/java/org/apache/catalina/tribes/transport/RxTaskPool.java
@@ -41,7 +41,8 @@ public class RxTaskPool {
*
* @param maxTasks maximum number of tasks in the pool
* @param minTasks minimum number of tasks in the pool
- * @param creator factory for creating new receive tasks
+ * @param creator factory for creating new receive tasks
+ *
* @throws Exception if initialization fails
*/
public RxTaskPool(int maxTasks, int minTasks, TaskCreator creator) throws
Exception {
@@ -64,7 +65,7 @@ public class RxTaskPool {
/**
* Find an idle worker thread, creating a new one if needed. Blocks until
a worker is available.
*
- * @return a worker, or null if the pool has been stopped
+ * @return a worker, or null if the pool has been stopped or the calling
thread was interrupted
*/
public AbstractRxTask getRxTask() {
AbstractRxTask worker = null;
@@ -83,7 +84,10 @@ public class RxTaskPool {
try {
mutex.wait();
} catch (InterruptedException x) {
+ // Restore the interrupt status and stop waiting
rather than spin on a wait that would throw
+ // again immediately
Thread.currentThread().interrupt();
+ return null;
}
}
}
@@ -111,18 +115,14 @@ public class RxTaskPool {
* @param worker The worker
*/
public void returnWorker(AbstractRxTask worker) {
- if (running) {
- synchronized (mutex) {
- used.remove(worker);
- if (idle.size() < maxTasks && !idle.contains(worker)) {
- idle.add(worker); // let max be the upper limit
- } else {
- worker.close();
- }
- mutex.notifyAll();
+ synchronized (mutex) {
+ used.remove(worker);
+ if (running && idle.size() < maxTasks && !idle.contains(worker)) {
+ idle.add(worker); // let max be the upper limit
+ } else {
+ worker.close();
}
- } else {
- worker.close();
+ mutex.notifyAll();
}
}
@@ -148,8 +148,8 @@ public class RxTaskPool {
* Stop the pool, closing all idle tasks and preventing new tasks from
being acquired.
*/
public void stop() {
- running = false;
synchronized (mutex) {
+ running = false;
for (AbstractRxTask worker : idle) {
worker.close();
}
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
index cfd73dc64f..107e5ed1dc 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
@@ -192,6 +192,11 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
@Override
public void stop() {
+ RxTaskPool taskPool = getTaskPool();
+ if (taskPool != null) {
+ // Wake the listener if it is waiting for a receive task before
waiting for it to stop
+ taskPool.stop();
+ }
this.stopListening();
// Startup may fail before the channels are registered with the
selector or the listener thread is started.
if (serverChannel != null) {
@@ -739,15 +744,27 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
if (task == null) {
// No threads/tasks available, do nothing, the selection
// loop will keep calling this method. getRxTask() blocks until a
task
- // is released, so it only returns null after the pool has been
stopped.
+ // is released, so it only returns null after the pool has been
stopped
+ // or the calling thread has been interrupted.
if (log.isDebugEnabled()) {
log.debug(sm.getString("nioReceiver.noThread"));
}
} else {
// invoking this wakes up the worker thread then returns
// add task to thread pool
- task.serviceChannel(key);
- getExecutor().execute(task);
+ try {
+ task.serviceChannel(key);
+ getExecutor().execute(task);
+ } catch (RuntimeException | Error e) {
+ // The task will never run so cancel its key and hand it back,
otherwise the connection and task are
+ // lost
+ try {
+ task.cancel();
+ } catch (RuntimeException | Error cleanup) {
+ e.addSuppressed(cleanup);
+ }
+ throw e;
+ }
}
}
diff --git
a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
index e7eb825570..5d2587d3b9 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
@@ -42,9 +42,8 @@ import org.apache.juli.logging.Log;
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.
+ * 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.
*/
public class NioReplicationTask extends AbstractRxTask {
@@ -89,6 +88,9 @@ public class NioReplicationTask extends AbstractRxTask {
@Override
public synchronized void run() {
try {
+ if (key == null) {
+ return; // just in case
+ }
if (buffer == null) {
int size = getRxBufSize();
if (key.channel() instanceof DatagramChannel) {
@@ -102,9 +104,6 @@ public class NioReplicationTask extends AbstractRxTask {
} else {
buffer.clear();
}
- if (key == null) {
- return; // just in case
- }
if (log.isTraceEnabled()) {
log.trace("Servicing key:" + key);
}
@@ -141,6 +140,13 @@ public class NioReplicationTask extends AbstractRxTask {
}
cancelKey(key);
}
+ } catch (Error e) {
+ // The key no longer has any interest ops so nothing would ever
service it again. Cancel it so that the
+ // connection is released, then let the Error propagate.
+ if (key != null) {
+ cancelKey(key);
+ }
+ throw e;
} finally {
key = null;
// done, ready for more, return to pool -- done from a finally
@@ -150,6 +156,20 @@ public class NioReplicationTask extends AbstractRxTask {
}
}
+ /**
+ * Cancel the current selection key.
+ */
+ synchronized void cancel() {
+ try {
+ if (key != null) {
+ cancelKey(key);
+ }
+ } finally {
+ key = null;
+ getTaskPool().returnWorker(this);
+ }
+ }
+
/**
* Called to initiate a unit of work by this worker thread on the provided
SelectionKey object. This method is
* synchronized, as is the run() method, so only one key can be serviced
at a given time. Before waking the worker
@@ -288,7 +308,7 @@ public class NioReplicationTask extends AbstractRxTask {
/**
* Registers a selection key for read interest.
*
- * @param key The selection key
+ * @param key The selection key
* @param reader The object reader
*/
protected void registerForRead(final SelectionKey key, ObjectReader
reader) {
diff --git a/test/org/apache/catalina/tribes/transport/TestRxTaskPool.java
b/test/org/apache/catalina/tribes/transport/TestRxTaskPool.java
new file mode 100644
index 0000000000..399837fb95
--- /dev/null
+++ b/test/org/apache/catalina/tribes/transport/TestRxTaskPool.java
@@ -0,0 +1,159 @@
+/*
+ * 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.catalina.tribes.transport;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.transport.nio.NioReceiver;
+import org.apache.catalina.tribes.transport.nio.NioReplicationTask;
+
+public class TestRxTaskPool {
+
+ private static final long TIMEOUT = 10000;
+
+
+ private static class NoOpTask extends AbstractRxTask {
+ NoOpTask() {
+ super(null);
+ }
+
+ @Override
+ public void run() {
+ // NO-OP
+ }
+ }
+
+
+ /*
+ * A thread parked in getRxTask() because the pool is exhausted must be
woken, and told there is no task, when the
+ * pool is stopped.
+ */
+ @Test(timeout = 60000)
+ public void testStopWakesWaitingThread() throws Exception {
+ RxTaskPool pool = new RxTaskPool(1, 1, NoOpTask::new);
+ Assert.assertNotNull(pool.getRxTask());
+
+ AtomicReference<AbstractRxTask> result = new AtomicReference<>(new
NoOpTask());
+ Thread waiter = new Thread(() -> result.set(pool.getRxTask()));
+ waiter.start();
+ // Give the waiter time to park
+ Thread.sleep(500);
+ Assert.assertTrue(waiter.isAlive());
+
+ pool.stop();
+ waiter.join(TIMEOUT);
+ Assert.assertFalse("getRxTask() was not woken by stop()",
waiter.isAlive());
+ Assert.assertNull(result.get());
+ }
+
+
+ /*
+ * Stopping the receiver must stop its task pool before waiting for the
listener thread, since that thread may be
+ * parked in getRxTask() when the pool is exhausted.
+ */
+ @Test(timeout = 60000)
+ public void testReceiverStopWakesWaitingThread() throws Exception {
+ NioReceiver receiver = new NioReceiver();
+ RxTaskPool pool = new RxTaskPool(1, 1, NoOpTask::new);
+ receiver.setPool(pool);
+ Assert.assertNotNull(pool.getRxTask());
+
+ AtomicReference<AbstractRxTask> result = new AtomicReference<>(new
NoOpTask());
+ Thread waiter = new Thread(() -> result.set(pool.getRxTask()));
+ waiter.start();
+ Thread.sleep(500);
+ Assert.assertTrue(waiter.isAlive());
+
+ receiver.stop();
+ waiter.join(TIMEOUT);
+ Assert.assertFalse("receiver stop did not wake getRxTask()",
waiter.isAlive());
+ Assert.assertNull(result.get());
+ }
+
+
+ /*
+ * A worker that started to return before stop() acquired the pool monitor
must still observe that the pool was
+ * stopped and must not be added back to its idle list.
+ */
+ @Test(timeout = 60000)
+ public void testWorkerIsNotReturnedAfterStop() throws Exception {
+ RxTaskPool pool = new RxTaskPool(1, 1, NoOpTask::new);
+ AbstractRxTask worker = pool.getRxTask();
+ Thread returner = new Thread(() -> pool.returnWorker(worker));
+
+ synchronized (pool.mutex) {
+ returner.start();
+ long deadline = System.currentTimeMillis() + TIMEOUT;
+ while (returner.getState() != Thread.State.BLOCKED &&
System.currentTimeMillis() < deadline) {
+ Thread.yield();
+ }
+ Assert.assertEquals(Thread.State.BLOCKED, returner.getState());
+ pool.stop();
+ }
+
+ returner.join(TIMEOUT);
+ Assert.assertFalse(returner.isAlive());
+ Assert.assertEquals(0, pool.available());
+ }
+
+
+ /*
+ * An interrupted waiter must give up, with its interrupt status restored,
rather than spin.
+ */
+ @Test(timeout = 60000)
+ public void testInterruptEndsWait() throws Exception {
+ RxTaskPool pool = new RxTaskPool(1, 1, NoOpTask::new);
+ Assert.assertNotNull(pool.getRxTask());
+
+ AtomicReference<AbstractRxTask> result = new AtomicReference<>(new
NoOpTask());
+ AtomicReference<Boolean> interrupted = new AtomicReference<>();
+ Thread waiter = new Thread(() -> {
+ result.set(pool.getRxTask());
+
interrupted.set(Boolean.valueOf(Thread.currentThread().isInterrupted()));
+ });
+ waiter.start();
+ Thread.sleep(500);
+ waiter.interrupt();
+ waiter.join(TIMEOUT);
+ Assert.assertFalse("getRxTask() ignored the interrupt",
waiter.isAlive());
+ Assert.assertNull(result.get());
+ Assert.assertEquals(Boolean.TRUE, interrupted.get());
+ }
+
+
+ /*
+ * A worker that is run without a key must still be returned to the pool.
+ */
+ @Test(timeout = 60000)
+ public void testWorkerWithoutKeyIsReturned() throws Exception {
+ NioReceiver receiver = new NioReceiver();
+ receiver.setChannel(new GroupChannel());
+ RxTaskPool pool = new RxTaskPool(1, 1, receiver::createRxTask);
+ NioReplicationTask task = (NioReplicationTask) pool.getRxTask();
+ Assert.assertNotNull(task);
+ Assert.assertEquals(0, pool.available());
+
+ task.run();
+
+ Assert.assertEquals(1, pool.available());
+ pool.stop();
+ }
+}
diff --git a/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
b/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
index 552d3daa84..f9b632324b 100644
--- a/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
+++ b/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
@@ -23,9 +23,13 @@ import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.nio.channels.DatagramChannel;
+import java.nio.channels.Pipe;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -294,6 +298,42 @@ public class TestNioReceiver {
}
}
+ @Test
+ public void testRejectedTaskIsResetBeforeReturningToPool() throws
Exception {
+ NioReceiver receiver = new NioReceiver();
+ receiver.setChannel(new GroupChannel());
+ RxTaskPool pool = new RxTaskPool(1, 1, receiver::createRxTask);
+ receiver.setPool(pool);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ executor.shutdown();
+ receiver.setExecutor(executor);
+
+ Pipe pipe = Pipe.open();
+ try (Pipe.SourceChannel source = pipe.source(); Pipe.SinkChannel sink
= pipe.sink();
+ Selector selector = Selector.open()) {
+ Field selectorField =
NioReceiver.class.getDeclaredField("selector");
+ selectorField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ AtomicReference<Selector> receiverSelector =
(AtomicReference<Selector>) selectorField.get(receiver);
+ receiverSelector.set(selector);
+ source.configureBlocking(false);
+ SelectionKey key = source.register(selector, SelectionKey.OP_READ);
+
+ Assert.assertThrows(RejectedExecutionException.class, () ->
receiver.readDataFromSocket(key));
+ Assert.assertEquals(1, pool.available());
+ receiver.events();
+ Assert.assertFalse(key.isValid());
+
+ Object task = pool.getRxTask();
+ Field keyField = NioReplicationTask.class.getDeclaredField("key");
+ keyField.setAccessible(true);
+ Assert.assertNull(keyField.get(task));
+ pool.returnWorker((NioReplicationTask) task);
+ } finally {
+ receiver.stop();
+ }
+ }
+
private void assertReleased(NioReceiver receiver) throws Exception {
Field serverChannelField =
NioReceiver.class.getDeclaredField("serverChannel");
serverChannelField.setAccessible(true);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
