Repository: netty/netty
Issue: Removed HTTP/2 streams should consume all bytes
Labels: defect

When a stream is removed from the connection, we should mark all of it's bytes as consumed so that the local flow controller will restore those bytes to the connection window. We'll also need to protect against the application trying to consume those bytes as well (probably just make it a no-op for closed streams).


Issue comments:
Thanks @nmittler.  I see you left this as a comment in https://github.com/netty/netty/pull/3616.  Are you pulling together a separate PR for this?

I'm working on a PR to address this and history ... should have something ready shortly.

@nmittler - OK. I would like to get a consolidated understanding as to why we need to keep stream history before you invest a lot of time in it.  See https://github.com/netty/netty/issues/3653#issuecomment-94580186.  Should we open another issue for stream history...or do you want to just list the rational here?

@Scottmitch I've put together #3672 to address this.

I have been putting together a PR for stream history, but you're right we should finish our discussion on this: #3673.  Let's sort this out quickly ... grpc needs this.  @Scottmitch can you provide a dump of your thoughts on how we can avoid history on a case-by-case basis in the decoder?


Pull request review comments:
Do we have to do this?  If we are forcing the unconsumed bytes to 0 in the connection listener, is that good enough...or is this method used while the stream is being closed before the listener is notified?

Was this intentional? I looked through the call path down to the FrameWriter and I don't see a null check.  Am I missing something?

`{@link #unconsumedBytes}` -> `{@link #unconsumedBytes(Http2Stream)}`

Hum I'm not sure if one way is better/different from another but should we save this reference in the `consumeBytes` method, because that is what it is used for?

Maybe we should make a static method to do this...instead of having 1 state object reach out and impact another state object?  Then we could either call either the static method, or directly call the connection state's consume bytes.

``` java
public static void consumeBytesConnectionAndStream(FlowState streamState) {
FlowState connectionState = state(connectionId);
connectionState.doStuff;

streamState.doStuff;
}
```

nit: consider linking comment to code more explicitly...
`If {@stream}'s {@link Http2Stream#state()} method returns {@link Http2Stream$State.CLOSED}...`

`closed/terminated` -> `closed`.

Was this done for development purposes or was this causing a problem?

No, not intentional.  Restored this check.

We probably don't need to do this in this PR. This was the result of some of my work for adding stream history, where closed streams would no longer have a `FlowState` property.  Since we still need to apply flow control for closed streams, I didn't want these methods to require that the `FlowState` exist.  WDYT ... revert this for now?

Actually, the `consumeBytes` method isn't guaranteed to have been called before the stream is removed. The logic for putting it here is that if there is any data to be consumed, then it must have gone through this method.

Agreed this is a little ugly.  I got rid of the special connection logic and changed the code to call `consumeBytes` on both states separately.

done.

done.

done.

This was an artifact of the work I was doing regarding stream history. Reverted.

@Scottmitch this is somewhat unrelated, but I noticed that this seemed wrong.  The previous logic would only let us close/remove if the stream had at one point become active.  This change lifts this restriction.

Yah lets revert for now. 

SGTM

@nmittler - Good catch.  I'm wondering if this should even be in this method.  Would it make sense in `DefaultStream.close()` after `state = CLOSED`?

do we need to fully qualify this? `io.netty.handler.codec.http2.Http2Stream.State` -> `Http2Stream.State`?

@Scottmitch possibly.  Although when `onClosed()` is called I would like it to be also removed from the active list.  WDYT?

No ... fixed.

