merge with head

7 files changed, 481 insertions, 31 deletions

diff --git a/src/core/profiling/basic_timers.c b/src/core/profiling/basic_timers.c
index 527a160101..f0fce7858d 100644
--- a/src/core/profiling/basic_timers.c
+++ b/src/core/profiling/basic_timers.c
@@ -53,50 +53,186 @@ typedef struct gpr_timer_entry {
   short line;
   char type;
   gpr_uint8 important;
+  int thd;
 } gpr_timer_entry;
 
-#define MAX_COUNT (5 * 1024 * 1024 / sizeof(gpr_timer_entry))
+#define MAX_COUNT 1000000
 
-static __thread gpr_timer_entry g_log[MAX_COUNT];
-static __thread int g_count;
+typedef struct gpr_timer_log {
+  size_t num_entries;
+  struct gpr_timer_log *next;
+  struct gpr_timer_log *prev;
+  gpr_timer_entry log[MAX_COUNT];
+} gpr_timer_log;
+
+typedef struct gpr_timer_log_list {
+  gpr_timer_log *head;
+  /* valid iff head!=NULL */
+  gpr_timer_log *tail;
+} gpr_timer_log_list;
+
+static __thread gpr_timer_log *g_thread_log;
 static gpr_once g_once_init = GPR_ONCE_INIT;
 static FILE *output_file;
+static const char *output_filename = "latency_trace.txt";
+static pthread_mutex_t g_mu;
+static pthread_cond_t g_cv;
+static gpr_timer_log_list g_in_progress_logs;
+static gpr_timer_log_list g_done_logs;
+static int g_shutdown;
+static gpr_thd_id g_writing_thread;
+static __thread int g_thread_id;
+static int g_next_thread_id;
 
-static void close_output() { fclose(output_file); }
+static int timer_log_push_back(gpr_timer_log_list *list, gpr_timer_log *log) {
+  if (list->head == NULL) {
+    list->head = list->tail = log;
+    log->next = log->prev = NULL;
+    return 1;
+  } else {
+    log->prev = list->tail;
+    log->next = NULL;
+    list->tail->next = log;
+    list->tail = log;
+    return 0;
+  }
+}
 
