Issue #12272 - Potential deadlock with Vaadin.

sbordet · sbordet · commit 713a3085e863 · 2024-11-11T20:07:36.000+01:00
Fixed the case where a GOAWAY followed by a TCP FIN was causing a race between closing the `EndPoint` and running the failure `Runnable` task.

The TCP FIN after the GOAWAY causes the streams to be failed on the server;
in turn, failing the streams generates failure `Runnable` tasks that are submitted to the HTTP/2 execution strategy;
however, the streams were destroyed before the failure `Runnable` tasks actually ran, so the `EndPoint` was closed;
closing the `EndPoint` would close the `HTTP2Connection`, which in turn would stop the execution strategy;
this lead to the fact that the failure `Runnable` tasks were never run.

Now, the failure `Runnable` tasks are invoked via `ThreadPool.executeImmediately()` rather than being submitted to the execution strategy.
This ensures that they would be run and not queued, even in case of lack of threads, so that they could unblock blocked reads or writes, freeing up blocked threads.

Additionally, improved `HTTP2Stream.onFailure()` to destroy the stream only after the failure tasks have completed.

Smaller other fixes to improve the code.

Signed-off-by: Simone Bordet &lt;simone.bordet@gmail.com&gt;
diff --git a/jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Connection.java b/jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Connection.java
@@ -145,7 +145,6 @@ public void onClose(Throwable cause)
         if (LOG.isDebugEnabled())
             LOG.debug("HTTP2 Close {} ", this);
         super.onClose(cause);
-
         LifeCycle.stop(strategy);
     }
 
@@ -375,7 +374,8 @@ else if (filled == 0)
                     {
                         shutdown = true;
                         session.onShutdown();
-                        return null;
+                        // The onShutDown() call above may have produced a task.
+                        return pollTask();
                     }
                 }
             }
diff --git a/jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Stream.java b/jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Stream.java
@@ -615,13 +615,17 @@ private void onFailure(FailureFrame frame, Callback callback)
             failure = frame.getFailure();
             flowControlLength = drain();
         }
-        close();
-        boolean removed = session.removeStream(this);
         session.dataConsumed(this, flowControlLength);
-        if (removed)
-            notifyFailure(this, frame, callback);
-        else
-            callback.succeeded();
+        close();
+
+        notifyFailure(this, frame, new Nested(callback)
+        {
+            @Override
+            public void completed()
+            {
+                session.removeStream(HTTP2Stream.this);
+            }
+        });
     }
 
     private int drain()
diff --git a/jetty-core/jetty-http2/jetty-http2-server/src/main/java/org/eclipse/jetty/http2/server/internal/HTTP2ServerConnection.java b/jetty-core/jetty-http2/jetty-http2-server/src/main/java/org/eclipse/jetty/http2/server/internal/HTTP2ServerConnection.java
@@ -54,6 +54,7 @@
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.Promise;
 import org.eclipse.jetty.util.StringUtil;
+import org.eclipse.jetty.util.thread.ThreadPool;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -202,12 +203,9 @@ public void onStreamFailure(Stream stream, Throwable failure, Callback callback)
         if (channel != null)
         {
             Runnable task = channel.onFailure(failure, callback);
-            if (task != null)
-            {
-                // We must dispatch to another thread because the task
-                // may call application code that performs blocking I/O.
-                offerTask(task, true);
-            }
+            // The task may unblock a blocked read or write, so it cannot be
+            // queued, because there may be no threads available to run it.
+            ThreadPool.executeImmediately(getExecutor(), task);
         }
         else
         {
diff --git a/jetty-core/jetty-http2/jetty-http2-server/src/main/java/org/eclipse/jetty/http2/server/internal/HttpStreamOverHTTP2.java b/jetty-core/jetty-http2/jetty-http2-server/src/main/java/org/eclipse/jetty/http2/server/internal/HttpStreamOverHTTP2.java
@@ -591,12 +591,32 @@ public Runnable onFailure(Throwable failure, Callback callback)
     {
         boolean remote = failure instanceof EOFException;
         Runnable runnable = remote ? _httpChannel.onRemoteFailure(new EofException(failure)) : _httpChannel.onFailure(failure);
-        return () ->
+
+        class FailureTask implements Runnable
         {
-            if (runnable != null)
-                runnable.run();
-            callback.succeeded();
-        };
+            @Override
+            public void run()
+            {
+                try
+                {
+                    if (runnable != null)
+                        runnable.run();
+                    callback.succeeded();
+                }
+                catch (Throwable x)
+                {
+                    callback.failed(x);
+                }
+            }
+
+            @Override
+            public String toString()
+            {
+                return "%s[%s]".formatted(getClass().getSimpleName(), runnable);
+            }
+        }
+
+        return new FailureTask();
     }
 
     @Override
