netty: increase message quantum

carl-mastrangelo · web-flow · commit 7b821a0e50a4 · 2017-05-26T16:08:26.000-07:00
This is a minor change setting the size of data frames sent when
interleaving RPCs.  The size was ~1024bytes previously, which
resulted in the `writev` syscalls sending many smaller chunks
before hitting the low water mark.  The end effect is larger calls
to `writev`, as seen with strace.

The effect of this is noticeable when sending a lot of data.  When
sending as many 1MB messages as possible it nearly doubles the
rate.

Before:

```
INFO: single throughput GRPC
50.0%ile Latency (in nanos):		280856575
90.0%ile Latency (in nanos):		349618175
95.0%ile Latency (in nanos):		380444671
99.0%ile Latency (in nanos):		455172095
99.9%ile Latency (in nanos):		537198591
100.0%ile Latency (in nanos):		566886399
QPS:                           346
Count:                         103984
```

After:
```
gRPC
50.0%ile Latency (in nanos):		125948927
90.0%ile Latency (in nanos):		166322175
95.0%ile Latency (in nanos):		177276927
99.0%ile Latency (in nanos):		193840127
99.9%ile Latency (in nanos):		226841599
100.0%ile Latency (in nanos):		256110591
QPS:                           774
Count:                         232340
```
diff --git a/netty/src/main/java/io/grpc/netty/NettyClientHandler.java b/netty/src/main/java/io/grpc/netty/NettyClientHandler.java
@@ -61,6 +61,7 @@
 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.DefaultHttp2RemoteFlowController;
 import io.netty.handler.codec.http2.Http2Connection;
 import io.netty.handler.codec.http2.Http2ConnectionAdapter;
 import io.netty.handler.codec.http2.Http2ConnectionDecoder;
@@ -78,6 +79,7 @@
 import io.netty.handler.codec.http2.Http2Stream;
 import io.netty.handler.codec.http2.Http2StreamVisitor;
 import io.netty.handler.codec.http2.StreamBufferingEncoder;
+import io.netty.handler.codec.http2.WeightedFairQueueByteDistributor;
 import io.netty.handler.logging.LogLevel;
 import java.nio.channels.ClosedChannelException;
 import java.util.concurrent.Executor;
@@ -121,6 +123,11 @@ static NettyClientHandler newHandler(ClientTransportLifecycleManager lifecycleMa
     Http2FrameReader frameReader = new DefaultHttp2FrameReader(headersDecoder);
     Http2FrameWriter frameWriter = new DefaultHttp2FrameWriter();
     Http2Connection connection = new DefaultHttp2Connection(false);
+    WeightedFairQueueByteDistributor dist = new WeightedFairQueueByteDistributor(connection);
+    dist.allocationQuantum(16 * 1024); // Make benchmarks fast again.
+    DefaultHttp2RemoteFlowController controller =
+        new DefaultHttp2RemoteFlowController(connection, dist);
+    connection.remote().flowController(controller);
 
     return newHandler(connection, frameReader, frameWriter, lifecycleManager, keepAliveManager,
         flowControlWindow, maxHeaderListSize, ticker, tooManyPingsRunnable);
diff --git a/netty/src/main/java/io/grpc/netty/NettyServerHandler.java b/netty/src/main/java/io/grpc/netty/NettyServerHandler.java
@@ -69,6 +69,7 @@
 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.DefaultHttp2RemoteFlowController;
 import io.netty.handler.codec.http2.Http2Connection;
 import io.netty.handler.codec.http2.Http2ConnectionAdapter;
 import io.netty.handler.codec.http2.Http2ConnectionDecoder;
@@ -87,6 +88,7 @@
 import io.netty.handler.codec.http2.Http2Settings;
 import io.netty.handler.codec.http2.Http2Stream;
 import io.netty.handler.codec.http2.Http2StreamVisitor;
+import io.netty.handler.codec.http2.WeightedFairQueueByteDistributor;
 import io.netty.handler.logging.LogLevel;
 import io.netty.util.AsciiString;
 import io.netty.util.ReferenceCountUtil;
@@ -180,6 +182,11 @@ static NettyServerHandler newHandler(
     Preconditions.checkArgument(maxMessageSize > 0, "maxMessageSize must be positive");
 
     final Http2Connection connection = new DefaultHttp2Connection(true);
+    WeightedFairQueueByteDistributor dist = new WeightedFairQueueByteDistributor(connection);
+    dist.allocationQuantum(16 * 1024); // Make benchmarks fast again.
+    DefaultHttp2RemoteFlowController controller =
+        new DefaultHttp2RemoteFlowController(connection, dist);
+    connection.remote().flowController(controller);
     final KeepAliveEnforcer keepAliveEnforcer = new KeepAliveEnforcer(
         permitKeepAliveWithoutCalls, permitKeepAliveTimeInNanos, TimeUnit.NANOSECONDS);
 
