This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 9.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 847c8d2e219ab8835dca57f60fa0dfe2a1e4c279 Author: opencode <[email protected]> AuthorDate: Tue Oct 13 13:17:37 2026 +0200 Fix end-of-input and ReadListener call-back contract for upgraded NIO connections Reporting a clean client EOF (peer FIN) by throwing an EOFException from NioSocketWrapper.fillReadBuffer() meant that upgraded connections could not implement the expected input stream contract: a blocking read threw instead of returning -1 and, for non-blocking reads, the clean end of the input was reported to the ReadListener as onError(EOFException) rather than as the normal completion call-back onAllDataRead(). It also made isReadyForRead() throw at the end of the input. NioSocketWrapper now returns -1 from fillReadBuffer() at the end of the input, which read() propagates to the caller, and isReadyForRead() reports the socket as ready in that case since a subsequent read can immediately report the end of the input. All callers of these methods (HTTP/1.1 input buffer, AJP, HTTP/2, WebSocket and the upgrade streams) already handle a -1 return, so the change restores the contract those consumers were written against. The socket wrapper and processor read javadoc was updated to document the end-of-input return value. In UpgradeServletInputStream, an error reported to the ReadListener is now terminal: processing stops immediately when checking read readiness fails and the new errorReported flag suppresses any further call-backs on later dispatches, fixing the previously possible onAllDataRead() after onError() ordering. The onAllDataRead() call-back is now also reported at most once via the new allDataReadReported flag. Tests cover the blocking end-of-input return, the non-blocking clean onAllDataRead() path and the terminal nature of the error and all-data-read call-backs. --- .../http11/upgrade/UpgradeServletInputStream.java | 24 +- java/org/apache/tomcat/util/net/NioEndpoint.java | 13 +- .../apache/tomcat/util/net/SocketWrapperBase.java | 10 +- .../apache/coyote/http11/upgrade/TestUpgrade.java | 303 +++++++++++++++++++++ webapps/docs/changelog.xml | 10 + 5 files changed, 350 insertions(+), 10 deletions(-) diff --git a/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java b/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java index d0331d9d6a..3ad23974c6 100644 --- a/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java +++ b/java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java @@ -68,6 +68,18 @@ public class UpgradeServletInputStream extends ServletInputStream { */ private volatile Boolean ready = Boolean.TRUE; + /** + * Whether the error call-back has been triggered. No further call-backs + * are permitted once the error call-back has been triggered. + */ + private volatile boolean errorReported = false; + + /** + * Whether the all data read call-back has been triggered. It is only + * permitted to trigger the all data read call-back once. + */ + private volatile boolean allDataReadReported = false; + /** * The async read listener. */ @@ -245,12 +257,20 @@ public class UpgradeServletInputStream extends ServletInputStream { final void onDataAvailable() { + if (errorReported || allDataReadReported) { + // No further call-backs are permitted once an error has been + // reported or once the end of the data has been reported. + return; + } try { if (listener == null || !socketWrapper.isReadyForRead()) { return; } } catch (IOException ioe) { onError(ioe); + // Once an error has been reported to the listener, no further + // listener call-backs are permitted so stop processing here. + return; } ready = Boolean.TRUE; ClassLoader oldCL = processor.getUpgradeToken().getContextBind().bind(false, null); @@ -258,7 +278,8 @@ public class UpgradeServletInputStream extends ServletInputStream { if (!eof) { listener.onDataAvailable(); } - if (eof) { + if (eof && !errorReported) { + allDataReadReported = true; listener.onAllDataRead(); } } catch (Throwable t) { @@ -274,6 +295,7 @@ public class UpgradeServletInputStream extends ServletInputStream { if (listener == null) { return; } + errorReported = true; ClassLoader oldCL = processor.getUpgradeToken().getContextBind().bind(false, null); try { listener.onError(t); diff --git a/java/org/apache/tomcat/util/net/NioEndpoint.java b/java/org/apache/tomcat/util/net/NioEndpoint.java index f6167447e8..9fc0ad0092 100644 --- a/java/org/apache/tomcat/util/net/NioEndpoint.java +++ b/java/org/apache/tomcat/util/net/NioEndpoint.java @@ -1615,9 +1615,13 @@ public class NioEndpoint extends AbstractJsseEndpoint<NioChannel,SocketChannel> return true; } - fillReadBuffer(false); + int nRead = fillReadBuffer(false); - return socketBufferHandler.getReadBuffer().position() > 0; + // A return of -1 signals the end of the input. Report that the + // socket is ready in that case since a subsequent read is able to + // report the end of the data to the reader immediately rather than + // waiting for data that will never arrive. + return nRead == -1 || socketBufferHandler.getReadBuffer().position() > 0; } @@ -1754,7 +1758,7 @@ public class NioEndpoint extends AbstractJsseEndpoint<NioChannel,SocketChannel> synchronized (readLock) { n = getSocket().read(buffer); if (n == -1) { - throw new EOFException(); + return -1; } else if (n == 0) { // Ensure a spurious wake-up doesn't trigger a duplicate registration if (!readBlocking) { @@ -1779,9 +1783,6 @@ public class NioEndpoint extends AbstractJsseEndpoint<NioChannel,SocketChannel> } while (n == 0); // TLS needs to loop as reading zero application bytes is possible } else { n = getSocket().read(buffer); - if (n == -1) { - throw new EOFException(); - } } return n; } diff --git a/java/org/apache/tomcat/util/net/SocketWrapperBase.java b/java/org/apache/tomcat/util/net/SocketWrapperBase.java index 9409eee359..5540cd77f2 100644 --- a/java/org/apache/tomcat/util/net/SocketWrapperBase.java +++ b/java/org/apache/tomcat/util/net/SocketWrapperBase.java @@ -580,7 +580,8 @@ public abstract class SocketWrapperBase<E> { * @param off The offset in the array * @param len The number of bytes to read * - * @return the number of bytes read + * @return the number of bytes read, or -1 if the end of the input has been + * reached * * @throws IOException If an I/O error occurs */ @@ -592,14 +593,17 @@ public abstract class SocketWrapperBase<E> { * @param block Whether to block * @param to The ByteBuffer to read into * - * @return the number of bytes read + * @return the number of bytes read, or -1 if the end of the input has been + * reached * * @throws IOException If an I/O error occurs */ public abstract int read(boolean block, ByteBuffer to) throws IOException; /** - * Checks if the socket is ready for reading. + * Checks if the socket is ready for reading. Note that this method may + * return {@code true} at the end of the input since, in that case, a + * subsequent read is able to report the end of the input immediately. * * @return {@code true} if data is available to read * diff --git a/test/org/apache/coyote/http11/upgrade/TestUpgrade.java b/test/org/apache/coyote/http11/upgrade/TestUpgrade.java index ae46f81013..db993e4510 100644 --- a/test/org/apache/coyote/http11/upgrade/TestUpgrade.java +++ b/test/org/apache/coyote/http11/upgrade/TestUpgrade.java @@ -48,6 +48,15 @@ import static org.apache.catalina.startup.SimpleHttpClient.CRLF; import org.apache.catalina.Context; import org.apache.catalina.startup.Tomcat; import org.apache.catalina.startup.TomcatBaseTest; +import org.apache.coyote.UpgradeToken; +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.SSLSupport; +import org.apache.tomcat.util.net.SocketEvent; +import org.apache.tomcat.util.net.SocketWrapperBase; +import org.easymock.EasyMock; public class TestUpgrade extends TomcatBaseTest { @@ -102,6 +111,128 @@ public class TestUpgrade extends TomcatBaseTest { doTestFixedResponse(FixedResponseNonBlocking.class); } + @Test + public void testCleanEofWithBlockingRead() throws Exception { + UpgradeConnection uc = doUpgrade(BlockingEof.class); + + // Half-close the connection from the client side. The blocking read in + // the upgrade handler should then see the end of the input. + uc.shutdownOutput(); + + int c = uc.getReader().read(); + + // End of input should be signalled by read() returning -1 so the + // handler reports normal termination. + Assert.assertEquals(BlockingEof.EOF_MARKER, c); + + uc.shutdownInput(); + } + + @Test + public void testCleanEofWithNonBlockingRead() throws Exception { + UpgradeConnection uc = doUpgrade(NonBlockingEof.class); + + uc.getWriter().write(MESSAGE); + uc.getWriter().flush(); + + // Half-close the connection from the client side. The read listener in + // the upgrade handler should then see the end of the input. + uc.shutdownOutput(); + + int c = uc.getReader().read(); + + // End of input should be reported via onAllDataRead() (normal + // termination marker) rather than via onError(). + Assert.assertEquals(NonBlockingEof.EOF_MARKER, c); + + uc.shutdownInput(); + } + + @Test + @SuppressWarnings("unchecked") + public void testNoCallbacksAfterError() throws Exception { + SocketWrapperBase<?> socketWrapper = EasyMock.createMock(SocketWrapperBase.class); + ReadListener listener = EasyMock.createMock(ReadListener.class); + + ContextBind contextBind = new ContextBind() { + @Override + public ClassLoader bind(ClassLoader originalClassLoader) { + return null; + } + + @Override + public void unbind(ClassLoader originalClassLoader) { + // NO-OP + } + }; + UpgradeProcessorBase processor = + new TesterUpgradeProcessor(new UpgradeToken(null, contextBind, null, "test")); + + IOException error = new IOException("Deliberately triggered for the test"); + + socketWrapper.registerReadInterest(); + // The first call reports an error. No further socket state check or + // listener call-back may occur once the error has been reported. + EasyMock.expect(socketWrapper.isReadyForRead()).andThrow(error); + listener.onError(error); + EasyMock.expectLastCall().once(); + + EasyMock.replay(socketWrapper, listener); + + UpgradeServletInputStream sis = new UpgradeServletInputStream(processor, socketWrapper, + new UpgradeInfo()); + sis.setReadListener(listener); + + sis.onDataAvailable(); + sis.onDataAvailable(); + + EasyMock.verify(socketWrapper, listener); + } + + @Test + @SuppressWarnings("unchecked") + public void testOnAllDataReadReportedOnlyOnce() throws Exception { + SocketWrapperBase<?> socketWrapper = EasyMock.createMock(SocketWrapperBase.class); + + ContextBind contextBind = new ContextBind() { + @Override + public ClassLoader bind(ClassLoader originalClassLoader) { + return null; + } + + @Override + public void unbind(ClassLoader originalClassLoader) { + // NO-OP + } + }; + UpgradeProcessorBase processor = + new TesterUpgradeProcessor(new UpgradeToken(null, contextBind, null, "test")); + + // The first dispatch reports the socket as ready and the subsequent + // read by the listener triggers the end of the data. The all data read + // call-back must only be reported once, even if further dispatches + // occur. + socketWrapper.registerReadInterest(); + EasyMock.expect(socketWrapper.isReadyForRead()).andReturn(Boolean.TRUE); + EasyMock.expect(socketWrapper.read(EasyMock.eq(false), EasyMock.anyObject(byte[].class), + EasyMock.anyInt(), EasyMock.anyInt())).andReturn(-1); + + EasyMock.replay(socketWrapper); + + UpgradeServletInputStream sis = new UpgradeServletInputStream(processor, socketWrapper, + new UpgradeInfo()); + RecordingReadListener listener = new RecordingReadListener(sis); + sis.setReadListener(listener); + + sis.onDataAvailable(); + sis.onDataAvailable(); + + EasyMock.verify(socketWrapper); + Assert.assertEquals(1, listener.dataAvailableCount); + Assert.assertEquals(1, listener.allDataReadCount); + Assert.assertNull(listener.error); + } + private void doTestCheckClosed( Class<? extends HttpUpgradeHandler> upgradeHandlerClass) throws Exception { @@ -496,6 +627,102 @@ public class TestUpgrade extends TomcatBaseTest { } + public static class BlockingEof implements HttpUpgradeHandler { + + public static final char EOF_MARKER = 'O'; + + @Override + public void init(WebConnection connection) { + try (ServletInputStream sis = connection.getInputStream(); + ServletOutputStream sos = connection.getOutputStream()) { + byte[] buffer = new byte[8192]; + int read; + while ((read = sis.read(buffer)) >= 0) { + // Discard any data until the end of the input is reached + } + // A blocking read should report the end of the input by + // returning -1 rather than by throwing an exception. + if (read == -1) { + sos.write(EOF_MARKER); + sos.flush(); + } + } catch (IOException ioe) { + throw new IllegalStateException(ioe); + } + } + + @Override + public void destroy() { + // NO-OP + } + } + + + public static class NonBlockingEof implements HttpUpgradeHandler { + + public static final char EOF_MARKER = 'O'; + public static final char ERROR_MARKER = 'E'; + + @Override + public void init(WebConnection connection) { + ServletInputStream sis; + ServletOutputStream sos; + try { + sis = connection.getInputStream(); + sos = connection.getOutputStream(); + } catch (IOException ioe) { + throw new IllegalStateException(ioe); + } + sis.setReadListener(new EofReadListener(sis, sos)); + } + + @Override + public void destroy() { + // NO-OP + } + + private static class EofReadListener implements ReadListener { + + private final ServletInputStream sis; + private final ServletOutputStream sos; + + EofReadListener(ServletInputStream sis, ServletOutputStream sos) { + this.sis = sis; + this.sos = sos; + } + + @Override + public void onDataAvailable() throws IOException { + byte[] buffer = new byte[8192]; + while (sis.isReady()) { + if (sis.read(buffer) == -1) { + break; + } + } + } + + @Override + public void onAllDataRead() throws IOException { + // Normal end of input. Report that all data was read. + sos.write(EOF_MARKER); + sos.flush(); + sis.close(); + sos.close(); + } + + @Override + public void onError(Throwable throwable) { + try { + sos.write(ERROR_MARKER); + sos.flush(); + } catch (IOException ioe) { + // Ignore + } + } + } + } + + private static class NoOpReadListener implements ReadListener { @Override @@ -527,4 +754,80 @@ public class TestUpgrade extends TomcatBaseTest { // NO-OP } } + + + private static class RecordingReadListener implements ReadListener { + + private final ServletInputStream sis; + private final byte[] buffer = new byte[8192]; + + private int dataAvailableCount = 0; + private int allDataReadCount = 0; + private Throwable error = null; + + RecordingReadListener(ServletInputStream sis) { + this.sis = sis; + } + + @Override + public void onDataAvailable() throws IOException { + dataAvailableCount++; + while (sis.isReady() && sis.read(buffer) > 0) { + // Discard the data + } + } + + @Override + public void onAllDataRead() { + allDataReadCount++; + } + + @Override + public void onError(Throwable throwable) { + error = throwable; + } + } + + + private static class TesterUpgradeProcessor extends UpgradeProcessorBase { + + TesterUpgradeProcessor(UpgradeToken upgradeToken) { + super(upgradeToken); + } + + @Override + protected Log getLog() { + return LogFactory.getLog(TesterUpgradeProcessor.class); + } + + @Override + protected SocketState dispatch(SocketEvent status) { + return null; + } + + @Override + public void close() { + // NO-OP + } + + @Override + public ServletInputStream getInputStream() { + return null; + } + + @Override + public ServletOutputStream getOutputStream() { + return null; + } + + @Override + public void pause() { + // NO-OP + } + + @Override + public void setSslSupport(SSLSupport sslSupport) { + // NO-OP + } + } } diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index 24ab661204..ab875767c3 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -263,6 +263,16 @@ an application loop that polls <code>available()</code> running. (remm) </fix> + <fix> + Report a clean end of input on an upgraded connection as end of + stream, with blocking and non-blocking reads returning + <code>-1</code> and the non-blocking <code>ReadListener</code> + receiving <code>onAllDataRead()</code>, rather than as an error with + an <code>EOFException</code>. Also ensure that no further + <code>ReadListener</code> call-backs occur once an error has been + reported to the listener, or once the end of the data has been + reported. (remm) + </fix> </changelog> </subsection> <subsection name="Web applications"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