diff --git a/jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/SocketChannelEndPoint.java b/jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/SocketChannelEndPoint.java
@@ -89,7 +89,8 @@ else if (filled == -1)
         }
         catch (IOException e)
         {
-            LOG.debug("Unable to shutdown output", e);
+            if (LOG.isDebugEnabled())
+                LOG.debug("Unable to shutdown input", e);
             shutdownInput();
             filled = -1;
         }
diff --git a/jetty-ee10/jetty-ee10-tests/jetty-ee10-test-client-transports/src/test/java/org/eclipse/jetty/ee10/test/client/transport/Http2AsyncIOServletTest.java b/jetty-ee10/jetty-ee10-tests/jetty-ee10-test-client-transports/src/test/java/org/eclipse/jetty/ee10/test/client/transport/Http2AsyncIOServletTest.java
@@ -15,6 +15,8 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -33,23 +35,29 @@
 import org.eclipse.jetty.http.HttpURI;
 import org.eclipse.jetty.http.HttpVersion;
 import org.eclipse.jetty.http.MetaData;
+import org.eclipse.jetty.http2.CloseState;
 import org.eclipse.jetty.http2.ErrorCode;
+import org.eclipse.jetty.http2.HTTP2Session;
 import org.eclipse.jetty.http2.api.Session;
 import org.eclipse.jetty.http2.api.Stream;
 import org.eclipse.jetty.http2.client.HTTP2Client;
+import org.eclipse.jetty.http2.frames.DataFrame;
 import org.eclipse.jetty.http2.frames.HeadersFrame;
 import org.eclipse.jetty.http2.frames.ResetFrame;
 import org.eclipse.jetty.http2.server.HTTP2CServerConnectionFactory;
 import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.server.HttpConfiguration;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.FuturePromise;
 import org.eclipse.jetty.util.component.LifeCycle;
+import org.eclipse.jetty.util.thread.QueuedThreadPool;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.awaitility.Awaitility.await;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.nullValue;
@@ -64,15 +72,20 @@ public class Http2AsyncIOServletTest
 
     private void start(HttpServlet httpServlet) throws Exception
     {
-        server = new Server();
+        QueuedThreadPool serverThreads = new QueuedThreadPool();
+        serverThreads.setName("server");
+        server = new Server(serverThreads);
         connector = new ServerConnector(server, 1, 1, new HTTP2CServerConnectionFactory(httpConfig));
         server.addConnector(connector);
         ServletContextHandler servletContextHandler = new ServletContextHandler("/");
         servletContextHandler.addServlet(new ServletHolder(httpServlet), "/*");
         server.setHandler(servletContextHandler);
         server.start();
 
+        QueuedThreadPool clientThreads = new QueuedThreadPool();
+        clientThreads.setName("client");
         client = new HTTP2Client();
+        client.setExecutor(clientThreads);
         client.start();
     }
 
@@ -218,4 +231,206 @@ public void onStartAsync(AsyncEvent event)
 
         assertTrue(errorLatch.await(5, TimeUnit.SECONDS));
     }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    public void testSessionCloseWithPendingRequestThenReset(boolean useReaderWriter) throws Exception
