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 af887f1a01 Review "Fix end-of-input and ReadListener call-back
contract ..."
af887f1a01 is described below
commit af887f1a0150f660c0a7b74d546510565476de45
Author: Mark Thomas <[email protected]>
AuthorDate: Mon Nov 2 09:36:07 2026 +0000
Review "Fix end-of-input and ReadListener call-back contract ..."
Fix a regression that would cause the CPU to enter a tight loop
Also smaller related changes:
- reduce duplicate code for closing upgrade input/output streams
- document that upgrade processors are not recycled
Co-authored by: Codex / GPT-6.1-Sol
---
java/org/apache/coyote/AbstractProtocol.java | 5 +
.../http11/upgrade/UpgradeProcessorBase.java | 6 +
.../http11/upgrade/UpgradeProcessorExternal.java | 70 +++++---
.../http11/upgrade/UpgradeServletInputStream.java | 5 +
.../apache/tomcat/util/net/AbstractEndpoint.java | 4 +
.../apache/coyote/http11/upgrade/TestUpgrade.java | 184 +++++++++++++++++++++
webapps/docs/changelog.xml | 7 +
7 files changed, 257 insertions(+), 24 deletions(-)
diff --git a/java/org/apache/coyote/AbstractProtocol.java
b/java/org/apache/coyote/AbstractProtocol.java
index cd2371a7a4..abc0c5ebc7 100644
--- a/java/org/apache/coyote/AbstractProtocol.java
+++ b/java/org/apache/coyote/AbstractProtocol.java
@@ -1411,6 +1411,11 @@ public abstract class AbstractProtocol<S> implements
ProtocolHandler, MBeanRegis
longPoll(wrapper, processor);
getProtocol().addWaitingProcessor(processor);
}
+ } else if (state == SocketState.UPGRADED_HALF_CLOSED) {
+ // Don't add sockets back to the poller.
+ // The handler will register for write as required
+ // Add to waiting processors so the processor is visible
if the protocol is stopped
+ getProtocol().addWaitingProcessor(processor);
} else if (state == SocketState.ASYNC_IO) {
// Don't add sockets back to the poller.
// The handler will initiate all further I/O
diff --git a/java/org/apache/coyote/http11/upgrade/UpgradeProcessorBase.java
b/java/org/apache/coyote/http11/upgrade/UpgradeProcessorBase.java
index 62e0ae1a99..432b4240e8 100644
--- a/java/org/apache/coyote/http11/upgrade/UpgradeProcessorBase.java
+++ b/java/org/apache/coyote/http11/upgrade/UpgradeProcessorBase.java
@@ -66,6 +66,12 @@ public abstract class UpgradeProcessorBase extends
AbstractProcessorLight implem
}
+ /**
+ * {@inheritDoc}
+ * <p>
+ * Upgrade processors are not recycled and re-used. This method is only
used to trigger any necessary clean-up once
+ * the upgrade processor is no longer required.
+ */
@Override
public void recycle() {
// NO-OP. Sub-classes override this to release resources when the
diff --git
a/java/org/apache/coyote/http11/upgrade/UpgradeProcessorExternal.java
b/java/org/apache/coyote/http11/upgrade/UpgradeProcessorExternal.java
index 9b42804c89..5cc0426f54 100644
--- a/java/org/apache/coyote/http11/upgrade/UpgradeProcessorExternal.java
+++ b/java/org/apache/coyote/http11/upgrade/UpgradeProcessorExternal.java
@@ -17,6 +17,7 @@
package org.apache.coyote.http11.upgrade;
import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
import jakarta.servlet.ServletInputStream;
import jakarta.servlet.ServletOutputStream;
@@ -24,6 +25,7 @@ import jakarta.servlet.ServletOutputStream;
import org.apache.coyote.UpgradeToken;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
+import org.apache.tomcat.util.ExceptionUtils;
import org.apache.tomcat.util.net.AbstractEndpoint.Handler.SocketState;
import org.apache.tomcat.util.net.SSLSupport;
import org.apache.tomcat.util.net.SocketEvent;
@@ -58,11 +60,13 @@ public class UpgradeProcessorExternal extends
UpgradeProcessorBase {
*/
private final SocketWrapperBase<?> wrapper;
+ private AtomicBoolean stopCalled = new AtomicBoolean();
+
/**
* Constructs a new UpgradeProcessorExternal.
*
- * @param wrapper the socket wrapper
- * @param upgradeToken the upgrade token
+ * @param wrapper the socket wrapper
+ * @param upgradeToken the upgrade token
* @param upgradeGroupInfo the group info for statistics
*/
public UpgradeProcessorExternal(SocketWrapperBase<?> wrapper, UpgradeToken
upgradeToken,
@@ -105,10 +109,16 @@ public class UpgradeProcessorExternal extends
UpgradeProcessorBase {
@Override
public void close() throws Exception {
- upgradeServletInputStream.close();
- upgradeServletOutputStream.close();
+ closeStreamsQuietly();
// Triggers update of stats from UpgradeInfo to UpgradeGroupInfo
upgradeInfo.setGroupInfo(null);
+ if (stopCalled.compareAndSet(false, true)) {
+ /*
+ * Read polling may have stopped at EOF. Ensure an application
thread closing the connection still
+ * triggers socket closure and handler cleanup on a container
thread.
+ */
+ wrapper.processSocket(SocketEvent.STOP, true);
+ }
}
@@ -137,16 +147,7 @@ public class UpgradeProcessorExternal extends
UpgradeProcessorBase {
if (log.isTraceEnabled()) {
log.trace(sm.getString("upgradeProcessor.stop"));
}
- try {
- upgradeServletInputStream.close();
- } catch (IOException ioe) {
- log.debug(sm.getString("upgradeProcessor.isCloseFail", ioe));
- }
- try {
- upgradeServletOutputStream.close();
- } catch (IOException ioe) {
- log.debug(sm.getString("upgradeProcessor.osCloseFail", ioe));
- }
+ closeStreamsQuietly();
return SocketState.CLOSED;
} else if (status == SocketEvent.ERROR) {
// Give the listeners (if any) the chance to handle the error
before
@@ -157,16 +158,7 @@ public class UpgradeProcessorExternal extends
UpgradeProcessorBase {
}
upgradeServletInputStream.onError(error);
upgradeServletOutputStream.onError(error);
- try {
- upgradeServletInputStream.close();
- } catch (IOException ioe) {
- log.debug(sm.getString("upgradeProcessor.isCloseFail", ioe));
- }
- try {
- upgradeServletOutputStream.close();
- } catch (IOException ioe) {
- log.debug(sm.getString("upgradeProcessor.osCloseFail", ioe));
- }
+ closeStreamsQuietly();
return SocketState.CLOSED;
} else {
// Unexpected state
@@ -183,10 +175,40 @@ public class UpgradeProcessorExternal extends
UpgradeProcessorBase {
}
return SocketState.CLOSED;
}
+ if (upgradeServletInputStream.isEof()) {
+ /*
+ * Do not re-register read interest after EOF, but allow the
output stream to register write interest while
+ * the application finishes its response.
+ */
+ return SocketState.UPGRADED_HALF_CLOSED;
+ }
return SocketState.UPGRADED;
}
+ /*
+ * Close the input and output streams if not already closed, logging any
exceptions at debug level.
+ */
+ private void closeStreamsQuietly() {
+ try {
+ if (!upgradeServletInputStream.isClosed()) {
+ upgradeServletInputStream.close();
+ }
+ } catch (Throwable t) {
+ ExceptionUtils.handleThrowable(t);
+ log.debug(sm.getString("upgradeProcessor.isCloseFail", t));
+ }
+ try {
+ if (!upgradeServletOutputStream.isClosed()) {
+ upgradeServletOutputStream.close();
+ }
+ } catch (Throwable t) {
+ ExceptionUtils.handleThrowable(t);
+ log.debug(sm.getString("upgradeProcessor.osCloseFail", t));
+ }
+ }
+
+
// ----------------------------------------- Unimplemented Processor
methods
@Override
diff --git
a/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java
b/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java
index 7ae52006c6..90505e6b2b 100644
--- a/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java
+++ b/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java
@@ -324,4 +324,9 @@ public class UpgradeServletInputStream extends
ServletInputStream {
final boolean isClosed() {
return closed;
}
+
+
+ final boolean isEof() {
+ return eof;
+ }
}
diff --git a/java/org/apache/tomcat/util/net/AbstractEndpoint.java
b/java/org/apache/tomcat/util/net/AbstractEndpoint.java
index 522efbee76..1e1d8d3656 100644
--- a/java/org/apache/tomcat/util/net/AbstractEndpoint.java
+++ b/java/org/apache/tomcat/util/net/AbstractEndpoint.java
@@ -129,6 +129,10 @@ public abstract class AbstractEndpoint<S, U> {
* The connection has been upgraded.
*/
UPGRADED,
+ /**
+ * The connection has been closed for reads from the client but
writes to the client may continue.
+ */
+ UPGRADED_HALF_CLOSED,
/**
* Asynchronous I/O is in progress.
*/
diff --git a/test/org/apache/coyote/http11/upgrade/TestUpgrade.java
b/test/org/apache/coyote/http11/upgrade/TestUpgrade.java
index 22414a1cac..167e49c93e 100644
--- a/test/org/apache/coyote/http11/upgrade/TestUpgrade.java
+++ b/test/org/apache/coyote/http11/upgrade/TestUpgrade.java
@@ -30,6 +30,9 @@ import java.net.Socket;
import java.net.SocketException;
import java.nio.charset.StandardCharsets;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;
import javax.net.SocketFactory;
@@ -50,14 +53,18 @@ import org.junit.Test;
import static org.apache.catalina.startup.SimpleHttpClient.CRLF;
import org.apache.catalina.Context;
+import org.apache.catalina.connector.Connector;
import org.apache.catalina.startup.Tomcat;
import org.apache.catalina.startup.TomcatBaseTest;
import org.apache.coyote.UpgradeToken;
import org.apache.coyote.http11.AbstractHttp11Protocol;
+import org.apache.coyote.http11.Http11NioProtocol;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
import org.apache.tomcat.ContextBind;
import org.apache.tomcat.util.net.AbstractEndpoint.Handler.SocketState;
+import org.apache.tomcat.util.net.NioChannel;
+import org.apache.tomcat.util.net.NioEndpoint;
import org.apache.tomcat.util.net.SSLSupport;
import org.apache.tomcat.util.net.SocketEvent;
import org.apache.tomcat.util.net.SocketWrapperBase;
@@ -153,6 +160,105 @@ public class TestUpgrade extends TomcatBaseTest {
uc.shutdownInput();
}
+ @Test
+ public void testDelayedResponseAfterEofNonBlockingReadNonBlockingWrite()
throws Exception {
+ doTestDelayedResponseAfterEof(false, false);
+ }
+
+ @Test
+ public void testDelayedResponseAfterEofNonBlockingReadBlockingWrite()
throws Exception {
+ doTestDelayedResponseAfterEof(false, true);
+ }
+
+ @Test
+ public void testDelayedResponseAfterEofBlockingReadNonBlockingWrite()
throws Exception {
+ doTestDelayedResponseAfterEof(true, false);
+ }
+
+ @Test
+ public void testDelayedResponseAfterEofBlockingReadBlockingWrite() throws
Exception {
+ doTestDelayedResponseAfterEof(true, true);
+ }
+
+ private void doTestDelayedResponseAfterEof(boolean blockingRead, boolean
blockingWrite) throws Exception {
+ Tomcat tomcat = getTomcatInstance();
+ CountingEndpoint endpoint = new CountingEndpoint();
+ Connector connector = new Connector(new Http11NioProtocol(endpoint));
+ connector.setPort(0);
+ connector.setThrowOnFailure(true);
+ Assert.assertTrue(connector.setProperty("address", "localhost"));
+ tomcat.getService().removeConnector(tomcat.getConnector());
+ tomcat.setConnector(connector);
+ Context context = getProgrammaticRootContext();
+ AtomicReference<EofHandler> handlerRef = new AtomicReference<>();
+ Tomcat.addServlet(context, "upgrade", new HttpServlet() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ protected void doGet(HttpServletRequest request,
HttpServletResponse response)
+ throws ServletException, IOException {
+ response.setHeader("Upgrade", "test");
+ EofHandler handler = request.upgrade(EofHandler.class);
+ handler.blockingRead = blockingRead;
+ handlerRef.set(handler);
+ }
+ });
+ context.addServletMapping("/", "upgrade");
+ tomcat.start();
+
+ try (Socket socket = new Socket("localhost", getPort())) {
+ socket.setSoTimeout(5000);
+ socket.getOutputStream().write(
+ "GET / HTTP/1.1\r\nHost: localhost\r\nUpgrade:
test\r\n\r\n".getBytes(StandardCharsets.US_ASCII));
+ BufferedReader reader = new BufferedReader(new
InputStreamReader(socket.getInputStream(),
+ StandardCharsets.US_ASCII));
+ Assert.assertTrue(reader.readLine().contains("101"));
+ while (!reader.readLine().isEmpty()) {
+ // Skip response headers
+ }
+ socket.shutdownOutput();
+
+ EofHandler handler = handlerRef.get();
+ try {
+ Assert.assertTrue(handler.inputFinished.await(5,
TimeUnit.SECONDS));
+ Assert.assertNull(handler.error);
+ endpoint.inputFinished = true;
+ /*
+ * A read event may already be queued, but EOF must not cause
repeated dispatches while output is open.
+ *
+ * There is a trade-off here. On slow systems a repeated read
may take a while to arrive but the longer
+ * the wait, the more delay it adds to the tests. 300ms is a
reasonable delay and while a heavily loaded
+ * test machine may miss the failure, the tests are run often
enough that that is acceptable.
+ */
+ Assert.assertFalse("Read polling continued after EOF",
+ endpoint.repeatedReads.await(300,
TimeUnit.MILLISECONDS));
+
+ if (blockingWrite) {
+ handler.writeResponse();
+ } else {
+ handler.connection.getOutputStream().setWriteListener(new
WriteListener() {
+ @Override
+ public void onWritePossible() throws IOException {
+ handler.writeResponse();
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ handler.error = t;
+ }
+ });
+ }
+ Assert.assertEquals('O', reader.read());
+ // Closing from an application thread must still close the
socket when read polling has stopped.
+ Assert.assertEquals(-1, reader.read());
+ Assert.assertTrue(handler.destroyed.await(5,
TimeUnit.SECONDS));
+ Assert.assertNull(handler.error);
+ } finally {
+ handler.connection.close();
+ }
+ }
+ }
+
@Test
public void testNoCallbacksAfterError() throws Exception {
SocketWrapperBase<?> socketWrapper =
EasyMock.createMock(SocketWrapperBase.class);
@@ -885,6 +991,84 @@ public class TestUpgrade extends TomcatBaseTest {
}
+ public static class CountingEndpoint extends NioEndpoint {
+ private volatile boolean inputFinished;
+ private final CountDownLatch repeatedReads = new CountDownLatch(2);
+
+ @Override
+ public boolean processSocket(SocketWrapperBase<NioChannel> socket,
SocketEvent event, boolean dispatch) {
+ if (inputFinished && event == SocketEvent.OPEN_READ) {
+ repeatedReads.countDown();
+ }
+ return super.processSocket(socket, event, dispatch);
+ }
+ }
+
+ public static class EofHandler implements HttpUpgradeHandler, ReadListener
{
+ private boolean blockingRead;
+ private WebConnection connection;
+ private ServletInputStream input;
+ private final CountDownLatch inputFinished = new CountDownLatch(1);
+ private final CountDownLatch destroyed = new CountDownLatch(1);
+ private volatile Throwable error;
+
+ @Override
+ public void init(WebConnection connection) {
+ this.connection = connection;
+ try {
+ input = connection.getInputStream();
+ if (blockingRead) {
+ byte[] buffer = new byte[1024];
+ while (input.read(buffer) != -1) {
+ // Discard data until EOF
+ }
+ inputFinished.countDown();
+ } else {
+ input.setReadListener(this);
+ }
+ } catch (IOException e) {
+ onError(e);
+ }
+ }
+
+ @Override
+ public void onDataAvailable() throws IOException {
+ byte[] buffer = new byte[1024];
+ while (input.isReady() && input.read(buffer) != -1) {
+ // Discard data until EOF
+ }
+ }
+
+ @Override
+ public void onAllDataRead() {
+ // Leave both streams open until the application is ready to send
a response.
+ inputFinished.countDown();
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ error = t;
+ inputFinished.countDown();
+ }
+
+ private void writeResponse() throws IOException {
+ ServletOutputStream output = connection.getOutputStream();
+ output.write('O');
+ output.flush();
+ try {
+ connection.close();
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void destroy() {
+ destroyed.countDown();
+ }
+ }
+
+
private static class NoOpReadListener implements ReadListener {
@Override
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index fc1d308edd..a3655efde8 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -664,6 +664,13 @@
Detect and disconnect a mis-behaving HTTP/2 client that sends large
numbers of unexpected settings acknowledgement frames. (markt)
</fix>
+ <fix>
+ Fix excessive CPU usage after a client half-closes an upgraded
+ connection while the application keeps the output stream open to finish
+ its response. Stop polling for input at end of stream and ensure that a
+ subsequent <code>WebConnection.close()</code> closes the connection.
+ (markt)
+ </fix>
</changelog>
</subsection>
<subsection name="Jasper">
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
