8343855: HTTP/2 ConnectionWindowUpdateSender may miss some unprocessed DataFrames from closed streams

Reviewed-by: phh, rschmelter
Backport-of: bd6152f5967107d7b32db9bcfa224fc07314f098

Author: GoeLin

src/java.net.http/share/classes/jdk/internal/net/http/Stream.java

@@ -46,6 +46,8 @@
 import java.util.concurrent.Flow.Subscription;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiPredicate;
 import java.net.http.HttpClient;
 import java.net.http.HttpHeaders;
@@ -157,6 +159,10 @@ class Stream<T> extends ExchangeImpl<T> {
 
     // send lock: prevent sending DataFrames after reset occurred.
     private final Object sendLock = new Object();
+    // inputQ lock: methods that take from the inputQ
+    //      must not run concurrently.
+    private final Lock inputQLock = new ReentrantLock();
+
     /**
      * A reference to this Stream's connection Send Window controller. The
      * stream MUST acquire the appropriate amount of Send Window before
@@ -177,6 +183,8 @@ HttpConnection connection() {
     private void schedule() {
         boolean onCompleteCalled = false;
         HttpResponse.BodySubscriber<T> subscriber = responseSubscriber;
+        // prevents drainInputQueue() from running concurrently
+        inputQLock.lock();
         try {
             if (subscriber == null) {
                 subscriber = responseSubscriber = pendingResponseSubscriber;
@@ -194,7 +202,7 @@ private void schedule() {
                     handleReset(rf, subscriber);
                     return;
                 }
-                DataFrame df = (DataFrame)frame;
+                DataFrame df = (DataFrame) frame;
                 boolean finished = df.getFlag(DataFrame.END_STREAM);
 
                 List<ByteBuffer> buffers = df.getData();
@@ -244,6 +252,7 @@ private void schedule() {
         } catch (Throwable throwable) {
             errorRef.compareAndSet(null, throwable);
         } finally {
+            inputQLock.unlock();
             if (sched.isStopped()) drainInputQueue();
         }
 
@@ -262,26 +271,36 @@ private void schedule() {
             } catch (Throwable x) {
                 Log.logError("Subscriber::onError threw exception: {0}", t);
             } finally {
+                // cancelImpl will eventually call drainInputQueue();
                 cancelImpl(t);
-                drainInputQueue();
             }
         }
     }
 
-    // must only be called from the scheduler schedule() loop.
-    // ensure that all received data frames are accounted for
+    // Called from the scheduler schedule() loop,
+    // or after resetting the stream.
+    // Ensures that all received data frames are accounted for
     // in the connection window flow control if the scheduler
     // is stopped before all the data is consumed.
+    // The inputQLock is used to prevent concurrently taking
+    // from the queue.
     private void drainInputQueue() {
         Http2Frame frame;
-        while ((frame = inputQ.poll()) != null) {
-            if (frame instanceof DataFrame df) {
-                // Data frames that have been added to the inputQ
-                // must be released using releaseUnconsumed() to
-                // account for the amount of unprocessed bytes
-                // tracked by the connection.windowUpdater.
-                connection.releaseUnconsumed(df);
+        // will wait until schedule() has finished taking
+        // from the queue, if needed.
+        inputQLock.lock();
+        try {
+            while ((frame = inputQ.poll()) != null) {
+                if (frame instanceof DataFrame df) {
+                    // Data frames that have been added to the inputQ
+                    // must be released using releaseUnconsumed() to
+                    // account for the amount of unprocessed bytes
+                    // tracked by the connection.windowUpdater.
+                    connection.releaseUnconsumed(df);
+                }
             }
+        } finally {
+            inputQLock.unlock();
         }
     }
 
@@ -393,12 +412,35 @@ private void receiveDataFrame(DataFrame df) {
                     return;
                 }
             }
-            inputQ.add(df);
+           pushDataFrame(len, df);
         } finally {
             sched.runOrSchedule();
         }
     }
 
+    // Ensures that no data frame is pushed on the inputQ
+    // after the stream is closed.
+    // Changes to the `closed` boolean are guarded by the
+    // stateLock. Contention should be low as only one
+    // thread at a time adds to the inputQ, and
+    // we can only contend when closing the stream.
+    // Note that this method can run concurrently with
+    // methods holding the inputQLock: that is OK.
+    // The inputQLock is there to ensure that methods
+    // taking from the queue are not running concurrently
+    // with each others, but concurrently adding at the
+    // end of the queue while peeking/polling at the head
+    // is OK.
+    private void pushDataFrame(int len, DataFrame df) {
+        boolean closed = false;
+        synchronized(this) {
+            if (!(closed = this.closed)) {
+                inputQ.add(df);
+            }
+        }
+        if (closed && len > 0) connection.releaseUnconsumed(df);
+    }
+
     /** Handles a RESET frame. RESET is always handled inline in the queue. */
     private void receiveResetFrame(ResetFrame frame) {
         inputQ.add(frame);
@@ -1475,6 +1517,8 @@ void cancelImpl(final Throwable e, final int resetFrameErrCode) {
             }
         } catch (Throwable ex) {
             Log.logError(ex);
+        } finally {
+            drainInputQueue();
         }
     }
 