Leave it per our discussion.


Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
index 6ddf71ee78e..d6b333c32ad 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
@@ -1121,19 +1121,19 @@ void addToActiveStreams(DefaultStream stream) {
 
         void removeFromActiveStreams(DefaultStream stream) {
             if (streams.remove(stream)) {
-                try {
-                    // Update the number of active streams initiated by the endpoint.
-                    stream.createdBy().numActiveStreams--;
+                // Update the number of active streams initiated by the endpoint.
+                stream.createdBy().numActiveStreams--;
+            }
+            notifyClosed(stream);
+            removeStream(stream);
+        }
 
-                    for (int i = 0; i < listeners.size(); i++) {
-                        try {
-                            listeners.get(i).onStreamClosed(stream);
-                        } catch (RuntimeException e) {
-                            logger.error("Caught RuntimeException from listener onStreamClosed.", e);
-                        }
-                    }
-                } finally {
-                    removeStream(stream);
+        private void notifyClosed(DefaultStream stream) {
+            for (int i = 0; i < listeners.size(); i++) {
+                try {
+                    listeners.get(i).onStreamClosed(stream);
+                } catch (RuntimeException e) {
+                    logger.error("Caught RuntimeException from listener onStreamClosed.", e);
                 }
             }
         }
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowController.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowController.java
index 2fa90dc7598..aca6a581c73 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowController.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowController.java
@@ -30,6 +30,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http2.Http2Exception.CompositeStreamException;
 import io.netty.handler.codec.http2.Http2Exception.StreamException;
+import io.netty.util.internal.PlatformDependent;
 
 /**
  * Basic implementation of {@link Http2LocalFlowController}.
@@ -43,6 +44,7 @@ public class DefaultHttp2LocalFlowController implements Http2LocalFlowController
 
     private final Http2Connection connection;
     private final Http2FrameWriter frameWriter;
+    private ChannelHandlerContext ctx;
     private volatile float windowUpdateRatio;
     private volatile int initialWindowSize = DEFAULT_WINDOW_SIZE;
 
@@ -73,6 +75,22 @@ public void onStreamActive(Http2Stream stream) {
                 // frames which may have been exchanged while it was in IDLE
                 state(stream).window(initialWindowSize);
             }
+
+            @Override
+            public void onStreamClosed(Http2Stream stream) {
+                try {
+                    // When a stream is closed, consume any remaining bytes so that they
+                    // are restored to the connection window.
+                    FlowState state = state(stream);
+                    int unconsumedBytes = state.unconsumedBytes();
+                    if (ctx != null && unconsumedBytes > 0) {
+                        connectionState().consumeBytes(ctx, unconsumedBytes);
+                        state.consumeBytes(ctx, unconsumedBytes);
+                    }
+                } catch (Http2Exception e) {
+                    PlatformDependent.throwException(e);
+                }
+            }
         });
     }
 
@@ -109,7 +127,19 @@ public void incrementWindowSize(ChannelHandlerContext ctx, Http2Stream stream, i
     @Override
     public void consumeBytes(ChannelHandlerContext ctx, Http2Stream stream, int numBytes)
             throws Http2Exception {
-        state(stream).consumeBytes(ctx, numBytes);
+        if (stream.id() == CONNECTION_STREAM_ID) {
+            throw new UnsupportedOperationException("Returning bytes for the connection window is not supported");
+        }
+        if (numBytes <= 0) {
+            throw new IllegalArgumentException("numBytes must be positive");
+        }
+
+        // Streams automatically consume all remaining bytes when they are closed, so just ignore
+        // if already closed.
+        if (!isClosed(stream)) {
+            connectionState().consumeBytes(ctx, numBytes);
+            state(stream).consumeBytes(ctx, numBytes);
+        }
     }
 
     @Override
@@ -178,15 +208,22 @@ public float windowUpdateRatio(Http2Stream stream) throws Http2Exception {
     @Override
     public void receiveFlowControlledFrame(ChannelHandlerContext ctx, Http2Stream stream, ByteBuf data,
             int padding, boolean endOfStream) throws Http2Exception {
+        this.ctx = checkNotNull(ctx, "ctx");
         int dataLength = data.readableBytes() + padding;
 
         // Apply the connection-level flow control
-        connectionState().receiveFlowControlledFrame(dataLength);
-
-        // Apply the stream-level flow control
-        FlowState state = state(stream);
-        state.endOfStream(endOfStream);
-        state.receiveFlowControlledFrame(dataLength);
+        FlowState connectionState = connectionState();
+        connectionState.receiveFlowControlledFrame(dataLength);
+
+        if (!isClosed(stream)) {
+            // Apply the stream-level flow control
+            FlowState state = state(stream);
+            state.endOfStream(endOfStream);
+            state.receiveFlowControlledFrame(dataLength);
+        } else if (dataLength > 0) {
+            // Immediately consume the bytes for the connection window.
+            connectionState.consumeBytes(ctx, dataLength);
+        }
     }
 
     private FlowState connectionState() {
@@ -198,6 +235,10 @@ private static FlowState state(Http2Stream stream) {
         return stream.getProperty(FlowState.class);
     }
 
+    private static boolean isClosed(Http2Stream stream) {
+        return stream.state() == Http2Stream.State.CLOSED;
+    }
+
     /**
      * Flow control window state for an individual stream.
      */
@@ -322,18 +363,6 @@ void returnProcessedBytes(int delta) throws Http2Exception {
         }
 
         void consumeBytes(ChannelHandlerContext ctx, int numBytes) throws Http2Exception {
-            if (stream.id() == CONNECTION_STREAM_ID) {
-                throw new UnsupportedOperationException("Returning bytes for the connection window is not supported");
-            }
-            if (numBytes <= 0) {
-                throw new IllegalArgumentException("numBytes must be positive");
-            }
-
-            // Return bytes to the connection window
-            FlowState connectionState = connectionState();
-            connectionState.returnProcessedBytes(numBytes);
-            connectionState.writeWindowUpdateIfNeeded(ctx);
-
             // Return the bytes processed and update the window.
             returnProcessedBytes(numBytes);
             writeWindowUpdateIfNeeded(ctx);
@@ -347,7 +376,7 @@ int unconsumedBytes() {
          * Updates the flow control window for this stream if it is appropriate.
          */
         void writeWindowUpdateIfNeeded(ChannelHandlerContext ctx) throws Http2Exception {
-            if (endOfStream || initialStreamWindowSize <= 0) {
+            if (endOfStream || initialStreamWindowSize <= 0 || isClosed(stream)) {
                 return;
             }
 
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2LocalFlowController.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2LocalFlowController.java
index b852b249f59..83d193350e3 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2LocalFlowController.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2LocalFlowController.java
@@ -28,6 +28,8 @@ public interface Http2LocalFlowController extends Http2FlowController {
      * policies to it for both the {@code stream} as well as the connection. If any flow control
      * policies have been violated, an exception is raised immediately, otherwise the frame is
      * considered to have "passed" flow control.
+     * <p/>
+     * If {@code stream} is closed, flow control should only be applied to the connection window.
      *
      * @param ctx the context from the handler where the frame was read.
      * @param stream the subject stream for the received frame. The connection stream object must
@@ -39,22 +41,24 @@ public interface Http2LocalFlowController extends Http2FlowController {
      * @throws Http2Exception if any flow control errors are encountered.
      */
     void receiveFlowControlledFrame(ChannelHandlerContext ctx, Http2Stream stream, ByteBuf data, int padding,
-            boolean endOfStream) throws Http2Exception;
+                                    boolean endOfStream) throws Http2Exception;
 
     /**
-     * Indicates that the application has consumed a number of bytes for the given stream and is
-     * therefore ready to receive more data from the remote endpoint. The application must consume
-     * any bytes that it receives or the flow control window will collapse. Consuming bytes enables
-     * the flow controller to send {@code WINDOW_UPDATE} to restore a portion of the flow control
-     * window for the stream.
+     * Indicates that the application has consumed a number of bytes for the given stream and is therefore ready to
+     * receive more data from the remote endpoint. The application must consume any bytes that it receives or the flow
+     * control window will collapse. Consuming bytes enables the flow controller to send {@code WINDOW_UPDATE} to
+     * restore a portion of the flow control window for the stream.
+     * <p/>
+     * If {@code stream} is closed (i.e. {@link Http2Stream#state()} method returns {@link Http2Stream.State#CLOSED}),
+     * the consumed bytes are only restored to the connection window. When a stream is closed, the flow controller
+     * automatically restores any unconsumed bytes for that stream to the connection window. This is done to ensure that
+     * the connection window does not degrade over time as streams are closed.
      *
-     * @param ctx the channel handler context to use when sending a {@code WINDOW_UPDATE} if
-     *            appropriate
-     * @param stream the stream for which window space should be freed. The connection stream object
-     *            must not be used.
+     * @param ctx the channel handler context to use when sending a {@code WINDOW_UPDATE} if appropriate
+     * @param stream the stream for which window space should be freed. The connection stream object must not be used.
      * @param numBytes the number of bytes to be returned to the flow control window.
-     * @throws Http2Exception if the number of bytes returned exceeds the {@link #unconsumedBytes}
-     *             for the stream.
+     * @throws Http2Exception if the number of bytes returned exceeds the {@link #unconsumedBytes(Http2Stream)} for the
+     * stream.
      */
     void consumeBytes(ChannelHandlerContext ctx, Http2Stream stream, int numBytes) throws Http2Exception;
 


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowControllerTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowControllerTest.java
index 1e324426d37..94d8b14d4d1 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowControllerTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2LocalFlowControllerTest.java
@@ -201,6 +201,22 @@ public void connectionWindowShouldAdjustWithMultipleStreams() throws Http2Except
         }
     }
 
+    @Test
+    public void closeShouldConsumeBytes() throws Http2Exception {
+        receiveFlowControlledFrame(STREAM_ID, 10, 0, false);
+        assertEquals(10, controller.unconsumedBytes(connection.connectionStream()));
+        stream(STREAM_ID).close();
+        assertEquals(0, controller.unconsumedBytes(connection.connectionStream()));
+    }
+
+    @Test
+    public void dataReceivedForClosedStreamShouldImmediatelyConsumeBytes() throws Http2Exception {
+        Http2Stream stream = stream(STREAM_ID);
+        stream.close();
+        receiveFlowControlledFrame(stream, 10, 0, false);
+        assertEquals(0, controller.unconsumedBytes(connection.connectionStream()));
+    }
+
     @Test
     public void globalRatioShouldImpactStreams() throws Http2Exception {
         float ratio = 0.6f;
@@ -254,10 +270,15 @@ private static int getWindowDelta(int initialSize, int windowSize, int dataSize)
     }
 
     private void receiveFlowControlledFrame(int streamId, int dataSize, int padding,
-            boolean endOfStream) throws Http2Exception {
+                                            boolean endOfStream) throws Http2Exception {
+        receiveFlowControlledFrame(stream(streamId), dataSize, padding, endOfStream);
+    }
+
+    private void receiveFlowControlledFrame(Http2Stream stream, int dataSize, int padding,
+                                            boolean endOfStream) throws Http2Exception {
         final ByteBuf buf = dummyData(dataSize);
         try {
-            controller.receiveFlowControlledFrame(ctx, stream(streamId), buf, padding, endOfStream);
+            controller.receiveFlowControlledFrame(ctx, stream, buf, padding, endOfStream);
         } finally {
             buf.release();
         }