Repository: netty/netty
Issue: HTTP2 padding merging breaks flow control
Labels: defect

@Scottmitch 
@nmittler 

In DefaultHttp2Controller.enqueueFrame we increment the pendingBytes based on the size of the frame passed regardless of whether it was merged. Since merging uses padding = max(padding, new.padding) the number of bytes added to the merged frame may be less than newFrame.size().

This eventually leads to the flow controller allocating bytes to write when there are no frames which causes an assertion.

As a general rule it would make sense for the flow-controller unit tests to always use padding.

Working on fix for the immediate issue. Found this while working on stress test to find memory leak

```
Caused by: io.netty.handler.codec.http2.Http2Exception: byte distribution write error
    at io.netty.handler.codec.http2.Http2Exception.connectionError(Http2Exception.java:96) ~[classes/:na]
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor$WriteVisitor.visit(PriorityStreamByteDistributor.java:416) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams.forEachActiveStream(DefaultHttp2Connection.java:1134) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2Connection.forEachActiveStream(DefaultHttp2Connection.java:135) ~[classes/:na]
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor$WriteVisitor.writeAllocatedBytes(PriorityStreamByteDistributor.java:396) ~[classes/:na]
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor.distribute(PriorityStreamByteDistributor.java:93) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$WritabilityMonitor.writePendingBytes(DefaultHttp2RemoteFlowController.java:743) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$DefaultWritabilityMonitor.writePendingBytes(DefaultHttp2RemoteFlowController.java:792) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController.writePendingBytes(DefaultHttp2RemoteFlowController.java:275) ~[classes/:na]
    at io.netty.handler.codec.http2.Http2ConnectionHandler.flush(Http2ConnectionHandler.java:349) ~[classes/:na]
    at io.netty.handler.codec.http2.FindTheLeakTest$1.run(FindTheLeakTest.java:103) ~[test-classes/:na]
    at io.netty.handler.codec.http2.Http2TestUtil$1.run(Http2TestUtil.java:46) ~[test-classes/:na]
    ... 4 common frames omitted
Caused by: java.lang.AssertionError: null
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor$PriorityState.updateStreamableBytes(PriorityStreamByteDistributor.java:349) ~[classes/:na]
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor.updateStreamableBytes(PriorityStreamByteDistributor.java:81) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$DefaultState.incrementStreamWindow(DefaultHttp2RemoteFlowController.java:399) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$DefaultState.decrementFlowControlWindow(DefaultHttp2RemoteFlowController.java:505) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$DefaultState.writeAllocatedBytes(DefaultHttp2RemoteFlowController.java:380) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$DefaultWritabilityMonitor$1.write(DefaultHttp2RemoteFlowController.java:786) ~[classes/:na]
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor$WriteVisitor.visit(PriorityStreamByteDistributor.java:412) ~[classes/:na]
    ... 14 common frames omitted
18:51:59.807 [defaultEventLoopGroup-3-1] WARN  i.n.channel.DefaultChannelPipeline - An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
java.lang.AssertionError: null
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor$PriorityState.updateStreamableBytes(PriorityStreamByteDistributor.java:349) ~[classes/:na]
    at io.netty.handler.codec.http2.PriorityStreamByteDistributor.updateStreamableBytes(PriorityStreamByteDistributor.java:81) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$DefaultState.cancel(DefaultHttp2RemoteFlowController.java:476) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$DefaultState.cancel(DefaultHttp2RemoteFlowController.java:454) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$1.onStreamClosed(DefaultHttp2RemoteFlowController.java:111) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2Connection.notifyClosed(DefaultHttp2Connection.java:263) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams.removeFromActiveStreams(DefaultHttp2Connection.java:1177) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams$2.process(DefaultHttp2Connection.java:1124) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams.forEachActiveStream(DefaultHttp2Connection.java:1148) ~[classes/:na]
    at io.netty.handler.codec.http2.DefaultHttp2Connection.forEachActiveStream(DefaultHttp2Connection.java:135) ~[classes/:na]
    at io.netty.handler.codec.http2.Http2ConnectionHandler$BaseDecoder.channelInactive(Http2ConnectionHandler.java:371) ~[classes/:na]
    at io.netty.handler.codec.http2.Http2ConnectionHandler.channelInactive(Http2ConnectionHandler.java:569) ~[classes/:na]
    at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelInactiveNow(ChannelHandlerInvokerUtil.java:56) ~[classes/:na]
    at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelInactive(DefaultChannelHandlerInvoker.java:102) [classes/:na]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:132) [classes/:na]
    at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:910) [classes/:na]
    at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:678) [classes/:na]
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:310) [classes/:na]
    at io.netty.util.concurrent.SingleThreadEventExecutor.confirmShutdown(SingleThreadEventExecutor.java:609) [classes/:na]
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:765) [classes/:na]
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137) [classes/:na]
    at java.lang.Thread.run(Thread.java:745)
```