@@ -1700,7 +1744,7 @@ String dbgString() {
         @Override
         protected boolean windowSizeExceeded(long received) {
             onProtocolError(new ProtocolException("stream %s flow control window exceeded"
-                            .formatted(streamid)), ResetFrame.FLOW_CONTROL_ERROR);
+                        .formatted(streamid)), ResetFrame.FLOW_CONTROL_ERROR);
             return true;
         }
     }

test/jdk/java/net/httpclient/http2/ConnectionFlowControlTest.java

@@ -175,7 +175,11 @@ void test(String uri) throws Exception {
                         var response = responses.get(keys[i]);
                         String ckey = response.headers().firstValue("X-Connection-Key").get();
                         if (label == null) label = ckey;
-                        assertEquals(ckey, label, "Unexpected key for " + query);
+                        if (i < max - 1) {
+                            // the connection window might be exceeded at i == max - 2, which
+                            // means that the last request could go on a new connection.
+                            assertEquals(ckey, label, "Unexpected key for " + query);
+                        }
                         int wait = uri.startsWith("https://") ? 500 : 250;
                         try (InputStream is = response.body()) {
                             Thread.sleep(Utils.adjustTimeout(wait));

test/jdk/java/net/httpclient/http2/StreamFlowControlTest.java

@@ -23,7 +23,7 @@
 
 /*
  * @test
- * @bug 8342075
+ * @bug 8342075 8343855
  * @library /test/lib /test/jdk/java/net/httpclient/lib
  * @build jdk.httpclient.test.lib.http2.Http2TestServer jdk.test.lib.net.SimpleSSLContext
  *        jdk.httpclient.test.lib.common.TestServerConfigurator
@@ -56,6 +56,7 @@
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLSession;
 
+import jdk.httpclient.test.lib.common.HttpServerAdapters.HttpHeadOrGetHandler;
 import jdk.httpclient.test.lib.common.HttpServerAdapters.HttpTestServer;
 import jdk.httpclient.test.lib.http2.BodyOutputStream;
 import jdk.httpclient.test.lib.http2.Http2Handler;
@@ -72,6 +73,7 @@
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.fail;
 
@@ -95,6 +97,19 @@ public Object[][] variants() {
         };
     }
 
+    static void sleep(long wait) throws InterruptedException {
+        if (wait <= 0) return;
+        long remaining = Utils.adjustTimeout(wait);
+        long start = System.nanoTime();
+        while (remaining > 0) {
+            Thread.sleep(remaining);
+            long end = System.nanoTime();
+            remaining = remaining - NANOSECONDS.toMillis(end - start);
+        }
+        System.out.printf("Waited %s ms%n",
+                NANOSECONDS.toMillis(System.nanoTime() - start));
+    }
+
 
     @Test(dataProvider = "variants")
     void test(String uri,
@@ -121,7 +136,7 @@ void test(String uri,
                 CompletableFuture<String> sent = new CompletableFuture<>();
                 responseSent.put(query, sent);
                 HttpRequest request = HttpRequest.newBuilder(uriWithQuery)
-                        .POST(BodyPublishers.ofString("Hello there!"))
+                        .GET()
                         .build();
                 System.out.println("\nSending request:" + uriWithQuery);
                 final HttpClient cc = client;
@@ -136,9 +151,9 @@ void test(String uri,
                     // we have to pull to get the exception, but slow enough
                     // so that DataFrames are buffered up to the point that
                     // the window is exceeded...
-                    int wait = uri.startsWith("https://") ? 500 : 350;
+                    long wait = uri.startsWith("https://") ? 800 : 350;
                     try (InputStream is = response.body()) {
-                        Thread.sleep(Utils.adjustTimeout(wait));
+                        sleep(wait);
                         is.readAllBytes();
                     }
                     // we could fail here if we haven't waited long enough
@@ -187,7 +202,7 @@ void testAsync(String uri,
                 CompletableFuture<String> sent = new CompletableFuture<>();
                 responseSent.put(query, sent);
                 HttpRequest request = HttpRequest.newBuilder(uriWithQuery)
-                        .POST(BodyPublishers.ofString("Hello there!"))
+                        .GET()
                         .build();
                 System.out.println("\nSending request:" + uriWithQuery);
                 final HttpClient cc = client;
@@ -201,9 +216,9 @@ void testAsync(String uri,
                         assertEquals(key, label, "Unexpected key for " + query);
                     }
                     sent.join();
-                    int wait = uri.startsWith("https://") ? 600 : 300;
+                    long wait = uri.startsWith("https://") ? 800 : 350;
                     try (InputStream is = response.body()) {
-                        Thread.sleep(Utils.adjustTimeout(wait));
+                        sleep(wait);
                         is.readAllBytes();
                     }
                     // we could fail here if we haven't waited long enough
@@ -269,7 +284,9 @@ public void setup() throws Exception {
         var https2TestServer = new Http2TestServer("localhost", true, sslContext);
         https2TestServer.addHandler(new Http2TestHandler(), "/https2/");
         this.https2TestServer = HttpTestServer.of(https2TestServer);
+        this.https2TestServer.addHandler(new HttpHeadOrGetHandler(), "/https2/head/");
         https2URI = "https://" + this.https2TestServer.serverAuthority() + "/https2/x";
+        String h2Head = "https://" + this.https2TestServer.serverAuthority() + "/https2/head/z";
 
         // Override the default exchange supplier with a custom one to enable
         // particular test scenarios
@@ -278,6 +295,17 @@ public void setup() throws Exception {
 
         this.http2TestServer.start();
         this.https2TestServer.start();
+
+        // warmup to eliminate delay due to SSL class loading and initialization.
+        HttpClient client = HttpClient.newBuilder().executor(Executors.newCachedThreadPool()).sslContext(sslContext).build();
+        try {
+            var request = HttpRequest.newBuilder(URI.create(h2Head)).method("HEAD", BodyPublishers.noBody()).build();
+            var resp = client.send(request, BodyHandlers.discarding());
+            assertEquals(resp.statusCode(), 200);
+        } finally {
+            ExecutorService exec = (ExecutorService)client.executor().get();
+            exec.shutdownNow();
+        }
     }
 
     @AfterTest
@@ -296,11 +324,19 @@ public void handle(Http2TestExchange t) throws IOException {
                  OutputStream os = t.getResponseBody()) {
 
                 byte[] bytes = is.readAllBytes();
-                System.out.println("Server " + t.getLocalAddress() + " received:\n"
-                        + t.getRequestURI() + ": " + new String(bytes, StandardCharsets.UTF_8));
+                if (bytes.length != 0) {
+                    System.out.println("Server " + t.getLocalAddress() + " received:\n"
+                            + t.getRequestURI() + ": " + new String(bytes, StandardCharsets.UTF_8));
+                } else {
+                    System.out.println("No request body for " + t.getRequestMethod());
+                }
+
                 t.getResponseHeaders().setHeader("X-Connection-Key", t.getConnectionKey());
 
-                if (bytes.length == 0) bytes = "no request body!".getBytes(StandardCharsets.UTF_8);
+                if (bytes.length == 0) {
+                    bytes = "no request body!"
+                            .repeat(100).getBytes(StandardCharsets.UTF_8);
+                }
                 int window = Integer.getInteger("jdk.httpclient.windowsize", 2 * 16 * 1024);
                 final int maxChunkSize;
                 if (t instanceof FCHttp2TestExchange fct) {
@@ -324,13 +360,22 @@ public void handle(Http2TestExchange t) throws IOException {
                             // ignore and continue...
                         }
                     }
-                    ((BodyOutputStream) os).writeUncontrolled(resp, 0, resp.length);
+                    try {
+                        ((BodyOutputStream) os).writeUncontrolled(resp, 0, resp.length);
+                    } catch (IOException x) {
+                        if (t instanceof FCHttp2TestExchange fct) {
+                            fct.conn.updateConnectionWindow(resp.length);
+                        }
+                    }
+                }
+            } finally {
+                if (t instanceof FCHttp2TestExchange fct) {
+                    fct.responseSent(query);
+                } else {
+                    fail("Exchange is not %s but %s"
+                            .formatted(FCHttp2TestExchange.class.getName(), t.getClass().getName()));
                 }
             }
-            if (t instanceof FCHttp2TestExchange fct) {
-                fct.responseSent(query);
-            } else fail("Exchange is not %s but %s"
-                    .formatted(FCHttp2TestExchange.class.getName(), t.getClass().getName()));
         }
     }