HTTP2: Allow to manually manage window update frames when using Http2… (#14980)

…… (#14974)

…StreamChannel

Motivation:

In some advanced use-cases it might be useful for the end-user to
manually manage the window update frames and so have more control over
the backpressure.

Modifications:

- Add a new Http2StreamChannelOption that allows to configure that
window update frames will be send manually and adjust channel
implementation to allow it
- Add test case

Result:

Fixes #14945

Co-authored-by: Jonas Konrad <[email protected]>

Author: normanmaurer

codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java

@@ -24,6 +24,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelId;
 import io.netty.channel.ChannelMetadata;
+import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelOutboundBuffer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelProgressivePromise;
@@ -41,6 +42,7 @@
 import io.netty.handler.ssl.SslCloseCompletionEvent;
 import io.netty.util.DefaultAttributeMap;
 import io.netty.util.ReferenceCountUtil;
+import io.netty.util.internal.ObjectUtil;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
@@ -49,6 +51,7 @@
 import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayDeque;
+import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -906,24 +909,19 @@ void readEOS() {
             readEOS = true;
         }
 
-        private void updateLocalWindowIfNeeded() {
-            if (flowControlledBytes != 0 && !parentContext().isRemoved()) {
+        private boolean updateLocalWindowIfNeeded() {
+            if (flowControlledBytes != 0 && !parentContext().isRemoved() && config.autoStreamFlowControl) {
                 int bytes = flowControlledBytes;
                 flowControlledBytes = 0;
-                ChannelFuture future = write0(parentContext(), new DefaultHttp2WindowUpdateFrame(bytes).stream(stream));
-                // window update frames are commonly swallowed by the Http2FrameCodec and the promise is synchronously
-                // completed but the flow controller _may_ have generated a wire level WINDOW_UPDATE. Therefore we need,
-                // to assume there was a write done that needs to be flushed or we risk flow control starvation.
-                writeDoneAndNoFlush = true;
-                // Add a listener which will notify and teardown the stream
-                // when a window update fails if needed or check the result of the future directly if it was completed
-                // already.
-                // See https://github.com/netty/netty/issues/9663
-                if (future.isDone()) {
-                    windowUpdateFrameWriteComplete(future, AbstractHttp2StreamChannel.this);
-                } else {
-                    future.addListener(windowUpdateFrameWriteListener);
-                }
+                writeWindowUpdateFrame(new DefaultHttp2WindowUpdateFrame(bytes).stream(stream));
+                return true;
+            }
+            return false;
+        }
+
+        void updateLocalWindowIfNeededAndFlush() {
+            if (updateLocalWindowIfNeeded()) {
+                flush();
             }
         }
 
@@ -982,6 +980,24 @@ void doRead0(Http2Frame frame, RecvByteBufAllocator.Handle allocHandle) {
             pipeline().fireChannelRead(frame);
         }
 
+        private ChannelFuture writeWindowUpdateFrame(Http2WindowUpdateFrame windowUpdateFrame) {
+            ChannelFuture future = write0(parentContext(), windowUpdateFrame);
+            // window update frames are commonly swallowed by the Http2FrameCodec and the promise is synchronously
+            // completed but the flow controller _may_ have generated a wire level WINDOW_UPDATE. Therefore we need,
+            // to assume there was a write done that needs to be flushed or we risk flow control starvation.
+            writeDoneAndNoFlush = true;
+            // Add a listener which will notify and teardown the stream
+            // when a window update fails if needed or check the result of the future directly if it was completed
+            // already.
+            // See https://github.com/netty/netty/issues/9663
+            if (future.isDone()) {
+                windowUpdateFrameWriteComplete(future, AbstractHttp2StreamChannel.this);
+            } else {
+                future.addListener(windowUpdateFrameWriteListener);
+            }
+            return future;
+        }
+
         @Override
         public void write(Object msg, final ChannelPromise promise) {
             // After this point its not possible to cancel a write anymore.
@@ -1001,7 +1017,42 @@ public void write(Object msg, final ChannelPromise promise) {
             try {
                 if (msg instanceof Http2StreamFrame) {
                     Http2StreamFrame frame = validateStreamFrame((Http2StreamFrame) msg).stream(stream());
-                    writeHttp2StreamFrame(frame, promise);
+                    if (msg instanceof Http2WindowUpdateFrame) {
+                        Http2WindowUpdateFrame updateFrame = (Http2WindowUpdateFrame) msg;
+                        if (config.autoStreamFlowControl) {
+                            ReferenceCountUtil.release(msg);
+                            promise.setFailure(new UnsupportedOperationException(
+                                    Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL + " is set to false"));
+                            return;
+                        }
+                        try {
+                            ObjectUtil.checkInRange(updateFrame.windowSizeIncrement(), 0,
+                                    flowControlledBytes, "windowSizeIncrement");
+                        } catch (RuntimeException e) {
+                            ReferenceCountUtil.release(updateFrame);
+                            promise.setFailure(e);
+                            return;
+                        }
+                        flowControlledBytes -= updateFrame.windowSizeIncrement();
+                        if (parentContext().isRemoved()) {
+                            ReferenceCountUtil.release(msg);
+                            promise.setFailure(new ClosedChannelException());
+                            return;
+                        }
+                        ChannelFuture f = writeWindowUpdateFrame(updateFrame);
+                        if (f.isDone()) {
+                            writeComplete(f, promise);
+                        } else {
+                            f.addListener(new ChannelFutureListener() {
+                                @Override
+                                public void operationComplete(ChannelFuture future) {
+                                    writeComplete(future, promise);
+                                }
+                            });
+                        }
+                    } else {
+                        writeHttp2StreamFrame(frame, promise);
+                    }
                 } else {
                     String msgStr = msg.toString();
                     ReferenceCountUtil.release(msg);
@@ -1152,6 +1203,8 @@ public ChannelOutboundBuffer outboundBuffer() {
      * changes.
      */
     private static final class Http2StreamChannelConfig extends DefaultChannelConfig {
+
+        volatile boolean autoStreamFlowControl = true;
         Http2StreamChannelConfig(Channel channel) {
             super(channel);
         }
@@ -1175,6 +1228,49 @@ public ChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
             super.setRecvByteBufAllocator(allocator);
             return this;
         }
+
+        @Override
+        public Map<ChannelOption<?>, Object> getOptions() {
+            return getOptions(
+                    super.getOptions(),
+                    Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL);
+        }
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public <T> T getOption(ChannelOption<T> option) {
+            if (option == Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL) {
+                return (T) Boolean.valueOf(autoStreamFlowControl);
+            }
+            return super.getOption(option);
+        }
+
+        @Override
+        public <T> boolean setOption(ChannelOption<T> option, T value) {
+            validate(option, value);
+            if (option == Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL) {
+                boolean newValue = (Boolean) value;
+                boolean changed = newValue && !autoStreamFlowControl;
+                autoStreamFlowControl = (Boolean) value;
+                if (changed) {
+                    if (channel.isRegistered()) {
+                        final Http2ChannelUnsafe unsafe = (Http2ChannelUnsafe) channel.unsafe();
+                        if (channel.eventLoop().inEventLoop()) {
+                            unsafe.updateLocalWindowIfNeededAndFlush();
+                        } else {
+                            channel.eventLoop().execute(new Runnable() {
+                                @Override
+                                public void run() {
+                                    unsafe.updateLocalWindowIfNeededAndFlush();
+                                }
+                            });
+                        }
+                    }
+                }
+                return true;
+            }
+            return super.setOption(option, value);
+        }
     }
 
     private void maybeAddChannelToReadCompletePendingQueue() {

codec-http2/src/main/java/io/netty/handler/codec/http2/Http2StreamChannelOption.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2025 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.codec.http2;
+
+import io.netty.channel.ChannelOption;
+
+/**
+ * {@link ChannelOption}s that are specific to {@link Http2StreamChannel}s.
+ *
+ * @param <T>   the type of the value which is valid for the {@link ChannelOption}
+ */
+public final class Http2StreamChannelOption<T> extends ChannelOption<T> {
+    private Http2StreamChannelOption(String name) {
+        super(name);
+    }
+
+    /**
+     * When set to {@code true} {@link Http2WindowUpdateFrame}s will be automatically be generated and written for
+     * {@link Http2StreamChannel}s as soon as frames are passed to the user via
+     * {@link io.netty.channel.ChannelPipeline#fireChannelRead(Object)}. If the user wants more control on when a
+     * window update is send its possible to set it to {@code false}. In this case the user is responsible to
+     * generate the correct {@link Http2WindowUpdateFrame}s and eventually write these to the channel.
+     * <p>
+     * See <a href="https://datatracker.ietf.org/doc/html/rfc9113#section-5.2">RFC9113 5.2. Flow Control</a> for more
+     * details.
+     */
+    public static final ChannelOption<Boolean> AUTO_STREAM_FLOW_CONTROL =
+            valueOf("AUTO_STREAM_FLOW_CONTROL");
+}

codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexTest.java

@@ -45,7 +45,6 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.MethodSource;
-import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.ArgumentMatcher;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
@@ -1563,11 +1562,23 @@ public void flush(ChannelHandlerContext ctx) throws Exception {
 
     @Test
     public void windowUpdatesAreFlushed() {
+        windowUpdatesAreFlushed(true);
+    }
+
+    @Test
+    public void windowUpdatesNotDoneAutomatically() {
+        windowUpdatesAreFlushed(false);
+    }
+
+    private void windowUpdatesAreFlushed(boolean autoWriteWindowUpdateFrames) {
         LastInboundHandler inboundHandler = new LastInboundHandler();
         FlushSniffer flushSniffer = new FlushSniffer();
         parentChannel.pipeline().addFirst(flushSniffer);
 
         Http2StreamChannel childChannel = newInboundStream(3, false, inboundHandler);
+        childChannel.config().setOption(
+                Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL, autoWriteWindowUpdateFrames);
+
         assertTrue(childChannel.config().isAutoRead());
         childChannel.config().setAutoRead(false);
         assertFalse(childChannel.config().isAutoRead());
@@ -1591,16 +1602,90 @@ public void windowUpdatesAreFlushed() {
         // Trigger a read of the second frame.
         childChannel.read();
         verifyFramesMultiplexedToCorrectChannel(childChannel, inboundHandler, 1);
-        // We expect a flush here because the StreamChannel will flush the smaller increment but the
-        // connection will collect the bytes and decide not to send a wire level frame until more are consumed.
-        assertTrue(flushSniffer.checkFlush());
+        if (autoWriteWindowUpdateFrames) {
+            // We expect a flush here because the StreamChannel will flush the smaller increment but the
+            // connection will collect the bytes and decide not to send a wire level frame until more are consumed.
+            assertTrue(flushSniffer.checkFlush());
+        } else {
+            assertFalse(flushSniffer.checkFlush());
+        }
+
         verify(frameWriter, never()).writeWindowUpdate(eqCodecCtx(), anyInt(), anyInt(), anyChannelPromise());
 
         // Call read one more time which should trigger the writing of the flow control update.
         childChannel.read();
+        if (autoWriteWindowUpdateFrames) {
+            verify(frameWriter).writeWindowUpdate(eqCodecCtx(), eq(0), eq(32 * 1024), anyChannelPromise());
+            verify(frameWriter).writeWindowUpdate(
+                    eqCodecCtx(), eq(childChannel.stream().id()), eq(32 * 1024), anyChannelPromise());
+            assertTrue(flushSniffer.checkFlush());
+        } else {
+            verify(frameWriter, never()).writeWindowUpdate(eqCodecCtx(), anyInt(), anyInt(), anyChannelPromise());
+            assertFalse(flushSniffer.checkFlush());
+
+            // Let's manually send a window update frame now.
+            ChannelFuture f = childChannel.writeAndFlush(new DefaultHttp2WindowUpdateFrame(32 * 1024)
+                    .stream(childChannel.stream()));
+            assertTrue(f.isSuccess());
+            verify(frameWriter).writeWindowUpdate(eqCodecCtx(), eq(0), eq(32 * 1024), anyChannelPromise());
+            verify(frameWriter).writeWindowUpdate(
+                    eqCodecCtx(), eq(childChannel.stream().id()), eq(32 * 1024), anyChannelPromise());
+            assertTrue(flushSniffer.checkFlush());
+
+            // Let's try to send one more even though there are no more pending bytes
+            f = childChannel.writeAndFlush(new DefaultHttp2WindowUpdateFrame(32 * 1024)
+                    .stream(childChannel.stream()));
+            assertNotNull(f.cause());
+        }
+    }
+
+    @Test
+    public void windowUpdatesSendWhenAutoReadEnabled() {
+        LastInboundHandler inboundHandler = new LastInboundHandler();
+        FlushSniffer flushSniffer = new FlushSniffer();
+        parentChannel.pipeline().addFirst(flushSniffer);
+
+        Http2StreamChannel childChannel = newInboundStream(3, false, inboundHandler);
+        childChannel.config().setOption(
+                Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL, false);
+
+        assertTrue(childChannel.config().isAutoRead());
+        childChannel.config().setAutoRead(false);
+        assertFalse(childChannel.config().isAutoRead());
+
+        Http2HeadersFrame headersFrame = inboundHandler.readInbound();
+        assertNotNull(headersFrame);
+
+        assertTrue(flushSniffer.checkFlush());
+
+        // Write some bytes to get the channel into the idle state with buffered data and also verify we
+        // do not dispatch it until we receive a read() call.
+        frameInboundWriter.writeInboundData(childChannel.stream().id(), bb(16 * 1024), 0, false);
+        frameInboundWriter.writeInboundData(childChannel.stream().id(), bb(16 * 1024), 0, false);
+        assertTrue(flushSniffer.checkFlush());
+
+        verify(frameWriter, never()).writeWindowUpdate(eqCodecCtx(), anyInt(), anyInt(), anyChannelPromise());
+        // only the first one was read because it was legacy auto-read behavior.
+        verifyFramesMultiplexedToCorrectChannel(childChannel, inboundHandler, 1);
+        assertFalse(flushSniffer.checkFlush());
+
+        // Trigger a read of the second frame.
+        childChannel.read();
+        verifyFramesMultiplexedToCorrectChannel(childChannel, inboundHandler, 1);
+        assertFalse(flushSniffer.checkFlush());
+
+        verify(frameWriter, never()).writeWindowUpdate(eqCodecCtx(), anyInt(), anyInt(), anyChannelPromise());
+
+        childChannel.read();
+
+        verify(frameWriter, never()).writeWindowUpdate(eqCodecCtx(), anyInt(), anyInt(), anyChannelPromise());
+        assertFalse(flushSniffer.checkFlush());
+
+        childChannel.config().setOption(
+                Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL, true);
         verify(frameWriter).writeWindowUpdate(eqCodecCtx(), eq(0), eq(32 * 1024), anyChannelPromise());
         verify(frameWriter).writeWindowUpdate(
-            eqCodecCtx(), eq(childChannel.stream().id()), eq(32 * 1024), anyChannelPromise());
+                eqCodecCtx(), eq(childChannel.stream().id()), eq(32 * 1024), anyChannelPromise());
         assertTrue(flushSniffer.checkFlush());
     }