This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 10.1.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit deab50e2374ec7cc0f5c0169f5ddb25b4f588ee5 Author: opencode <[email protected]> AuthorDate: Sun Oct 11 17:57:12 2026 +0200 Stop the listener thread from reviving the listening flag of a stopped receiver The listener thread of the NioReceiver set the listening flag unconditionally, then looked at the selector and, finding it already closed and nulled by a stop that had completed before the thread got scheduled, skipped the select loop and returned. The flag stayed set without anything listening, so every later start exited at the already started guard and the node silently stopped receiving cluster traffic. A closely related interleaving, in which the thread set the flag while stop was waiting for it, made the thread select on the closed selector it held as a local copy and spin forever at full CPU. The thread now only sets the flag when the selector is still there to select on and clears it on exit, an exceptional one, only when it had set it itself, so a stale thread of a previous start can neither clear the flag of a live listener nor leave a stale one behind. The loop condition re-reads the selector field, so a stop that closes and nulls the selector ends the loop even if the flag had been re-set meanwhile. The cleanup at the end of the select now also runs when the thread never got to select, and it no longer assumes the server channel to be present, which closes the bound but never registered datagram channel of a stop that raced with the start. Add TestNioReceiver covering the flag behavior after a stop, the stale-thread case, the normal stop of a running listener and the datagram channel cleanup. --- .../catalina/tribes/transport/nio/NioReceiver.java | 154 +++++++++++++-------- .../tribes/transport/nio/TestNioReceiver.java | 124 +++++++++++++++++ 2 files changed, 218 insertions(+), 60 deletions(-) diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java index 80cd5ad0ed..260748e1b1 100644 --- a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java +++ b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java @@ -304,80 +304,114 @@ public class NioReceiver extends ReceiverBase implements Runnable, NioReceiverMB protected void listen() throws Exception { if (isListening()) { log.warn(sm.getString("nioReceiver.alreadyStarted")); + // The flag belongs to the listener that is running, do not clear + // it here return; } - setListen(true); - // Avoid NPEs if selector is set to null on stop. Selector selector = this.selector.get(); - if (selector != null && datagramChannel != null) { - ObjectReader oreader = new ObjectReader(MAX_UDP_SIZE); // max size for a datagram packet - registerChannel(selector, datagramChannel, SelectionKey.OP_READ, oreader); + // 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 + // listening, so that a later start is not blocked by the "already + // started" guard above. + boolean ownsListenFlag = selector != null; + if (ownsListenFlag) { + setListen(true); } - while (isListening() && selector != null) { - // this may block for a long time, upon return the - // selected set contains keys of the ready channels - try { - events(); - socketTimeouts(); - int n = selector.select(getSelectorTimeout()); - if (n == 0) { - // there is a good chance that we got here - // because the TcpReplicationThread called - // selector wakeup(). - // if that happens, we must ensure that that - // thread has enough time to call interestOps - // synchronized (interestOpsMutex) { - // if we got the lock, means there are no - // keys trying to register for the - // interestOps method - // } - continue; // nothing to do - } - // get an iterator over the set of selected keys - Iterator<SelectionKey> it = selector.selectedKeys().iterator(); - // look at each key in the selected set - while (it.hasNext()) { - SelectionKey key = it.next(); - // Is a new connection coming in? - if (key.isAcceptable()) { - ServerSocketChannel server = (ServerSocketChannel) key.channel(); - SocketChannel channel = server.accept(); - channel.socket().setReceiveBufferSize(getRxBufSize()); - channel.socket().setSendBufferSize(getTxBufSize()); - channel.socket().setTcpNoDelay(getTcpNoDelay()); - channel.socket().setKeepAlive(getSoKeepAlive()); - channel.socket().setOOBInline(getOoBInline()); - channel.socket().setReuseAddress(getSoReuseAddress()); - channel.socket().setSoLinger(getSoLingerOn(), getSoLingerTime()); - channel.socket().setSoTimeout(getTimeout()); - Object attach = new ObjectReader(channel); - registerChannel(selector, channel, SelectionKey.OP_READ, attach); + try { + if (ownsListenFlag && datagramChannel != null) { + ObjectReader oreader = new ObjectReader(MAX_UDP_SIZE); // max size for a datagram packet + registerChannel(selector, datagramChannel, SelectionKey.OP_READ, oreader); + } + + // Test the field, not only the local reference, so that a stop + // 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) { + // this may block for a long time, upon return the + // selected set contains keys of the ready channels + try { + events(); + socketTimeouts(); + int n = selector.select(getSelectorTimeout()); + if (n == 0) { + // there is a good chance that we got here + // because the TcpReplicationThread called + // selector wakeup(). + // if that happens, we must ensure that that + // thread has enough time to call interestOps + // synchronized (interestOpsMutex) { + // if we got the lock, means there are no + // keys trying to register for the + // interestOps method + // } + continue; // nothing to do } - // is there data to read on this channel? - if (key.isReadable()) { - readDataFromSocket(key); - } else { - key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE)); + // get an iterator over the set of selected keys + Iterator<SelectionKey> it = selector.selectedKeys().iterator(); + // look at each key in the selected set + while (it.hasNext()) { + SelectionKey key = it.next(); + // Is a new connection coming in? + if (key.isAcceptable()) { + ServerSocketChannel server = (ServerSocketChannel) key.channel(); + SocketChannel channel = server.accept(); + try { + channel.socket().setReceiveBufferSize(getRxBufSize()); + channel.socket().setSendBufferSize(getTxBufSize()); + channel.socket().setTcpNoDelay(getTcpNoDelay()); + channel.socket().setKeepAlive(getSoKeepAlive()); + channel.socket().setOOBInline(getOoBInline()); + channel.socket().setReuseAddress(getSoReuseAddress()); + channel.socket().setSoLinger(getSoLingerOn(), getSoLingerTime()); + channel.socket().setSoTimeout(getTimeout()); + Object attach = new ObjectReader(channel); + registerChannel(selector, channel, SelectionKey.OP_READ, attach); + } catch (Throwable t) { + ExceptionUtils.handleThrowable(t); + try { + channel.close(); + } catch (IOException ioe) { + t.addSuppressed(ioe); + } + throw t; + } + } + // is there data to read on this channel? + if (key.isReadable()) { + readDataFromSocket(key); + } else { + key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE)); + } + + // remove key from selected set, it's been handled + it.remove(); } - - // remove key from selected set, it's been handled - it.remove(); + } catch (ClosedSelectorException cse) { + // ignore is normal at shutdown or stop listen socket + } catch (CancelledKeyException nx) { + log.warn(sm.getString("nioReceiver.clientDisconnect")); + } catch (Throwable t) { + ExceptionUtils.handleThrowable(t); + log.error(sm.getString("nioReceiver.requestError"), t); } - } catch (ClosedSelectorException cse) { - // ignore is normal at shutdown or stop listen socket - } catch (CancelledKeyException nx) { - log.warn(sm.getString("nioReceiver.clientDisconnect")); - } catch (Throwable t) { - ExceptionUtils.handleThrowable(t); - log.error(sm.getString("nioReceiver.requestError"), t); + } + } finally { + if (ownsListenFlag) { + // This thread set the flag itself, so it clears it on every + // exit path, an exceptional exit included + setListen(false); + } + } + if (serverChannel != null) { + serverChannel.close(); } - serverChannel.close(); if (datagramChannel != null) { try { datagramChannel.close(); diff --git a/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java b/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java new file mode 100644 index 0000000000..89e9360819 --- /dev/null +++ b/test/org/apache/catalina/tribes/transport/nio/TestNioReceiver.java @@ -0,0 +1,124 @@ +/* + * 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.nio; + +import java.lang.reflect.Field; +import java.nio.channels.DatagramChannel; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.catalina.tribes.group.GroupChannel; + +public class TestNioReceiver { + + /* + * A stop that raced with the start of the listener thread used to leave + * the listening flag set, because the thread re-set it after the stop had + * cleared it. Every later start then exited at the "already started" + * guard and the receiver never listened again. The thread now only sets + * the flag when the selector is still there to select on. + */ + @Test + public void testListenWithANullSelectorDoesNotSetTheFlag() throws Exception { + NioReceiver receiver = new NioReceiver(); + receiver.setChannel(new GroupChannel()); + receiver.listen(); + Assert.assertFalse(receiver.isListening()); + } + + /* + * A stale listener thread of a previous start that hits the "already + * started" guard must leave the flag of the live listener alone, so that + * it cannot tear a running receiver down. + */ + @Test + public void testStaleThreadDoesNotClearTheFlagOfALiveListener() { + NioReceiver receiver = new NioReceiver(); + receiver.setChannel(new GroupChannel()); + receiver.setListen(true); + receiver.run(); + Assert.assertTrue(receiver.isListening()); + } + + /* + * The headline race: the receiver is fully stopped before the listener + * thread gets to run. The thread must then not set the listening flag, so + * the receiver is not left marked as listening without anything selecting, + * which would block every later start at the "already started" guard. + */ + @Test + public void testListenAfterStopDoesNotSetTheFlag() throws Exception { + NioReceiver receiver = new NioReceiver(); + receiver.setChannel(new GroupChannel()); + receiver.setAddress("127.0.0.1"); + receiver.setPort(0); + receiver.bind(); + receiver.stopListening(); + receiver.listen(); + Assert.assertFalse(receiver.isListening()); + } + + /* + * The thread that set the listening flag itself clears it again when the + * selector is closed and nulled underneath it by a stop. + */ + @Test + public void testListenerThreadClearsTheFlagOnStop() throws Exception { + NioReceiver receiver = new NioReceiver(); + receiver.setChannel(new GroupChannel()); + receiver.setAddress("127.0.0.1"); + receiver.setPort(0); + receiver.bind(); + Thread thread = new Thread(receiver); + thread.start(); + long deadline = System.currentTimeMillis() + 10000; + while (!receiver.isListening() && System.currentTimeMillis() < deadline) { + Thread.sleep(50); + } + Assert.assertTrue("the listener thread never started listening", receiver.isListening()); + receiver.stopListening(); + thread.join(10000); + Assert.assertFalse("the listener thread did not stop", thread.isAlive()); + Assert.assertFalse(receiver.isListening()); + } + + /* + * The datagram channel is bound by bind() but only registered with the + * selector in listen(), so stopListening() cannot reach it. The listener + * thread that runs after the stop must still close it during its cleanup, + * even when it never got to select. + */ + @Test + public void testListenAfterStopClosesTheBoundDatagramChannel() throws Exception { + NioReceiver receiver = new NioReceiver(); + receiver.setChannel(new GroupChannel()); + receiver.setAddress("127.0.0.1"); + receiver.setPort(0); + receiver.setUdpPort(45678); + receiver.bind(); + Field field = NioReceiver.class.getDeclaredField("datagramChannel"); + field.setAccessible(true); + DatagramChannel datagramChannel = (DatagramChannel) field.get(receiver); + Assert.assertNotNull("bind() did not open the datagram channel", datagramChannel); + receiver.stopListening(); + receiver.listen(); + Assert.assertFalse("the datagram channel was not closed", datagramChannel.isOpen()); + Assert.assertNull("the datagram channel reference was not cleared", field.get(receiver)); + Assert.assertFalse(receiver.isListening()); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
