Merge pull request #4136 from ctiller/configurable-lookahead

Add a (temporary) facility to tune stream read-ahead size

include/grpc/grpc.h

@@ -127,6 +127,11 @@ typedef struct {
 /** Initial sequence number for http2 transports */
 #define GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER \
   "grpc.http2.initial_sequence_number"
+/** Amount to read ahead on individual streams. Defaults to 64kb, larger
+    values can help throughput on high-latency connections.
+    NOTE: at some point we'd like to auto-tune this, and this parameter
+    will become a no-op. */
+#define GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES "grpc.http2.lookahead_bytes"
 /** How much memory to use for hpack decoding */
 #define GRPC_ARG_HTTP2_HPACK_TABLE_SIZE_DECODER \
   "grpc.http2.hpack_table_size.decoder"

src/core/transport/chttp2/internal.h

@@ -191,6 +191,9 @@ typedef struct {
       copied to next_stream_id in parsing when parsing commences */
   gpr_uint32 next_stream_id;
 
+  /** how far to lookahead in a stream? */
+  gpr_uint32 stream_lookahead;
+
   /** last received stream id */
   gpr_uint32 last_incoming_stream_id;
 

src/core/transport/chttp2_transport.c

@@ -52,7 +52,6 @@
 #include "src/core/transport/transport_impl.h"
 
 #define DEFAULT_WINDOW 65535
-#define GRPC_CHTTP2_STREAM_LOOKAHEAD DEFAULT_WINDOW
 #define DEFAULT_CONNECTION_WINDOW_TARGET (1024 * 1024)
 #define MAX_WINDOW 0x7fffffffu
 
@@ -245,6 +244,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   t->global.is_client = is_client;
   t->writing.outgoing_window = DEFAULT_WINDOW;
   t->parsing.incoming_window = DEFAULT_WINDOW;
+  t->global.stream_lookahead = DEFAULT_WINDOW;
   t->global.connection_window_target = DEFAULT_CONNECTION_WINDOW_TARGET;
   t->global.ping_counter = 1;
   t->global.pings.next = t->global.pings.prev = &t->global.pings;
@@ -338,6 +338,18 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
           t->global.next_stream_id =
               (gpr_uint32)channel_args->args[i].value.integer;
         }
+      } else if (0 == strcmp(channel_args->args[i].key,
+                             GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES)) {
+        if (channel_args->args[i].type != GRPC_ARG_INTEGER) {
+          gpr_log(GPR_ERROR, "%s: must be an integer",
+                  GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES);
+        } else if (channel_args->args[i].value.integer <= 5) {
+          gpr_log(GPR_ERROR, "%s: must be at least 5",
+                  GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES);
+        } else {
+          t->global.stream_lookahead =
+              (gpr_uint32)channel_args->args[i].value.integer;
+        }
       } else if (0 == strcmp(channel_args->args[i].key,
                              GRPC_ARG_HTTP2_HPACK_TABLE_SIZE_DECODER)) {
         if (channel_args->args[i].type != GRPC_ARG_INTEGER) {
@@ -1404,8 +1416,8 @@ static void incoming_byte_stream_update_flow_control(
   gpr_uint32 max_recv_bytes;
 
   /* clamp max recv hint to an allowable size */
-  if (max_size_hint >= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD) {
-    max_recv_bytes = GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD;
+  if (max_size_hint >= GPR_UINT32_MAX - transport_global->stream_lookahead) {
+    max_recv_bytes = GPR_UINT32_MAX - transport_global->stream_lookahead;
   } else {
     max_recv_bytes = (gpr_uint32)max_size_hint;
   }
@@ -1418,8 +1430,9 @@ static void incoming_byte_stream_update_flow_control(
   }
 
   /* add some small lookahead to keep pipelines flowing */
-  GPR_ASSERT(max_recv_bytes <= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD);
-  max_recv_bytes += GRPC_CHTTP2_STREAM_LOOKAHEAD;
+  GPR_ASSERT(max_recv_bytes <=
+             GPR_UINT32_MAX - transport_global->stream_lookahead);
+  max_recv_bytes += transport_global->stream_lookahead;
   if (stream_global->max_recv_bytes < max_recv_bytes) {
     gpr_uint32 add_max_recv_bytes =
         max_recv_bytes - stream_global->max_recv_bytes;