-static void init_output() {
-  output_file = fopen("latency_trace.txt", "w");
-  GPR_ASSERT(output_file);
-  atexit(close_output);
+static gpr_timer_log *timer_log_pop_front(gpr_timer_log_list *list) {
+  gpr_timer_log *out = list->head;
+  if (out != NULL) {
+    list->head = out->next;
+    if (list->head != NULL) {
+      list->head->prev = NULL;
+    } else {
+      list->tail = NULL;
+    }
+  }
+  return out;
 }
 
-static void log_report() {
-  int i;
-  gpr_once_init(&g_once_init, init_output);
-  for (i = 0; i < g_count; i++) {
-    gpr_timer_entry *entry = &(g_log[i]);
+static void timer_log_remove(gpr_timer_log_list *list, gpr_timer_log *log) {
+  if (log->prev == NULL) {
+    list->head = log->next;
+    if (list->head != NULL) {
+      list->head->prev = NULL;
+    }
+  } else {
+    log->prev->next = log->next;
+  }
+  if (log->next == NULL) {
+    list->tail = log->prev;
+    if (list->tail != NULL) {
+      list->tail->next = NULL;
+    }
+  } else {
+    log->next->prev = log->prev;
+  }
+}
+
+static void write_log(gpr_timer_log *log) {
+  size_t i;
+  if (output_file == NULL) {
+    output_file = fopen(output_filename, "w");
+  }
+  for (i = 0; i < log->num_entries; i++) {
+    gpr_timer_entry *entry = &(log->log[i]);
+    if (gpr_time_cmp(entry->tm, gpr_time_0(entry->tm.clock_type)) < 0) {
+      entry->tm = gpr_time_0(entry->tm.clock_type);
+    }
     fprintf(output_file,
-            "{\"t\": %ld.%09d, \"thd\": \"%p\", \"type\": \"%c\", \"tag\": "
+            "{\"t\": %ld.%09d, \"thd\": \"%d\", \"type\": \"%c\", \"tag\": "
             "\"%s\", \"file\": \"%s\", \"line\": %d, \"imp\": %d}\n",
-            entry->tm.tv_sec, entry->tm.tv_nsec,
-            (void *)(gpr_intptr)gpr_thd_currentid(), entry->type, entry->tagstr,
-            entry->file, entry->line, entry->important);
+            entry->tm.tv_sec, entry->tm.tv_nsec, entry->thd, entry->type,
+            entry->tagstr, entry->file, entry->line, entry->important);
+  }
+}
+
+static void writing_thread(void *unused) {
+  gpr_timer_log *log;
+  pthread_mutex_lock(&g_mu);
+  for (;;) {
+    while ((log = timer_log_pop_front(&g_done_logs)) == NULL && !g_shutdown) {
+      pthread_cond_wait(&g_cv, &g_mu);
+    }
+    if (log != NULL) {
+      pthread_mutex_unlock(&g_mu);
+      write_log(log);
+      free(log);
+      pthread_mutex_lock(&g_mu);
+    }
+    if (g_shutdown) {
+      pthread_mutex_unlock(&g_mu);
+      return;
+    }
   }
+}
 
-  /* Now clear out the log */
-  g_count = 0;
+static void flush_logs(gpr_timer_log_list *list) {
+  gpr_timer_log *log;
+  while ((log = timer_log_pop_front(list)) != NULL) {
+    write_log(log);
+    free(log);
+  }
+}
+
+static void finish_writing() {
+  pthread_mutex_lock(&g_mu);
+  g_shutdown = 1;
+  pthread_cond_signal(&g_cv);
+  pthread_mutex_unlock(&g_mu);
+  gpr_thd_join(g_writing_thread);
+
+  gpr_log(GPR_INFO, "flushing logs");
+
+  pthread_mutex_lock(&g_mu);
+  flush_logs(&g_done_logs);
+  flush_logs(&g_in_progress_logs);
+  pthread_mutex_unlock(&g_mu);
+
+  if (output_file) {
+    fclose(output_file);
+  }
+}
+
+void gpr_timers_set_log_filename(const char *filename) {
+  output_filename = filename;
+}
+
+static void init_output() {
+  gpr_thd_options options = gpr_thd_options_default();
+  gpr_thd_options_set_joinable(&options);
+  gpr_thd_new(&g_writing_thread, writing_thread, NULL, &options);
+  atexit(finish_writing);
+}
+
+static void rotate_log() {
+  gpr_timer_log *new = malloc(sizeof(*new));
+  gpr_once_init(&g_once_init, init_output);
+  new->num_entries = 0;
+  pthread_mutex_lock(&g_mu);
+  if (g_thread_log != NULL) {
+    timer_log_remove(&g_in_progress_logs, g_thread_log);
+    if (timer_log_push_back(&g_done_logs, g_thread_log)) {
+      pthread_cond_signal(&g_cv);
+    }
+  } else {
+    g_thread_id = g_next_thread_id++;
+  }
+  timer_log_push_back(&g_in_progress_logs, new);
+  pthread_mutex_unlock(&g_mu);
+  g_thread_log = new;
 }
 
 static void gpr_timers_log_add(const char *tagstr, marker_type type,
                                int important, const char *file, int line) {
   gpr_timer_entry *entry;
 
-  /* TODO (vpai) : Improve concurrency */
-  if (g_count == MAX_COUNT) {
-    log_report();
+  if (g_thread_log == NULL || g_thread_log->num_entries == MAX_COUNT) {
+    rotate_log();
   }
 
-  entry = &g_log[g_count++];
+  entry = &g_thread_log->log[g_thread_log->num_entries++];
 
   entry->tm = gpr_now(GPR_CLOCK_PRECISE);
   entry->tagstr = tagstr;
@@ -104,6 +240,7 @@ static void gpr_timers_log_add(const char *tagstr, marker_type type,
   entry->file = file;
   entry->line = (short)line;
   entry->important = important != 0;
+  entry->thd = g_thread_id;
 }
 
 /* Latency profiler API implementation. */
@@ -131,4 +268,6 @@ void gpr_timers_global_destroy(void) {}
 void gpr_timers_global_init(void) {}
 
 void gpr_timers_global_destroy(void) {}
+
+void gpr_timers_set_log_filename(const char *filename) {}
 #endif /* GRPC_BASIC_PROFILER */
diff --git a/src/core/profiling/timers.h b/src/core/profiling/timers.h
index 0d112e7248..6a188dc566 100644
--- a/src/core/profiling/timers.h
+++ b/src/core/profiling/timers.h
@@ -48,6 +48,8 @@ void gpr_timer_begin(const char *tagstr, int important, const char *file,
 void gpr_timer_end(const char *tagstr, int important, const char *file,
                    int line);
 
+void gpr_timers_set_log_filename(const char *filename);
+
 #if !(defined(GRPC_STAP_PROFILER) + defined(GRPC_BASIC_PROFILER))
 /* No profiling. No-op all the things. */
 #define GPR_TIMER_MARK(tag, important) \
diff --git a/src/core/support/avl.c b/src/core/support/avl.c
new file mode 100644
index 0000000000..9734c9987f
--- /dev/null
+++ b/src/core/support/avl.c
@@ -0,0 +1,288 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc/support/avl.h>
+
+#include <assert.h>
+#include <stdlib.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/useful.h>
+
+gpr_avl gpr_avl_create(const gpr_avl_vtable *vtable) {
+  gpr_avl out;
+  out.vtable = vtable;
+  out.root = NULL;
+  return out;
+}
+
+static gpr_avl_node *ref_node(gpr_avl_node *node) {
+  if (node) {
+    gpr_ref(&node->refs);
+  }
+  return node;
+}
+
+static void unref_node(const gpr_avl_vtable *vtable, gpr_avl_node *node) {
+  if (node == NULL) {
+    return;
+  }
+  if (gpr_unref(&node->refs)) {
+    vtable->destroy_key(node->key);
+    vtable->destroy_value(node->value);
+    unref_node(vtable, node->left);
+    unref_node(vtable, node->right);
+    gpr_free(node);
+  }
+}
+
+static long node_height(gpr_avl_node *node) {
+  return node == NULL ? 0 : node->height;
+}
+
+#ifndef NDEBUG
+static long calculate_height(gpr_avl_node *node) {
+  return node == NULL ? 0 : 1 + GPR_MAX(calculate_height(node->left),
+                                        calculate_height(node->right));
+}
+
+static gpr_avl_node *assert_invariants(gpr_avl_node *n) {
+  if (n == NULL) return NULL;
+  assert_invariants(n->left);
+  assert_invariants(n->right);
+  assert(calculate_height(n) == n->height);
+  assert(labs(node_height(n->left) - node_height(n->right)) <= 1);
+  return n;
+}
+#else
+static gpr_avl_node *assert_invariants(gpr_avl_node *n) { return n; }
+#endif
+
+gpr_avl_node *new_node(void *key, void *value, gpr_avl_node *left,
+                       gpr_avl_node *right) {
+  gpr_avl_node *node = gpr_malloc(sizeof(*node));
+  gpr_ref_init(&node->refs, 1);
+  node->key = key;
+  node->value = value;
+  node->left = assert_invariants(left);
+  node->right = assert_invariants(right);
+  node->height = 1 + GPR_MAX(node_height(left), node_height(right));
+  return node;
+}
+
+static gpr_avl_node *get(const gpr_avl_vtable *vtable, gpr_avl_node *node,
+                         void *key) {
+  long cmp;
+
+  if (node == NULL) {
+    return NULL;
+  }
+
+  cmp = vtable->compare_keys(node->key, key);
+  if (cmp == 0) {
+    return node;
+  } else if (cmp > 0) {
+    return get(vtable, node->left, key);
+  } else {
+    return get(vtable, node->right, key);
+  }
+}
+
+void *gpr_avl_get(gpr_avl avl, void *key) {
+  gpr_avl_node *node = get(avl.vtable, avl.root, key);
+  return node ? node->value : NULL;
+}
+
+static gpr_avl_node *rotate_left(const gpr_avl_vtable *vtable, void *key,
+                                 void *value, gpr_avl_node *left,
+                                 gpr_avl_node *right) {
+  gpr_avl_node *n =
+      new_node(vtable->copy_key(right->key), vtable->copy_value(right->value),
+               new_node(key, value, left, ref_node(right->left)),
+               ref_node(right->right));
+  unref_node(vtable, right);
+  return n;
+}
+
+static gpr_avl_node *rotate_right(const gpr_avl_vtable *vtable, void *key,
+                                  void *value, gpr_avl_node *left,
+                                  gpr_avl_node *right) {
+  gpr_avl_node *n = new_node(
+      vtable->copy_key(left->key), vtable->copy_value(left->value),
+      ref_node(left->left), new_node(key, value, ref_node(left->right), right));
+  unref_node(vtable, left);
+  return n;
+}
+
+static gpr_avl_node *rotate_left_right(const gpr_avl_vtable *vtable, void *key,
+                                       void *value, gpr_avl_node *left,
+                                       gpr_avl_node *right) {
+  /* rotate_right(..., rotate_left(left), right) */
+  gpr_avl_node *n = new_node(
+      vtable->copy_key(left->right->key),
+      vtable->copy_value(left->right->value),
+      new_node(vtable->copy_key(left->key), vtable->copy_value(left->value),
+               ref_node(left->left), ref_node(left->right->left)),
+      new_node(key, value, ref_node(left->right->right), right));
+  unref_node(vtable, left);
+  return n;
+}
+
+static gpr_avl_node *rotate_right_left(const gpr_avl_vtable *vtable, void *key,
+                                       void *value, gpr_avl_node *left,
+                                       gpr_avl_node *right) {
+  /* rotate_left(..., left, rotate_right(right)) */
+  gpr_avl_node *n = new_node(
+      vtable->copy_key(right->left->key),
+      vtable->copy_value(right->left->value),
+      new_node(key, value, left, ref_node(right->left->left)),
+      new_node(vtable->copy_key(right->key), vtable->copy_key(right->value),
+               ref_node(right->left->right), ref_node(right->right)));
+  unref_node(vtable, right);
+  return n;
+}
+
+static gpr_avl_node *rebalance(const gpr_avl_vtable *vtable, void *key,
+                               void *value, gpr_avl_node *left,
+                               gpr_avl_node *right) {
+  switch (node_height(left) - node_height(right)) {
+    case 2:
+      if (node_height(left->left) - node_height(left->right) == -1) {
+        return assert_invariants(
+            rotate_left_right(vtable, key, value, left, right));
+      } else {
+        return assert_invariants(rotate_right(vtable, key, value, left, right));
+      }
+    case -2:
+      if (node_height(right->left) - node_height(right->right) == 1) {
+        return assert_invariants(
+            rotate_right_left(vtable, key, value, left, right));
+      } else {
+        return assert_invariants(rotate_left(vtable, key, value, left, right));
+      }
+    default:
+      return assert_invariants(new_node(key, value, left, right));
+  }
+}
+
+static gpr_avl_node *add(const gpr_avl_vtable *vtable, gpr_avl_node *node,
+                         void *key, void *value) {
+  long cmp;
+  if (node == NULL) {
+    return new_node(key, value, NULL, NULL);
+  }
+  cmp = vtable->compare_keys(node->key, key);
+  if (cmp == 0) {
+    return new_node(key, value, ref_node(node->left), ref_node(node->right));
+  } else if (cmp > 0) {
+    return rebalance(
+        vtable, vtable->copy_key(node->key), vtable->copy_value(node->value),
+        add(vtable, node->left, key, value), ref_node(node->right));
+  } else {
+    return rebalance(vtable, vtable->copy_key(node->key),
+                     vtable->copy_value(node->value), ref_node(node->left),
+                     add(vtable, node->right, key, value));
+  }
+}
+
+gpr_avl gpr_avl_add(gpr_avl avl, void *key, void *value) {
+  gpr_avl_node *old_root = avl.root;
+  avl.root = add(avl.vtable, avl.root, key, value);
+  assert_invariants(avl.root);
+  unref_node(avl.vtable, old_root);
+  return avl;
+}
+
+static gpr_avl_node *in_order_head(gpr_avl_node *node) {
+  while (node->left != NULL) {
+    node = node->left;
+  }
+  return node;
+}
+
+static gpr_avl_node *in_order_tail(gpr_avl_node *node) {
+  while (node->right != NULL) {
+    node = node->right;
+  }
+  return node;
+}
+
+static gpr_avl_node *remove(const gpr_avl_vtable *vtable, gpr_avl_node *node,
+                            void *key) {
+  long cmp;
+  if (node == NULL) {
+    return NULL;
+  }
+  cmp = vtable->compare_keys(node->key, key);
+  if (cmp == 0) {
+    if (node->left == NULL) {
+      return ref_node(node->right);
+    } else if (node->right == NULL) {
+      return ref_node(node->left);
+    } else if (node->left->height < node->right->height) {
+      gpr_avl_node *h = in_order_head(node->right);
+      return rebalance(vtable, vtable->copy_key(h->key),
+                       vtable->copy_value(h->value), ref_node(node->left),
+                       remove(vtable, node->right, h->key));
+    } else {
+      gpr_avl_node *h = in_order_tail(node->left);
+      return rebalance(
+          vtable, vtable->copy_key(h->key), vtable->copy_value(h->value),
+          remove(vtable, node->left, h->key), ref_node(node->right));
+    }
+  } else if (cmp > 0) {
+    return rebalance(vtable, vtable->copy_key(node->key),
+                     vtable->copy_value(node->value),
+                     remove(vtable, node->left, key), ref_node(node->right));
+  } else {
+    return rebalance(vtable, vtable->copy_key(node->key),
+                     vtable->copy_value(node->value), ref_node(node->left),
+                     remove(vtable, node->right, key));
+  }
+}
+
+gpr_avl gpr_avl_remove(gpr_avl avl, void *key) {
+  gpr_avl_node *old_root = avl.root;
+  avl.root = remove(avl.vtable, avl.root, key);
+  assert_invariants(avl.root);
+  unref_node(avl.vtable, old_root);
+  return avl;
+}
+
+gpr_avl gpr_avl_ref(gpr_avl avl) {
+  ref_node(avl.root);
+  return avl;
+}
+
+void gpr_avl_unref(gpr_avl avl) { unref_node(avl.vtable, avl.root); }
diff --git a/src/core/support/thd_posix.c b/src/core/support/thd_posix.c
index c36d94d044..653a1c88c1 100644
--- a/src/core/support/thd_posix.c
+++ b/src/core/support/thd_posix.c
@@ -53,7 +53,7 @@ struct thd_arg {
 /* Body of every thread started via gpr_thd_new. */
 static void *thread_body(void *v) {
   struct thd_arg a = *(struct thd_arg *)v;
-  gpr_free(v);
+  free(v);
   (*a.body)(a.arg);
   return NULL;
 }
@@ -63,7 +63,10 @@ int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg,
   int thread_started;
   pthread_attr_t attr;
   pthread_t p;
-  struct thd_arg *a = gpr_malloc(sizeof(*a));
+  /* don't use gpr_malloc as we may cause an infinite recursion with
+   * the profiling code */
+  struct thd_arg *a = malloc(sizeof(*a));
+  GPR_ASSERT(a != NULL);
   a->body = thd_body;
   a->arg = arg;
 
@@ -78,7 +81,7 @@ int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg,
   thread_started = (pthread_create(&p, &attr, &thread_body, a) == 0);
   GPR_ASSERT(pthread_attr_destroy(&attr) == 0);
   if (!thread_started) {
-    gpr_free(a);
+    free(a);
   }
   *t = (gpr_thd_id)p;
   return thread_started;
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index 216439549c..b2ac8ee965 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/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;
 
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index 3cb28f8f0c..805d05222d 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -320,6 +320,8 @@ static void finalize_outbuf(grpc_exec_ctx *exec_ctx,
       grpc_chttp2_list_add_written_stream(transport_writing, stream_writing);
     }
   }
+
+  GPR_TIMER_END("finalize_outbuf", 0);
 }
 
 void grpc_chttp2_cleanup_writing(
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index fdd835def9..77c68c9df0 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/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;
@@ -339,6 +339,18 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
               (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) {
           gpr_log(GPR_ERROR, "%s: must be an 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;