+    {
+        // Disable output aggregation for Servlets, so each byte is echoed back.
+        httpConfig.setOutputAggregationSize(0);
+        CountDownLatch serverFailureLatch = new CountDownLatch(1);
+        start(new HttpServlet()
+        {
+            @Override
+            protected void service(HttpServletRequest request, HttpServletResponse response) throws IOException
+            {
+                try
+                {
+                    if (useReaderWriter)
+                        request.getReader().transferTo(response.getWriter());
+                    else
+                        request.getInputStream().transferTo(response.getOutputStream());
+                }
+                catch (Throwable x)
+                {
+                    serverFailureLatch.countDown();
+                    throw x;
+                }
+            }
+        });
+
+        HTTP2Session session = (HTTP2Session)client.connect(new InetSocketAddress("localhost", connector.getLocalPort()), new Session.Listener() {})
+            .get(5, TimeUnit.SECONDS);
+        Queue<Stream.Data> dataList = new ConcurrentLinkedQueue<>();
+        MetaData.Request request = new MetaData.Request("POST", HttpURI.from("/"), HttpVersion.HTTP_2, HttpFields.EMPTY);
+        Stream stream = session.newStream(new HeadersFrame(request, null, false), new Stream.Listener()
+        {
+            @Override
+            public void onDataAvailable(Stream stream)
+            {
+                while (true)
+                {
+                    Stream.Data data = stream.readData();
+                    if (data == null)
+                    {
+                        stream.demand();
+                        return;
+                    }
+                    dataList.offer(data);
+                    if (data.frame().isEndStream())
+                        return;
+                }
+            }
+        }).get(5, TimeUnit.SECONDS);
+
+        stream.data(new DataFrame(stream.getId(), UTF_8.encode("Hello Jetty"), false))
+            .get(5, TimeUnit.SECONDS);
+        stream.demand();
+
+        await().atMost(5, TimeUnit.SECONDS).until(() -> !dataList.isEmpty());
+
+        // Initiates graceful close, waits for the streams to finish as per specification.
+        session.close(ErrorCode.NO_ERROR.code, "client_close", Callback.NOOP);
+
+        // Finish the pending stream, either by resetting or sending the last frame.
+        stream.reset(new ResetFrame(stream.getId(), ErrorCode.CANCEL_STREAM_ERROR.code));
+
+        // The server should see the effects of the reset.
+        assertTrue(serverFailureLatch.await(5, TimeUnit.SECONDS));
+        // The session must eventually be closed.
+        await().atMost(5, TimeUnit.SECONDS).until(() -> session.getCloseState() == CloseState.CLOSED);
+        // The endPoint must eventually be closed.
+        await().atMost(5, TimeUnit.SECONDS).until(() -> !session.getEndPoint().isOpen());
+
+        // Cleanup.
+        dataList.forEach(Stream.Data::release);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    public void testSessionCloseWithPendingRequestServerIdleTimeout(boolean useReaderWriter) throws Exception
+    {
+        // Disable output aggregation for Servlets, so each byte is echoed back.
+        httpConfig.setOutputAggregationSize(0);
+        CountDownLatch serverFailureLatch = new CountDownLatch(1);
+        start(new HttpServlet()
+        {
+            @Override
+            protected void service(HttpServletRequest request, HttpServletResponse response) throws IOException
+            {
+                try
+                {
+                    if (useReaderWriter)
+                        request.getReader().transferTo(response.getWriter());
+                    else
+                        request.getInputStream().transferTo(response.getOutputStream());
+                }
+                catch (Throwable x)
+                {
+                    serverFailureLatch.countDown();
+                    throw x;
+                }
+            }
+        });
+        long idleTimeout = 1000;
+        connector.setIdleTimeout(idleTimeout);
+
+        HTTP2Session session = (HTTP2Session)client.connect(new InetSocketAddress("localhost", connector.getLocalPort()), new Session.Listener() {})
+            .get(5, TimeUnit.SECONDS);
+        Queue<Stream.Data> dataList = new ConcurrentLinkedQueue<>();
+        MetaData.Request request = new MetaData.Request("POST", HttpURI.from("/"), HttpVersion.HTTP_2, HttpFields.EMPTY);
+        Stream stream = session.newStream(new HeadersFrame(request, null, false), new Stream.Listener()
+        {
+            @Override
+            public void onDataAvailable(Stream stream)
+            {
+                while (true)
+                {
+                    Stream.Data data = stream.readData();
+                    if (data == null)
+                    {
+                        stream.demand();
+                        return;
+                    }
+                    dataList.offer(data);
+                    if (data.frame().isEndStream())
+                        return;
+                }
+            }
+        }).get(5, TimeUnit.SECONDS);
+
+        stream.data(new DataFrame(stream.getId(), UTF_8.encode("Hello Jetty"), false))
+            .get(5, TimeUnit.SECONDS);
+        stream.demand();
+
+        await().atMost(5, TimeUnit.SECONDS).until(() -> !dataList.isEmpty());
+
+        // Initiates graceful close, waits for the streams to finish as per specification.
+        session.close(ErrorCode.NO_ERROR.code, "client_close", Callback.NOOP);
+
+        // Do not finish the streams, the server must idle timeout.
+        assertTrue(serverFailureLatch.await(2 * idleTimeout, TimeUnit.SECONDS));
+        // The session must eventually be closed.
+        await().atMost(5, TimeUnit.SECONDS).until(() -> session.getCloseState() == CloseState.CLOSED);
+        // The endPoint must eventually be closed.
+        await().atMost(5, TimeUnit.SECONDS).until(() -> !session.getEndPoint().isOpen());
+
+        // Cleanup.
+        dataList.forEach(Stream.Data::release);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    public void testSessionCloseWithPendingRequestThenClientDisconnectThenServerIdleTimeout(boolean useReaderWriter) throws Exception
+    {
+        AtomicReference<Thread> serverThreadRef = new AtomicReference<>();
+        CountDownLatch serverFailureLatch = new CountDownLatch(1);
+        start(new HttpServlet()
+        {
+            @Override
+            protected void service(HttpServletRequest request, HttpServletResponse response) throws IOException
+            {
+                try
+                {
+                    serverThreadRef.set(Thread.currentThread());
+                    if (useReaderWriter)
+                        request.getReader().transferTo(response.getWriter());
+                    else
+                        request.getInputStream().transferTo(response.getOutputStream());
+                }
+                catch (Throwable x)
+                {
+                    serverFailureLatch.countDown();
+                    throw x;
+                }
+            }
+        });
+        long idleTimeout = 1000;
+        connector.setIdleTimeout(idleTimeout);
+
+        HTTP2Session session = (HTTP2Session)client.connect(new InetSocketAddress("localhost", connector.getLocalPort()), new Session.Listener() {})
+            .get(5, TimeUnit.SECONDS);
+        MetaData.Request request = new MetaData.Request("POST", HttpURI.from("/"), HttpVersion.HTTP_2, HttpFields.EMPTY);
+        Stream stream = session.newStream(new HeadersFrame(request, null, false), new Stream.Listener() {}).get(5, TimeUnit.SECONDS);
+
+        stream.data(new DataFrame(stream.getId(), UTF_8.encode("Hello Jetty"), false))
+            .get(5, TimeUnit.SECONDS);
+        stream.demand();
+
+        await().atMost(5, TimeUnit.SECONDS).until(() ->
+        {
+            Thread serverThread = serverThreadRef.get();
+            return serverThread != null && serverThread.getState() == Thread.State.WAITING;
+        });
+
+        // Initiates graceful close, then immediately disconnect.
+        session.close(ErrorCode.NO_ERROR.code, "client_close", Callback.from(session::disconnect));
+
+        // Do not finish the streams, the server must idle timeout.
+        assertTrue(serverFailureLatch.await(2 * idleTimeout, TimeUnit.SECONDS));
+        // The session must eventually be closed.
+        await().atMost(5, TimeUnit.SECONDS).until(() -> session.getCloseState() == CloseState.CLOSED);
+        // The endPoint must eventually be closed.
+        await().atMost(5, TimeUnit.SECONDS).until(() -> !session.getEndPoint().isOpen());
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -145,7 +145,6 @@ public void onClose(Throwable cause)`
`145`	`145`	`if (LOG.isDebugEnabled())`
`146`	`146`	`LOG.debug("HTTP2 Close {} ", this);`
`147`	`147`	`super.onClose(cause);`
`148`		`-`
`149`	`148`	`LifeCycle.stop(strategy);`
`150`	`149`	`}`
`151`	`150`
`@@ -375,7 +374,8 @@ else if (filled == 0)`
`375`	`374`	`{`
`376`	`375`	`shutdown = true;`
`377`	`376`	`session.onShutdown();`
`378`		`- return null;`
	`377`	`+ // The onShutDown() call above may have produced a task.`
	`378`	`+ return pollTask();`
`379`	`379`	`}`
`380`	`380`	`}`
`381`	`381`	`}`
Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,8 @@ else if (filled == -1)`
`89`	`89`	`}`
`90`	`90`	`catch (IOException e)`
`91`	`91`	`{`
`92`		`- LOG.debug("Unable to shutdown output", e);`
	`92`	`+ if (LOG.isDebugEnabled())`
	`93`	`+ LOG.debug("Unable to shutdown input", e);`
`93`	`94`	`shutdownInput();`
`94`	`95`	`filled = -1;`
`95`	`96`	`}`