Issue comments:
@louiscryan - makes sense.

@louiscryan - Any updates on this? It would be nice to get this in before the next release. I can take over if you don't have cycles.

@louiscryan - I'll take this over for now. We can improve tests later.

Sorry Scott. Let me dig up my PR tomorrow and at least send that over so
you can take it over. It's pretty small

On Wed, Jan 20, 2016 at 5:12 PM, Scott Mitchell notifications@github.com
wrote:

> @louiscryan https://github.com/louiscryan - I'll take this over for
> now. We can improve tests later.
> 
> —
> Reply to this email directly or view it on GitHub
> https://github.com/netty/netty/issues/4573#issuecomment-173418841.

@louiscryan - No problem. I have a PR pending, but if you push yours so I can make sure I covered everything. I'll also ping you for a review of my PR.


Pull request review comments:
Why wouldn't we just increment by frame.size()?  Is the padding not merged?

The padding is currently not necessarily additive when a merge occurs. It is currently taken as `max`.

See https://github.com/netty/netty/issues/4573 for details too.

@nmittler - PTAL. This is to address https://github.com/netty/netty/issues/4773.

I could add a unit test, but we don't have an environment setup to mock buffers that go through this code so it would require a huge allocation amount or take a bit more time to mock...

@Scottmitch perfect :)

