Ensure HTTP/2 flow control to work at the stream level (#6266)

Motivation:

Since `InboundTrafficController` operates at the TCP level, it isn't
aligned with HTTP/2 flow control. As a result, a stream may receive more
than its stream window size allows and interfering with other streams
from receiving data.

Related: #6253

Modifications:

- Integrated `InboundTrafficController` with HTTP/2
`Http2LocalFlowController`
- Invokes `Http2LocalFlowController.consumeBytes()` when steam data is
consumed to send a `WINDOW_UPDATE` frame.
- Introduced
`{ServerBuilder,ClientFactoryBuilder}.http2StreamWindowUpdateRatio` to
customize the threshold to send WINDOW_UPDATE frames.
- Fixed `client.Http2ResponseDecoder` and `server.Http2RequestDecoder`
to defer reporting the consumed data length.
- The length is not reported via `InboundTrafficController` when the
data is consumed in userland.

Result:

- Fixed a bug where HTTP/2 flow control did not work properly, and
stream-level windowing was ignored.
- Closes #6253

---------

Co-authored-by: Yizhou Feng <yizhou.feng@databricks.com>

Author: ikhoon

core/src/main/java/com/linecorp/armeria/client/AbstractHttpRequestHandler.java

@@ -187,6 +187,7 @@ final boolean tryInitialize() {
         }
 
         this.session = session;
+        originalRes.setId(id);
         responseWrapper = responseDecoder.addResponse(this, id, originalRes, ctx, ch.eventLoop());
 
         if (timeoutMillis > 0) {

core/src/main/java/com/linecorp/armeria/client/ClientFactoryBuilder.java

@@ -76,6 +76,7 @@
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
+import io.netty.handler.codec.http2.DefaultHttp2LocalFlowController;
 import io.netty.handler.codec.http2.Http2CodecUtil;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
@@ -634,6 +635,26 @@ public ClientFactoryBuilder http2InitialStreamWindowSize(int http2InitialStreamW
         return this;
     }
 
+    /**
+     * Sets the threshold ratio of the HTTP/2 stream flow-control window at which a
+     * <a href="https://datatracker.ietf.org/doc/html/rfc7540#section-6.9">WINDOW_UPDATE</a> frame will be sent.
+     * When the size of the flow-control window drops below the specified ratio (relative to the initial window
+     * size), a {@code WINDOW_UPDATE} frame is triggered to replenish the window.
+     *
+     * <p>The default value is {@value DefaultHttp2LocalFlowController#DEFAULT_WINDOW_UPDATE_RATIO}.
+     * The value must be greater than 0 and less than 1.0.
+     *
+     * <p>Note: Do not change this value unless you know what you are doing.
+     */
+    @UnstableApi
+    public ClientFactoryBuilder http2StreamWindowUpdateRatio(float http2StreamWindowUpdateRatio) {
+        checkArgument(http2StreamWindowUpdateRatio > 0 && http2StreamWindowUpdateRatio < 1.0f,
+                      "http2StreamWindowUpdateRatio: %s (expected: > 0 and < 1.0)",
+                      http2StreamWindowUpdateRatio);
+        option(ClientFactoryOptions.HTTP2_STREAM_WINDOW_UPDATE_RATIO, http2StreamWindowUpdateRatio);
+        return this;
+    }
+
     /**
      * Sets the
      * <a href="https://datatracker.ietf.org/doc/html/rfc7540#section-6.5.2">SETTINGS_MAX_FRAME_SIZE</a>

core/src/main/java/com/linecorp/armeria/client/ClientFactoryOptions.java

@@ -160,6 +160,17 @@ public final class ClientFactoryOptions
             ClientFactoryOption.define("HTTP2_INITIAL_STREAM_WINDOW_SIZE",
                                        Flags.defaultHttp2InitialStreamWindowSize());
 
+    /**
+     * The threshold ratio of the HTTP/2 stream flow-control window at which a
+     * <a href="https://datatracker.ietf.org/doc/html/rfc7540#section-6.9">WINDOW_UPDATE</a> frame will be sent.
+     * When the size of the flow-control window drops below the specified ratio (relative to the initial window
+     * size), a {@code WINDOW_UPDATE} frame is triggered to replenish the window.
+     */
+    @UnstableApi
+    public static final ClientFactoryOption<Float> HTTP2_STREAM_WINDOW_UPDATE_RATIO =
+            ClientFactoryOption.define("HTTP2_STREAM_WINDOW_UPDATE_RATIO",
+                                       Flags.defaultHttp2StreamWindowUpdateRatio());
+
     /**
      * The <a href="https://datatracker.ietf.org/doc/html/rfc7540#section-6.5.2">SETTINGS_MAX_FRAME_SIZE</a>
      * that indicates the size of the largest frame payload that this client is willing to receive.
@@ -504,6 +515,17 @@ public int http2InitialStreamWindowSize() {
         return get(HTTP2_INITIAL_STREAM_WINDOW_SIZE);
     }
 
+    /**
+     * Returns the threshold ratio of the HTTP/2 stream flow-control window at which a
+     * <a href="https://datatracker.ietf.org/doc/html/rfc7540#section-6.9">WINDOW_UPDATE</a> frame will be sent.
+     * When the size of the flow-control window drops below the specified ratio (relative to the initial window
+     * size), a {@code WINDOW_UPDATE} frame is triggered to replenish the window.
+     */
+    @UnstableApi
+    public float http2StreamWindowUpdateRatio() {
+        return get(HTTP2_STREAM_WINDOW_UPDATE_RATIO);
+    }
+
     /**
      * Returns the <a href="https://datatracker.ietf.org/doc/html/rfc7540#section-6.5.2">SETTINGS_MAX_FRAME_SIZE</a>
      * that indicates the size of the largest frame payload that this client is willing to receive.

core/src/main/java/com/linecorp/armeria/client/Http2ClientConnectionHandler.java

@@ -45,9 +45,10 @@ final class Http2ClientConnectionHandler extends AbstractHttp2ConnectionHandler
         super(decoder, encoder, initialSettings,
               newKeepAliveHandler(encoder, channel, clientFactory, protocol));
 
-        responseDecoder = new Http2ResponseDecoder(channel, encoder(), clientFactory, keepAliveHandler());
+        responseDecoder = new Http2ResponseDecoder(channel, encoder(), decoder, clientFactory,
+                                                   keepAliveHandler());
         connection().addListener(responseDecoder);
-        decoder().frameListener(responseDecoder);
+        decoder.frameListener(responseDecoder);
     }
 
     private static KeepAliveHandler newKeepAliveHandler(

core/src/main/java/com/linecorp/armeria/client/Http2ResponseDecoder.java

@@ -44,6 +44,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.EventLoop;
 import io.netty.handler.codec.http2.Http2Connection;
+import io.netty.handler.codec.http2.Http2ConnectionDecoder;
 import io.netty.handler.codec.http2.Http2ConnectionEncoder;
 import io.netty.handler.codec.http2.Http2Error;
 import io.netty.handler.codec.http2.Http2Exception;
@@ -63,10 +64,12 @@ final class Http2ResponseDecoder extends AbstractHttpResponseDecoder implements
     private final Http2GoAwayHandler goAwayHandler;
     private final KeepAliveHandler keepAliveHandler;
 
-    Http2ResponseDecoder(Channel channel, Http2ConnectionEncoder encoder, HttpClientFactory clientFactory,
+    Http2ResponseDecoder(Channel channel, Http2ConnectionEncoder encoder, Http2ConnectionDecoder decoder,
+                         HttpClientFactory clientFactory,
                          KeepAliveHandler keepAliveHandler) {
         super(channel,
-              InboundTrafficController.ofHttp2(channel, clientFactory.http2InitialConnectionWindowSize()));
+              InboundTrafficController.ofHttp2(channel, decoder,
+                                               clientFactory.http2InitialConnectionWindowSize()));
         conn = encoder.connection();
         this.encoder = encoder;
         assert keepAliveHandler instanceof Http2ClientKeepAliveHandler ||
@@ -284,8 +287,8 @@ public int onDataRead(
             res.close();
         }
 
-        // All bytes have been processed.
-        return dataLength + padding;
+        // The data length will be reported to InboundTrafficController upon consumption for flow control.
+        return padding;
     }
 
     /**

core/src/main/java/com/linecorp/armeria/client/HttpClientFactory.java

@@ -111,6 +111,7 @@ private static void setupTlsMetrics(List<X509Certificate> certificates, MeterReg
     private final AddressResolverGroup<InetSocketAddress> addressResolverGroup;
     private final int http2InitialConnectionWindowSize;
     private final int http2InitialStreamWindowSize;
+    private final float http2StreamWindowUpdateRatio;
     private final int http2MaxFrameSize;
     private final long http2MaxHeaderListSize;
     private final int http1MaxInitialLineLength;
@@ -208,6 +209,7 @@ private static void setupTlsMetrics(List<X509Certificate> certificates, MeterReg
 
         http2InitialConnectionWindowSize = options.http2InitialConnectionWindowSize();
         http2InitialStreamWindowSize = options.http2InitialStreamWindowSize();
+        http2StreamWindowUpdateRatio = options.http2StreamWindowUpdateRatio();
         http2MaxFrameSize = options.http2MaxFrameSize();
         http2MaxHeaderListSize = options.http2MaxHeaderListSize();
         pingIntervalMillis = options.pingIntervalMillis();
@@ -265,6 +267,10 @@ int http2InitialStreamWindowSize() {
         return http2InitialStreamWindowSize;
     }
 
+    float http2StreamWindowUpdateRatio() {
+        return http2StreamWindowUpdateRatio;
+    }
+
     int http2MaxFrameSize() {
         return http2MaxFrameSize;
     }

core/src/main/java/com/linecorp/armeria/client/HttpClientPipelineConfigurator.java

@@ -95,6 +95,7 @@
 import io.netty.handler.codec.http2.DefaultHttp2ConnectionEncoder;
 import io.netty.handler.codec.http2.DefaultHttp2FrameReader;
 import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
+import io.netty.handler.codec.http2.DefaultHttp2LocalFlowController;
 import io.netty.handler.codec.http2.Http2ClientUpgradeCodec;
 import io.netty.handler.codec.http2.Http2CodecUtil;
 import io.netty.handler.codec.http2.Http2Connection;
@@ -539,7 +540,9 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
             final Http2ResponseDecoder responseDecoder = this.responseDecoder;
             final DecodedHttpResponse res = new DecodedHttpResponse(ctx.channel().eventLoop());
 
+            final int id = 0;
             res.init(responseDecoder.inboundTrafficController());
+            res.setId(id);
             res.subscribe(new Subscriber<HttpObject>() {
 
                 private boolean notified;
@@ -580,7 +583,7 @@ public void onComplete() {}
                     System.nanoTime(), SystemInfo.currentTimeMicros());
 
             // NB: No need to set the response timeout because we have session creation timeout.
-            responseDecoder.addResponse(null, 0, res, reqCtx, ctx.channel().eventLoop());
+            responseDecoder.addResponse(null, id, res, reqCtx, ctx.channel().eventLoop());
             ctx.fireChannelActive();
         }
 
@@ -781,6 +784,10 @@ private Http2ConnectionDecoder decoder(Http2Connection connection, Http2Connecti
                 /* validateHeaders */ false, clientFactory.http2MaxHeaderListSize());
         Http2FrameReader reader = new DefaultHttp2FrameReader(headersDecoder);
         reader = new Http2InboundFrameLogger(reader, frameLogger);
+        final DefaultHttp2LocalFlowController flowController =
+                new DefaultHttp2LocalFlowController(connection, clientFactory.http2StreamWindowUpdateRatio(),
+                                                    false);
+        connection.local().flowController(flowController);
         return new DefaultHttp2ConnectionDecoder(connection, encoder, reader);
     }
 

