Resolved changes

Author: GoeLin

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

@@ -157,14 +157,13 @@ class Stream<T> extends ExchangeImpl<T> {
 
     // send lock: prevent sending DataFrames after reset occurred.
     private final Object sendLock = new Object();
-
     /**
      * A reference to this Stream's connection Send Window controller. The
      * stream MUST acquire the appropriate amount of Send Window before
      * sending any data. Will be null for PushStreams, as they cannot send data.
      */
     private final WindowController windowController;
-    private final WindowUpdateSender windowUpdater;
+    private final WindowUpdateSender streamWindowUpdater;
 
     @Override
     HttpConnection connection() {
@@ -510,6 +509,8 @@ void incoming(Http2Frame frame) throws IOException {
                 if (debug.on()) {
                     debug.log("request cancelled or stream closed: dropping data frame");
                 }
+                // Data frames that have not been added to the inputQ
+                // can be released using dropDataFrame
                 connection.dropDataFrame((DataFrame) frame);
             } else {
                 receiveDataFrame((DataFrame) frame);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2024, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -31,15 +31,30 @@
 import jdk.internal.net.http.frame.WindowUpdateFrame;
 import jdk.internal.net.http.common.Utils;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
+/**
+ * A class that tracks the amount of flow controlled
+ * data received on an HTTP/2 connection
+ */
 abstract class WindowUpdateSender {
 
     final Logger debug =
             Utils.getDebugLogger(this::dbgString, Utils.DEBUG);
 
+    // The threshold at which window updates are sent in bytes
     final int limit;
+    // The flow control window in bytes
+    final int windowSize;
     final Http2Connection connection;
-    final AtomicInteger received = new AtomicInteger();
+    // The amount of flow controlled data received and processed, in bytes,
+    // since the start of the window.
+    // The window is exhausted when received + unprocessed >= windowSize
+    final AtomicLong received = new AtomicLong();
+    // The amount of flow controlled data received and unprocessed, in bytes,
+    // since the start of the window.
+    // The window is exhausted when received + unprocessed >= windowSize
+    final AtomicLong unprocessed = new AtomicLong();
 
     WindowUpdateSender(Http2Connection connection) {
         this(connection, connection.clientSettings.getParameter(SettingsFrame.INITIAL_WINDOW_SIZE));
@@ -65,15 +80,117 @@ abstract class WindowUpdateSender {
                       maxFrameSize, initWindowSize, limit);
     }
 
+    // O for the connection window, > 0 for a stream window
     abstract int getStreamId();
 
+    /**
+     * {@return {@code true} if buffering the given amount of
+     * flow controlled data would not exceed the flow control
+     * window}
+     * <p>
+     * This method is called before buffering and processing
+     * a DataFrame. The count of unprocessed bytes is incremented
+     * by the given amount, and checked against the number of
+     * available bytes in the flow control window.
+     * <p>
+     * This method returns {@code true} if the bytes can be buffered
+     * without exceeding the flow control window, {@code false}
+     * if the flow control window is exceeded and corrective
+     * action (close/reset) has been taken.
+     * <p>
+     * When this method returns true, either {@link #processed(int)}
+     * or {@link #released(int)} must eventually be called to release
+     * the bytes from the flow control window.
+     *
+     * @implSpec
+     * an HTTP/2 endpoint may disable its own flow control
+     * (see <a href="https://www.rfc-editor.org/rfc/rfc9113.html#section-5.2.1">
+     *     RFC 9113, section 5.2.1</a>), in which case this
+     * method may return true even if the flow control window would
+     * be exceeded: that is, the flow control window is exceeded but
+     * the endpoint decided to take no corrective action.
+     *
+     * @param  len a number of unprocessed bytes, which
+     *             the caller wants to buffer.
+     */
+    boolean canBufferUnprocessedBytes(int len) {
+        return !checkWindowSizeExceeded(unprocessed.addAndGet(len));
+    }
+
+    // adds the provided amount to the amount of already
+    // received and processed bytes and checks whether the
+    // flow control window is exceeded. If so, take
+    // corrective actions and return true.
+    private boolean checkWindowSizeExceeded(long len) {
+        // because windowSize is bound by Integer.MAX_VALUE
+        // we will never reach the point where received.get() + len
+        // could overflow
+        long rcv = received.get() + len;
+        return rcv > windowSize && windowSizeExceeded(rcv);
+    }
+
+    /**
+     * Called after unprocessed buffered bytes have been
+     * processed, to release part of the flow control window
+     *
+     * @apiNote this method is called only when releasing bytes
+     * that where buffered after calling
+     * {@link #canBufferUnprocessedBytes(int)}.
+     *
+     * @param delta the amount of processed bytes to release
+     */
+    void processed(int delta) {
+        long rest = unprocessed.addAndGet(-delta);
+        assert rest >= 0;
+        update(delta);
+    }
+
+    /**
+     * Called when it is desired to release unprocessed bytes
+     * without processing them, or without triggering the
+     * sending of a window update. This method can be called
+     * instead of calling {@link #processed(int)}.
+     * When this method is called instead of calling {@link #processed(int)},
+     * it should generally be followed by a call to {@link #update(int)},
+     * unless the stream or connection is being closed.
+     *
+     * @apiNote this method should only be called to release bytes that
+     * have been buffered after calling {@link
+     * #canBufferUnprocessedBytes(int)}.
+     *
+     * @param delta the amount of bytes to release from the window
+     *
+     * @return the amount of remaining unprocessed bytes
+     */
+    long released(int delta) {
+        long rest = unprocessed.addAndGet(-delta);
+        assert rest >= 0;
+        return rest;
+    }
+
+    /**
+     * This method is called to update the flow control window,
+     * and possibly send a window update
+     *
+     * @apiNote this method can be called directly if a frame is
+     * dropped before calling {@link #canBufferUnprocessedBytes(int)}.
+     * Otherwise, either {@link #processed(int)} or {@link #released(int)}
+     * should be called, depending on whether sending a window update
+     * is desired or not. It is typically not desired to send an update
+     * if the stream or connection is being closed.
+     *
+     * @param delta the amount of bytes released from the window.
+     */
     void update(int delta) {
-        int rcv = received.addAndGet(delta);
+        long rcv = received.addAndGet(delta);
         if (debug.on()) debug.log("update: %d, received: %d, limit: %d", delta, rcv, limit);
+        if (rcv > windowSize && windowSizeExceeded(rcv)) {
+            return;
+        }
         if (rcv > limit) {
             synchronized (this) {
-                int tosend = received.get();
-                if( tosend > limit) {
+                int tosend = (int)Math.min(received.get(), Integer.MAX_VALUE);
+                if (tosend > limit) {
                     received.getAndAdd(-tosend);
                     sendWindowUpdate(tosend);
                 }