Repository: netty/netty
Issue: HTTP2 should coalesce small writes into a single DATA frame

@Scottmitch @nmittler 

This is the analog of 

https://github.com/netty/netty/issues/3737

but is probably more useful in general though possibly more complicated to implement. The goal is to allow for more small data writes to fit into a single DATA frame. Given that we already implement a queue of DATA & HEADER frames in the outbound flow-controller we could just merge contiguous DATA frames in writePendingBytes when flushing the channel to achieve the desired effect.

The complication is the relative ordering of non flow-controlled frames such as RST_STREAM relative to those in the flow-controller. In this case we can address by simply draining the flow-controller for the relevant stream and then writing the RST.


Issue comments:
@louiscryan - Let me summarize the face value tradeoffs and follow up with some questions.  Feel free to elaborate.  It seems like goal is to potentially reduce the amount of data frames written (less data frame header overhead on the wire) at the cost of an extra iteration step before writing out data frames (until a different frame type, or a frame for a different stream is encountered).

Questions:
- What does "small" mean?  How many "small" frames should we combine?  Is this based upon network conditions (i.e. is "small" for a cell network the same as for corp network), and/or defined by the application?
- Seems like a goal is the peer will potentially generate less WINDOW_UPDATE frames to replenish bytes to flow control (as stated in #3737)?  I guess the "small" qualification is to avoid the risk of  sacrificing responsiveness for some use-cases where a WINDOW_UPDATE could be sent more immediately when we don't combine?
- Would this require some sort of "look ahead" step in the flow controller to determine "if we combine this next frame, would that exceed my allotted allocation"? 

> In this case we can address by simply draining the flow-controller for the relevant stream and then writing the RST.

Does this mean we need to potentially be able to queue every frame type in the flow control queues? For example what if we can't "drain" the queues because we are blocked due to flow control constraints?

Thats a fair summary of the goal with the associated benefit of larger DATA frame reads on the receiving side. 

I don't know that we necessarily need to 'flush' DATA for a stream if a different frame type or stream output is encountered. Enqueueing data should consume flow-control which would prevent ordering problems in the consumption of flow-control on eventual write. In the case of RST_STREAM we already discard any pending DATA waiting for flow-control so discarding any pending DATA waiting for flush would not be a significant semantic shift. The processing of a WINDOW_UPDATE after a write should similarly be consumed by the already enqueued but not written data.
- I don't think we need to define 'small'. We don't actually commit anything to the wire without flushing the channel anyway.
- Responsiveness is really dictated by how often the application calls 'flush' which I believe is the desired behavior.
- I think this is moot given that we don't need to define small.

See note above about ordering semantics of RST, I'm not sure we actually have an issue. A RST would cause the stream to become inactive and any pending-consumed window could be returned without actually writing the bytes.

I should point out that a perfectly reasonable option is that we don't want to do any of this and applications are responsible for making their writes chunky.

See parallel discussion in GRPC about doing just that
https://github.com/grpc/grpc-java/issues/502

I think it may make sense.  Just a bunch of things to think about :)  For example does this negatively impact responsiveness (see below), is it possible users would want the different data frames (different application layer semantics), etc...  I can't think of anything but seems like you have thought about this a bit more than I have. 

Does this negatively impact a lossy channel (i.e. cellular)?  For lossy channels it may be more likely to get the "smaller" chunks up to the application in a timely fashion.  However I guess having less overall data (less frames, less window updates) may be better in a lossy environment?  Maybe these endpoints are already limiting their flow control window sizes as well?

