Add CRONET_READ_BUFFER_SIZE_KEY API to CronetClientStream

By default, CronetClientStreams would use a 4KB buffer to read data
from Cronet. This can be inefficient especially if the amount of data
being read is huge (~MBs) as each read callback operation incur overhead
from Cronet itself (e.g. Context switch, JNI calls). The alternative
would be to immediately bump the default to a bigger number but that
would incur an increase in memory usage.

So in order to safely experiment on this, An OptionKey is introduced
which allows setting a per-stream value which will be controlled in a
controlled environment to ensure we find the best new default.

Author: aymanm-google

cronet/src/main/java/io/grpc/cronet/CronetClientStream.java

@@ -69,6 +69,12 @@ class CronetClientStream extends AbstractClientStream {
 
   static final CallOptions.Key<Collection<Object>> CRONET_ANNOTATIONS_KEY =
       CallOptions.Key.create("cronet-annotations");
+  /**
+   * Sets the read buffer size which the GRPC layer will use to read data from Cronet. Higher buffer
+   * size leads to less overhead but more memory consumption. The current default value is 4KB.
+   */
+  public static final CallOptions.Key<Integer> CRONET_READ_BUFFER_SIZE_KEY =
+      CallOptions.Key.createWithDefault("cronet-read-buffer-size", 4 * 1024);
 
   private final String url;
   private final String userAgent;
@@ -85,6 +91,8 @@ class CronetClientStream extends AbstractClientStream {
   private final Collection<Object> annotations;
   private final TransportState state;
   private final Sink sink = new Sink();
+  @VisibleForTesting
+  final int readBufferSize;
   private StreamBuilderFactory streamFactory;
 
   CronetClientStream(
@@ -120,6 +128,7 @@ class CronetClientStream extends AbstractClientStream {
     this.annotations = callOptions.getOption(CRONET_ANNOTATIONS_KEY);
     this.state = new TransportState(maxMessageSize, statsTraceCtx, lock, transportTracer,
             callOptions);
+    this.readBufferSize = callOptions.getOption(CRONET_READ_BUFFER_SIZE_KEY);
 
     // Tests expect the "plain" deframer behavior, not MigratingDeframer
     // https://github.com/grpc/grpc-java/issues/7140

cronet/src/test/java/io/grpc/cronet/CronetChannelBuilderTest.java

@@ -16,6 +16,7 @@
 
 package io.grpc.cronet;
 
+import static io.grpc.cronet.CronetClientStream.CRONET_READ_BUFFER_SIZE_KEY;
 import static io.grpc.internal.GrpcUtil.TIMER_SERVICE;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertSame;
@@ -92,6 +93,41 @@ public void alwaysUsePut_defaultsToFalse() throws Exception {
     assertFalse(stream.idempotent);
   }
 
+  @Test
+  public void channelBuilderReadBufferSize_defaultsTo4Kb() throws Exception {
+    CronetChannelBuilder builder = CronetChannelBuilder.forAddress("address", 1234, mockEngine);
+    CronetTransportFactory transportFactory =
+        (CronetTransportFactory) builder.buildTransportFactory();
+    CronetClientTransport transport =
+        (CronetClientTransport)
+            transportFactory.newClientTransport(
+                new InetSocketAddress("localhost", 443),
+                new ClientTransportOptions(),
+                channelLogger);
+    CronetClientStream stream = transport.newStream(
+        method, new Metadata(), CallOptions.DEFAULT, tracers);
+
+    assertEquals(4 * 1024, stream.readBufferSize);
+  }
+
+  @Test
+  public void channelBuilderReadBufferSize_changeReflected() throws Exception {
+    CronetChannelBuilder builder = CronetChannelBuilder.forAddress("address", 1234, mockEngine);
+    CronetTransportFactory transportFactory =
+        (CronetTransportFactory) builder.buildTransportFactory();
+    CronetClientTransport transport =
+        (CronetClientTransport)
+            transportFactory.newClientTransport(
+                new InetSocketAddress("localhost", 443),
+                new ClientTransportOptions(),
+                channelLogger);
+    CronetClientStream stream = transport.newStream(
+        method, new Metadata(),
+        CallOptions.DEFAULT.withOption(CRONET_READ_BUFFER_SIZE_KEY, 32 * 1024), tracers);
+
+    assertEquals(32 * 1024, stream.readBufferSize);
+  }
+
   @Test
   public void scheduledExecutorService_default() {
     CronetChannelBuilder builder = CronetChannelBuilder.forAddress("address", 1234, mockEngine);