This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/main by this push:
new 78b32b1a5b Review "Stop the listener thread from reviving the
listening flag..."
78b32b1a5b is described below
commit 78b32b1a5b35ed0700672893695afefc44f88986
Author: Mark Thomas <[email protected]>
AuthorDate: Fri Oct 23 21:02:07 2026 +0100
Review "Stop the listener thread from reviving the listening flag..."
Fix similar issue for running
Make clean-up more robust
---
.../catalina/tribes/transport/nio/NioReceiver.java | 237 ++++++++++++++-------
.../tribes/transport/nio/TestNioReceiver.java | 105 +++++++++
2 files changed, 261 insertions(+), 81 deletions(-)
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
index 8ca3477163..e6668a3470 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
@@ -35,6 +35,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.catalina.tribes.ChannelMessage;
@@ -63,8 +64,11 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
*/
protected static final StringManager sm =
StringManager.getManager(NioReceiver.class);
- private volatile boolean running = false;
+ // Number of listener threads in run(). A counter rather than a flag so
that a stale thread of an earlier start
+ // that exits late cannot hide a live listener of a later start.
+ private final AtomicInteger runningListeners = new AtomicInteger();
+ private final Object channelLock = new Object();
private final AtomicReference<Selector> selector = new AtomicReference<>();
private ServerSocketChannel serverChannel = null;
private volatile ServerSocketChannel secureServerChannel = null;
@@ -198,22 +202,24 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
taskPool.stop();
}
this.stopListening();
- // Startup may fail before the channels are registered with the
selector or the listener thread is started.
- if (serverChannel != null) {
- try {
- serverChannel.close();
- } catch (IOException ioe) {
- log.debug(sm.getString("nioReceiver.closeError"), ioe);
+ synchronized (channelLock) {
+ // Startup may fail before the channels are registered with the
selector or the listener thread is started.
+ if (serverChannel != null) {
+ try {
+ serverChannel.close();
+ } catch (IOException ioe) {
+ log.debug(sm.getString("nioReceiver.closeError"), ioe);
+ }
+ serverChannel = null;
}
- serverChannel = null;
- }
- if (datagramChannel != null) {
- try {
- datagramChannel.close();
- } catch (IOException ioe) {
- log.debug(sm.getString("nioReceiver.closeError"), ioe);
+ if (datagramChannel != null) {
+ try {
+ datagramChannel.close();
+ } catch (IOException ioe) {
+ log.debug(sm.getString("nioReceiver.closeError"), ioe);
+ }
+ datagramChannel = null;
}
- datagramChannel = null;
}
if (secureServerChannel != null) {
try {
@@ -300,56 +306,58 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
* @throws IOException If binding fails
*/
protected void bind() throws IOException {
- try {
- // allocate an unbound server socket channel
- serverChannel = ServerSocketChannel.open();
- // Get the associated ServerSocket to bind it with
- ServerSocket serverSocket = serverChannel.socket();
- // create a new Selector for use below
- this.selector.set(Selector.open());
- // set the port the server channel will listen to
- // serverSocket.bind(new InetSocketAddress(getBind(),
getTcpListenPort()));
- bind(serverSocket, getPort(), getAutoBind(), false);
- // set non-blocking mode for the listening socket
- serverChannel.configureBlocking(false);
- // register the ServerSocketChannel with the Selector
- serverChannel.register(this.selector.get(),
SelectionKey.OP_ACCEPT);
-
- // set up the datagram channel
- if (this.getUdpPort() >= 0) {
- datagramChannel = DatagramChannel.open();
- configureDatagramChannel();
- // bind to the address to avoid security checks
- bindUdp(datagramChannel.socket(), getUdpPort(), getAutoBind());
- }
- } catch (IOException | RuntimeException e) {
- // The caller of start() only sees the exception, so release
- // everything that was opened up to the failure point. Without
- // this, every failed start leaks the server channel descriptor,
- // the selector and, when the UDP bind is what failed, a bound
- // datagram channel that nothing will ever listen on.
- if (datagramChannel != null) {
- try {
- datagramChannel.close();
- } catch (IOException ioe) {
- e.addSuppressed(ioe);
+ synchronized (channelLock) {
+ try {
+ // allocate an unbound server socket channel
+ serverChannel = ServerSocketChannel.open();
+ // Get the associated ServerSocket to bind it with
+ ServerSocket serverSocket = serverChannel.socket();
+ // create a new Selector for use below
+ this.selector.set(Selector.open());
+ // set the port the server channel will listen to
+ // serverSocket.bind(new InetSocketAddress(getBind(),
getTcpListenPort()));
+ bind(serverSocket, getPort(), getAutoBind(), false);
+ // set non-blocking mode for the listening socket
+ serverChannel.configureBlocking(false);
+ // register the ServerSocketChannel with the Selector
+ serverChannel.register(this.selector.get(),
SelectionKey.OP_ACCEPT);
+
+ // set up the datagram channel
+ if (this.getUdpPort() >= 0) {
+ datagramChannel = DatagramChannel.open();
+ configureDatagramChannel();
+ // bind to the address to avoid security checks
+ bindUdp(datagramChannel.socket(), getUdpPort(),
getAutoBind());
+ }
+ } catch (IOException | RuntimeException e) {
+ // The caller of start() only sees the exception, so release
+ // everything that was opened up to the failure point. Without
+ // this, every failed start leaks the server channel
descriptor,
+ // the selector and, when the UDP bind is what failed, a bound
+ // datagram channel that nothing will ever listen on.
+ if (datagramChannel != null) {
+ try {
+ datagramChannel.close();
+ } catch (IOException ioe) {
+ e.addSuppressed(ioe);
+ }
+ datagramChannel = null;
+ }
+ if (serverChannel != null) {
+ try {
+ serverChannel.close();
+ } catch (IOException ioe) {
+ e.addSuppressed(ioe);
+ }
+ serverChannel = null;
}
- datagramChannel = null;
- }
- if (serverChannel != null) {
try {
- serverChannel.close();
+ closeSelector();
} catch (IOException ioe) {
e.addSuppressed(ioe);
}
- serverChannel = null;
- }
- try {
- closeSelector();
- } catch (IOException ioe) {
- e.addSuppressed(ioe);
+ throw e;
}
- throw e;
}
}
@@ -514,8 +522,15 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
return;
}
- // Avoid NPEs if selector is set to null on stop.
- Selector selector = this.selector.get();
+ Selector selector;
+ ServerSocketChannel serverChannel;
+ DatagramChannel datagramChannel;
+ synchronized (channelLock) {
+ // Capture one receiver generation so a delayed listener cannot
clean up resources created by a later start.
+ selector = this.selector.get();
+ serverChannel = this.serverChannel;
+ datagramChannel = this.datagramChannel;
+ }
// Set the flag only if this thread gets to select. A receiver that
// was stopped before this thread got here must not end up marked as
@@ -526,6 +541,8 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
setListen(true);
}
+ Exception exception = null;
+ Error error = null;
try {
if (ownsListenFlag && datagramChannel != null) {
ObjectReader oreader = new ObjectReader(MAX_UDP_SIZE); // max
size for a datagram packet
@@ -536,7 +553,7 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
// which closes and nulls the selector while this thread is in the
// select also ends this loop even if the thread had re-set the
// listening flag in the meantime
- while (ownsListenFlag && isListening() && this.selector.get() !=
null) {
+ while (ownsListenFlag && isListening() && this.selector.get() ==
selector) {
// this may block for a long time, upon return the
// selected set contains keys of the ready channels
try {
@@ -607,28 +624,75 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
}
}
+ } catch (Exception e) {
+ exception = e;
+ throw e;
+ } catch (Error e) {
+ error = e;
+ throw e;
} finally {
if (ownsListenFlag) {
- // This thread set the flag itself, so it clears it on every
- // exit path, an exceptional exit included
- setListen(false);
+ synchronized (channelLock) {
+ Selector currentSelector = this.selector.get();
+ if (currentSelector == selector || currentSelector ==
null) {
+ // Do not clear the flag of a listener for a newer
selector generation.
+ setListen(false);
+ }
+ }
+ }
+ try {
+ closeListenerResources(selector, serverChannel,
datagramChannel);
+ } catch (IOException ioe) {
+ if (exception != null) {
+ exception.addSuppressed(ioe);
+ } else if (error != null) {
+ error.addSuppressed(ioe);
+ } else {
+ throw ioe;
+ }
}
}
+ }
- if (serverChannel != null) {
- serverChannel.close();
- }
- if (datagramChannel != null) {
+ void closeListenerResources(Selector selector, ServerSocketChannel
serverChannel, DatagramChannel datagramChannel)
+ throws IOException {
+ synchronized (channelLock) {
+ IOException failure = null;
+ if (serverChannel != null) {
+ try {
+ serverChannel.close();
+ } catch (IOException ioe) {
+ failure = ioe;
+ }
+ if (this.serverChannel == serverChannel) {
+ this.serverChannel = null;
+ }
+ }
+ if (datagramChannel != null) {
+ try {
+ datagramChannel.close();
+ } catch (Exception e) {
+ if (log.isDebugEnabled()) {
+ log.debug(sm.getString("nioReceiver.closeError"), e);
+ }
+ }
+ if (this.datagramChannel == datagramChannel) {
+ this.datagramChannel = null;
+ }
+ }
try {
- datagramChannel.close();
- } catch (Exception e) {
- if (log.isDebugEnabled()) {
- log.debug(sm.getString("nioReceiver.closeError"), e);
+ closeSelector(selector);
+ } catch (IOException ioe) {
+ if (failure == null) {
+ failure = ioe;
+ } else {
+ failure.addSuppressed(ioe);
}
}
- datagramChannel = null;
+ if (failure != null) {
+ throw failure;
+ }
}
- closeSelector();
}
@@ -646,27 +710,38 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
selector.wakeup();
// Wait for the receiver thread to finish
int count = 0;
- while (running && count < 50) {
+ while (runningListeners.get() > 0 && count < 50) {
Thread.sleep(100);
count++;
}
- if (running) {
+ if (runningListeners.get() > 0) {
log.warn(sm.getString("nioReceiver.stop.threadRunning"));
}
- closeSelector();
+ closeSelector(selector);
} catch (Exception e) {
log.error(sm.getString("nioReceiver.stop.fail"), e);
} finally {
- this.selector.set(null);
+ this.selector.compareAndSet(selector, null);
}
}
}
private void closeSelector() throws IOException {
Selector selector = this.selector.getAndSet(null);
+ closeSelector(selector, false);
+ }
+
+ private void closeSelector(Selector selector) throws IOException {
+ closeSelector(selector, true);
+ }
+
+ private void closeSelector(Selector selector, boolean claim) throws
IOException {
if (selector == null) {
return;
}
+ if (claim && !this.selector.compareAndSet(selector, null)) {
+ return;
+ }
try {
// look at each key in the selected set
for (SelectionKey key : selector.keys()) {
@@ -718,13 +793,13 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
*/
@Override
public void run() {
- running = true;
+ runningListeners.incrementAndGet();
try {
listen();
} catch (Exception e) {
log.error(sm.getString("nioReceiver.run.fail"), e);
} finally {
- running = false;
+ runningListeners.decrementAndGet();
}
}
diff --git a/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
b/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
index f9b632324b..36f17746b3 100644
--- a/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
+++ b/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java
@@ -24,6 +24,7 @@ import java.net.InetAddress;
import java.net.ServerSocket;
import java.nio.channels.DatagramChannel;
import java.nio.channels.Pipe;
+import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
@@ -31,6 +32,7 @@ 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.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import javax.management.MBeanServer;
@@ -141,6 +143,109 @@ public class TestNioReceiver {
Assert.assertFalse(receiver.isListening());
}
+ /*
+ * A stale listener thread that exits must not make stopListening()
believe that the live listener has finished.
+ */
+ @Test
+ public void testStaleThreadExitKeepsLiveListenerCounted() throws Exception
{
+ NioReceiver receiver = new NioReceiver();
+ receiver.setChannel(new GroupChannel());
+ receiver.setAddress("127.0.0.1");
+ receiver.setPort(0);
+ receiver.bind();
+ Field field = NioReceiver.class.getDeclaredField("runningListeners");
+ field.setAccessible(true);
+ AtomicInteger running = (AtomicInteger) field.get(receiver);
+ Thread live = new Thread(receiver);
+ live.start();
+ try {
+ long deadline = System.currentTimeMillis() + 10000;
+ while (!receiver.isListening() && System.currentTimeMillis() <
deadline) {
+ Thread.sleep(50);
+ }
+ Assert.assertTrue(receiver.isListening());
+ Assert.assertEquals(1, running.get());
+ // Hits the already started guard and returns
+ receiver.run();
+ Assert.assertEquals(1, running.get());
+ Assert.assertTrue(receiver.isListening());
+ } finally {
+ receiver.stopListening();
+ live.join(10000);
+ }
+ Assert.assertEquals(0, running.get());
+ }
+
+ @Test
+ public void testStaleListenerDoesNotCloseRestartedReceiver() throws
Exception {
+ NioReceiver receiver = new NioReceiver();
+ receiver.setChannel(new GroupChannel());
+ receiver.setAddress("127.0.0.1");
+ receiver.setPort(0);
+ receiver.setUdpPort(0);
+
+ Field serverChannelField =
NioReceiver.class.getDeclaredField("serverChannel");
+ serverChannelField.setAccessible(true);
+ Field datagramChannelField =
NioReceiver.class.getDeclaredField("datagramChannel");
+ datagramChannelField.setAccessible(true);
+ Field selectorField = NioReceiver.class.getDeclaredField("selector");
+ selectorField.setAccessible(true);
+
+ try {
+ receiver.bind();
+ ServerSocketChannel oldServer = (ServerSocketChannel)
serverChannelField.get(receiver);
+ DatagramChannel oldDatagram = (DatagramChannel)
datagramChannelField.get(receiver);
+ @SuppressWarnings("unchecked")
+ AtomicReference<Selector> selectorReference =
+ (AtomicReference<Selector>) selectorField.get(receiver);
+ Selector oldSelector = selectorReference.get();
+
+ receiver.stopListening();
+ receiver.bind();
+ ServerSocketChannel newServer = (ServerSocketChannel)
serverChannelField.get(receiver);
+ DatagramChannel newDatagram = (DatagramChannel)
datagramChannelField.get(receiver);
+ Selector newSelector = selectorReference.get();
+
+ receiver.closeListenerResources(oldSelector, oldServer,
oldDatagram);
+
+ Assert.assertTrue(newServer.isOpen());
+ Assert.assertTrue(newDatagram.isOpen());
+ Assert.assertTrue(newSelector.isOpen());
+ Assert.assertSame(newServer, serverChannelField.get(receiver));
+ Assert.assertSame(newDatagram, datagramChannelField.get(receiver));
+ Assert.assertSame(newSelector, selectorReference.get());
+ } finally {
+ receiver.stop();
+ }
+ }
+
+ @Test
+ public void testListenerSetupFailureClosesCapturedResources() throws
Exception {
+ NioReceiver receiver = new NioReceiver() {
+ @Override
+ protected void registerChannel(Selector selector,
SelectableChannel channel, int ops, Object attach)
+ throws Exception {
+ if (channel instanceof DatagramChannel) {
+ throw new IOException("Datagram registration failed");
+ }
+ super.registerChannel(selector, channel, ops, attach);
+ }
+ };
+ receiver.setChannel(new GroupChannel());
+ receiver.setAddress("127.0.0.1");
+ receiver.setPort(0);
+ receiver.setUdpPort(0);
+
+ try {
+ receiver.bind();
+ Assert.assertThrows(IOException.class, receiver::listen);
+ assertReleased(receiver);
+ Assert.assertFalse(receiver.isListening());
+ } finally {
+ receiver.stop();
+ }
+ }
+
/*
* A TCP bind that exhausts the auto-bind retries throws out of bind()
* after the server channel and the selector were already opened. Both
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
