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
commit 5da23da5ed8b8f0fc015e31625c19c38c7b537cb 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 71b34d8e1a..68dee8f455 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 { @@ -301,56 +307,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; } } @@ -515,8 +523,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 @@ -527,6 +542,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 @@ -537,7 +554,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 { @@ -608,28 +625,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(); } @@ -647,27 +711,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()) { @@ -719,13 +794,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]