core/src/main/java/com/linecorp/armeria/common/DefaultFlagsProvider.java

@@ -36,6 +36,7 @@
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
+import io.netty.handler.codec.http2.DefaultHttp2LocalFlowController;
 
 /**
  * Implementation of {@link FlagsProvider} which provides default values to {@link Flags}.
@@ -81,6 +82,8 @@ final class DefaultFlagsProvider implements FlagsProvider {
     static final long DEFAULT_CLIENT_HTTP2_GRACEFUL_SHUTDOWN_TIMEOUT_MILLIS = 1000;
     static final int DEFAULT_HTTP2_INITIAL_CONNECTION_WINDOW_SIZE = 1024 * 1024; // 1MiB
     static final int DEFAULT_HTTP2_INITIAL_STREAM_WINDOW_SIZE = 1024 * 1024; // 1MiB
+    static final float DEFAULT_HTTP2_STREAM_WINDOW_UPDATE_RATIO =
+            DefaultHttp2LocalFlowController.DEFAULT_WINDOW_UPDATE_RATIO; // 0.5f
     static final int DEFAULT_HTTP2_MAX_FRAME_SIZE = 16384; // From HTTP/2 specification
 
     // Can't use 0xFFFFFFFFL because some implementations use a signed 32-bit integer to store HTTP/2 SETTINGS
@@ -330,6 +333,11 @@ public Integer defaultHttp2InitialStreamWindowSize() {
         return DEFAULT_HTTP2_INITIAL_STREAM_WINDOW_SIZE;
     }
 
+    @Override
+    public Float defaultHttp2StreamWindowUpdateRatio() {
+        return DEFAULT_HTTP2_STREAM_WINDOW_UPDATE_RATIO;
+    }
+
     @Override
     public Integer defaultHttp2MaxFrameSize() {
         return DEFAULT_HTTP2_MAX_FRAME_SIZE;

core/src/main/java/com/linecorp/armeria/common/Flags.java

@@ -297,6 +297,10 @@ private static boolean validateTransportType(TransportType transportType, String
             getValue(FlagsProvider::defaultHttp2InitialStreamWindowSize,
                      "defaultHttp2InitialStreamWindowSize", value -> value > 0);
 
+    private static final float DEFAULT_HTTP2_STREAM_WINDOW_UPDATE_RATIO =
+            getValue(FlagsProvider::defaultHttp2StreamWindowUpdateRatio,
+                     "defaultHttp2InitialStreamWindowSize", value -> value > 0 && value <= 1.0f);
+
     private static final int DEFAULT_HTTP2_MAX_FRAME_SIZE =
             getValue(FlagsProvider::defaultHttp2MaxFrameSize, "defaultHttp2MaxFrameSize",
                      value -> value >= Http2CodecUtil.MAX_FRAME_SIZE_LOWER_BOUND &&
@@ -1112,6 +1116,23 @@ public static int defaultHttp2InitialStreamWindowSize() {
         return DEFAULT_HTTP2_INITIAL_STREAM_WINDOW_SIZE;
     }
 
+    /**
+     * Returns the default value of the {@link ServerBuilder#http2StreamWindowUpdateRatio(float)} and
+     * {@link ClientFactoryBuilder#http2StreamWindowUpdateRatio(float)}.
+     * Note that this flag has no effect if a user specified the value explicitly via
+     * {@link ServerBuilder#http2StreamWindowUpdateRatio(float)} or
+     * {@link ClientFactoryBuilder#http2StreamWindowUpdateRatio(float)}.
+     *
+     * <p>The default value of this flag is
+     * {@value DefaultFlagsProvider#DEFAULT_HTTP2_STREAM_WINDOW_UPDATE_RATIO}.
+     * Specify the {@code -Dcom.linecorp.armeria.defaultHttp2StreamWindowUpdateRatio=<float>} JVM option
+     * to override the default value.
+     */
+    @UnstableApi
+    public static Float defaultHttp2StreamWindowUpdateRatio() {
+        return DEFAULT_HTTP2_STREAM_WINDOW_UPDATE_RATIO;
+    }
+
     /**
      * Returns the default value of the {@link ServerBuilder#http2MaxFrameSize(int)} and
      * {@link ClientFactoryBuilder#http2MaxFrameSize(int)} option.

core/src/main/java/com/linecorp/armeria/common/FlagsProvider.java

@@ -702,6 +702,23 @@ default Integer defaultHttp2InitialStreamWindowSize() {
         return null;
     }
 
+    /**
+     * Returns the default value of the {@link ServerBuilder#http2StreamWindowUpdateRatio(float)} and
+     * {@link ClientFactoryBuilder#http2StreamWindowUpdateRatio(float)}.
+     * Note that this flag has no effect if a user specified the value explicitly via
+     * {@link ServerBuilder#http2StreamWindowUpdateRatio(float)} or
+     * {@link ClientFactoryBuilder#http2StreamWindowUpdateRatio(float)}.
+     *
+     * <p>The default value of this flag is
+     * {@value DefaultFlagsProvider#DEFAULT_HTTP2_STREAM_WINDOW_UPDATE_RATIO}.
+     * Specify the {@code -Dcom.linecorp.armeria.defaultHttp2StreamWindowUpdateRatio=<float>} JVM option
+     * to override the default value.
+     */
+    @Nullable
+    default Float defaultHttp2StreamWindowUpdateRatio() {
+        return null;
+    }
+
     /**
      * Returns the default value of the {@link ServerBuilder#http2MaxFrameSize(int)} and
      * {@link ClientFactoryBuilder#http2MaxFrameSize(int)} option.