@louiscryan looking at the [current API](https://github.com/netty/netty/blob/master/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2RemoteFlowController.java#L44), this would be a fairly significant redesign, since `FlowControllable`s would need to support merging. This gets a bit hairier when considering that `HEADERS` are `FlowControllable` too, and we wouldn't support merging them with other headers or with data frames. It certainly possible to do this, but I wonder what the resulting API would look like.  Any thoughts?

Something like

boolean merge(FlowControllable next)

if return is true then they were merged successfully, if false they werent and 'next' should be added to the queue

@louiscryan right, that's pretty much what my initial thought was as well. Were you thinking of cooking up a PR to see what this would look like?

I started hacking on something, primarily so I can compare the performance of implicit merging vs explicit merging at the application layer. If implicit is close then it's probably a good idea, if it's not then perhaps explicit application merging is the recommendation and Netty remains unchanged other than adding doc.

SGTM

Just an update on this. I have most of this working but at the moment it performs worse than before in GRPC. This is largely due to the fact that for each DATA write we attach a listener to the future and use that to return flow-control to the application. This causes flow-control to be returned to a stream for each small message rather than as a block for the coalesced write.

To fix this we need to be notified of how much data the flow-controller wrote when it was flushed. Ill look into that tomorrow.

OK. Fixed performance issues by allowing for more granular notifications out of the flow-controller.

In GRPC I see about a 2x throughput improvement for streams of small messages.

@Scottmitch @ejona86 @nmittler 

An update on whether to support window update coalescing.... 
In short I'm inclined not to make the change as it only seems to help when the flow-control window is improperly sized to the network.

I've started using netem with our benchmarks so we can better simulate real network behaviors over loopback. For 1gb links we can saturate them pretty easily, for 10gb links we currently get to about 60% which isn't too shabby but we'd like to do better.

1gb link, 200us latency
sudo tc qdisc change dev lo root netem delay 0.2ms rate 1gbit

With WU coalescing 4.1B6

```
Benchmark                                                     (clientInboundFlowWindow)  (maxConcurrentStreams)  (responseSize)   Mode  Cnt    Score    Error  Units
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                     MEDIUM                       1           LARGE  thrpt  100  404.407 ±  3.885  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      LARGE                       1           LARGE  thrpt  100  889.155 ±  2.494  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      JUMBO                       1           LARGE  thrpt  100  949.817 ±  0.400  ops/s
```

Head

```
Benchmark                                                     (clientInboundFlowWindow)  (maxConcurrentStreams)  (responseSize)   Mode  Cnt    Score    Error  Units
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                     MEDIUM                       1           LARGE  thrpt  100  398.030 ±  3.349  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      LARGE                       1           LARGE  thrpt  100  869.634 ±  2.837  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      JUMBO                       1           LARGE  thrpt  100  950.606 ±  0.270  ops/s
```

10gb link
sudo tc qdisc change dev lo root netem delay 0.2ms rate 10gbit

With WU coalescing 4.1B6

```
Benchmark                                                     (clientInboundFlowWindow)  (maxConcurrentStreams)  (responseSize)   Mode  Cnt     Score     Error  Units
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                     MEDIUM                       1           LARGE  thrpt  100   653.040 ±   4.803  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      LARGE                       1           LARGE  thrpt  100  5337.066 ±  49.104  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      JUMBO                       1           LARGE  thrpt  100  4277.931 ± 711.438  ops/s
```

Master

```
Benchmark                                                     (clientInboundFlowWindow)  (maxConcurrentStreams)  (responseSize)   Mode  Cnt     Score      Error  Units
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                     MEDIUM                       1           LARGE  thrpt   30   489.693 ±    6.204  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      LARGE                       1           LARGE  thrpt   30  5916.676 ±  173.040  ops/s
StreamingResponseBandwidthBenchmark.stream:megabitsPerSecond                      JUMBO                       1           LARGE  thrpt   30  5048.846 ± 1229.370  ops/s
```

@normanmaurer - Random aside, have you seen much use of netem with Netty before? I wonder if it would be useful to specify a standard 'network' configuration that folks can use to run benchmarks with to improve comparison across environments. E..g setup a tuntap network with a well known name, apply netem to it and then detect its presence in the benchmark using java.net.NetworkInterface

@louiscryan - Thanks for the update. It seemed the PR which coalesced data was assuming this change would rectify some of the degradation it introduced for larger payload sizes. Do we need to take any follow-up action to correct this?

No, the degradation for larger payload was caused by using a payload
ByteBuf that was too large to be allocated out of the pool. I don't think
there is any followup specific to HTTP2 here. The appropraite fix has been
made in the GRPC benchmarks https://github.com/grpc/grpc-java/pull/563

On Thu, Jun 25, 2015 at 12:08 PM, Scott Mitchell notifications@github.com
wrote:

> @louiscryan https://github.com/louiscryan - Thanks for the update. It
> seemed the PR which coalesced data was assuming this change would rectify
> some of the degradation it introduced for larger payload sizes. Do we need
> to take any follow-up action to correct this?
> 
> —
> Reply to this email directly or view it on GitHub
> https://github.com/netty/netty/issues/3852#issuecomment-115365415.

@louiscryan - I'm not sure how reducing the payload size by itself would mitigate an issue, just seems like we are taking a different data point, maybe you should also add another size for what was `JUMBO`?  I recall you saying there was some extra copy operations going on in the benchmarks?  I could see may if the benchmarks were somehow copying the now bigger payloads that may cause some overhead in this situation.

@louiscryan - I think this is already done. Re-open if there is more work needed here.


Pull request review comments:

Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameWriter.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameWriter.java
index 53066118b2a..4c410e52e36 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameWriter.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameWriter.java
@@ -15,6 +15,13 @@
 
 package io.netty.handler.codec.http2;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http2.Http2CodecUtil.SimpleChannelPromiseAggregator;
+import io.netty.handler.codec.http2.Http2FrameWriter.Configuration;
+
 import static io.netty.buffer.Unpooled.directBuffer;
 import static io.netty.buffer.Unpooled.unmodifiableBuffer;
 import static io.netty.buffer.Unpooled.unreleasableBuffer;
@@ -53,16 +60,8 @@
 import static io.netty.handler.codec.http2.Http2FrameTypes.SETTINGS;
 import static io.netty.handler.codec.http2.Http2FrameTypes.WINDOW_UPDATE;
 import static io.netty.util.internal.ObjectUtil.checkNotNull;
-import static java.lang.Math.max;
 import static java.lang.Math.min;
 
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelPromise;
-import io.netty.handler.codec.http2.Http2CodecUtil.SimpleChannelPromiseAggregator;
-import io.netty.handler.codec.http2.Http2FrameWriter.Configuration;
-
 /**
  * A {@link Http2FrameWriter} that supports all frame types defined by the HTTP/2 specification.
  */
@@ -122,55 +121,71 @@ public void close() { }
 
     @Override
     public ChannelFuture writeData(ChannelHandlerContext ctx, int streamId, ByteBuf data,
-            int padding, boolean endStream, ChannelPromise promise) {
+            final int padding, boolean endStream, ChannelPromise promise) {
         final SimpleChannelPromiseAggregator promiseAggregator =
                 new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
-        final DataFrameHeader header = new DataFrameHeader(ctx, streamId);
-        boolean needToReleaseHeaders = true;
-        boolean needToReleaseData = true;
+        int i = 0;
+        ByteBuf headerBuffer = null;
+        final Http2Flags flags = new Http2Flags().paddingPresent(padding > 0);
+        final int dataPerFrame = maxFrameSize - (padding + flags.getPaddingPresenceFieldLength());
+        final int numFrames = data.readableBytes() / dataPerFrame + (data.readableBytes() % dataPerFrame == 0 ? 0 : 1);
+
         try {
             verifyStreamId(streamId, STREAM_ID);
             verifyPadding(padding);
 
-            boolean lastFrame;
-            int remainingData = data.readableBytes();
-            do {
-                // Determine how much data and padding to write in this frame. Put all padding at the end.
-                int frameDataBytes = min(remainingData, maxFrameSize);
-                int framePaddingBytes = min(padding, max(0, (maxFrameSize - 1) - frameDataBytes));
-
-                // Decrement the remaining counters.
-                padding -= framePaddingBytes;
-                remainingData -= frameDataBytes;
-
-                // Determine whether or not this is the last frame to be sent.
-                lastFrame = remainingData == 0 && padding == 0;
-
-                // Only the last frame is not retained. Until then, the outer finally must release.
-                ByteBuf frameHeader = header.slice(frameDataBytes, framePaddingBytes, lastFrame && endStream);
-                needToReleaseHeaders = !lastFrame;
-                ctx.write(lastFrame ? frameHeader : frameHeader.retain(), promiseAggregator.newPromise());
-
-                // Write the frame data.
-                ByteBuf frameData = data.readSlice(frameDataBytes);
-                // Only the last frame is not retained. Until then, the outer finally must release.
-                needToReleaseData = !lastFrame;
-                ctx.write(lastFrame ? frameData : frameData.retain(), promiseAggregator.newPromise());
-
-                // Write the frame padding.
-                if (framePaddingBytes > 0) {
-                    ctx.write(ZERO_BUFFER.slice(0, framePaddingBytes), promiseAggregator.newPromise());
+            // Initialize the header portion of the frame which will be shared by frames (0, n).
+            // * 2 because we may have at most 2 different headers frames.
+            headerBuffer = ctx.alloc().buffer(DATA_FRAME_HEADER_LENGTH << 1);
+            final int numFramesLessOne = numFrames - 1;
+            if (i < numFramesLessOne) {
+                writeFrameHeaderInternal(headerBuffer,
+                                         dataPerFrame + padding + flags.getPaddingPresenceFieldLength(),
+                                         DATA, flags, streamId);
+                writePaddingLength(headerBuffer, padding);
+
+                // Write frame header + data + padding for frames (0, n).
+                if (padding > 0) {
+                    do {
+                        ctx.write(headerBuffer.slice().retain(), promiseAggregator.newPromise());
+                        ctx.write(data.readSlice(dataPerFrame).retain(), promiseAggregator.newPromise());
+                        ctx.write(ZERO_BUFFER.slice(0, padding), promiseAggregator.newPromise());
+                    } while (++i < numFramesLessOne);
+                } else {
+                    do {
+                        ctx.write(headerBuffer.slice().retain(), promiseAggregator.newPromise());
+                        ctx.write(data.readSlice(dataPerFrame).retain(), promiseAggregator.newPromise());
+                    } while (++i < numFramesLessOne);
                 }
-            } while (!lastFrame);
 
+                // Skip enough bytes so we are on the unused 2nd half portion of the buffer.
+                headerBuffer.readerIndex(headerBuffer.writerIndex());
+            }
+
+            // Initialize the header portion of the frame for frame [n].
+            flags.endOfStream(endStream);
+            writeFrameHeaderInternal(headerBuffer,
+                                     data.readableBytes() + padding + flags.getPaddingPresenceFieldLength(),
+                                     DATA, flags, streamId);
+            writePaddingLength(headerBuffer, padding);
+
+            // Write frame header + data + padding for frame [n].
+            ByteBuf headerBuffer2 = headerBuffer; // make sure headerBuffer isn't released in the catch block.
+            headerBuffer = null;
+            ctx.write(headerBuffer2, promiseAggregator.newPromise());
+            ++i; // increment i so we don't release in the catch block.
+            ctx.write(data.readSlice(data.readableBytes()), promiseAggregator.newPromise());
+            if (padding > 0) { // Write the frame padding.
+                ctx.write(ZERO_BUFFER.slice(0, padding), promiseAggregator.newPromise());
+            }
             return promiseAggregator.doneAllocatingPromises();
         } catch (Throwable t) {
-            if (needToReleaseHeaders) {
-                header.release();
-            }
-            if (needToReleaseData) {
+            if (i < numFrames) {
                 data.release();
             }
+            if (headerBuffer != null) {
+                headerBuffer.release();
+            }
             return promiseAggregator.setFailure(t);
         }
     }
@@ -566,51 +581,4 @@ private static void verifyPingPayload(ByteBuf data) {
             throw new IllegalArgumentException("Opaque data must be " + PING_FRAME_PAYLOAD_LENGTH + " bytes");
         }
     }
-
-    /**
-     * Utility class that manages the creation of frame header buffers for {@code DATA} frames. Attempts
-     * to reuse the same buffer repeatedly when splitting data into multiple frames.
-     */
-    private static final class DataFrameHeader {
-        private final int streamId;
-        private final ByteBuf buffer;
-        private final Http2Flags flags = new Http2Flags();
-        private int prevData;
-        private int prevPadding;
-        private ByteBuf frameHeader;
-
-        DataFrameHeader(ChannelHandlerContext ctx, int streamId) {
-            // All padding will be put at the end, so in the worst case we need 3 headers:
-            // a repeated no-padding frame of maxFrameSize, a frame that has part data and part
-            // padding, and a frame that has the remainder of the padding.
-            buffer = ctx.alloc().buffer(3 * DATA_FRAME_HEADER_LENGTH);
-            this.streamId = streamId;
-        }
-
-        /**
-         * Gets the frame header buffer configured for the current frame.
-         */
-        ByteBuf slice(int data, int padding, boolean endOfStream) {
-            // Since we're reusing the current frame header whenever possible, check if anything changed
-            // that requires a new header.
-            if (data != prevData || padding != prevPadding
-                    || endOfStream != flags.endOfStream() || frameHeader == null) {
-                // Update the header state.
-                prevData = data;
-                prevPadding = padding;
-                flags.paddingPresent(padding > 0);
-                flags.endOfStream(endOfStream);
-                frameHeader = buffer.readSlice(DATA_FRAME_HEADER_LENGTH).writerIndex(0);
-
-                int payloadLength = data + padding + flags.getPaddingPresenceFieldLength();
-                writeFrameHeaderInternal(frameHeader, payloadLength, DATA, flags, streamId);
-                writePaddingLength(frameHeader, padding);
-            }
-            return frameHeader.slice();
-        }
-
-        void release() {
-            buffer.release();
-        }
-    }
 }


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameRoundtripTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameRoundtripTest.java
index 29cdc162bf5..297fa57bb33 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameRoundtripTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameRoundtripTest.java
@@ -197,7 +197,7 @@ public void largeDataFrameShouldMatch() throws Exception {
         for (int framePadding : paddingCaptor.getAllValues()) {
             totalReadPadding += framePadding;
         }
-        assertEquals(originalPadding, totalReadPadding);
+        assertEquals(originalPadding * paddingCaptor.getAllValues().size(), totalReadPadding);
     }
 
     @Test