sgtm


Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java
index f3435061360..8c5718a3713 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java
@@ -357,10 +357,11 @@ public void write(ChannelHandlerContext ctx, int allowedBytes) {
 
         @Override
         public boolean merge(ChannelHandlerContext ctx, Http2RemoteFlowController.FlowControlled next) {
-            if (FlowControlledData.class != next.getClass()) {
+            FlowControlledData nextData;
+            if (FlowControlledData.class != next.getClass() ||
+                Integer.MAX_VALUE - (nextData = (FlowControlledData) next).size() < size()) {
                 return false;
             }
-            FlowControlledData nextData = (FlowControlledData) next;
             nextData.queue.copyTo(queue);
             // Given that we're merging data into a frame it doesn't really make sense to accumulate padding.
             padding = Math.max(padding, nextData.padding);
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
index c99f86d6278..0acedebdcfa 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
@@ -404,9 +404,21 @@ public int pendingBytes() {
         @Override
         void enqueueFrame(FlowControlled frame) {
             FlowControlled last = pendingWriteQueue.peekLast();
-            if (last == null || !last.merge(ctx, frame)) {
-                pendingWriteQueue.offer(frame);
+            if (last == null) {
+                enqueueFrameWithoutMerge(frame);
+                return;
+            }
+
+            int lastSize = last.size();
+            if (last.merge(ctx, frame)) {
+                incrementPendingBytes(last.size() - lastSize, true);
+                return;
             }
+            enqueueFrameWithoutMerge(frame);
+        }
+
+        private void enqueueFrameWithoutMerge(FlowControlled frame) {
+            pendingWriteQueue.offer(frame);
             // This must be called after adding to the queue in order so that hasFrame() is
             // updated before updating the stream state.
             incrementPendingBytes(frame.size(), true);


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowControllerTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowControllerTest.java
index 58742311dd2..d7f26f58765 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowControllerTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowControllerTest.java
@@ -212,6 +212,23 @@ public void payloadsShouldMerge() throws Http2Exception {
         assertFalse(controller.isWritable(stream(STREAM_A)));
     }
 
+    @Test
+    public void flowControllerCorrectlyAccountsForBytesWithMerge() throws Http2Exception {
+        controller.initialWindowSize(112); // This must be more than the total merged frame size 110
+        FakeFlowControlled data1 = new FakeFlowControlled(5, 2, true);
+        FakeFlowControlled data2 = new FakeFlowControlled(5, 100, true);
+        sendData(STREAM_A, data1);
+        sendData(STREAM_A, data2);
+        data1.assertNotWritten();
+        data1.assertNotWritten();
+        data2.assertMerged();
+        controller.writePendingBytes();
+        data1.assertFullyWritten();
+        data2.assertNotWritten();
+        verify(listener, never()).writabilityChanged(stream(STREAM_A));
+        assertTrue(controller.isWritable(stream(STREAM_A)));
+    }
+
     @Test
     public void stalledStreamShouldQueuePayloads() throws Http2Exception {
         controller.initialWindowSize(0);
@@ -953,9 +970,10 @@ private void setChannelWritability(boolean isWritable) throws Http2Exception {
     }
 
     private static final class FakeFlowControlled implements Http2RemoteFlowController.FlowControlled {
-
-        private int currentSize;
-        private int originalSize;
+        private int currentPadding;
+        private int currentPayloadSize;
+        private int originalPayloadSize;
+        private int originalPadding;
         private boolean writeCalled;
         private final boolean mergeable;
         private boolean merged;
@@ -963,20 +981,26 @@ private static final class FakeFlowControlled implements Http2RemoteFlowControll
         private Throwable t;
 
         private FakeFlowControlled(int size) {
-            this.currentSize = size;
-            this.originalSize = size;
-            this.mergeable = false;
+            this(size, false);
         }
 
         private FakeFlowControlled(int size, boolean mergeable) {
-            this.currentSize = size;
-            this.originalSize = size;
+            this(size, 0, mergeable);
+        }
+
+        private FakeFlowControlled(int payloadSize, int padding, boolean mergeable) {
+            currentPayloadSize = originalPayloadSize = payloadSize;
+            currentPadding = originalPadding = padding;
             this.mergeable = mergeable;
         }
 
         @Override
         public int size() {
-            return currentSize;
+            return currentPayloadSize + currentPadding;
+        }
+
+        private int originalSize() {
+            return originalPayloadSize + originalPadding;
         }
 
         @Override
@@ -990,28 +1014,36 @@ public void writeComplete() {
 
         @Override
         public void write(ChannelHandlerContext ctx, int allowedBytes) {
-            if (allowedBytes <= 0 && currentSize != 0) {
+            if (allowedBytes <= 0 && size() != 0) {
                 // Write has been called but no data can be written
                 return;
             }
             writeCalled = true;
-            int written = Math.min(currentSize, allowedBytes);
-            currentSize -= written;
+            int written = Math.min(size(), allowedBytes);
+            if (written > currentPayloadSize) {
+                written -= currentPayloadSize;
+                currentPayloadSize = 0;
+                currentPadding -= written;
+            } else {
+                currentPayloadSize -= written;
+            }
         }
 
         @Override
         public boolean merge(ChannelHandlerContext ctx, Http2RemoteFlowController.FlowControlled next) {
             if (mergeable && next instanceof FakeFlowControlled) {
-                this.originalSize += ((FakeFlowControlled) next).originalSize;
-                this.currentSize += ((FakeFlowControlled) next).originalSize;
-                ((FakeFlowControlled) next).merged = true;
+                FakeFlowControlled ffcNext = (FakeFlowControlled) next;
+                originalPayloadSize += ffcNext.originalPayloadSize;
+                currentPayloadSize += ffcNext.originalPayloadSize;
+                currentPadding = originalPadding = Math.max(originalPadding, ffcNext.originalPadding);
+                ffcNext.merged = true;
                 return true;
             }
             return false;
         }
 
         public int written() {
-            return originalSize - currentSize;
+            return originalSize() - size();
         }
 
         public void assertNotWritten() {
@@ -1029,7 +1061,8 @@ public void assertPartiallyWritten(int expectedWritten, int delta) {
 
         public void assertFullyWritten() {
             assertTrue(writeCalled);
-            assertEquals(0, currentSize);
+            assertEquals(0, currentPayloadSize);
+            assertEquals(0, currentPadding);
         }
 
         public boolean assertMerged() {