Merge remote-tracking branch 'upstream/master' into enable-cronet-compression

61 files changed, 1693 insertions, 798 deletions

diff --git a/src/core/ext/census/grpc_filter.c b/src/core/ext/census/grpc_filter.c
index b80d831557..fc29dbd454 100644
--- a/src/core/ext/census/grpc_filter.c
+++ b/src/core/ext/census/grpc_filter.c
@@ -138,7 +138,7 @@ static grpc_error *client_init_call_elem(grpc_exec_ctx *exec_ctx,
 static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_call_element *elem,
                                      const grpc_call_final_info *final_info,
-                                     void *ignored) {
+                                     grpc_closure *ignored) {
   call_data *d = elem->call_data;
   GPR_ASSERT(d != NULL);
   /* TODO(hongyu): record rpc client stats and census_rpc_end_op here */
@@ -160,7 +160,7 @@ static grpc_error *server_init_call_elem(grpc_exec_ctx *exec_ctx,
 static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_call_element *elem,
                                      const grpc_call_final_info *final_info,
-                                     void *ignored) {
+                                     grpc_closure *ignored) {
   call_data *d = elem->call_data;
   GPR_ASSERT(d != NULL);
   /* TODO(hongyu): record rpc server stats and census_tracing_end_op here */
diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c
index bf64f84772..00c20913b0 100644
--- a/src/core/ext/client_channel/client_channel.c
+++ b/src/core/ext/client_channel/client_channel.c
@@ -47,6 +47,7 @@
 #include "src/core/ext/client_channel/lb_policy_registry.h"
 #include "src/core/ext/client_channel/proxy_mapper_registry.h"
 #include "src/core/ext/client_channel/resolver_registry.h"
+#include "src/core/ext/client_channel/retry_throttle.h"
 #include "src/core/ext/client_channel/subchannel.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/connected_channel.h"
@@ -71,7 +72,8 @@
  */
 
 typedef enum {
-  WAIT_FOR_READY_UNSET,
+  /* zero so it can be default initialized */
+  WAIT_FOR_READY_UNSET = 0,
   WAIT_FOR_READY_FALSE,
   WAIT_FOR_READY_TRUE
 } wait_for_ready_value;
@@ -188,6 +190,8 @@ typedef struct client_channel_channel_data {
   grpc_combiner *combiner;
   /** currently active load balancer */
   grpc_lb_policy *lb_policy;
+  /** retry throttle data */
+  grpc_server_retry_throttle_data *retry_throttle_data;
   /** maps method names to method_parameters structs */
   grpc_slice_hash_table *method_params_table;
   /** incoming resolver result - set by resolver.next() */
@@ -283,6 +287,65 @@ static void watch_lb_policy_locked(grpc_exec_ctx *exec_ctx, channel_data *chand,
                                                &w->on_changed);
 }
 
+typedef struct {
+  char *server_name;
+  grpc_server_retry_throttle_data *retry_throttle_data;
+} service_config_parsing_state;
+
+static void parse_retry_throttle_params(const grpc_json *field, void *arg) {
+  service_config_parsing_state *parsing_state = arg;
+  if (strcmp(field->key, "retryThrottling") == 0) {
+    if (parsing_state->retry_throttle_data != NULL) return;  // Duplicate.
+    if (field->type != GRPC_JSON_OBJECT) return;
+    int max_milli_tokens = 0;
+    int milli_token_ratio = 0;
+    for (grpc_json *sub_field = field->child; sub_field != NULL;
+         sub_field = sub_field->next) {
+      if (sub_field->key == NULL) return;
+      if (strcmp(sub_field->key, "maxTokens") == 0) {
+        if (max_milli_tokens != 0) return;  // Duplicate.
+        if (sub_field->type != GRPC_JSON_NUMBER) return;
+        max_milli_tokens = gpr_parse_nonnegative_int(sub_field->value);
+        if (max_milli_tokens == -1) return;
+        max_milli_tokens *= 1000;
+      } else if (strcmp(sub_field->key, "tokenRatio") == 0) {
+        if (milli_token_ratio != 0) return;  // Duplicate.
+        if (sub_field->type != GRPC_JSON_NUMBER) return;
+        // We support up to 3 decimal digits.
+        size_t whole_len = strlen(sub_field->value);
+        uint32_t multiplier = 1;
+        uint32_t decimal_value = 0;
+        const char *decimal_point = strchr(sub_field->value, '.');
+        if (decimal_point != NULL) {
+          whole_len = (size_t)(decimal_point - sub_field->value);
+          multiplier = 1000;
+          size_t decimal_len = strlen(decimal_point + 1);
+          if (decimal_len > 3) decimal_len = 3;
+          if (!gpr_parse_bytes_to_uint32(decimal_point + 1, decimal_len,
+                                         &decimal_value)) {
+            return;
+          }
+          uint32_t decimal_multiplier = 1;
+          for (size_t i = 0; i < (3 - decimal_len); ++i) {
+            decimal_multiplier *= 10;
+          }
+          decimal_value *= decimal_multiplier;
+        }
+        uint32_t whole_value;
+        if (!gpr_parse_bytes_to_uint32(sub_field->value, whole_len,
+                                       &whole_value)) {
+          return;
+        }
+        milli_token_ratio = (int)((whole_value * multiplier) + decimal_value);
+        if (milli_token_ratio <= 0) return;
+      }
+    }
+    parsing_state->retry_throttle_data =
+        grpc_retry_throttle_map_get_data_for_server(
+            parsing_state->server_name, max_milli_tokens, milli_token_ratio);
+  }
+}
+
 static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx,
                                               void *arg, grpc_error *error) {
   channel_data *chand = arg;
@@ -294,6 +357,8 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx,
   bool exit_idle = false;
   grpc_error *state_error = GRPC_ERROR_CREATE("No load balancing policy");
   char *service_config_json = NULL;
+  service_config_parsing_state parsing_state;
+  memset(&parsing_state, 0, sizeof(parsing_state));
 
   if (chand->resolver_result != NULL) {
     // Find LB policy name.
@@ -354,6 +419,19 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx,
       grpc_service_config *service_config =
           grpc_service_config_create(service_config_json);
       if (service_config != NULL) {
+        channel_arg =
+            grpc_channel_args_find(chand->resolver_result, GRPC_ARG_SERVER_URI);
+        GPR_ASSERT(channel_arg != NULL);
+        GPR_ASSERT(channel_arg->type == GRPC_ARG_STRING);
+        grpc_uri *uri =
+            grpc_uri_parse(exec_ctx, channel_arg->value.string, true);
+        GPR_ASSERT(uri->path[0] != '\0');
+        parsing_state.server_name =
+            uri->path[0] == '/' ? uri->path + 1 : uri->path;
+        grpc_service_config_parse_global_params(
+            service_config, parse_retry_throttle_params, &parsing_state);
+        parsing_state.server_name = NULL;
+        grpc_uri_destroy(uri);
         method_params_table = grpc_service_config_create_method_config_table(
             exec_ctx, service_config, method_parameters_create_from_json,
             &method_parameters_vtable);
@@ -385,6 +463,11 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx,
     chand->info_service_config_json = service_config_json;
   }
   gpr_mu_unlock(&chand->info_mu);
+
+  if (chand->retry_throttle_data != NULL) {
+    grpc_server_retry_throttle_data_unref(chand->retry_throttle_data);
+  }
+  chand->retry_throttle_data = parsing_state.retry_throttle_data;
   if (chand->method_params_table != NULL) {
     grpc_slice_hash_table_unref(exec_ctx, chand->method_params_table);
   }
@@ -612,6 +695,9 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
   }
   gpr_free(chand->info_lb_policy_name);
   gpr_free(chand->info_service_config_json);
+  if (chand->retry_throttle_data != NULL) {
+    grpc_server_retry_throttle_data_unref(chand->retry_throttle_data);
+  }
   if (chand->method_params_table != NULL) {
     grpc_slice_hash_table_unref(exec_ctx, chand->method_params_table);
   }
@@ -631,7 +717,8 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
 #define CANCELLED_CALL ((grpc_subchannel_call *)1)
 
 typedef enum {
-  GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING,
+  /* zero so that it can be default-initialized */
+  GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING = 0,
   GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL
 } subchannel_creation_phase;
 
@@ -652,14 +739,15 @@ typedef struct client_channel_call_data {
   grpc_slice path;  // Request path.
   gpr_timespec call_start_time;
   gpr_timespec deadline;
+  grpc_server_retry_throttle_data *retry_throttle_data;
   method_parameters *method_params;
-  grpc_closure read_service_config;
 
   grpc_error *cancel_error;
 
   /** either 0 for no call, 1 for cancelled, or a pointer to a
       grpc_subchannel_call */
   gpr_atm subchannel_call;
+  gpr_arena *arena;
 
   subchannel_creation_phase creation_phase;
   grpc_connected_subchannel *connected_subchannel;
@@ -674,6 +762,9 @@ typedef struct client_channel_call_data {
   grpc_call_stack *owning_call;
 
   grpc_linked_mdelem lb_token_mdelem;
+
+  grpc_closure on_complete;
+  grpc_closure *original_on_complete;
 } call_data;
 
 grpc_subchannel_call *grpc_client_channel_get_subchannel_call(
@@ -726,6 +817,51 @@ static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
   gpr_free(ops);
 }
 
+// Sets calld->method_params and calld->retry_throttle_data.
+// If the method params specify a timeout, populates
+// *per_method_deadline and returns true.
+static bool set_call_method_params_from_service_config_locked(
+    grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+    gpr_timespec *per_method_deadline) {
+  channel_data *chand = elem->channel_data;
+  call_data *calld = elem->call_data;
+  if (chand->retry_throttle_data != NULL) {
+    calld->retry_throttle_data =
+        grpc_server_retry_throttle_data_ref(chand->retry_throttle_data);
+  }
+  if (chand->method_params_table != NULL) {
+    calld->method_params = grpc_method_config_table_get(
+        exec_ctx, chand->method_params_table, calld->path);
+    if (calld->method_params != NULL) {
+      method_parameters_ref(calld->method_params);
+      if (gpr_time_cmp(calld->method_params->timeout,
+                       gpr_time_0(GPR_TIMESPAN)) != 0) {
+        *per_method_deadline =
+            gpr_time_add(calld->call_start_time, calld->method_params->timeout);
+        return true;
+      }
+    }
+  }
+  return false;
+}
+
+static void apply_final_configuration_locked(grpc_exec_ctx *exec_ctx,
+                                             grpc_call_element *elem) {
+  /* apply service-config level configuration to the call (now that we're
+   * certain it exists) */
+  call_data *calld = elem->call_data;
+  gpr_timespec per_method_deadline;
+  if (set_call_method_params_from_service_config_locked(exec_ctx, elem,
+                                                        &per_method_deadline)) {
+    // If the deadline from the service config is shorter than the one
+    // from the client API, reset the deadline timer.
+    if (gpr_time_cmp(per_method_deadline, calld->deadline) < 0) {
+      calld->deadline = per_method_deadline;
+      grpc_deadline_state_reset(exec_ctx, elem, calld->deadline);
+    }
+  }
+}
+
 static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
                                     grpc_error *error) {
   grpc_call_element *elem = arg;
@@ -754,9 +890,14 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
   } else {
     /* Create call on subchannel. */
     grpc_subchannel_call *subchannel_call = NULL;
+    const grpc_connected_subchannel_call_args call_args = {
+        .pollent = calld->pollent,
+        .path = calld->path,
+        .start_time = calld->call_start_time,
+        .deadline = calld->deadline,
+        .arena = calld->arena};
     grpc_error *new_error = grpc_connected_subchannel_create_call(
-        exec_ctx, calld->connected_subchannel, calld->pollent, calld->path,
-        calld->call_start_time, calld->deadline, &subchannel_call);
+        exec_ctx, calld->connected_subchannel, &call_args, &subchannel_call);
     if (new_error != GRPC_ERROR_NONE) {
       new_error = grpc_error_add_child(new_error, error);
       subchannel_call = CANCELLED_CALL;
@@ -851,6 +992,7 @@ static bool pick_subchannel_locked(
   }
   GPR_ASSERT(error == GRPC_ERROR_NONE);
   if (chand->lb_policy != NULL) {
+    apply_final_configuration_locked(exec_ctx, elem);
     grpc_lb_policy *lb_policy = chand->lb_policy;
     GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel");
     // If the application explicitly set wait_for_ready, use that.
@@ -982,9 +1124,14 @@ static void start_transport_stream_op_locked_inner(grpc_exec_ctx *exec_ctx,
   if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING &&
       calld->connected_subchannel != NULL) {
     grpc_subchannel_call *subchannel_call = NULL;
+    const grpc_connected_subchannel_call_args call_args = {
+        .pollent = calld->pollent,
+        .path = calld->path,
+        .start_time = calld->call_start_time,
+        .deadline = calld->deadline,
+        .arena = calld->arena};
     grpc_error *error = grpc_connected_subchannel_create_call(
-        exec_ctx, calld->connected_subchannel, calld->pollent, calld->path,
-        calld->call_start_time, calld->deadline, &subchannel_call);
+        exec_ctx, calld->connected_subchannel, &call_args, &subchannel_call);
     if (error != GRPC_ERROR_NONE) {
       subchannel_call = CANCELLED_CALL;
       fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error));
@@ -1002,6 +1149,26 @@ static void start_transport_stream_op_locked_inner(grpc_exec_ctx *exec_ctx,
   add_waiting_locked(calld, op);
 }
 
+static void on_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
+  grpc_call_element *elem = arg;
+  call_data *calld = elem->call_data;
+  if (calld->retry_throttle_data != NULL) {
+    if (error == GRPC_ERROR_NONE) {
+      grpc_server_retry_throttle_data_record_success(
+          calld->retry_throttle_data);
+    } else {
+      // TODO(roth): In a subsequent PR, check the return value here and
+      // decide whether or not to retry.  Note that we should only
+      // record failures whose statuses match the configured retryable
+      // or non-fatal status codes.
+      grpc_server_retry_throttle_data_record_failure(
+          calld->retry_throttle_data);
+    }
+  }
+  grpc_closure_run(exec_ctx, calld->original_on_complete,
+                   GRPC_ERROR_REF(error));
+}
+
 static void start_transport_stream_op_locked(grpc_exec_ctx *exec_ctx, void *arg,
                                              grpc_error *error_ignored) {
   GPR_TIMER_BEGIN("start_transport_stream_op_locked", 0);
@@ -1010,6 +1177,14 @@ static void start_transport_stream_op_locked(grpc_exec_ctx *exec_ctx, void *arg,
   grpc_call_element *elem = op->handler_private.args[0];
   call_data *calld = elem->call_data;
 
+  if (op->recv_trailing_metadata != NULL) {
+    GPR_ASSERT(op->on_complete != NULL);
+    calld->original_on_complete = op->on_complete;
+    grpc_closure_init(&calld->on_complete, on_complete, elem,
+                      grpc_schedule_on_exec_ctx);
+    op->on_complete = &calld->on_complete;
+  }
+
   start_transport_stream_op_locked_inner(exec_ctx, op, elem);
 
   GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call,
@@ -1060,114 +1235,19 @@ static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
   GPR_TIMER_END("cc_start_transport_stream_op", 0);
 }
 
-// Sets calld->method_params.
-// If the method params specify a timeout, populates
-// *per_method_deadline and returns true.
-static bool set_call_method_params_from_service_config_locked(
-    grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-    gpr_timespec *per_method_deadline) {
-  channel_data *chand = elem->channel_data;
-  call_data *calld = elem->call_data;
-  if (chand->method_params_table != NULL) {
-    calld->method_params = grpc_method_config_table_get(
-        exec_ctx, chand->method_params_table, calld->path);
-    if (calld->method_params != NULL) {
-      method_parameters_ref(calld->method_params);
-      if (gpr_time_cmp(calld->method_params->timeout,
-                       gpr_time_0(GPR_TIMESPAN)) != 0) {
-        *per_method_deadline =
-            gpr_time_add(calld->call_start_time, calld->method_params->timeout);
-        return true;
-      }
-    }
-  }
-  return false;
-}
-
-// Gets data from the service config.  Invoked when the resolver returns
-// its initial result.
-static void read_service_config_locked(grpc_exec_ctx *exec_ctx, void *arg,
-                                       grpc_error *error) {
-  grpc_call_element *elem = arg;
-  call_data *calld = elem->call_data;
-  // If this is an error, there's no point in looking at the service config.
-  if (error == GRPC_ERROR_NONE) {
-    gpr_timespec per_method_deadline;
-    if (set_call_method_params_from_service_config_locked(
-            exec_ctx, elem, &per_method_deadline)) {
-      // If the deadline from the service config is shorter than the one
-      // from the client API, reset the deadline timer.
-      if (gpr_time_cmp(per_method_deadline, calld->deadline) < 0) {
-        calld->deadline = per_method_deadline;
-        grpc_deadline_state_reset(exec_ctx, elem, calld->deadline);
-      }
-    }
-  }
-  GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config");
-}
-
-static void initial_read_service_config_locked(grpc_exec_ctx *exec_ctx,
-                                               void *arg,
-                                               grpc_error *error_ignored) {
-  grpc_call_element *elem = arg;
-  channel_data *chand = elem->channel_data;
-  call_data *calld = elem->call_data;
-  // If the resolver has already returned results, then we can access
-  // the service config parameters immediately.  Otherwise, we need to
-  // defer that work until the resolver returns an initial result.
-  if (chand->lb_policy != NULL) {
-    // We already have a resolver result, so check for service config.
-    gpr_timespec per_method_deadline;
-    if (set_call_method_params_from_service_config_locked(
-            exec_ctx, elem, &per_method_deadline)) {
-      calld->deadline = gpr_time_min(calld->deadline, per_method_deadline);
-    }
-  } else {
-    // We don't yet have a resolver result, so register a callback to
-    // get the service config data once the resolver returns.
-    // Take a reference to the call stack to be owned by the callback.
-    GRPC_CALL_STACK_REF(calld->owning_call, "read_service_config");
-    grpc_closure_init(&calld->read_service_config, read_service_config_locked,
-                      elem, grpc_combiner_scheduler(chand->combiner, false));
-    grpc_closure_list_append(&chand->waiting_for_config_closures,
-                             &calld->read_service_config, GRPC_ERROR_NONE);
-  }
-  // Start the deadline timer with the current deadline value.  If we
-  // do not yet have service config data, then the timer may be reset
-  // later.
-  grpc_deadline_state_start(exec_ctx, elem, calld->deadline);
-  GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call,
-                        "initial_read_service_config");
-}
-
 /* Constructor for call_data */
 static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_call_element *elem,
                                      const grpc_call_element_args *args) {
-  channel_data *chand = elem->channel_data;
   call_data *calld = elem->call_data;
   // Initialize data members.
   grpc_deadline_state_init(exec_ctx, elem, args->call_stack);
   calld->path = grpc_slice_ref_internal(args->path);
   calld->call_start_time = args->start_time;
   calld->deadline = gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC);
-  calld->method_params = NULL;
-  calld->cancel_error = GRPC_ERROR_NONE;
-  gpr_atm_rel_store(&calld->subchannel_call, 0);
-  calld->connected_subchannel = NULL;
-  calld->waiting_ops = NULL;
-  calld->waiting_ops_count = 0;
-  calld->waiting_ops_capacity = 0;
-  calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
   calld->owning_call = args->call_stack;
-  calld->pollent = NULL;
-  GRPC_CALL_STACK_REF(calld->owning_call, "initial_read_service_config");
-  grpc_closure_sched(
-      exec_ctx,
-      grpc_closure_init(&calld->read_service_config,
-                        initial_read_service_config_locked, elem,
-                        grpc_combiner_scheduler(chand->combiner, false)),
-      GRPC_ERROR_NONE);
+  calld->arena = args->arena;
+  grpc_deadline_state_start(exec_ctx, elem, calld->deadline);
   return GRPC_ERROR_NONE;
 }
 
@@ -1175,7 +1255,7 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
 static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                  grpc_call_element *elem,
                                  const grpc_call_final_info *final_info,
-                                 void *and_free_memory) {
+                                 grpc_closure *then_schedule_closure) {
   call_data *calld = elem->call_data;
   grpc_deadline_state_destroy(exec_ctx, elem);
   grpc_slice_unref_internal(exec_ctx, calld->path);
@@ -1185,6 +1265,8 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
   GRPC_ERROR_UNREF(calld->cancel_error);
   grpc_subchannel_call *call = GET_CALL(calld);
   if (call != NULL && call != CANCELLED_CALL) {
+    grpc_subchannel_call_set_cleanup_closure(call, then_schedule_closure);
+    then_schedule_closure = NULL;
     GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call");
   }
   GPR_ASSERT(calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING);
@@ -1194,7 +1276,7 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                     "picked");
   }
   gpr_free(calld->waiting_ops);
-  gpr_free(and_free_memory);
+  grpc_closure_sched(exec_ctx, then_schedule_closure, GRPC_ERROR_NONE);
 }
 
 static void cc_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/ext/client_channel/client_channel_plugin.c b/src/core/ext/client_channel/client_channel_plugin.c
index 28d3b63f99..f51277d0b2 100644
--- a/src/core/ext/client_channel/client_channel_plugin.c
+++ b/src/core/ext/client_channel/client_channel_plugin.c
@@ -43,6 +43,7 @@
 #include "src/core/ext/client_channel/lb_policy_registry.h"
 #include "src/core/ext/client_channel/proxy_mapper_registry.h"
 #include "src/core/ext/client_channel/resolver_registry.h"
+#include "src/core/ext/client_channel/retry_throttle.h"
 #include "src/core/ext/client_channel/subchannel_index.h"
 #include "src/core/lib/surface/channel_init.h"
 
@@ -82,6 +83,7 @@ static bool set_default_host_if_unset(grpc_exec_ctx *exec_ctx,
 void grpc_client_channel_init(void) {
   grpc_lb_policy_registry_init();
   grpc_resolver_registry_init();
+  grpc_retry_throttle_map_init();
   grpc_proxy_mapper_registry_init();
   grpc_register_http_proxy_mapper();
   grpc_subchannel_index_init();
@@ -96,6 +98,7 @@ void grpc_client_channel_shutdown(void) {
   grpc_subchannel_index_shutdown();
   grpc_channel_init_shutdown();
   grpc_proxy_mapper_registry_shutdown();
+  grpc_retry_throttle_map_shutdown();
   grpc_resolver_registry_shutdown();
   grpc_lb_policy_registry_shutdown();
 }
diff --git a/src/core/ext/client_channel/proxy_mapper_registry.c b/src/core/ext/client_channel/proxy_mapper_registry.c
index 2c44b9d490..0935ddbdbd 100644
--- a/src/core/ext/client_channel/proxy_mapper_registry.c
+++ b/src/core/ext/client_channel/proxy_mapper_registry.c
@@ -94,6 +94,14 @@ static void grpc_proxy_mapper_list_destroy(grpc_proxy_mapper_list* list) {
     grpc_proxy_mapper_destroy(list->list[i]);
   }
   gpr_free(list->list);
+  // Clean up in case we re-initialze later.
+  // TODO(ctiller): This should ideally live in
+  // grpc_proxy_mapper_registry_init().  However, if we did this there,
+  // then we would do it AFTER we start registering proxy mappers from
+  // third-party plugins, so they'd never show up (and would leak memory).
+  // We probably need some sort of dependency system for plugins to fix
+  // this.
+  memset(list, 0, sizeof(*list));
 }
 
 //
@@ -102,9 +110,7 @@ static void grpc_proxy_mapper_list_destroy(grpc_proxy_mapper_list* list) {
 
 static grpc_proxy_mapper_list g_proxy_mapper_list;
 
-void grpc_proxy_mapper_registry_init() {
-  memset(&g_proxy_mapper_list, 0, sizeof(g_proxy_mapper_list));
-}
+void grpc_proxy_mapper_registry_init() {}
 
 void grpc_proxy_mapper_registry_shutdown() {
   grpc_proxy_mapper_list_destroy(&g_proxy_mapper_list);
diff --git a/src/core/ext/client_channel/retry_throttle.c b/src/core/ext/client_channel/retry_throttle.c
new file mode 100644
index 0000000000..8926c3d782
--- /dev/null
+++ b/src/core/ext/client_channel/retry_throttle.c
@@ -0,0 +1,210 @@
+/*
+ *
+ * Copyright 2017, 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 "src/core/ext/client_channel/retry_throttle.h"
+
+#include <limits.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/atm.h>
+#include <grpc/support/avl.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/sync.h>
+
+//
+// server_retry_throttle_data
+//
+
+struct grpc_server_retry_throttle_data {
+  gpr_refcount refs;
+  int max_milli_tokens;
+  int milli_token_ratio;
+  gpr_atm milli_tokens;
+  // A pointer to the replacement for this grpc_server_retry_throttle_data
+  // entry.  If non-NULL, then this entry is stale and must not be used.
+  // We hold a reference to the replacement.
+  gpr_atm replacement;
+};
+
+static void get_replacement_throttle_data_if_needed(
+    grpc_server_retry_throttle_data** throttle_data) {
+  while (true) {
+    grpc_server_retry_throttle_data* new_throttle_data =
+        (grpc_server_retry_throttle_data*)gpr_atm_acq_load(
+            &(*throttle_data)->replacement);
+    if (new_throttle_data == NULL) return;
+    *throttle_data = new_throttle_data;
+  }
+}
+
+bool grpc_server_retry_throttle_data_record_failure(
+    grpc_server_retry_throttle_data* throttle_data) {
+  // First, check if we are stale and need to be replaced.
+  get_replacement_throttle_data_if_needed(&throttle_data);
+  // We decrement milli_tokens by 1000 (1 token) for each failure.
+  const int new_value = (int)gpr_atm_no_barrier_clamped_add(
+      &throttle_data->milli_tokens, (gpr_atm)-1000, (gpr_atm)0,
+      (gpr_atm)throttle_data->max_milli_tokens);
+  // Retries are allowed as long as the new value is above the threshold
+  // (max_milli_tokens / 2).
+  return new_value > throttle_data->max_milli_tokens / 2;
+}
+
+void grpc_server_retry_throttle_data_record_success(
+    grpc_server_retry_throttle_data* throttle_data) {
+  // First, check if we are stale and need to be replaced.
+  get_replacement_throttle_data_if_needed(&throttle_data);
+  // We increment milli_tokens by milli_token_ratio for each success.
+  gpr_atm_no_barrier_clamped_add(
+      &throttle_data->milli_tokens, (gpr_atm)throttle_data->milli_token_ratio,
+      (gpr_atm)0, (gpr_atm)throttle_data->max_milli_tokens);
+}
+
+grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_ref(
+    grpc_server_retry_throttle_data* throttle_data) {
+  gpr_ref(&throttle_data->refs);
+  return throttle_data;
+}
+
+void grpc_server_retry_throttle_data_unref(
+    grpc_server_retry_throttle_data* throttle_data) {
+  if (gpr_unref(&throttle_data->refs)) {
+    grpc_server_retry_throttle_data* replacement =
+        (grpc_server_retry_throttle_data*)gpr_atm_acq_load(
+            &throttle_data->replacement);
+    if (replacement != NULL) {
+      grpc_server_retry_throttle_data_unref(replacement);
+    }
+    gpr_free(throttle_data);
+  }
+}
+
+static grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_create(
+    int max_milli_tokens, int milli_token_ratio,
+    grpc_server_retry_throttle_data* old_throttle_data) {
+  grpc_server_retry_throttle_data* throttle_data =
+      gpr_malloc(sizeof(*throttle_data));
+  memset(throttle_data, 0, sizeof(*throttle_data));
+  gpr_ref_init(&throttle_data->refs, 1);
+  throttle_data->max_milli_tokens = max_milli_tokens;
+  throttle_data->milli_token_ratio = milli_token_ratio;
+  int initial_milli_tokens = max_milli_tokens;
+  // If there was a pre-existing entry for this server name, initialize
+  // the token count by scaling proportionately to the old data.  This
+  // ensures that if we're already throttling retries on the old scale,
+  // we will start out doing the same thing on the new one.
+  if (old_throttle_data != NULL) {
+    double token_fraction =
+        (int)gpr_atm_acq_load(&old_throttle_data->milli_tokens) /
+        (double)old_throttle_data->max_milli_tokens;
+    initial_milli_tokens = (int)(token_fraction * max_milli_tokens);
+  }
+  gpr_atm_rel_store(&throttle_data->milli_tokens,
+                    (gpr_atm)initial_milli_tokens);
+  // If there was a pre-existing entry, mark it as stale and give it a
+  // pointer to the new entry, which is its replacement.
+  if (old_throttle_data != NULL) {
+    grpc_server_retry_throttle_data_ref(throttle_data);
+    gpr_atm_rel_store(&old_throttle_data->replacement, (gpr_atm)throttle_data);
+  }
+  return throttle_data;
+}
+
+//
+// avl vtable for string -> server_retry_throttle_data map
+//
+
+static void* copy_server_name(void* key) { return gpr_strdup(key); }
+
+static long compare_server_name(void* key1, void* key2) {
+  return strcmp(key1, key2);
+}
+
+static void destroy_server_retry_throttle_data(void* value) {
+  grpc_server_retry_throttle_data* throttle_data = value;
+  grpc_server_retry_throttle_data_unref(throttle_data);
+}
+
+static void* copy_server_retry_throttle_data(void* value) {
+  grpc_server_retry_throttle_data* throttle_data = value;
+  return grpc_server_retry_throttle_data_ref(throttle_data);
+}
+
+static const gpr_avl_vtable avl_vtable = {
+    gpr_free /* destroy_key */, copy_server_name, compare_server_name,
+    destroy_server_retry_throttle_data, copy_server_retry_throttle_data};
+
+//
+// server_retry_throttle_map
+//
+
+static gpr_mu g_mu;
+static gpr_avl g_avl;
+
+void grpc_retry_throttle_map_init() {
+  gpr_mu_init(&g_mu);
+  g_avl = gpr_avl_create(&avl_vtable);
+}
+
+void grpc_retry_throttle_map_shutdown() {
+  gpr_mu_destroy(&g_mu);
+  gpr_avl_unref(g_avl);
+}
+
+grpc_server_retry_throttle_data* grpc_retry_throttle_map_get_data_for_server(
+    const char* server_name, int max_milli_tokens, int milli_token_ratio) {
+  gpr_mu_lock(&g_mu);
+  grpc_server_retry_throttle_data* throttle_data =
+      gpr_avl_get(g_avl, (char*)server_name);
+  if (throttle_data == NULL) {
+    // Entry not found.  Create a new one.
+    throttle_data = grpc_server_retry_throttle_data_create(
+        max_milli_tokens, milli_token_ratio, NULL);
+    g_avl = gpr_avl_add(g_avl, (char*)server_name, throttle_data);
+  } else {
+    if (throttle_data->max_milli_tokens != max_milli_tokens ||
+        throttle_data->milli_token_ratio != milli_token_ratio) {
+      // Entry found but with old parameters.  Create a new one based on
+      // the original one.
+      throttle_data = grpc_server_retry_throttle_data_create(
+          max_milli_tokens, milli_token_ratio, throttle_data);
+      g_avl = gpr_avl_add(g_avl, (char*)server_name, throttle_data);
+    } else {
+      // Entry found.  Increase refcount.
+      grpc_server_retry_throttle_data_ref(throttle_data);
+    }
+  }
+  gpr_mu_unlock(&g_mu);
+  return throttle_data;
+}
diff --git a/src/core/ext/client_channel/retry_throttle.h b/src/core/ext/client_channel/retry_throttle.h
new file mode 100644
index 0000000000..f9971faf65
--- /dev/null
+++ b/src/core/ext/client_channel/retry_throttle.h
@@ -0,0 +1,65 @@
+/*
+ *
+ * Copyright 2017, 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.
+ *
+ */
+
+#ifndef GRPC_CORE_EXT_CLIENT_CHANNEL_RETRY_THROTTLE_H
+#define GRPC_CORE_EXT_CLIENT_CHANNEL_RETRY_THROTTLE_H
+
+#include <stdbool.h>
+
+/// Tracks retry throttling data for an individual server name.
+typedef struct grpc_server_retry_throttle_data grpc_server_retry_throttle_data;
+
+/// Records a failure.  Returns true if it's okay to send a retry.
+bool grpc_server_retry_throttle_data_record_failure(
+    grpc_server_retry_throttle_data* throttle_data);
+/// Records a success.
+void grpc_server_retry_throttle_data_record_success(
+    grpc_server_retry_throttle_data* throttle_data);
+
+grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_ref(
+    grpc_server_retry_throttle_data* throttle_data);
+void grpc_server_retry_throttle_data_unref(
+    grpc_server_retry_throttle_data* throttle_data);
+
+/// Initializes global map of failure data for each server name.
+void grpc_retry_throttle_map_init();
+/// Shuts down global map of failure data for each server name.
+void grpc_retry_throttle_map_shutdown();
+
+/// Returns a reference to the failure data for \a server_name, creating
+/// a new entry if needed.
+/// Caller must eventually unref via \a grpc_server_retry_throttle_data_unref().
+grpc_server_retry_throttle_data* grpc_retry_throttle_map_get_data_for_server(
+    const char* server_name, int max_milli_tokens, int milli_token_ratio);
+
+#endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_RETRY_THROTTLE_H */
diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c
index 5df0a9060d..ed5029ea9a 100644
--- a/src/core/ext/client_channel/subchannel.c
+++ b/src/core/ext/client_channel/subchannel.c
@@ -148,6 +148,7 @@ struct grpc_subchannel {
 
 struct grpc_subchannel_call {
   grpc_connected_subchannel *connection;
+  grpc_closure *schedule_closure_after_destroy;
 };
 
 #define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1))
@@ -340,17 +341,15 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
     GPR_ASSERT(new_address != NULL);
     gpr_free(addr);
     addr = new_address;
-    if (new_args != NULL) c->args = new_args;
-  }
-  if (c->args == NULL) {
-    static const char *keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS};
-    grpc_arg new_arg = grpc_create_subchannel_address_arg(addr);
-    c->args = grpc_channel_args_copy_and_add_and_remove(
-        args->args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), &new_arg,
-        1);
-    gpr_free(new_arg.value.string);
   }
+  static const char *keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS};
+  grpc_arg new_arg = grpc_create_subchannel_address_arg(addr);
   gpr_free(addr);
+  c->args = grpc_channel_args_copy_and_add_and_remove(
+      new_args != NULL ? new_args : args->args, keys_to_remove,
+      GPR_ARRAY_SIZE(keys_to_remove), &new_arg, 1);
+  gpr_free(new_arg.value.string);
+  if (new_args != NULL) grpc_channel_args_destroy(exec_ctx, new_args);
   c->root_external_state_watcher.next = c->root_external_state_watcher.prev =
       &c->root_external_state_watcher;
   grpc_closure_init(&c->connected, subchannel_connected, c,
@@ -719,13 +718,22 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg,
 static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call,
                                     grpc_error *error) {
   grpc_subchannel_call *c = call;
+  GPR_ASSERT(c->schedule_closure_after_destroy != NULL);
   GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0);
   grpc_connected_subchannel *connection = c->connection;
-  grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), NULL, c);
+  grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), NULL,
+                          c->schedule_closure_after_destroy);
   GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, connection, "subchannel_call");
   GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0);
 }
 
+void grpc_subchannel_call_set_cleanup_closure(grpc_subchannel_call *call,
+                                              grpc_closure *closure) {
+  GPR_ASSERT(call->schedule_closure_after_destroy == NULL);
+  GPR_ASSERT(closure != NULL);
+  call->schedule_closure_after_destroy = closure;
+}
+
 void grpc_subchannel_call_ref(
     grpc_subchannel_call *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
   GRPC_CALL_STACK_REF(SUBCHANNEL_CALL_TO_CALL_STACK(c), REF_REASON);
@@ -761,15 +769,22 @@ grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
 
 grpc_error *grpc_connected_subchannel_create_call(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
-    grpc_polling_entity *pollent, grpc_slice path, gpr_timespec start_time,
-    gpr_timespec deadline, grpc_subchannel_call **call) {
+    const grpc_connected_subchannel_call_args *args,
+    grpc_subchannel_call **call) {
   grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
-  *call = gpr_zalloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
+  *call = gpr_arena_alloc(
+      args->arena, sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
   grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(*call);
   (*call)->connection = con;  // Ref is added below.
-  grpc_error *error =
-      grpc_call_stack_init(exec_ctx, chanstk, 1, subchannel_call_destroy, *call,
-                           NULL, NULL, path, start_time, deadline, callstk);
+  const grpc_call_element_args call_args = {.call_stack = callstk,
+                                            .server_transport_data = NULL,
+                                            .context = NULL,
+                                            .path = args->path,
+                                            .start_time = args->start_time,
+                                            .deadline = args->deadline,
+                                            .arena = args->arena};
+  grpc_error *error = grpc_call_stack_init(
+      exec_ctx, chanstk, 1, subchannel_call_destroy, *call, &call_args);
   if (error != GRPC_ERROR_NONE) {
     const char *error_string = grpc_error_string(error);
     gpr_log(GPR_ERROR, "error: %s", error_string);
@@ -778,7 +793,7 @@ grpc_error *grpc_connected_subchannel_create_call(
     return error;
   }
   GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call");
-  grpc_call_stack_set_pollset_or_pollset_set(exec_ctx, callstk, pollent);
+  grpc_call_stack_set_pollset_or_pollset_set(exec_ctx, callstk, args->pollent);
   return GRPC_ERROR_NONE;
 }
 
diff --git a/src/core/ext/client_channel/subchannel.h b/src/core/ext/client_channel/subchannel.h
index 6a70a76467..3e64a2507c 100644
--- a/src/core/ext/client_channel/subchannel.h
+++ b/src/core/ext/client_channel/subchannel.h
@@ -37,6 +37,7 @@
 #include "src/core/ext/client_channel/connector.h"
 #include "src/core/lib/channel/channel_stack.h"
 #include "src/core/lib/iomgr/polling_entity.h"
+#include "src/core/lib/support/arena.h"
 #include "src/core/lib/transport/connectivity_state.h"
 #include "src/core/lib/transport/metadata.h"
 
@@ -112,10 +113,18 @@ void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
                                     GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 
 /** construct a subchannel call */
+typedef struct {
+  grpc_polling_entity *pollent;
+  grpc_slice path;
+  gpr_timespec start_time;
+  gpr_timespec deadline;
+  gpr_arena *arena;
+} grpc_connected_subchannel_call_args;
+
 grpc_error *grpc_connected_subchannel_create_call(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *connected_subchannel,
-    grpc_polling_entity *pollent, grpc_slice path, gpr_timespec start_time,
-    gpr_timespec deadline, grpc_subchannel_call **subchannel_call);
+    const grpc_connected_subchannel_call_args *args,
+    grpc_subchannel_call **subchannel_call);
 
 /** process a transport level op */
 void grpc_connected_subchannel_process_transport_op(
@@ -154,6 +163,11 @@ void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx,
 char *grpc_subchannel_call_get_peer(grpc_exec_ctx *exec_ctx,
                                     grpc_subchannel_call *subchannel_call);
 
+/** Must be called once per call. Sets the 'then_schedule_closure' argument for
+    call stack destruction. */
+void grpc_subchannel_call_set_cleanup_closure(
+    grpc_subchannel_call *subchannel_call, grpc_closure *closure);
+
 grpc_call_stack *grpc_subchannel_call_get_call_stack(
     grpc_subchannel_call *subchannel_call);
 
diff --git a/src/core/ext/load_reporting/load_reporting_filter.c b/src/core/ext/load_reporting/load_reporting_filter.c
index c2750634a5..4ed832671d 100644
--- a/src/core/ext/load_reporting/load_reporting_filter.c
+++ b/src/core/ext/load_reporting/load_reporting_filter.c
@@ -123,7 +123,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {
+                              grpc_closure *ignored) {
   call_data *calld = elem->call_data;
 
   /* TODO(dgq): do something with the data
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 042a89277d..89659e7464 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -575,7 +575,7 @@ void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s) {
 
 static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                        grpc_stream *gs, grpc_stream_refcount *refcount,
-                       const void *server_data) {
+                       const void *server_data, gpr_arena *arena) {
   GPR_TIMER_BEGIN("init_stream", 0);
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
   grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs;
@@ -588,8 +588,8 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
   gpr_ref_init(&s->active_streams, 1);
   GRPC_CHTTP2_STREAM_REF(s, "chttp2");
 
-  grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[0]);
-  grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[1]);
+  grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[0], arena);
+  grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[1], arena);
   grpc_chttp2_data_parser_init(&s->data_parser);
   grpc_slice_buffer_init(&s->flow_controlled_buffer);
   s->deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
@@ -665,16 +665,17 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
 
   GPR_TIMER_END("destroy_stream", 0);
 
-  gpr_free(s->destroy_stream_arg);
+  grpc_closure_sched(exec_ctx, s->destroy_stream_arg, GRPC_ERROR_NONE);
 }
 
 static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
-                           grpc_stream *gs, void *and_free_memory) {
+                           grpc_stream *gs,
+                           grpc_closure *then_schedule_closure) {
   GPR_TIMER_BEGIN("destroy_stream", 0);
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
   grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs;
 
-  s->destroy_stream_arg = and_free_memory;
+  s->destroy_stream_arg = then_schedule_closure;
   grpc_closure_sched(
       exec_ctx, grpc_closure_init(&s->destroy_stream, destroy_stream_locked, s,
                                   grpc_combiner_scheduler(t->combiner, false)),
@@ -1629,15 +1630,19 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
       s->recv_trailing_metadata_finished != NULL) {
     char status_string[GPR_LTOA_MIN_BUFSIZE];
     gpr_ltoa(status, status_string);
-    grpc_chttp2_incoming_metadata_buffer_replace_or_add(
-        exec_ctx, &s->metadata_buffer[1],
-        grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_GRPC_STATUS,
-                                grpc_slice_from_copied_string(status_string)));
+    GRPC_LOG_IF_ERROR("add_status",
+                      grpc_chttp2_incoming_metadata_buffer_replace_or_add(
+                          exec_ctx, &s->metadata_buffer[1],
+                          grpc_mdelem_from_slices(
+                              exec_ctx, GRPC_MDSTR_GRPC_STATUS,
+                              grpc_slice_from_copied_string(status_string))));
     if (msg != NULL) {
-      grpc_chttp2_incoming_metadata_buffer_replace_or_add(
-          exec_ctx, &s->metadata_buffer[1],
-          grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_GRPC_MESSAGE,
-                                  grpc_slice_from_copied_string(msg)));
+      GRPC_LOG_IF_ERROR(
+          "add_status_message",
+          grpc_chttp2_incoming_metadata_buffer_replace_or_add(
+              exec_ctx, &s->metadata_buffer[1],
+              grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_GRPC_MESSAGE,
+                                      grpc_slice_from_copied_string(msg))));
     }
     s->published_metadata[1] = GRPC_METADATA_SYNTHESIZED_FROM_FAKE;
     grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, t, s);
diff --git a/src/core/ext/transport/chttp2/transport/hpack_parser.c b/src/core/ext/transport/chttp2/transport/hpack_parser.c
index 40f5120308..1865b997b7 100644
--- a/src/core/ext/transport/chttp2/transport/hpack_parser.c
+++ b/src/core/ext/transport/chttp2/transport/hpack_parser.c
@@ -1620,13 +1620,18 @@ void grpc_chttp2_hpack_parser_destroy(grpc_exec_ctx *exec_ctx,
 grpc_error *grpc_chttp2_hpack_parser_parse(grpc_exec_ctx *exec_ctx,
                                            grpc_chttp2_hpack_parser *p,
                                            grpc_slice slice) {
-  /* TODO(ctiller): limit the distance of end from beg, and perform multiple
-     steps in the event of a large chunk of data to limit
-     stack space usage when no tail call optimization is
-     available */
+/* max number of bytes to parse at a time... limits call stack depth on
+ * compilers without TCO */
+#define MAX_PARSE_LENGTH 1024
   p->current_slice_refcount = slice.refcount;
-  grpc_error *error = p->state(exec_ctx, p, GRPC_SLICE_START_PTR(slice),
-                               GRPC_SLICE_END_PTR(slice));
+  uint8_t *start = GRPC_SLICE_START_PTR(slice);
+  uint8_t *end = GRPC_SLICE_END_PTR(slice);
+  grpc_error *error = GRPC_ERROR_NONE;
+  while (start != end && error == GRPC_ERROR_NONE) {
+    uint8_t *target = start + GPR_MIN(MAX_PARSE_LENGTH, end - start);
+    error = p->state(exec_ctx, p, start, target);
+    start = target;
+  }
   p->current_slice_refcount = NULL;
   return error;
 }
diff --git a/src/core/ext/transport/chttp2/transport/incoming_metadata.c b/src/core/ext/transport/chttp2/transport/incoming_metadata.c
index c91b019aa0..da0a34d32a 100644
--- a/src/core/ext/transport/chttp2/transport/incoming_metadata.c
+++ b/src/core/ext/transport/chttp2/transport/incoming_metadata.c
@@ -41,69 +41,48 @@
 #include <grpc/support/log.h>
 
 void grpc_chttp2_incoming_metadata_buffer_init(
-    grpc_chttp2_incoming_metadata_buffer *buffer) {
-  buffer->deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
+    grpc_chttp2_incoming_metadata_buffer *buffer, gpr_arena *arena) {
+  buffer->arena = arena;
+  grpc_metadata_batch_init(&buffer->batch);
+  buffer->batch.deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
 }
 
 void grpc_chttp2_incoming_metadata_buffer_destroy(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer) {
-  size_t i;
-  if (!buffer->published) {
-    for (i = 0; i < buffer->count; i++) {
-      GRPC_MDELEM_UNREF(exec_ctx, buffer->elems[i].md);
-    }
-  }
-  gpr_free(buffer->elems);
+  grpc_metadata_batch_destroy(exec_ctx, &buffer->batch);
 }
 
-void grpc_chttp2_incoming_metadata_buffer_add(
-    grpc_chttp2_incoming_metadata_buffer *buffer, grpc_mdelem elem) {
-  GPR_ASSERT(!buffer->published);
-  if (buffer->capacity == buffer->count) {
-    buffer->capacity = GPR_MAX(8, 2 * buffer->capacity);
-    buffer->elems =
-        gpr_realloc(buffer->elems, sizeof(*buffer->elems) * buffer->capacity);
-  }
-  buffer->elems[buffer->count++].md = elem;
+grpc_error *grpc_chttp2_incoming_metadata_buffer_add(
+    grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer,
+    grpc_mdelem elem) {
   buffer->size += GRPC_MDELEM_LENGTH(elem);
+  return grpc_metadata_batch_add_tail(
+      exec_ctx, &buffer->batch,
+      gpr_arena_alloc(buffer->arena, sizeof(grpc_linked_mdelem)), elem);
 }
 
-void grpc_chttp2_incoming_metadata_buffer_replace_or_add(
+grpc_error *grpc_chttp2_incoming_metadata_buffer_replace_or_add(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer,
     grpc_mdelem elem) {
-  for (size_t i = 0; i < buffer->count; i++) {
-    if (grpc_slice_eq(GRPC_MDKEY(buffer->elems[i].md), GRPC_MDKEY(elem))) {
-      GRPC_MDELEM_UNREF(exec_ctx, buffer->elems[i].md);
-      buffer->elems[i].md = elem;
-      return;
+  for (grpc_linked_mdelem *l = buffer->batch.list.head; l != NULL;
+       l = l->next) {
+    if (grpc_slice_eq(GRPC_MDKEY(l->md), GRPC_MDKEY(elem))) {
+      GRPC_MDELEM_UNREF(exec_ctx, l->md);
+      l->md = elem;
+      return GRPC_ERROR_NONE;
     }
   }
-  grpc_chttp2_incoming_metadata_buffer_add(buffer, elem);
+  return grpc_chttp2_incoming_metadata_buffer_add(exec_ctx, buffer, elem);
 }
 
 void grpc_chttp2_incoming_metadata_buffer_set_deadline(
     grpc_chttp2_incoming_metadata_buffer *buffer, gpr_timespec deadline) {
-  GPR_ASSERT(!buffer->published);
-  buffer->deadline = deadline;
+  buffer->batch.deadline = deadline;
 }
 
 void grpc_chttp2_incoming_metadata_buffer_publish(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer,
     grpc_metadata_batch *batch) {
-  GPR_ASSERT(!buffer->published);
-  buffer->published = 1;
-  if (buffer->count > 0) {
-    size_t i;
-    for (i = 0; i < buffer->count; i++) {
-      /* TODO(ctiller): do something better here */
-      if (!GRPC_LOG_IF_ERROR("grpc_chttp2_incoming_metadata_buffer_publish",
-                             grpc_metadata_batch_link_tail(
-                                 exec_ctx, batch, &buffer->elems[i]))) {
-        GRPC_MDELEM_UNREF(exec_ctx, buffer->elems[i].md);
-      }
-    }
-  } else {
-    batch->list.head = batch->list.tail = NULL;
-  }
-  batch->deadline = buffer->deadline;
+  *batch = buffer->batch;
+  grpc_metadata_batch_init(&buffer->batch);
 }
diff --git a/src/core/ext/transport/chttp2/transport/incoming_metadata.h b/src/core/ext/transport/chttp2/transport/incoming_metadata.h
index 1eac6fc150..288c917e65 100644
--- a/src/core/ext/transport/chttp2/transport/incoming_metadata.h
+++ b/src/core/ext/transport/chttp2/transport/incoming_metadata.h
@@ -37,28 +37,26 @@
 #include "src/core/lib/transport/transport.h"
 
 typedef struct {
-  grpc_linked_mdelem *elems;
-  size_t count;
-  size_t capacity;
-  gpr_timespec deadline;
-  int published;
+  gpr_arena *arena;
+  grpc_metadata_batch batch;
   size_t size;  // total size of metadata
 } grpc_chttp2_incoming_metadata_buffer;
 
 /** assumes everything initially zeroed */
 void grpc_chttp2_incoming_metadata_buffer_init(
-    grpc_chttp2_incoming_metadata_buffer *buffer);
+    grpc_chttp2_incoming_metadata_buffer *buffer, gpr_arena *arena);
 void grpc_chttp2_incoming_metadata_buffer_destroy(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer);
 void grpc_chttp2_incoming_metadata_buffer_publish(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer,
     grpc_metadata_batch *batch);
 
-void grpc_chttp2_incoming_metadata_buffer_add(
-    grpc_chttp2_incoming_metadata_buffer *buffer, grpc_mdelem elem);
-void grpc_chttp2_incoming_metadata_buffer_replace_or_add(
+grpc_error *grpc_chttp2_incoming_metadata_buffer_add(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer,
-    grpc_mdelem elem);
+    grpc_mdelem elem) GRPC_MUST_USE_RESULT;
+grpc_error *grpc_chttp2_incoming_metadata_buffer_replace_or_add(
+    grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer *buffer,
+    grpc_mdelem elem) GRPC_MUST_USE_RESULT;
 void grpc_chttp2_incoming_metadata_buffer_set_deadline(
     grpc_chttp2_incoming_metadata_buffer *buffer, gpr_timespec deadline);
 
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index d26812ad6b..3c56c21599 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -425,7 +425,7 @@ struct grpc_chttp2_stream {
   grpc_stream_refcount *refcount;
 
   grpc_closure destroy_stream;
-  void *destroy_stream_arg;
+  grpc_closure *destroy_stream_arg;
 
   grpc_chttp2_stream_link links[STREAM_LIST_COUNT];
   uint8_t included[STREAM_LIST_COUNT];
diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c
index e7f2597f89..7efc8c63c9 100644
--- a/src/core/ext/transport/chttp2/transport/parsing.c
+++ b/src/core/ext/transport/chttp2/transport/parsing.c
@@ -548,7 +548,14 @@ static void on_initial_header(grpc_exec_ctx *exec_ctx, void *tp,
       s->seen_error = true;
       GRPC_MDELEM_UNREF(exec_ctx, md);
     } else {
-      grpc_chttp2_incoming_metadata_buffer_add(&s->metadata_buffer[0], md);
+      grpc_error *error = grpc_chttp2_incoming_metadata_buffer_add(
+          exec_ctx, &s->metadata_buffer[0], md);
+      if (error != GRPC_ERROR_NONE) {
+        grpc_chttp2_cancel_stream(exec_ctx, t, s, error);
+        grpc_chttp2_parsing_become_skip_parser(exec_ctx, t);
+        s->seen_error = true;
+        GRPC_MDELEM_UNREF(exec_ctx, md);
+      }
     }
   }
 
@@ -598,7 +605,14 @@ static void on_trailing_header(grpc_exec_ctx *exec_ctx, void *tp,
     s->seen_error = true;
     GRPC_MDELEM_UNREF(exec_ctx, md);
   } else {
-    grpc_chttp2_incoming_metadata_buffer_add(&s->metadata_buffer[1], md);
+    grpc_error *error = grpc_chttp2_incoming_metadata_buffer_add(
+        exec_ctx, &s->metadata_buffer[1], md);
+    if (error != GRPC_ERROR_NONE) {
+      grpc_chttp2_cancel_stream(exec_ctx, t, s, error);
+      grpc_chttp2_parsing_become_skip_parser(exec_ctx, t);
+      s->seen_error = true;
+      GRPC_MDELEM_UNREF(exec_ctx, md);
+    }
   }
 
   GPR_TIMER_END("on_trailing_header", 0);
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index 67bf6bb151..36bb67869c 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -185,6 +185,7 @@ struct op_storage {
 };
 
 struct stream_obj {
+  gpr_arena *arena;
   struct op_and_state *oas;
   grpc_transport_stream_op *curr_op;
   grpc_cronet_transport *curr_ct;
@@ -273,7 +274,7 @@ static void maybe_flush_read(stream_obj *s) {
   /* Whenever the evaluation of any of the two condition is changed, we check
    * whether we should enter the flush read state. */
   if (s->state.pending_recv_trailing_metadata && s->state.fail_state) {
-    if (!s->state.flush_read) {
+    if (!s->state.flush_read && !s->state.rs.read_stream_closed) {
       CRONET_LOG(GPR_DEBUG, "%p: Flush read", s);
       s->state.flush_read = true;
       null_and_maybe_free_read_buffer(s);
@@ -497,15 +498,18 @@ static void on_response_headers_received(
   gpr_mu_lock(&s->mu);
   memset(&s->state.rs.initial_metadata, 0,
          sizeof(s->state.rs.initial_metadata));
-  grpc_chttp2_incoming_metadata_buffer_init(&s->state.rs.initial_metadata);
+  grpc_chttp2_incoming_metadata_buffer_init(&s->state.rs.initial_metadata,
+                                            s->arena);
   for (size_t i = 0; i < headers->count; i++) {
-    grpc_chttp2_incoming_metadata_buffer_add(
-        &s->state.rs.initial_metadata,
-        grpc_mdelem_from_slices(
-            &exec_ctx, grpc_slice_intern(grpc_slice_from_static_string(
-                           headers->headers[i].key)),
-            grpc_slice_intern(
-                grpc_slice_from_static_string(headers->headers[i].value))));
+    GRPC_LOG_IF_ERROR(
+        "on_response_headers_received",
+        grpc_chttp2_incoming_metadata_buffer_add(
+            &exec_ctx, &s->state.rs.initial_metadata,
+            grpc_mdelem_from_slices(
+                &exec_ctx, grpc_slice_intern(grpc_slice_from_static_string(
+                               headers->headers[i].key)),
+                grpc_slice_intern(grpc_slice_from_static_string(
+                    headers->headers[i].value)))));
   }
   s->state.state_callback_received[OP_RECV_INITIAL_METADATA] = true;
   if (!(s->state.state_op_done[OP_CANCEL_ERROR] ||
@@ -598,17 +602,20 @@ static void on_response_trailers_received(
   memset(&s->state.rs.trailing_metadata, 0,
          sizeof(s->state.rs.trailing_metadata));
   s->state.rs.trailing_metadata_valid = false;
-  grpc_chttp2_incoming_metadata_buffer_init(&s->state.rs.trailing_metadata);
+  grpc_chttp2_incoming_metadata_buffer_init(&s->state.rs.trailing_metadata,
+                                            s->arena);
   for (size_t i = 0; i < trailers->count; i++) {
     CRONET_LOG(GPR_DEBUG, "trailer key=%s, value=%s", trailers->headers[i].key,
                trailers->headers[i].value);
-    grpc_chttp2_incoming_metadata_buffer_add(
-        &s->state.rs.trailing_metadata,
-        grpc_mdelem_from_slices(
-            &exec_ctx, grpc_slice_intern(grpc_slice_from_static_string(
-                           trailers->headers[i].key)),
-            grpc_slice_intern(
-                grpc_slice_from_static_string(trailers->headers[i].value))));
+    GRPC_LOG_IF_ERROR(
+        "on_response_trailers_received",
+        grpc_chttp2_incoming_metadata_buffer_add(
+            &exec_ctx, &s->state.rs.trailing_metadata,
+            grpc_mdelem_from_slices(
+                &exec_ctx, grpc_slice_intern(grpc_slice_from_static_string(
+                               trailers->headers[i].key)),
+                grpc_slice_intern(grpc_slice_from_static_string(
+                    trailers->headers[i].value)))));
     s->state.rs.trailing_metadata_valid = true;
     if (0 == strcmp(trailers->headers[i].key, "grpc-status") &&
         0 != strcmp(trailers->headers[i].value, "0")) {
@@ -1240,7 +1247,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
 
 static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                        grpc_stream *gs, grpc_stream_refcount *refcount,
-                       const void *server_data) {
+                       const void *server_data, gpr_arena *arena) {
   stream_obj *s = (stream_obj *)gs;
   memset(&s->storage, 0, sizeof(s->storage));
   s->storage.head = NULL;
@@ -1262,6 +1269,7 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
 
   s->curr_gs = gs;
   s->curr_ct = (grpc_cronet_transport *)gt;
+  s->arena = arena;
 
   gpr_mu_init(&s->mu);
   return 0;
@@ -1298,10 +1306,12 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
 }
 
 static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
-                           grpc_stream *gs, void *and_free_memory) {
+                           grpc_stream *gs,
+                           grpc_closure *then_schedule_closure) {
   stream_obj *s = (stream_obj *)gs;
   null_and_maybe_free_read_buffer(s);
   GRPC_ERROR_UNREF(s->state.cancel_error);
+  grpc_closure_sched(exec_ctx, then_schedule_closure, GRPC_ERROR_NONE);
 }
 
 static void destroy_transport(grpc_exec_ctx *exec_ctx, grpc_transport *gt) {}
diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c
index 3fb2a60ac7..6d53b0576e 100644
--- a/src/core/lib/channel/channel_stack.c
+++ b/src/core/lib/channel/channel_stack.c
@@ -166,41 +166,32 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx,
   }
 }
 
-grpc_error *grpc_call_stack_init(
-    grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack,
-    int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg,
-    grpc_call_context_element *context, const void *transport_server_data,
-    grpc_slice path, gpr_timespec start_time, gpr_timespec deadline,
-    grpc_call_stack *call_stack) {
+grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx,
+                                 grpc_channel_stack *channel_stack,
+                                 int initial_refs, grpc_iomgr_cb_func destroy,
+                                 void *destroy_arg,
+                                 const grpc_call_element_args *elem_args) {
   grpc_channel_element *channel_elems = CHANNEL_ELEMS_FROM_STACK(channel_stack);
   size_t count = channel_stack->count;
   grpc_call_element *call_elems;
   char *user_data;
   size_t i;
 
-  call_stack->count = count;
-  GRPC_STREAM_REF_INIT(&call_stack->refcount, initial_refs, destroy,
+  elem_args->call_stack->count = count;
+  GRPC_STREAM_REF_INIT(&elem_args->call_stack->refcount, initial_refs, destroy,
                        destroy_arg, "CALL_STACK");
-  call_elems = CALL_ELEMS_FROM_STACK(call_stack);
+  call_elems = CALL_ELEMS_FROM_STACK(elem_args->call_stack);
   user_data = ((char *)call_elems) +
               ROUND_UP_TO_ALIGNMENT_SIZE(count * sizeof(grpc_call_element));
 
   /* init per-filter data */
   grpc_error *first_error = GRPC_ERROR_NONE;
-  const grpc_call_element_args args = {
-      .start_time = start_time,
-      .call_stack = call_stack,
-      .server_transport_data = transport_server_data,
-      .context = context,
-      .path = path,
-      .deadline = deadline,
-  };
   for (i = 0; i < count; i++) {
     call_elems[i].filter = channel_elems[i].filter;
     call_elems[i].channel_data = channel_elems[i].channel_data;
     call_elems[i].call_data = user_data;
-    grpc_error *error =
-        call_elems[i].filter->init_call_elem(exec_ctx, &call_elems[i], &args);
+    grpc_error *error = call_elems[i].filter->init_call_elem(
+        exec_ctx, &call_elems[i], elem_args);
     if (error != GRPC_ERROR_NONE) {
       if (first_error == GRPC_ERROR_NONE) {
         first_error = error;
@@ -241,15 +232,16 @@ void grpc_call_stack_ignore_set_pollset_or_pollset_set(
 
 void grpc_call_stack_destroy(grpc_exec_ctx *exec_ctx, grpc_call_stack *stack,
                              const grpc_call_final_info *final_info,
-                             void *and_free_memory) {
+                             grpc_closure *then_schedule_closure) {
   grpc_call_element *elems = CALL_ELEMS_FROM_STACK(stack);
   size_t count = stack->count;
   size_t i;
 
   /* destroy per-filter data */
   for (i = 0; i < count; i++) {
-    elems[i].filter->destroy_call_elem(exec_ctx, &elems[i], final_info,
-                                       i == count - 1 ? and_free_memory : NULL);
+    elems[i].filter->destroy_call_elem(
+        exec_ctx, &elems[i], final_info,
+        i == count - 1 ? then_schedule_closure : NULL);
   }
 }
 
diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h
index 6d3340bcbf..80e3603e8d 100644
--- a/src/core/lib/channel/channel_stack.h
+++ b/src/core/lib/channel/channel_stack.h
@@ -56,6 +56,7 @@
 
 #include "src/core/lib/debug/trace.h"
 #include "src/core/lib/iomgr/polling_entity.h"
+#include "src/core/lib/support/arena.h"
 #include "src/core/lib/transport/transport.h"
 
 #ifdef __cplusplus
@@ -84,6 +85,7 @@ typedef struct {
   grpc_slice path;
   gpr_timespec start_time;
   gpr_timespec deadline;
+  gpr_arena *arena;
 } grpc_call_element_args;
 
 typedef struct {
@@ -139,12 +141,12 @@ typedef struct {
   /* Destroy per call data.
      The filter does not need to do any chaining.
      The bottom filter of a stack will be passed a non-NULL pointer to
-     \a and_free_memory that should be passed to gpr_free when destruction
-     is complete. \a final_info contains data about the completed call, mainly
-     for reporting purposes. */
+     \a then_schedule_closure that should be passed to grpc_closure_sched when
+     destruction is complete. \a final_info contains data about the completed
+     call, mainly for reporting purposes. */
   void (*destroy_call_elem)(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                             const grpc_call_final_info *final_info,
-                            void *and_free_memory);
+                            grpc_closure *then_schedule_closure);
 
   /* sizeof(per channel data) */
   size_t sizeof_channel_data;
@@ -236,12 +238,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx,
 /* Initialize a call stack given a channel stack. transport_server_data is
    expected to be NULL on a client, or an opaque transport owned pointer on the
    server. */
-grpc_error *grpc_call_stack_init(
-    grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack,
-    int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg,
-    grpc_call_context_element *context, const void *transport_server_data,
-    grpc_slice path, gpr_timespec start_time, gpr_timespec deadline,
-    grpc_call_stack *call_stack);
+grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx,
+                                 grpc_channel_stack *channel_stack,
+                                 int initial_refs, grpc_iomgr_cb_func destroy,
+                                 void *destroy_arg,
+                                 const grpc_call_element_args *elem_args);
 /* Set a pollset or a pollset_set for a call stack: must occur before the first
  * op is started */
 void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx,
@@ -271,7 +272,7 @@ void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx,
 /* Destroy a call stack */
 void grpc_call_stack_destroy(grpc_exec_ctx *exec_ctx, grpc_call_stack *stack,
                              const grpc_call_final_info *final_info,
-                             void *and_free_memory);
+                             grpc_closure *then_schedule_closure);
 
 /* Ignore set pollset{_set} - used by filters if they don't care about pollsets
  * at all. Does nothing. */
diff --git a/src/core/lib/channel/compress_filter.c b/src/core/lib/channel/compress_filter.c
index aa41014a21..02dc479f3a 100644
--- a/src/core/lib/channel/compress_filter.c
+++ b/src/core/lib/channel/compress_filter.c
@@ -292,7 +292,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {
+                              grpc_closure *ignored) {
   /* grab pointers to our data from the call element */
   call_data *calld = elem->call_data;
   grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices);
diff --git a/src/core/lib/channel/connected_channel.c b/src/core/lib/channel/connected_channel.c
index 29796f7ca7..42ef7b7806 100644
--- a/src/core/lib/channel/connected_channel.c
+++ b/src/core/lib/channel/connected_channel.c
@@ -88,7 +88,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
   channel_data *chand = elem->channel_data;
   int r = grpc_transport_init_stream(
       exec_ctx, chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld),
-      &args->call_stack->refcount, args->server_transport_data);
+      &args->call_stack->refcount, args->server_transport_data, args->arena);
   return r == 0 ? GRPC_ERROR_NONE
                 : GRPC_ERROR_CREATE("transport stream initialization failed");
 }
@@ -105,12 +105,12 @@ static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx,
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *and_free_memory) {
+                              grpc_closure *then_schedule_closure) {
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
   grpc_transport_destroy_stream(exec_ctx, chand->transport,
                                 TRANSPORT_STREAM_FROM_CALL_DATA(calld),
-                                and_free_memory);
+                                then_schedule_closure);
 }
 
 /* Constructor for channel_data */
diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c
index 5a12d62f1d..34114bbebf 100644
--- a/src/core/lib/channel/deadline_filter.c
+++ b/src/core/lib/channel/deadline_filter.c
@@ -256,7 +256,7 @@ static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
 // Destructor for call_data.  Used for both client and server filters.
 static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
                               const grpc_call_final_info* final_info,
-                              void* and_free_memory) {
+                              grpc_closure* ignored) {
   grpc_deadline_state_destroy(exec_ctx, elem);
 }
 
diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c
index c031533dd8..f9d0d689ac 100644
--- a/src/core/lib/channel/http_client_filter.c
+++ b/src/core/lib/channel/http_client_filter.c
@@ -412,7 +412,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {
+                              grpc_closure *ignored) {
   call_data *calld = elem->call_data;
   grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices);
 }
diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c
index fb70de8e96..bebd3af335 100644
--- a/src/core/lib/channel/http_server_filter.c
+++ b/src/core/lib/channel/http_server_filter.c
@@ -358,7 +358,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {
+                              grpc_closure *ignored) {
   call_data *calld = elem->call_data;
   grpc_slice_buffer_destroy_internal(exec_ctx, &calld->read_slice_buffer);
 }
diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c
index b424c0d2ac..5ba13fe251 100644
--- a/src/core/lib/channel/message_size_filter.c
+++ b/src/core/lib/channel/message_size_filter.c
@@ -200,7 +200,7 @@ static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
 // Destructor for call_data.
 static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
                               const grpc_call_final_info* final_info,
-                              void* ignored) {}
+                              grpc_closure* ignored) {}
 
 // Constructor for channel_data.
 static grpc_error* init_channel_elem(grpc_exec_ctx* exec_ctx,
diff --git a/src/core/lib/iomgr/port.h b/src/core/lib/iomgr/port.h
index f1897bb91f..94a454c0b7 100644
--- a/src/core/lib/iomgr/port.h
+++ b/src/core/lib/iomgr/port.h
@@ -39,6 +39,7 @@
 #if defined(GRPC_UV)
 // Do nothing
 #elif defined(GPR_MANYLINUX1)
+#define GRPC_HAVE_IFADDRS 1
 #define GRPC_HAVE_IPV6_RECVPKTINFO 1
 #define GRPC_HAVE_IP_PKTINFO 1
 #define GRPC_HAVE_MSG_NOSIGNAL 1
@@ -65,6 +66,7 @@
 #define GRPC_POSIX_WAKEUP_FD 1
 #define GRPC_TIMER_USE_GENERIC 1
 #elif defined(GPR_LINUX)
+#define GRPC_HAVE_IFADDRS 1
 #define GRPC_HAVE_IPV6_RECVPKTINFO 1
 #define GRPC_HAVE_IP_PKTINFO 1
 #define GRPC_HAVE_MSG_NOSIGNAL 1
@@ -90,6 +92,7 @@
 #define GRPC_POSIX_SOCKETUTILS
 #endif
 #elif defined(GPR_APPLE)
+#define GRPC_HAVE_IFADDRS 1
 #define GRPC_HAVE_SO_NOSIGPIPE 1
 #define GRPC_HAVE_UNIX_SOCKET 1
 #define GRPC_MSG_IOVLEN_TYPE int
@@ -100,6 +103,7 @@
 #define GRPC_POSIX_WAKEUP_FD 1
 #define GRPC_TIMER_USE_GENERIC 1
 #elif defined(GPR_FREEBSD)
+#define GRPC_HAVE_IFADDRS 1
 #define GRPC_HAVE_IPV6_RECVPKTINFO 1
 #define GRPC_HAVE_SO_NOSIGPIPE 1
 #define GRPC_HAVE_UNIX_SOCKET 1
diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c
index 5f286a6723..e242631fc0 100644
--- a/src/core/lib/iomgr/tcp_server_posix.c
+++ b/src/core/lib/iomgr/tcp_server_posix.c
@@ -44,11 +44,8 @@
 
 #include <errno.h>
 #include <fcntl.h>
-#include <ifaddrs.h>
-#include <limits.h>
 #include <netinet/in.h>
 #include <netinet/tcp.h>
-#include <stdio.h>
 #include <string.h>
 #include <sys/socket.h>
 #include <sys/stat.h>
@@ -67,82 +64,10 @@
 #include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "src/core/lib/iomgr/socket_utils_posix.h"
 #include "src/core/lib/iomgr/tcp_posix.h"
+#include "src/core/lib/iomgr/tcp_server_utils_posix.h"
 #include "src/core/lib/iomgr/unix_sockets_posix.h"
 #include "src/core/lib/support/string.h"
 
-#define MIN_SAFE_ACCEPT_QUEUE_SIZE 100
-
-static gpr_once s_init_max_accept_queue_size;
-static int s_max_accept_queue_size;
-
-/* one listening port */
-typedef struct grpc_tcp_listener grpc_tcp_listener;
-struct grpc_tcp_listener {
-  int fd;
-  grpc_fd *emfd;
-  grpc_tcp_server *server;
-  grpc_resolved_address addr;
-  int port;
-  unsigned port_index;
-  unsigned fd_index;
-  grpc_closure read_closure;
-  grpc_closure destroyed_closure;
-  struct grpc_tcp_listener *next;
-  /* sibling is a linked list of all listeners for a given port. add_port and
-     clone_port place all new listeners in the same sibling list. A member of
-     the 'sibling' list is also a member of the 'next' list. The head of each
-     sibling list has is_sibling==0, and subsequent members of sibling lists
-     have is_sibling==1. is_sibling allows separate sibling lists to be
-     identified while iterating through 'next'. */
-  struct grpc_tcp_listener *sibling;
-  int is_sibling;
-};
-
-/* the overall server */
-struct grpc_tcp_server {
-  gpr_refcount refs;
-  /* Called whenever accept() succeeds on a server port. */
-  grpc_tcp_server_cb on_accept_cb;
-  void *on_accept_cb_arg;
-
-  gpr_mu mu;
-
-  /* active port count: how many ports are actually still listening */
-  size_t active_ports;
-  /* destroyed port count: how many ports are completely destroyed */
-  size_t destroyed_ports;
-
-  /* is this server shutting down? */
-  bool shutdown;
-  /* have listeners been shutdown? */
-  bool shutdown_listeners;
-  /* use SO_REUSEPORT */
-  bool so_reuseport;
-  /* expand wildcard addresses to a list of all local addresses */
-  bool expand_wildcard_addrs;
-
-  /* linked list of server ports */
-  grpc_tcp_listener *head;
-  grpc_tcp_listener *tail;
-  unsigned nports;
-
-  /* List of closures passed to shutdown_starting_add(). */
-  grpc_closure_list shutdown_starting;
-
-  /* shutdown callback */
-  grpc_closure *shutdown_complete;
-
-  /* all pollsets interested in new connections */
-  grpc_pollset **pollsets;
-  /* number of pollsets in the pollsets array */
-  size_t pollset_count;
-
-  /* next pollset to assign a channel to */
-  gpr_atm next_pollset_to_assign;
-
-  grpc_resource_quota *resource_quota;
-};
-
 static gpr_once check_init = GPR_ONCE_INIT;
 static bool has_so_reuseport = false;
 
@@ -301,99 +226,6 @@ static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
   }
 }
 
-/* get max listen queue size on linux */
-static void init_max_accept_queue_size(void) {
-  int n = SOMAXCONN;
-  char buf[64];
-  FILE *fp = fopen("/proc/sys/net/core/somaxconn", "r");
-  if (fp == NULL) {
-    /* 2.4 kernel. */
-    s_max_accept_queue_size = SOMAXCONN;
-    return;
-  }
-  if (fgets(buf, sizeof buf, fp)) {
-    char *end;
-    long i = strtol(buf, &end, 10);
-    if (i > 0 && i <= INT_MAX && end && *end == 0) {
-      n = (int)i;
-    }
-  }
-  fclose(fp);
-  s_max_accept_queue_size = n;
-
-  if (s_max_accept_queue_size < MIN_SAFE_ACCEPT_QUEUE_SIZE) {
-    gpr_log(GPR_INFO,
-            "Suspiciously small accept queue (%d) will probably lead to "
-            "connection drops",
-            s_max_accept_queue_size);
-  }
-}
-
-static int get_max_accept_queue_size(void) {
-  gpr_once_init(&s_init_max_accept_queue_size, init_max_accept_queue_size);
-  return s_max_accept_queue_size;
-}
-
-/* Prepare a recently-created socket for listening. */
-static grpc_error *prepare_socket(int fd, const grpc_resolved_address *addr,
-                                  bool so_reuseport, int *port) {
-  grpc_resolved_address sockname_temp;
-  grpc_error *err = GRPC_ERROR_NONE;
-
-  GPR_ASSERT(fd >= 0);
-
-  if (so_reuseport && !grpc_is_unix_socket(addr)) {
-    err = grpc_set_socket_reuse_port(fd, 1);
-    if (err != GRPC_ERROR_NONE) goto error;
-  }
-
-  err = grpc_set_socket_nonblocking(fd, 1);
-  if (err != GRPC_ERROR_NONE) goto error;
-  err = grpc_set_socket_cloexec(fd, 1);
-  if (err != GRPC_ERROR_NONE) goto error;
-  if (!grpc_is_unix_socket(addr)) {
-    err = grpc_set_socket_low_latency(fd, 1);
-    if (err != GRPC_ERROR_NONE) goto error;
-    err = grpc_set_socket_reuse_addr(fd, 1);
-    if (err != GRPC_ERROR_NONE) goto error;
-  }
-  err = grpc_set_socket_no_sigpipe_if_possible(fd);
-  if (err != GRPC_ERROR_NONE) goto error;
-
-  GPR_ASSERT(addr->len < ~(socklen_t)0);
-  if (bind(fd, (struct sockaddr *)addr->addr, (socklen_t)addr->len) < 0) {
-    err = GRPC_OS_ERROR(errno, "bind");
-    goto error;
-  }
-
-  if (listen(fd, get_max_accept_queue_size()) < 0) {
-    err = GRPC_OS_ERROR(errno, "listen");
-    goto error;
-  }
-
-  sockname_temp.len = sizeof(struct sockaddr_storage);
-
-  if (getsockname(fd, (struct sockaddr *)sockname_temp.addr,
-                  (socklen_t *)&sockname_temp.len) < 0) {
-    err = GRPC_OS_ERROR(errno, "getsockname");
-    goto error;
-  }
-
-  *port = grpc_sockaddr_get_port(&sockname_temp);
-  return GRPC_ERROR_NONE;
-
-error:
-  GPR_ASSERT(err != GRPC_ERROR_NONE);
-  if (fd >= 0) {
-    close(fd);
-  }
-  grpc_error *ret = grpc_error_set_int(
-      GRPC_ERROR_CREATE_REFERENCING("Unable to configure socket", &err, 1),
-      GRPC_ERROR_INT_FD, fd);
-  GRPC_ERROR_UNREF(err);
-  return ret;
-}
-
 /* event manager callback when reads are ready */
 static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *err) {
   grpc_tcp_listener *sp = arg;
@@ -477,216 +309,6 @@ error:
   }
 }
 
-static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd,
-                                        const grpc_resolved_address *addr,
-                                        unsigned port_index, unsigned fd_index,
-                                        grpc_tcp_listener **listener) {
-  grpc_tcp_listener *sp = NULL;
-  int port = -1;
-  char *addr_str;
-  char *name;
-
-  grpc_error *err = prepare_socket(fd, addr, s->so_reuseport, &port);
-  if (err == GRPC_ERROR_NONE) {
-    GPR_ASSERT(port > 0);
-    grpc_sockaddr_to_string(&addr_str, addr, 1);
-    gpr_asprintf(&name, "tcp-server-listener:%s", addr_str);
-    gpr_mu_lock(&s->mu);
-    s->nports++;
-    GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server");
-    sp = gpr_malloc(sizeof(grpc_tcp_listener));
-    sp->next = NULL;
-    if (s->head == NULL) {
-      s->head = sp;
-    } else {
-      s->tail->next = sp;
-    }
-    s->tail = sp;
-    sp->server = s;
-    sp->fd = fd;
-    sp->emfd = grpc_fd_create(fd, name);
-    memcpy(&sp->addr, addr, sizeof(grpc_resolved_address));
-    sp->port = port;
-    sp->port_index = port_index;
-    sp->fd_index = fd_index;
-    sp->is_sibling = 0;
-    sp->sibling = NULL;
-    GPR_ASSERT(sp->emfd);
-    gpr_mu_unlock(&s->mu);
-    gpr_free(addr_str);
-    gpr_free(name);
-  }
-
-  *listener = sp;
-  return err;
-}
-
-/* If successful, add a listener to s for addr, set *dsmode for the socket, and
-   return the *listener. */
-static grpc_error *add_addr_to_server(grpc_tcp_server *s,
-                                      const grpc_resolved_address *addr,
-                                      unsigned port_index, unsigned fd_index,
-                                      grpc_dualstack_mode *dsmode,
-                                      grpc_tcp_listener **listener) {
-  grpc_resolved_address addr4_copy;
-  int fd;
-  grpc_error *err =
-      grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, dsmode, &fd);
-  if (err != GRPC_ERROR_NONE) {
-    return err;
-  }
-  if (*dsmode == GRPC_DSMODE_IPV4 &&
-      grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) {
-    addr = &addr4_copy;
-  }
-  return add_socket_to_server(s, fd, addr, port_index, fd_index, listener);
-}
-
-/* Bind to "::" to get a port number not used by any address. */
-static grpc_error *get_unused_port(int *port) {
-  grpc_resolved_address wild;
-  grpc_sockaddr_make_wildcard6(0, &wild);
-  grpc_dualstack_mode dsmode;
-  int fd;
-  grpc_error *err =
-      grpc_create_dualstack_socket(&wild, SOCK_STREAM, 0, &dsmode, &fd);
-  if (err != GRPC_ERROR_NONE) {
-    return err;
-  }
-  if (dsmode == GRPC_DSMODE_IPV4) {
-    grpc_sockaddr_make_wildcard4(0, &wild);
-  }
-  if (bind(fd, (const struct sockaddr *)wild.addr, (socklen_t)wild.len) != 0) {
-    err = GRPC_OS_ERROR(errno, "bind");
-    close(fd);
-    return err;
-  }
-  if (getsockname(fd, (struct sockaddr *)wild.addr, (socklen_t *)&wild.len) !=
-      0) {
-    err = GRPC_OS_ERROR(errno, "getsockname");
-    close(fd);
-    return err;
-  }
-  close(fd);
-  *port = grpc_sockaddr_get_port(&wild);
-  return *port <= 0 ? GRPC_ERROR_CREATE("Bad port") : GRPC_ERROR_NONE;
-}
-
-/* Return the listener in s with address addr or NULL. */
-static grpc_tcp_listener *find_listener_with_addr(grpc_tcp_server *s,
-                                                  grpc_resolved_address *addr) {
-  grpc_tcp_listener *l;
-  gpr_mu_lock(&s->mu);
-  for (l = s->head; l != NULL; l = l->next) {
-    if (l->addr.len != addr->len) {
-      continue;
-    }
-    if (memcmp(l->addr.addr, addr->addr, addr->len) == 0) {
-      break;
-    }
-  }
-  gpr_mu_unlock(&s->mu);
-  return l;
-}
-
-/* Get all addresses assigned to network interfaces on the machine and create a
-   listener for each. requested_port is the port to use for every listener, or 0
-   to select one random port that will be used for every listener. Set *out_port
-   to the port selected. Return GRPC_ERROR_NONE only if all listeners were
-   added. */
-static grpc_error *add_all_local_addrs_to_server(grpc_tcp_server *s,
-                                                 unsigned port_index,
-                                                 int requested_port,
-                                                 int *out_port) {
-  struct ifaddrs *ifa = NULL;
-  struct ifaddrs *ifa_it;
-  unsigned fd_index = 0;
-  grpc_tcp_listener *sp = NULL;
-  grpc_error *err = GRPC_ERROR_NONE;
-  if (requested_port == 0) {
-    /* Note: There could be a race where some local addrs can listen on the
-       selected port and some can't. The sane way to handle this would be to
-       retry by recreating the whole grpc_tcp_server. Backing out individual
-       listeners and orphaning the FDs looks like too much trouble. */
-    if ((err = get_unused_port(&requested_port)) != GRPC_ERROR_NONE) {
-      return err;
-    } else if (requested_port <= 0) {
-      return GRPC_ERROR_CREATE("Bad get_unused_port()");
-    }
-    gpr_log(GPR_DEBUG, "Picked unused port %d", requested_port);
-  }
-  if (getifaddrs(&ifa) != 0 || ifa == NULL) {
-    return GRPC_OS_ERROR(errno, "getifaddrs");
-  }
-  for (ifa_it = ifa; ifa_it != NULL; ifa_it = ifa_it->ifa_next) {
-    grpc_resolved_address addr;
-    char *addr_str = NULL;
-    grpc_dualstack_mode dsmode;
-    grpc_tcp_listener *new_sp = NULL;
-    const char *ifa_name = (ifa_it->ifa_name ? ifa_it->ifa_name : "<unknown>");
-    if (ifa_it->ifa_addr == NULL) {
-      continue;
-    } else if (ifa_it->ifa_addr->sa_family == AF_INET) {
-      addr.len = sizeof(struct sockaddr_in);
-    } else if (ifa_it->ifa_addr->sa_family == AF_INET6) {
-      addr.len = sizeof(struct sockaddr_in6);
-    } else {
-      continue;
-    }
-    memcpy(addr.addr, ifa_it->ifa_addr, addr.len);
-    if (!grpc_sockaddr_set_port(&addr, requested_port)) {
-      /* Should never happen, because we check sa_family above. */
-      err = GRPC_ERROR_CREATE("Failed to set port");
-      break;
-    }
-    if (grpc_sockaddr_to_string(&addr_str, &addr, 0) < 0) {
-      addr_str = gpr_strdup("<error>");
-    }
-    gpr_log(GPR_DEBUG,
-            "Adding local addr from interface %s flags 0x%x to server: %s",
-            ifa_name, ifa_it->ifa_flags, addr_str);
-    /* We could have multiple interfaces with the same address (e.g., bonding),
-       so look for duplicates. */
-    if (find_listener_with_addr(s, &addr) != NULL) {
-      gpr_log(GPR_DEBUG, "Skipping duplicate addr %s on interface %s", addr_str,
-              ifa_name);
-      gpr_free(addr_str);
-      continue;
-    }
-    if ((err = add_addr_to_server(s, &addr, port_index, fd_index, &dsmode,
-                                  &new_sp)) != GRPC_ERROR_NONE) {
-      char *err_str = NULL;
-      grpc_error *root_err;
-      if (gpr_asprintf(&err_str, "Failed to add listener: %s", addr_str) < 0) {
-        err_str = gpr_strdup("Failed to add listener");
-      }
-      root_err = GRPC_ERROR_CREATE(err_str);
-      gpr_free(err_str);
-      gpr_free(addr_str);
-      err = grpc_error_add_child(root_err, err);
-      break;
-    } else {
-      GPR_ASSERT(requested_port == new_sp->port);
-      ++fd_index;
-      if (sp != NULL) {
-        new_sp->is_sibling = 1;
-        sp->sibling = new_sp;
-      }
-      sp = new_sp;
-    }
-    gpr_free(addr_str);
-  }
-  freeifaddrs(ifa);
-  if (err != GRPC_ERROR_NONE) {
-    return err;
-  } else if (sp == NULL) {
-    return GRPC_ERROR_CREATE("No local addresses");
-  } else {
-    *out_port = sp->port;
-    return GRPC_ERROR_NONE;
-  }
-}
-
 /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */
 static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s,
                                                 unsigned port_index,
@@ -701,14 +323,16 @@ static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s,
   grpc_error *v6_err = GRPC_ERROR_NONE;
   grpc_error *v4_err = GRPC_ERROR_NONE;
   *out_port = -1;
-  if (s->expand_wildcard_addrs) {
-    return add_all_local_addrs_to_server(s, port_index, requested_port,
-                                         out_port);
+
+  if (grpc_tcp_server_have_ifaddrs() && s->expand_wildcard_addrs) {
+    return grpc_tcp_server_add_all_local_addrs(s, port_index, requested_port,
+                                               out_port);
   }
+
   grpc_sockaddr_make_wildcards(requested_port, &wild4, &wild6);
   /* Try listening on IPv6 first. */
-  if ((v6_err = add_addr_to_server(s, &wild6, port_index, fd_index, &dsmode,
-                                   &sp)) == GRPC_ERROR_NONE) {
+  if ((v6_err = grpc_tcp_server_add_addr(s, &wild6, port_index, fd_index,
+                                         &dsmode, &sp)) == GRPC_ERROR_NONE) {
     ++fd_index;
     requested_port = *out_port = sp->port;
     if (dsmode == GRPC_DSMODE_DUALSTACK || dsmode == GRPC_DSMODE_IPV4) {
@@ -717,8 +341,8 @@ static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s,
   }
   /* If we got a v6-only socket or nothing, try adding 0.0.0.0. */
   grpc_sockaddr_set_port(&wild4, requested_port);
-  if ((v4_err = add_addr_to_server(s, &wild4, port_index, fd_index, &dsmode,
-                                   &sp2)) == GRPC_ERROR_NONE) {
+  if ((v4_err = grpc_tcp_server_add_addr(s, &wild4, port_index, fd_index,
+                                         &dsmode, &sp2)) == GRPC_ERROR_NONE) {
     *out_port = sp2->port;
     if (sp != NULL) {
       sp2->is_sibling = 1;
@@ -726,8 +350,20 @@ static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s,
     }
   }
   if (*out_port > 0) {
-    GRPC_LOG_IF_ERROR("Failed to add :: listener", v6_err);
-    GRPC_LOG_IF_ERROR("Failed to add 0.0.0.0 listener", v4_err);
+    if (v6_err != GRPC_ERROR_NONE) {
+      gpr_log(GPR_INFO,
+              "Failed to add :: listener, "
+              "the environment may not support IPv6: %s",
+              grpc_error_string(v6_err));
+      GRPC_ERROR_UNREF(v6_err);
+    }
+    if (v4_err != GRPC_ERROR_NONE) {
+      gpr_log(GPR_INFO,
+              "Failed to add 0.0.0.0 listener, "
+              "the environment may not support IPv4: %s",
+              grpc_error_string(v4_err));
+      GRPC_ERROR_UNREF(v4_err);
+    }
     return GRPC_ERROR_NONE;
   } else {
     grpc_error *root_err =
@@ -756,7 +392,7 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
     err = grpc_create_dualstack_socket(&listener->addr, SOCK_STREAM, 0, &dsmode,
                                        &fd);
     if (err != GRPC_ERROR_NONE) return err;
-    err = prepare_socket(fd, &listener->addr, true, &port);
+    err = grpc_tcp_server_prepare_socket(fd, &listener->addr, true, &port);
     if (err != GRPC_ERROR_NONE) return err;
     listener->server->nports++;
     grpc_sockaddr_to_string(&addr_str, &listener->addr, 1);
@@ -828,7 +464,7 @@ grpc_error *grpc_tcp_server_add_port(grpc_tcp_server *s,
   if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
     addr = &addr6_v4mapped;
   }
-  if ((err = add_addr_to_server(s, addr, port_index, 0, &dsmode, &sp)) ==
+  if ((err = grpc_tcp_server_add_addr(s, addr, port_index, 0, &dsmode, &sp)) ==
       GRPC_ERROR_NONE) {
     *out_port = sp->port;
   }
diff --git a/src/core/lib/iomgr/tcp_server_utils_posix.h b/src/core/lib/iomgr/tcp_server_utils_posix.h
new file mode 100644
index 0000000000..f5dc8532f9
--- /dev/null
+++ b/src/core/lib/iomgr/tcp_server_utils_posix.h
@@ -0,0 +1,134 @@
+/*
+ *
+ * Copyright 2017, 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.
+ *
+ */
+
+#ifndef GRPC_CORE_LIB_IOMGR_TCP_SERVER_UTILS_POSIX_H
+#define GRPC_CORE_LIB_IOMGR_TCP_SERVER_UTILS_POSIX_H
+
+#include "src/core/lib/iomgr/ev_posix.h"
+#include "src/core/lib/iomgr/resolve_address.h"
+#include "src/core/lib/iomgr/socket_utils_posix.h"
+#include "src/core/lib/iomgr/tcp_server.h"
+
+/* one listening port */
+typedef struct grpc_tcp_listener {
+  int fd;
+  grpc_fd *emfd;
+  grpc_tcp_server *server;
+  grpc_resolved_address addr;
+  int port;
+  unsigned port_index;
+  unsigned fd_index;
+  grpc_closure read_closure;
+  grpc_closure destroyed_closure;
+  struct grpc_tcp_listener *next;
+  /* sibling is a linked list of all listeners for a given port. add_port and
+     clone_port place all new listeners in the same sibling list. A member of
+     the 'sibling' list is also a member of the 'next' list. The head of each
+     sibling list has is_sibling==0, and subsequent members of sibling lists
+     have is_sibling==1. is_sibling allows separate sibling lists to be
+     identified while iterating through 'next'. */
+  struct grpc_tcp_listener *sibling;
+  int is_sibling;
+} grpc_tcp_listener;
+
+/* the overall server */
+struct grpc_tcp_server {
+  gpr_refcount refs;
+  /* Called whenever accept() succeeds on a server port. */
+  grpc_tcp_server_cb on_accept_cb;
+  void *on_accept_cb_arg;
+
+  gpr_mu mu;
+
+  /* active port count: how many ports are actually still listening */
+  size_t active_ports;
+  /* destroyed port count: how many ports are completely destroyed */
+  size_t destroyed_ports;
+
+  /* is this server shutting down? */
+  bool shutdown;
+  /* have listeners been shutdown? */
+  bool shutdown_listeners;
+  /* use SO_REUSEPORT */
+  bool so_reuseport;
+  /* expand wildcard addresses to a list of all local addresses */
+  bool expand_wildcard_addrs;
+
+  /* linked list of server ports */
+  grpc_tcp_listener *head;
+  grpc_tcp_listener *tail;
+  unsigned nports;
+
+  /* List of closures passed to shutdown_starting_add(). */
+  grpc_closure_list shutdown_starting;
+
+  /* shutdown callback */
+  grpc_closure *shutdown_complete;
+
+  /* all pollsets interested in new connections */
+  grpc_pollset **pollsets;
+  /* number of pollsets in the pollsets array */
+  size_t pollset_count;
+
+  /* next pollset to assign a channel to */
+  gpr_atm next_pollset_to_assign;
+
+  grpc_resource_quota *resource_quota;
+};
+
+/* If successful, add a listener to \a s for \a addr, set \a dsmode for the
+   socket, and return the \a listener. */
+grpc_error *grpc_tcp_server_add_addr(grpc_tcp_server *s,
+                                     const grpc_resolved_address *addr,
+                                     unsigned port_index, unsigned fd_index,
+                                     grpc_dualstack_mode *dsmode,
+                                     grpc_tcp_listener **listener);
+
+/* Get all addresses assigned to network interfaces on the machine and create a
+   listener for each. requested_port is the port to use for every listener, or 0
+   to select one random port that will be used for every listener. Set *out_port
+   to the port selected. Return GRPC_ERROR_NONE only if all listeners were
+   added. */
+grpc_error *grpc_tcp_server_add_all_local_addrs(grpc_tcp_server *s,
+                                                unsigned port_index,
+                                                int requested_port,
+                                                int *out_port);
+
+/* Prepare a recently-created socket for listening. */
+grpc_error *grpc_tcp_server_prepare_socket(int fd,
+                                           const grpc_resolved_address *addr,
+                                           bool so_reuseport, int *port);
+/* Ruturn true if the platform supports ifaddrs */
+bool grpc_tcp_server_have_ifaddrs(void);
+
+#endif /* GRPC_CORE_LIB_IOMGR_TCP_SERVER_UTILS_POSIX_H */
diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_common.c b/src/core/lib/iomgr/tcp_server_utils_posix_common.c
new file mode 100644
index 0000000000..e45e27d5ab
--- /dev/null
+++ b/src/core/lib/iomgr/tcp_server_utils_posix_common.c
@@ -0,0 +1,220 @@
+/*
+ *
+ * Copyright 2017, 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 "src/core/lib/iomgr/port.h"
+
+#ifdef GRPC_HAVE_IFADDRS
+
+#include "src/core/lib/iomgr/tcp_server_utils_posix.h"
+
+#include <errno.h>
+#include <limits.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/sync.h>
+
+#include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/iomgr/sockaddr.h"
+#include "src/core/lib/iomgr/sockaddr_utils.h"
+#include "src/core/lib/iomgr/unix_sockets_posix.h"
+
+#define MIN_SAFE_ACCEPT_QUEUE_SIZE 100
+
+static gpr_once s_init_max_accept_queue_size;
+static int s_max_accept_queue_size;
+
+/* get max listen queue size on linux */
+static void init_max_accept_queue_size(void) {
+  int n = SOMAXCONN;
+  char buf[64];
+  FILE *fp = fopen("/proc/sys/net/core/somaxconn", "r");
+  if (fp == NULL) {
+    /* 2.4 kernel. */
+    s_max_accept_queue_size = SOMAXCONN;
+    return;
+  }
+  if (fgets(buf, sizeof buf, fp)) {
+    char *end;
+    long i = strtol(buf, &end, 10);
+    if (i > 0 && i <= INT_MAX && end && *end == 0) {
+      n = (int)i;
+    }
+  }
+  fclose(fp);
+  s_max_accept_queue_size = n;
+
+  if (s_max_accept_queue_size < MIN_SAFE_ACCEPT_QUEUE_SIZE) {
+    gpr_log(GPR_INFO,
+            "Suspiciously small accept queue (%d) will probably lead to "
+            "connection drops",
+            s_max_accept_queue_size);
+  }
+}
+
+static int get_max_accept_queue_size(void) {
+  gpr_once_init(&s_init_max_accept_queue_size, init_max_accept_queue_size);
+  return s_max_accept_queue_size;
+}
+
+static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd,
+                                        const grpc_resolved_address *addr,
+                                        unsigned port_index, unsigned fd_index,
+                                        grpc_tcp_listener **listener) {
+  grpc_tcp_listener *sp = NULL;
+  int port = -1;
+  char *addr_str;
+  char *name;
+
+  grpc_error *err =
+      grpc_tcp_server_prepare_socket(fd, addr, s->so_reuseport, &port);
+  if (err == GRPC_ERROR_NONE) {
+    GPR_ASSERT(port > 0);
+    grpc_sockaddr_to_string(&addr_str, addr, 1);
+    gpr_asprintf(&name, "tcp-server-listener:%s", addr_str);
+    gpr_mu_lock(&s->mu);
+    s->nports++;
+    GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server");
+    sp = gpr_malloc(sizeof(grpc_tcp_listener));
+    sp->next = NULL;
+    if (s->head == NULL) {
+      s->head = sp;
+    } else {
+      s->tail->next = sp;
+    }
+    s->tail = sp;
+    sp->server = s;
+    sp->fd = fd;
+    sp->emfd = grpc_fd_create(fd, name);
+    memcpy(&sp->addr, addr, sizeof(grpc_resolved_address));
+    sp->port = port;
+    sp->port_index = port_index;
+    sp->fd_index = fd_index;
+    sp->is_sibling = 0;
+    sp->sibling = NULL;
+    GPR_ASSERT(sp->emfd);
+    gpr_mu_unlock(&s->mu);
+    gpr_free(addr_str);
+    gpr_free(name);
+  }
+
+  *listener = sp;
+  return err;
+}
+
+/* If successful, add a listener to s for addr, set *dsmode for the socket, and
+   return the *listener. */
+grpc_error *grpc_tcp_server_add_addr(grpc_tcp_server *s,
+                                     const grpc_resolved_address *addr,
+                                     unsigned port_index, unsigned fd_index,
+                                     grpc_dualstack_mode *dsmode,
+                                     grpc_tcp_listener **listener) {
+  grpc_resolved_address addr4_copy;
+  int fd;
+  grpc_error *err =
+      grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, dsmode, &fd);
+  if (err != GRPC_ERROR_NONE) {
+    return err;
+  }
+  if (*dsmode == GRPC_DSMODE_IPV4 &&
+      grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) {
+    addr = &addr4_copy;
+  }
+  return add_socket_to_server(s, fd, addr, port_index, fd_index, listener);
+}
+
+/* Prepare a recently-created socket for listening. */
+grpc_error *grpc_tcp_server_prepare_socket(int fd,
+                                           const grpc_resolved_address *addr,
+                                           bool so_reuseport, int *port) {
+  grpc_resolved_address sockname_temp;
+  grpc_error *err = GRPC_ERROR_NONE;
+
+  GPR_ASSERT(fd >= 0);
+
+  if (so_reuseport && !grpc_is_unix_socket(addr)) {
+    err = grpc_set_socket_reuse_port(fd, 1);
+    if (err != GRPC_ERROR_NONE) goto error;
+  }
+
+  err = grpc_set_socket_nonblocking(fd, 1);
+  if (err != GRPC_ERROR_NONE) goto error;
+  err = grpc_set_socket_cloexec(fd, 1);
+  if (err != GRPC_ERROR_NONE) goto error;
+  if (!grpc_is_unix_socket(addr)) {
+    err = grpc_set_socket_low_latency(fd, 1);
+    if (err != GRPC_ERROR_NONE) goto error;
+    err = grpc_set_socket_reuse_addr(fd, 1);
+    if (err != GRPC_ERROR_NONE) goto error;
+  }
+  err = grpc_set_socket_no_sigpipe_if_possible(fd);
+  if (err != GRPC_ERROR_NONE) goto error;
+
+  GPR_ASSERT(addr->len < ~(socklen_t)0);
+  if (bind(fd, (struct sockaddr *)addr->addr, (socklen_t)addr->len) < 0) {
+    err = GRPC_OS_ERROR(errno, "bind");
+    goto error;
+  }
+
+  if (listen(fd, get_max_accept_queue_size()) < 0) {
+    err = GRPC_OS_ERROR(errno, "listen");
+    goto error;
+  }
+
+  sockname_temp.len = sizeof(struct sockaddr_storage);
+
+  if (getsockname(fd, (struct sockaddr *)sockname_temp.addr,
+                  (socklen_t *)&sockname_temp.len) < 0) {
+    err = GRPC_OS_ERROR(errno, "getsockname");
+    goto error;
+  }
+
+  *port = grpc_sockaddr_get_port(&sockname_temp);
+  return GRPC_ERROR_NONE;
+
+error:
+  GPR_ASSERT(err != GRPC_ERROR_NONE);
+  if (fd >= 0) {
+    close(fd);
+  }
+  grpc_error *ret = grpc_error_set_int(
+      GRPC_ERROR_CREATE_REFERENCING("Unable to configure socket", &err, 1),
+      GRPC_ERROR_INT_FD, fd);
+  GRPC_ERROR_UNREF(err);
+  return ret;
+}
+
+#endif /* GRPC_HAVE_IFADDRS */
diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.c b/src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.c
new file mode 100644
index 0000000000..6354a6bdc1
--- /dev/null
+++ b/src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.c
@@ -0,0 +1,195 @@
+/*
+ *
+ * Copyright 2017, 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 "src/core/lib/iomgr/port.h"
+
+#ifdef GRPC_HAVE_IFADDRS
+
+#include "src/core/lib/iomgr/tcp_server_utils_posix.h"
+
+#include <errno.h>
+#include <ifaddrs.h>
+#include <stddef.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+
+#include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/iomgr/sockaddr.h"
+#include "src/core/lib/iomgr/sockaddr_utils.h"
+
+/* Return the listener in s with address addr or NULL. */
+static grpc_tcp_listener *find_listener_with_addr(grpc_tcp_server *s,
+                                                  grpc_resolved_address *addr) {
+  grpc_tcp_listener *l;
+  gpr_mu_lock(&s->mu);
+  for (l = s->head; l != NULL; l = l->next) {
+    if (l->addr.len != addr->len) {
+      continue;
+    }
+    if (memcmp(l->addr.addr, addr->addr, addr->len) == 0) {
+      break;
+    }
+  }
+  gpr_mu_unlock(&s->mu);
+  return l;
+}
+
+/* Bind to "::" to get a port number not used by any address. */
+static grpc_error *get_unused_port(int *port) {
+  grpc_resolved_address wild;
+  grpc_sockaddr_make_wildcard6(0, &wild);
+  grpc_dualstack_mode dsmode;
+  int fd;
+  grpc_error *err =
+      grpc_create_dualstack_socket(&wild, SOCK_STREAM, 0, &dsmode, &fd);
+  if (err != GRPC_ERROR_NONE) {
+    return err;
+  }
+  if (dsmode == GRPC_DSMODE_IPV4) {
+    grpc_sockaddr_make_wildcard4(0, &wild);
+  }
+  if (bind(fd, (const struct sockaddr *)wild.addr, (socklen_t)wild.len) != 0) {
+    err = GRPC_OS_ERROR(errno, "bind");
+    close(fd);
+    return err;
+  }
+  if (getsockname(fd, (struct sockaddr *)wild.addr, (socklen_t *)&wild.len) !=
+      0) {
+    err = GRPC_OS_ERROR(errno, "getsockname");
+    close(fd);
+    return err;
+  }
+  close(fd);
+  *port = grpc_sockaddr_get_port(&wild);
+  return *port <= 0 ? GRPC_ERROR_CREATE("Bad port") : GRPC_ERROR_NONE;
+}
+
+grpc_error *grpc_tcp_server_add_all_local_addrs(grpc_tcp_server *s,
+                                                unsigned port_index,
+                                                int requested_port,
+                                                int *out_port) {
+  struct ifaddrs *ifa = NULL;
+  struct ifaddrs *ifa_it;
+  unsigned fd_index = 0;
+  grpc_tcp_listener *sp = NULL;
+  grpc_error *err = GRPC_ERROR_NONE;
+  if (requested_port == 0) {
+    /* Note: There could be a race where some local addrs can listen on the
+       selected port and some can't. The sane way to handle this would be to
+       retry by recreating the whole grpc_tcp_server. Backing out individual
+       listeners and orphaning the FDs looks like too much trouble. */
+    if ((err = get_unused_port(&requested_port)) != GRPC_ERROR_NONE) {
+      return err;
+    } else if (requested_port <= 0) {
+      return GRPC_ERROR_CREATE("Bad get_unused_port()");
+    }
+    gpr_log(GPR_DEBUG, "Picked unused port %d", requested_port);
+  }
+  if (getifaddrs(&ifa) != 0 || ifa == NULL) {
+    return GRPC_OS_ERROR(errno, "getifaddrs");
+  }
+  for (ifa_it = ifa; ifa_it != NULL; ifa_it = ifa_it->ifa_next) {
+    grpc_resolved_address addr;
+    char *addr_str = NULL;
+    grpc_dualstack_mode dsmode;
+    grpc_tcp_listener *new_sp = NULL;
+    const char *ifa_name = (ifa_it->ifa_name ? ifa_it->ifa_name : "<unknown>");
+    if (ifa_it->ifa_addr == NULL) {
+      continue;
+    } else if (ifa_it->ifa_addr->sa_family == AF_INET) {
+      addr.len = sizeof(struct sockaddr_in);
+    } else if (ifa_it->ifa_addr->sa_family == AF_INET6) {
+      addr.len = sizeof(struct sockaddr_in6);
+    } else {
+      continue;
+    }
+    memcpy(addr.addr, ifa_it->ifa_addr, addr.len);
+    if (!grpc_sockaddr_set_port(&addr, requested_port)) {
+      /* Should never happen, because we check sa_family above. */
+      err = GRPC_ERROR_CREATE("Failed to set port");
+      break;
+    }
+    if (grpc_sockaddr_to_string(&addr_str, &addr, 0) < 0) {
+      addr_str = gpr_strdup("<error>");
+    }
+    gpr_log(GPR_DEBUG,
+            "Adding local addr from interface %s flags 0x%x to server: %s",
+            ifa_name, ifa_it->ifa_flags, addr_str);
+    /* We could have multiple interfaces with the same address (e.g., bonding),
+       so look for duplicates. */
+    if (find_listener_with_addr(s, &addr) != NULL) {
+      gpr_log(GPR_DEBUG, "Skipping duplicate addr %s on interface %s", addr_str,
+              ifa_name);
+      gpr_free(addr_str);
+      continue;
+    }
+    if ((err = grpc_tcp_server_add_addr(s, &addr, port_index, fd_index, &dsmode,
+                                        &new_sp)) != GRPC_ERROR_NONE) {
+      char *err_str = NULL;
+      grpc_error *root_err;
+      if (gpr_asprintf(&err_str, "Failed to add listener: %s", addr_str) < 0) {
+        err_str = gpr_strdup("Failed to add listener");
+      }
+      root_err = GRPC_ERROR_CREATE(err_str);
+      gpr_free(err_str);
+      gpr_free(addr_str);
+      err = grpc_error_add_child(root_err, err);
+      break;
+    } else {
+      GPR_ASSERT(requested_port == new_sp->port);
+      ++fd_index;
+      if (sp != NULL) {
+        new_sp->is_sibling = 1;
+        sp->sibling = new_sp;
+      }
+      sp = new_sp;
+    }
+    gpr_free(addr_str);
+  }
+  freeifaddrs(ifa);
+  if (err != GRPC_ERROR_NONE) {
+    return err;
+  } else if (sp == NULL) {
+    return GRPC_ERROR_CREATE("No local addresses");
+  } else {
+    *out_port = sp->port;
+    return GRPC_ERROR_NONE;
+  }
+}
+
+bool grpc_tcp_server_have_ifaddrs(void) { return true; }
+
+#endif /* GRPC_HAVE_IFADDRS */
diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.c b/src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.c
new file mode 100644
index 0000000000..95c3198be6
--- /dev/null
+++ b/src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.c
@@ -0,0 +1,49 @@
+/*
+ *
+ * Copyright 2017, 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 "src/core/lib/iomgr/port.h"
+
+#if defined(GRPC_POSIX_SOCKET) && !defined(GRPC_HAVE_IFADDRS)
+
+#include "src/core/lib/iomgr/tcp_server_utils_posix.h"
+
+grpc_error *grpc_tcp_server_add_all_local_addrs(grpc_tcp_server *s,
+                                                unsigned port_index,
+                                                int requested_port,
+                                                int *out_port) {
+  return GRPC_ERROR_CREATE("no ifaddrs available");
+}
+
+bool grpc_tcp_server_have_ifaddrs(void) { return false; }
+
+#endif /* defined(GRPC_POSIX_SOCKET) && !defined(GRPC_HAVE_IFADDRS) */
diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c
index d1bcd89af1..71e295770a 100644
--- a/src/core/lib/iomgr/udp_server.c
+++ b/src/core/lib/iomgr/udp_server.c
@@ -109,8 +109,8 @@ struct grpc_udp_server {
   grpc_pollset **pollsets;
   /* number of pollsets in the pollsets array */
   size_t pollset_count;
-  /* The parent grpc server */
-  grpc_server *grpc_server;
+  /* opaque object to pass to callbacks */
+  void *user_data;
 };
 
 grpc_udp_server *grpc_udp_server_create(void) {
@@ -178,7 +178,7 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
       /* Call the orphan_cb to signal that the FD is about to be closed and
        * should no longer be used. */
       GPR_ASSERT(sp->orphan_cb);
-      sp->orphan_cb(exec_ctx, sp->emfd);
+      sp->orphan_cb(exec_ctx, sp->emfd, sp->server->user_data);
 
       grpc_fd_orphan(exec_ctx, sp->emfd, &sp->destroyed_closure, NULL,
                      "udp_listener_shutdown");
@@ -204,7 +204,7 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
   if (s->active_ports) {
     for (sp = s->head; sp; sp = sp->next) {
       GPR_ASSERT(sp->orphan_cb);
-      sp->orphan_cb(exec_ctx, sp->emfd);
+      sp->orphan_cb(exec_ctx, sp->emfd, sp->server->user_data);
       grpc_fd_shutdown(exec_ctx, sp->emfd,
                        GRPC_ERROR_CREATE("Server destroyed"));
     }
@@ -299,7 +299,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
 
   /* Tell the registered callback that data is available to read. */
   GPR_ASSERT(sp->read_cb);
-  sp->read_cb(exec_ctx, sp->emfd, sp->server->grpc_server);
+  sp->read_cb(exec_ctx, sp->emfd, sp->server->user_data);
 
   /* Re-arm the notification event so we get another chance to read. */
   grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure);
@@ -322,7 +322,7 @@ static void on_write(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
 
   /* Tell the registered callback that the socket is writeable. */
   GPR_ASSERT(sp->write_cb);
-  sp->write_cb(exec_ctx, sp->emfd);
+  sp->write_cb(exec_ctx, sp->emfd, sp->server->user_data);
 
   /* Re-arm the notification event so we get another chance to write. */
   grpc_fd_notify_on_write(exec_ctx, sp->emfd, &sp->write_closure);
@@ -464,13 +464,13 @@ int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index) {
 
 void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
                            grpc_pollset **pollsets, size_t pollset_count,
-                           grpc_server *server) {
+                           void *user_data) {
   size_t i;
   gpr_mu_lock(&s->mu);
   grpc_udp_listener *sp;
   GPR_ASSERT(s->active_ports == 0);
   s->pollsets = pollsets;
-  s->grpc_server = server;
+  s->user_data = user_data;
 
   sp = s->head;
   while (sp != NULL) {
diff --git a/src/core/lib/iomgr/udp_server.h b/src/core/lib/iomgr/udp_server.h
index ed63fa7d81..90842a47f0 100644
--- a/src/core/lib/iomgr/udp_server.h
+++ b/src/core/lib/iomgr/udp_server.h
@@ -47,23 +47,23 @@ typedef struct grpc_udp_server grpc_udp_server;
 
 /* Called when data is available to read from the socket. */
 typedef void (*grpc_udp_server_read_cb)(grpc_exec_ctx *exec_ctx, grpc_fd *emfd,
-                                        struct grpc_server *server);
+                                        void *user_data);
 
 /* Called when the socket is writeable. */
-typedef void (*grpc_udp_server_write_cb)(grpc_exec_ctx *exec_ctx,
-                                         grpc_fd *emfd);
+typedef void (*grpc_udp_server_write_cb)(grpc_exec_ctx *exec_ctx, grpc_fd *emfd,
+                                         void *user_data);
 
 /* Called when the grpc_fd is about to be orphaned (and the FD closed). */
 typedef void (*grpc_udp_server_orphan_cb)(grpc_exec_ctx *exec_ctx,
-                                          grpc_fd *emfd);
+                                          grpc_fd *emfd, void *user_data);
 
 /* Create a server, initially not bound to any ports */
 grpc_udp_server *grpc_udp_server_create(void);
 
-/* Start listening to bound ports */
+/* Start listening to bound ports. user_data is passed to callbacks. */
 void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *udp_server,
                            grpc_pollset **pollsets, size_t pollset_count,
-                           struct grpc_server *server);
+                           void *user_data);
 
 int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index);
 
diff --git a/src/core/lib/security/transport/client_auth_filter.c b/src/core/lib/security/transport/client_auth_filter.c
index a23082a866..8dea1d98ff 100644
--- a/src/core/lib/security/transport/client_auth_filter.c
+++ b/src/core/lib/security/transport/client_auth_filter.c
@@ -318,7 +318,7 @@ static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx,
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {
+                              grpc_closure *ignored) {
   call_data *calld = elem->call_data;
   grpc_call_credentials_unref(exec_ctx, calld->creds);
   if (calld->have_host) {
diff --git a/src/core/lib/security/transport/server_auth_filter.c b/src/core/lib/security/transport/server_auth_filter.c
index 14619d97ca..01cb473177 100644
--- a/src/core/lib/security/transport/server_auth_filter.c
+++ b/src/core/lib/security/transport/server_auth_filter.c
@@ -227,7 +227,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {}
+                              grpc_closure *ignored) {}
 
 /* Constructor for channel_data */
 static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/lib/support/arena.c b/src/core/lib/support/arena.c
new file mode 100644
index 0000000000..7bcb983f24
--- /dev/null
+++ b/src/core/lib/support/arena.c
@@ -0,0 +1,98 @@
+/*
+ *
+ * Copyright 2017, 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 "src/core/lib/support/arena.h"
+#include <grpc/support/alloc.h>
+#include <grpc/support/atm.h>
+#include <grpc/support/log.h>
+#include <grpc/support/useful.h>
+
+#define ROUND_UP_TO_ALIGNMENT_SIZE(x) \
+  (((x) + GPR_MAX_ALIGNMENT - 1u) & ~(GPR_MAX_ALIGNMENT - 1u))
+
+typedef struct zone {
+  size_t size_begin;
+  size_t size_end;
+  gpr_atm next_atm;
+} zone;
+
+struct gpr_arena {
+  gpr_atm size_so_far;
+  zone initial_zone;
+};
+
+gpr_arena *gpr_arena_create(size_t initial_size) {
+  initial_size = ROUND_UP_TO_ALIGNMENT_SIZE(initial_size);
+  gpr_arena *a = gpr_zalloc(sizeof(gpr_arena) + initial_size);
+  a->initial_zone.size_end = initial_size;
+  return a;
+}
+
+size_t gpr_arena_destroy(gpr_arena *arena) {
+  gpr_atm size = gpr_atm_no_barrier_load(&arena->size_so_far);
+  zone *z = (zone *)gpr_atm_no_barrier_load(&arena->initial_zone.next_atm);
+  gpr_free(arena);
+  while (z) {
+    zone *next_z = (zone *)gpr_atm_no_barrier_load(&z->next_atm);
+    gpr_free(z);
+    z = next_z;
+  }
+  return (size_t)size;
+}
+
+void *gpr_arena_alloc(gpr_arena *arena, size_t size) {
+  size = ROUND_UP_TO_ALIGNMENT_SIZE(size);
+  size_t start =
+      (size_t)gpr_atm_no_barrier_fetch_add(&arena->size_so_far, size);
+  zone *z = &arena->initial_zone;
+  while (start > z->size_end) {
+    zone *next_z = (zone *)gpr_atm_acq_load(&z->next_atm);
+    if (next_z == NULL) {
+      size_t next_z_size = (size_t)gpr_atm_no_barrier_load(&arena->size_so_far);
+      next_z = gpr_zalloc(sizeof(zone) + next_z_size);
+      next_z->size_begin = z->size_end;
+      next_z->size_end = z->size_end + next_z_size;
+      if (!gpr_atm_rel_cas(&z->next_atm, (gpr_atm)NULL, (gpr_atm)next_z)) {
+        gpr_free(next_z);
+        next_z = (zone *)gpr_atm_acq_load(&z->next_atm);
+      }
+    }
+    z = next_z;
+  }
+  if (start + size > z->size_end) {
+    return gpr_arena_alloc(arena, size);
+  }
+  GPR_ASSERT(start >= z->size_begin);
+  GPR_ASSERT(start + size <= z->size_end);
+  return ((char *)(z + 1)) + start - z->size_begin;
+}
diff --git a/src/core/lib/support/arena.h b/src/core/lib/support/arena.h
new file mode 100644
index 0000000000..c28033ffc3
--- /dev/null
+++ b/src/core/lib/support/arena.h
@@ -0,0 +1,54 @@
+/*
+ *
+ * Copyright 2017, 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.
+ *
+ */
+
+// \file Arena based allocator
+// Allows very fast allocation of memory, but that memory cannot be freed until
+// the arena as a whole is freed
+// Tracks the total memory allocated against it, so that future arenas can
+// pre-allocate the right amount of memory
+
+#ifndef GRPC_CORE_LIB_SUPPORT_ARENA_H
+#define GRPC_CORE_LIB_SUPPORT_ARENA_H
+
+#include <stddef.h>
+
+typedef struct gpr_arena gpr_arena;
+
+// Create an arena, with \a initial_size bytes in the first allocated buffer
+gpr_arena *gpr_arena_create(size_t initial_size);
+// Allocate \a size bytes from the arena
+void *gpr_arena_alloc(gpr_arena *arena, size_t size);
+// Destroy an arena, returning the total number of bytes allocated
+size_t gpr_arena_destroy(gpr_arena *arena);
+
+#endif /* GRPC_CORE_LIB_SUPPORT_ARENA_H */
diff --git a/src/core/lib/support/atm.c b/src/core/lib/support/atm.c
new file mode 100644
index 0000000000..06e8432caf
--- /dev/null
+++ b/src/core/lib/support/atm.c
@@ -0,0 +1,47 @@
+/*
+ *
+ * Copyright 2017, 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/atm.h>
+#include <grpc/support/useful.h>
+
+gpr_atm gpr_atm_no_barrier_clamped_add(gpr_atm *value, gpr_atm delta,
+                                       gpr_atm min, gpr_atm max) {
+  gpr_atm current;
+  gpr_atm new;
+  do {
+    current = gpr_atm_no_barrier_load(value);
+    new = GPR_CLAMP(current + delta, min, max);
+    if (new == current) break;
+  } while (!gpr_atm_no_barrier_cas(value, current, new));
+  return new;
+}
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index c2547c5147..2c5d8c0ff3 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -51,6 +51,7 @@
 #include "src/core/lib/profiling/timers.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/slice/slice_string_helpers.h"
+#include "src/core/lib/support/arena.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/surface/api_trace.h"
 #include "src/core/lib/surface/call.h"
@@ -138,14 +139,15 @@ typedef struct batch_control {
 } batch_control;
 
 struct grpc_call {
+  gpr_arena *arena;
   grpc_completion_queue *cq;
   grpc_polling_entity pollent;
   grpc_channel *channel;
   grpc_call *parent;
   grpc_call *first_child;
   gpr_timespec start_time;
-  /* TODO(ctiller): share with cq if possible? */
-  gpr_mu mu;
+  /* protects first_child, and child next/prev links */
+  gpr_mu child_list_mu;
 
   /* client or server call */
   bool is_client;
@@ -160,8 +162,8 @@ struct grpc_call {
   bool received_initial_metadata;
   bool receiving_message;
   bool requested_final_op;
-  bool received_final_op;
-  bool sent_any_op;
+  gpr_atm any_ops_sent_atm;
+  gpr_atm received_final_op_atm;
 
   /* have we received initial metadata */
   bool has_initial_md_been_received;
@@ -212,6 +214,8 @@ struct grpc_call {
   grpc_closure receiving_initial_metadata_ready;
   uint32_t test_only_last_message_flags;
 
+  grpc_closure release_call;
+
   union {
     struct {
       grpc_status_code *status;
@@ -273,9 +277,13 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
       grpc_channel_get_channel_stack(args->channel);
   grpc_call *call;
   GPR_TIMER_BEGIN("grpc_call_create", 0);
-  call = gpr_zalloc(sizeof(grpc_call) + channel_stack->call_stack_size);
+  gpr_arena *arena =
+      gpr_arena_create(grpc_channel_get_call_size_estimate(args->channel));
+  call = gpr_arena_alloc(arena,
+                         sizeof(grpc_call) + channel_stack->call_stack_size);
+  call->arena = arena;
   *out_call = call;
-  gpr_mu_init(&call->mu);
+  gpr_mu_init(&call->child_list_mu);
   call->channel = args->channel;
   call->cq = args->cq;
   call->parent = args->parent_call;
@@ -313,7 +321,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
     GPR_ASSERT(call->is_client);
     GPR_ASSERT(!args->parent_call->is_client);
 
-    gpr_mu_lock(&args->parent_call->mu);
+    gpr_mu_lock(&args->parent_call->child_list_mu);
 
     if (args->propagation_mask & GRPC_PROPAGATE_DEADLINE) {
       send_deadline = gpr_time_min(
@@ -341,6 +349,10 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
     }
     if (args->propagation_mask & GRPC_PROPAGATE_CANCELLATION) {
       call->cancellation_is_inherited = 1;
+      if (gpr_atm_acq_load(&args->parent_call->received_final_op_atm)) {
+        cancel_with_error(exec_ctx, call, STATUS_FROM_API_OVERRIDE,
+                          GRPC_ERROR_CANCELLED);
+      }
     }
 
     if (args->parent_call->first_child == NULL) {
@@ -353,18 +365,23 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
           call;
     }
 
-    gpr_mu_unlock(&args->parent_call->mu);
+    gpr_mu_unlock(&args->parent_call->child_list_mu);
   }
 
   call->send_deadline = send_deadline;
 
   GRPC_CHANNEL_INTERNAL_REF(args->channel, "call");
   /* initial refcount dropped by grpc_call_destroy */
+  grpc_call_element_args call_args = {
+      .call_stack = CALL_STACK_FROM_CALL(call),
+      .server_transport_data = args->server_transport_data,
+      .context = call->context,
+      .path = path,
+      .start_time = call->start_time,
+      .deadline = send_deadline,
+      .arena = call->arena};
   add_init_error(&error, grpc_call_stack_init(exec_ctx, channel_stack, 1,
-                                              destroy_call, call, call->context,
-                                              args->server_transport_data, path,
-                                              call->start_time, send_deadline,
-                                              CALL_STACK_FROM_CALL(call)));
+                                              destroy_call, call, &call_args));
   if (error != GRPC_ERROR_NONE) {
     cancel_with_error(exec_ctx, call, STATUS_FROM_SURFACE,
                       GRPC_ERROR_REF(error));
@@ -421,6 +438,14 @@ void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c REF_ARG) {
   GRPC_CALL_STACK_UNREF(exec_ctx, CALL_STACK_FROM_CALL(c), REF_REASON);
 }
 
+static void release_call(grpc_exec_ctx *exec_ctx, void *call,
+                         grpc_error *error) {
+  grpc_call *c = call;
+  grpc_channel *channel = c->channel;
+  grpc_channel_update_call_size_estimate(channel, gpr_arena_destroy(c->arena));
+  GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel, "call");
+}
+
 static void set_status_value_directly(grpc_status_code status, void *dest);
 static void destroy_call(grpc_exec_ctx *exec_ctx, void *call,
                          grpc_error *error) {
@@ -435,7 +460,7 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call,
   if (c->receiving_stream != NULL) {
     grpc_byte_stream_destroy(exec_ctx, c->receiving_stream);
   }
-  gpr_mu_destroy(&c->mu);
+  gpr_mu_destroy(&c->child_list_mu);
   for (ii = 0; ii < c->send_extra_metadata_count; ii++) {
     GRPC_MDELEM_UNREF(exec_ctx, c->send_extra_metadata[ii].md);
   }
@@ -447,7 +472,6 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call,
   if (c->cq) {
     GRPC_CQ_INTERNAL_UNREF(c->cq, "bind");
   }
-  grpc_channel *channel = c->channel;
 
   get_final_status(call, set_status_value_directly, &c->final_info.final_status,
                    NULL);
@@ -456,11 +480,12 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call,
 
   for (i = 0; i < STATUS_SOURCE_COUNT; i++) {
     GRPC_ERROR_UNREF(
-        unpack_received_status(gpr_atm_no_barrier_load(&c->status[i])).error);
+        unpack_received_status(gpr_atm_acq_load(&c->status[i])).error);
   }
 
-  grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->final_info, c);
-  GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel, "call");
+  grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->final_info,
+                          grpc_closure_init(&c->release_call, release_call, c,
+                                            grpc_schedule_on_exec_ctx));
   GPR_TIMER_END("destroy_call", 0);
 }
 
@@ -473,7 +498,7 @@ void grpc_call_destroy(grpc_call *c) {
   GRPC_API_TRACE("grpc_call_destroy(c=%p)", 1, (c));
 
   if (parent) {
-    gpr_mu_lock(&parent->mu);
+    gpr_mu_lock(&parent->child_list_mu);
     if (c == parent->first_child) {
       parent->first_child = c->sibling_next;
       if (c == parent->first_child) {
@@ -482,15 +507,14 @@ void grpc_call_destroy(grpc_call *c) {
       c->sibling_prev->sibling_next = c->sibling_next;
       c->sibling_next->sibling_prev = c->sibling_prev;
     }
-    gpr_mu_unlock(&parent->mu);
+    gpr_mu_unlock(&parent->child_list_mu);
     GRPC_CALL_INTERNAL_UNREF(&exec_ctx, parent, "child");
   }
 
-  gpr_mu_lock(&c->mu);
   GPR_ASSERT(!c->destroy_called);
   c->destroy_called = 1;
-  cancel = c->sent_any_op && !c->received_final_op;
-  gpr_mu_unlock(&c->mu);
+  cancel = gpr_atm_acq_load(&c->any_ops_sent_atm) &&
+           !gpr_atm_acq_load(&c->received_final_op_atm);
   if (cancel) {
     cancel_with_error(&exec_ctx, c, STATUS_FROM_API_OVERRIDE,
                       GRPC_ERROR_CANCELLED);
@@ -555,53 +579,25 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
       "c=%p, status=%d, description=%s, reserved=%p)",
       4, (c, (int)status, description, reserved));
   GPR_ASSERT(reserved == NULL);
-  gpr_mu_lock(&c->mu);
   cancel_with_status(&exec_ctx, c, STATUS_FROM_API_OVERRIDE, status,
                      description);
-  gpr_mu_unlock(&c->mu);
   grpc_exec_ctx_finish(&exec_ctx);
   return GRPC_CALL_OK;
 }
 
-typedef struct termination_closure {
-  grpc_closure closure;
-  grpc_call *call;
-  grpc_transport_stream_op op;
-} termination_closure;
-
-static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp,
+static void done_termination(grpc_exec_ctx *exec_ctx, void *call,
                              grpc_error *error) {
-  termination_closure *tc = tcp;
-  GRPC_CALL_INTERNAL_UNREF(exec_ctx, tc->call, "termination");
-  gpr_free(tc);
-}
-
-static void send_termination(grpc_exec_ctx *exec_ctx, void *tcp,
-                             grpc_error *error) {
-  termination_closure *tc = tcp;
-  memset(&tc->op, 0, sizeof(tc->op));
-  tc->op.cancel_error = GRPC_ERROR_REF(error);
-  /* reuse closure to catch completion */
-  tc->op.on_complete = grpc_closure_init(&tc->closure, done_termination, tc,
-                                         grpc_schedule_on_exec_ctx);
-  execute_op(exec_ctx, tc->call, &tc->op);
-}
-
-static void terminate_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
-                                 grpc_error *error) {
-  termination_closure *tc = gpr_malloc(sizeof(*tc));
-  memset(tc, 0, sizeof(*tc));
-  tc->call = c;
-  GRPC_CALL_INTERNAL_REF(tc->call, "termination");
-  grpc_closure_sched(exec_ctx, grpc_closure_init(&tc->closure, send_termination,
-                                                 tc, grpc_schedule_on_exec_ctx),
-                     error);
+  GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "termination");
 }
 
 static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
                               status_source source, grpc_error *error) {
+  GRPC_CALL_INTERNAL_REF(c, "termination");
   set_status_from_error(exec_ctx, c, source, GRPC_ERROR_REF(error));
-  terminate_with_error(exec_ctx, c, error);
+  grpc_transport_stream_op *op = grpc_make_transport_stream_op(
+      grpc_closure_create(done_termination, c, grpc_schedule_on_exec_ctx));
+  op->cancel_error = error;
+  execute_op(exec_ctx, c, op);
 }
 
 static grpc_error *error_from_status(grpc_status_code status,
@@ -715,9 +711,7 @@ static void set_incoming_compression_algorithm(
 grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm(
     grpc_call *call) {
   grpc_compression_algorithm algorithm;
-  gpr_mu_lock(&call->mu);
   algorithm = call->incoming_compression_algorithm;
-  gpr_mu_unlock(&call->mu);
   return algorithm;
 }
 
@@ -729,9 +723,7 @@ static grpc_compression_algorithm compression_algorithm_for_level_locked(
 
 uint32_t grpc_call_test_only_get_message_flags(grpc_call *call) {
   uint32_t flags;
-  gpr_mu_lock(&call->mu);
   flags = call->test_only_last_message_flags;
-  gpr_mu_unlock(&call->mu);
   return flags;
 }
 
@@ -785,9 +777,7 @@ static void set_encodings_accepted_by_peer(grpc_exec_ctx *exec_ctx,
 
 uint32_t grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call) {
   uint32_t encodings_accepted_by_peer;
-  gpr_mu_lock(&call->mu);
   encodings_accepted_by_peer = call->encodings_accepted_by_peer;
-  gpr_mu_unlock(&call->mu);
   return encodings_accepted_by_peer;
 }
 
@@ -1056,7 +1046,7 @@ static void finish_batch_completion(grpc_exec_ctx *exec_ctx, void *user_data,
 }
 
 static grpc_error *consolidate_batch_errors(batch_control *bctl) {
-  size_t n = (size_t)gpr_atm_no_barrier_load(&bctl->num_errors);
+  size_t n = (size_t)gpr_atm_acq_load(&bctl->num_errors);
   if (n == 0) {
     return GRPC_ERROR_NONE;
   } else if (n == 1) {
@@ -1083,8 +1073,6 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
   grpc_call *call = bctl->call;
   grpc_error *error = consolidate_batch_errors(bctl);
 
-  gpr_mu_lock(&call->mu);
-
   if (bctl->send_initial_metadata) {
     grpc_metadata_batch_destroy(
         exec_ctx,
@@ -1103,20 +1091,23 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
         &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
     recv_trailing_filter(exec_ctx, call, md);
 
-    call->received_final_op = true;
     /* propagate cancellation to any interested children */
+    gpr_atm_rel_store(&call->received_final_op_atm, 1);
+    gpr_mu_lock(&call->child_list_mu);
     child_call = call->first_child;
     if (child_call != NULL) {
       do {
         next_child_call = child_call->sibling_next;
         if (child_call->cancellation_is_inherited) {
           GRPC_CALL_INTERNAL_REF(child_call, "propagate_cancel");
-          grpc_call_cancel(child_call, NULL);
+          cancel_with_error(exec_ctx, child_call, STATUS_FROM_API_OVERRIDE,
+                            GRPC_ERROR_CANCELLED);
           GRPC_CALL_INTERNAL_UNREF(exec_ctx, child_call, "propagate_cancel");
         }
         child_call = next_child_call;
       } while (child_call != call->first_child);
     }
+    gpr_mu_unlock(&call->child_list_mu);
 
     if (call->is_client) {
       get_final_status(call, set_status_value_directly,
@@ -1130,7 +1121,6 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
     GRPC_ERROR_UNREF(error);
     error = GRPC_ERROR_NONE;
   }
-  gpr_mu_unlock(&call->mu);
 
   if (bctl->is_notify_tag_closure) {
     /* unrefs bctl->error */
@@ -1221,7 +1211,6 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
                                    grpc_error *error) {
   batch_control *bctl = bctlp;
   grpc_call *call = bctl->call;
-  gpr_mu_lock(&bctl->call->mu);
   if (error != GRPC_ERROR_NONE) {
     if (call->receiving_stream != NULL) {
       grpc_byte_stream_destroy(exec_ctx, call->receiving_stream);
@@ -1233,11 +1222,9 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
   }
   if (call->has_initial_md_been_received || error != GRPC_ERROR_NONE ||
       call->receiving_stream == NULL) {
-    gpr_mu_unlock(&bctl->call->mu);
     process_data_after_md(exec_ctx, bctlp);
   } else {
     call->saved_receiving_stream_ready_bctlp = bctlp;
-    gpr_mu_unlock(&bctl->call->mu);
   }
 }
 
@@ -1296,7 +1283,7 @@ static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
 static void add_batch_error(grpc_exec_ctx *exec_ctx, batch_control *bctl,
                             grpc_error *error, bool has_cancelled) {
   if (error == GRPC_ERROR_NONE) return;
-  int idx = (int)gpr_atm_no_barrier_fetch_add(&bctl->num_errors, 1);
+  int idx = (int)gpr_atm_full_fetch_add(&bctl->num_errors, 1);
   if (idx == 0 && !has_cancelled) {
     cancel_with_error(exec_ctx, bctl->call, STATUS_FROM_CORE,
                       GRPC_ERROR_REF(error));
@@ -1309,8 +1296,6 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
   batch_control *bctl = bctlp;
   grpc_call *call = bctl->call;
 
-  gpr_mu_lock(&call->mu);
-
   add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), false);
   if (error == GRPC_ERROR_NONE) {
     grpc_metadata_batch *md =
@@ -1336,11 +1321,9 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
         receiving_stream_ready, call->saved_receiving_stream_ready_bctlp,
         grpc_schedule_on_exec_ctx);
     call->saved_receiving_stream_ready_bctlp = NULL;
-    grpc_closure_sched(exec_ctx, saved_rsr_closure, GRPC_ERROR_REF(error));
+    grpc_closure_run(exec_ctx, saved_rsr_closure, GRPC_ERROR_REF(error));
   }
 
-  gpr_mu_unlock(&call->mu);
-
   finish_batch_step(exec_ctx, bctl);
 }
 
@@ -1393,7 +1376,6 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
   bctl->notify_tag = notify_tag;
   bctl->is_notify_tag_closure = (uint8_t)(is_notify_tag_closure != 0);
 
-  gpr_mu_lock(&call->mu);
   grpc_transport_stream_op *stream_op = &bctl->op;
   memset(stream_op, 0, sizeof(*stream_op));
   stream_op->covered_by_poller = true;
@@ -1679,8 +1661,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
   grpc_closure_init(&bctl->finish_batch, finish_batch, bctl,
                     grpc_schedule_on_exec_ctx);
   stream_op->on_complete = &bctl->finish_batch;
-  call->sent_any_op = true;
-  gpr_mu_unlock(&call->mu);
+  gpr_atm_rel_store(&call->any_ops_sent_atm, 1);
 
   execute_op(exec_ctx, call, stream_op);
 
@@ -1711,7 +1692,6 @@ done_with_error:
   if (bctl->recv_final_op) {
     call->requested_final_op = 0;
   }
-  gpr_mu_unlock(&call->mu);
   goto done;
 }
 
@@ -1760,10 +1740,8 @@ uint8_t grpc_call_is_client(grpc_call *call) { return call->is_client; }
 
 grpc_compression_algorithm grpc_call_compression_for_level(
     grpc_call *call, grpc_compression_level level) {
-  gpr_mu_lock(&call->mu);
   grpc_compression_algorithm algo =
       compression_algorithm_for_level_locked(call, level);
-  gpr_mu_unlock(&call->mu);
   return algo;
 }
 
diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c
index d6acd392c1..2b700b2f67 100644
--- a/src/core/lib/surface/channel.c
+++ b/src/core/lib/surface/channel.c
@@ -68,6 +68,8 @@ struct grpc_channel {
   grpc_compression_options compression_options;
   grpc_mdelem default_authority;
 
+  gpr_atm call_size_estimate;
+
   gpr_mu registered_call_mu;
   registered_call *registered_calls;
 
@@ -115,6 +117,10 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
   gpr_mu_init(&channel->registered_call_mu);
   channel->registered_calls = NULL;
 
+  gpr_atm_no_barrier_store(
+      &channel->call_size_estimate,
+      (gpr_atm)CHANNEL_STACK_FROM_CHANNEL(channel)->call_stack_size);
+
   grpc_compression_options_init(&channel->compression_options);
   for (size_t i = 0; i < args->num_args; i++) {
     if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) {
@@ -177,6 +183,32 @@ done:
   return channel;
 }
 
+size_t grpc_channel_get_call_size_estimate(grpc_channel *channel) {
+#define ROUND_UP_SIZE 256
+  return ((size_t)gpr_atm_no_barrier_load(&channel->call_size_estimate) +
+          ROUND_UP_SIZE) &
+         ~(size_t)(ROUND_UP_SIZE - 1);
+}
+
+void grpc_channel_update_call_size_estimate(grpc_channel *channel,
+                                            size_t size) {
+  size_t cur = (size_t)gpr_atm_no_barrier_load(&channel->call_size_estimate);
+  if (cur < size) {
+    /* size grew: update estimate */
+    gpr_atm_no_barrier_cas(&channel->call_size_estimate, (gpr_atm)cur,
+                           (gpr_atm)size);
+    /* if we lose: never mind, something else will likely update soon enough */
+  } else if (cur == size) {
+    /* no change: holding pattern */
+  } else if (cur > 0) {
+    /* size shrank: decrease estimate */
+    gpr_atm_no_barrier_cas(
+        &channel->call_size_estimate, (gpr_atm)cur,
+        (gpr_atm)(GPR_MIN(cur - 1, (255 * cur + size) / 256)));
+    /* if we lose: never mind, something else will likely update soon enough */
+  }
+}
+
 char *grpc_channel_get_target(grpc_channel *channel) {
   GRPC_API_TRACE("grpc_channel_get_target(channel=%p)", 1, (channel));
   return gpr_strdup(channel->target);
diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h
index 3a441d7add..c4aebd8b9b 100644
--- a/src/core/lib/surface/channel.h
+++ b/src/core/lib/surface/channel.h
@@ -66,6 +66,9 @@ grpc_mdelem grpc_channel_get_reffed_status_elem(grpc_exec_ctx *exec_ctx,
                                                 grpc_channel *channel,
                                                 int status_code);
 
+size_t grpc_channel_get_call_size_estimate(grpc_channel *channel);
+void grpc_channel_update_call_size_estimate(grpc_channel *channel, size_t size);
+
 #ifdef GRPC_STREAM_REFCOUNT_DEBUG
 void grpc_channel_internal_ref(grpc_channel *channel, const char *reason);
 void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
diff --git a/src/core/lib/surface/lame_client.c b/src/core/lib/surface/lame_client.c
index 49bc4c114b..9ddb88bd11 100644
--- a/src/core/lib/surface/lame_client.c
+++ b/src/core/lib/surface/lame_client.c
@@ -130,8 +130,8 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *and_free_memory) {
-  gpr_free(and_free_memory);
+                              grpc_closure *then_schedule_closure) {
+  grpc_closure_sched(exec_ctx, then_schedule_closure, GRPC_ERROR_NONE);
 }
 
 static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index b360579553..1186a4af63 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -898,7 +898,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
 
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
-                              void *ignored) {
+                              grpc_closure *ignored) {
   channel_data *chand = elem->channel_data;
   call_data *calld = elem->call_data;
 
diff --git a/src/core/lib/surface/version.c b/src/core/lib/surface/version.c
index 1143a9e044..ba80bd801e 100644
--- a/src/core/lib/surface/version.c
+++ b/src/core/lib/surface/version.c
@@ -38,4 +38,4 @@
 
 const char *grpc_version_string(void) { return "3.0.0-dev"; }
 
-const char *grpc_g_stands_for(void) { return "green"; }
+const char *grpc_g_stands_for(void) { return "gentle"; }
diff --git a/src/core/lib/transport/service_config.c b/src/core/lib/transport/service_config.c
index 12da2a88fe..1195f75044 100644
--- a/src/core/lib/transport/service_config.c
+++ b/src/core/lib/transport/service_config.c
@@ -93,6 +93,18 @@ void grpc_service_config_destroy(grpc_service_config* service_config) {
   gpr_free(service_config);
 }
 
+void grpc_service_config_parse_global_params(
+    const grpc_service_config* service_config,
+    void (*process_json)(const grpc_json* json, void* arg), void* arg) {
+  const grpc_json* json = service_config->json_tree;
+  if (json->type != GRPC_JSON_OBJECT || json->key != NULL) return;
+  for (grpc_json* field = json->child; field != NULL; field = field->next) {
+    if (field->key == NULL) return;
+    if (strcmp(field->key, "methodConfig") == 0) continue;
+    process_json(field, arg);
+  }
+}
+
 const char* grpc_service_config_get_lb_policy_name(
     const grpc_service_config* service_config) {
   const grpc_json* json = service_config->json_tree;
diff --git a/src/core/lib/transport/service_config.h b/src/core/lib/transport/service_config.h
index cd739a593c..ebfc59b534 100644
--- a/src/core/lib/transport/service_config.h
+++ b/src/core/lib/transport/service_config.h
@@ -42,6 +42,12 @@ typedef struct grpc_service_config grpc_service_config;
 grpc_service_config* grpc_service_config_create(const char* json_string);
 void grpc_service_config_destroy(grpc_service_config* service_config);
 
+/// Invokes \a process_json() for each global parameter in the service
+/// config.  \a arg is passed as the second argument to \a process_json().
+void grpc_service_config_parse_global_params(
+    const grpc_service_config* service_config,
+    void (*process_json)(const grpc_json* json, void* arg), void* arg);
+
 /// Gets the LB policy name from \a service_config.
 /// Returns NULL if no LB policy name was specified.
 /// Caller does NOT take ownership.
diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c
index 165950e288..d56cb31ee0 100644
--- a/src/core/lib/transport/transport.c
+++ b/src/core/lib/transport/transport.c
@@ -162,9 +162,9 @@ void grpc_transport_destroy(grpc_exec_ctx *exec_ctx,
 int grpc_transport_init_stream(grpc_exec_ctx *exec_ctx,
                                grpc_transport *transport, grpc_stream *stream,
                                grpc_stream_refcount *refcount,
-                               const void *server_data) {
+                               const void *server_data, gpr_arena *arena) {
   return transport->vtable->init_stream(exec_ctx, transport, stream, refcount,
-                                        server_data);
+                                        server_data, arena);
 }
 
 void grpc_transport_perform_stream_op(grpc_exec_ctx *exec_ctx,
@@ -197,9 +197,10 @@ void grpc_transport_set_pops(grpc_exec_ctx *exec_ctx, grpc_transport *transport,
 
 void grpc_transport_destroy_stream(grpc_exec_ctx *exec_ctx,
                                    grpc_transport *transport,
-                                   grpc_stream *stream, void *and_free_memory) {
+                                   grpc_stream *stream,
+                                   grpc_closure *then_schedule_closure) {
   transport->vtable->destroy_stream(exec_ctx, transport, stream,
-                                    and_free_memory);
+                                    then_schedule_closure);
 }
 
 char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx,
@@ -254,8 +255,9 @@ typedef struct {
 static void destroy_made_transport_stream_op(grpc_exec_ctx *exec_ctx, void *arg,
                                              grpc_error *error) {
   made_transport_stream_op *op = arg;
-  grpc_closure_sched(exec_ctx, op->inner_on_complete, GRPC_ERROR_REF(error));
+  grpc_closure *c = op->inner_on_complete;
   gpr_free(op);
+  grpc_closure_run(exec_ctx, c, GRPC_ERROR_REF(error));
 }
 
 grpc_transport_stream_op *grpc_make_transport_stream_op(
diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h
index cc1c277b35..950b18aeda 100644
--- a/src/core/lib/transport/transport.h
+++ b/src/core/lib/transport/transport.h
@@ -41,6 +41,7 @@
 #include "src/core/lib/iomgr/polling_entity.h"
 #include "src/core/lib/iomgr/pollset.h"
 #include "src/core/lib/iomgr/pollset_set.h"
+#include "src/core/lib/support/arena.h"
 #include "src/core/lib/transport/byte_stream.h"
 #include "src/core/lib/transport/metadata_batch.h"
 
@@ -229,7 +230,7 @@ size_t grpc_transport_stream_size(grpc_transport *transport);
 int grpc_transport_init_stream(grpc_exec_ctx *exec_ctx,
                                grpc_transport *transport, grpc_stream *stream,
                                grpc_stream_refcount *refcount,
-                               const void *server_data);
+                               const void *server_data, gpr_arena *arena);
 
 void grpc_transport_set_pops(grpc_exec_ctx *exec_ctx, grpc_transport *transport,
                              grpc_stream *stream, grpc_polling_entity *pollent);
@@ -246,7 +247,8 @@ void grpc_transport_set_pops(grpc_exec_ctx *exec_ctx, grpc_transport *transport,
                  caller, but any child memory must be cleaned up) */
 void grpc_transport_destroy_stream(grpc_exec_ctx *exec_ctx,
                                    grpc_transport *transport,
-                                   grpc_stream *stream, void *and_free_memory);
+                                   grpc_stream *stream,
+                                   grpc_closure *then_schedule_closure);
 
 void grpc_transport_stream_op_finish_with_failure(grpc_exec_ctx *exec_ctx,
                                                   grpc_transport_stream_op *op,
diff --git a/src/core/lib/transport/transport_impl.h b/src/core/lib/transport/transport_impl.h
index 8553148c35..6f688bf8d2 100644
--- a/src/core/lib/transport/transport_impl.h
+++ b/src/core/lib/transport/transport_impl.h
@@ -47,7 +47,7 @@ typedef struct grpc_transport_vtable {
   /* implementation of grpc_transport_init_stream */
   int (*init_stream)(grpc_exec_ctx *exec_ctx, grpc_transport *self,
                      grpc_stream *stream, grpc_stream_refcount *refcount,
-                     const void *server_data);
+                     const void *server_data, gpr_arena *arena);
 
   /* implementation of grpc_transport_set_pollset */
   void (*set_pollset)(grpc_exec_ctx *exec_ctx, grpc_transport *self,
@@ -67,7 +67,8 @@ typedef struct grpc_transport_vtable {
 
   /* implementation of grpc_transport_destroy_stream */
   void (*destroy_stream)(grpc_exec_ctx *exec_ctx, grpc_transport *self,
-                         grpc_stream *stream, void *and_free_memory);
+                         grpc_stream *stream,
+                         grpc_closure *then_schedule_closure);
 
   /* implementation of grpc_transport_destroy */
   void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_transport *self);
diff --git a/src/cpp/common/channel_arguments.cc b/src/cpp/common/channel_arguments.cc
index 65f3277499..eddcacc332 100644
--- a/src/cpp/common/channel_arguments.cc
+++ b/src/cpp/common/channel_arguments.cc
@@ -81,6 +81,16 @@ ChannelArguments::ChannelArguments(const ChannelArguments& other)
   }
 }
 
+ChannelArguments::~ChannelArguments() {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  for (auto it = args_.begin(); it != args_.end(); ++it) {
+    if (it->type == GRPC_ARG_POINTER) {
+      it->value.pointer.vtable->destroy(&exec_ctx, it->value.pointer.p);
+    }
+  }
+  grpc_exec_ctx_finish(&exec_ctx);
+}
+
 void ChannelArguments::Swap(ChannelArguments& other) {
   args_.swap(other.args_);
   strings_.swap(other.strings_);
@@ -101,8 +111,10 @@ void ChannelArguments::SetSocketMutator(grpc_socket_mutator* mutator) {
   for (auto it = args_.begin(); it != args_.end(); ++it) {
     if (it->type == mutator_arg.type &&
         grpc::string(it->key) == grpc::string(mutator_arg.key)) {
+      GPR_ASSERT(!replaced);
       it->value.pointer.vtable->destroy(&exec_ctx, it->value.pointer.p);
       it->value.pointer = mutator_arg.value.pointer;
+      replaced = true;
     }
   }
   grpc_exec_ctx_finish(&exec_ctx);
@@ -185,7 +197,7 @@ void ChannelArguments::SetPointerWithVtable(
   arg.type = GRPC_ARG_POINTER;
   strings_.push_back(key);
   arg.key = const_cast<char*>(strings_.back().c_str());
-  arg.value.pointer.p = value;
+  arg.value.pointer.p = vtable->copy(value);
   arg.value.pointer.vtable = vtable;
   args_.push_back(arg);
 }
diff --git a/src/cpp/common/channel_filter.h b/src/cpp/common/channel_filter.h
index 79c4bab985..494d5d64d7 100644
--- a/src/cpp/common/channel_filter.h
+++ b/src/cpp/common/channel_filter.h
@@ -318,7 +318,8 @@ class ChannelFilter final {
   static void DestroyCallElement(grpc_exec_ctx *exec_ctx,
                                  grpc_call_element *elem,
                                  const grpc_call_final_info *final_info,
-                                 void *and_free_memory) {
+                                 grpc_closure *then_call_closure) {
+    GPR_ASSERT(then_call_closure == NULL);
     reinterpret_cast<CallDataType *>(elem->call_data)->~CallDataType();
   }
 
diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py
index a9f20e6d2a..8edab1b759 100644
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@@ -33,6 +33,8 @@ CORE_SOURCE_FILES = [
   'src/core/lib/profiling/basic_timers.c',
   'src/core/lib/profiling/stap_timers.c',
   'src/core/lib/support/alloc.c',
+  'src/core/lib/support/arena.c',
+  'src/core/lib/support/atm.c',
   'src/core/lib/support/avl.c',
   'src/core/lib/support/backoff.c',
   'src/core/lib/support/cmdline.c',
@@ -134,6 +136,9 @@ CORE_SOURCE_FILES = [
   'src/core/lib/iomgr/tcp_client_windows.c',
   'src/core/lib/iomgr/tcp_posix.c',
   'src/core/lib/iomgr/tcp_server_posix.c',
+  'src/core/lib/iomgr/tcp_server_utils_posix_common.c',
+  'src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.c',
+  'src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.c',
   'src/core/lib/iomgr/tcp_server_uv.c',
   'src/core/lib/iomgr/tcp_server_windows.c',
   'src/core/lib/iomgr/tcp_uv.c',
@@ -263,6 +268,7 @@ CORE_SOURCE_FILES = [
   'src/core/ext/client_channel/resolver.c',
   'src/core/ext/client_channel/resolver_factory.c',
   'src/core/ext/client_channel/resolver_registry.c',
+  'src/core/ext/client_channel/retry_throttle.c',
   'src/core/ext/client_channel/subchannel.c',
   'src/core/ext/client_channel/subchannel_index.c',
   'src/core/ext/client_channel/uri_parser.c',
diff --git a/src/python/grpcio_health_checking/grpc_health/v1/health.py b/src/python/grpcio_health_checking/grpc_health/v1/health.py
index f0f11cf84b..e92c2659b7 100644
--- a/src/python/grpcio_health_checking/grpc_health/v1/health.py
+++ b/src/python/grpcio_health_checking/grpc_health/v1/health.py
@@ -33,9 +33,10 @@ import threading
 import grpc
 
 from grpc_health.v1 import health_pb2
+from grpc_health.v1 import health_pb2_grpc
 
 
-class HealthServicer(health_pb2.HealthServicer):
+class HealthServicer(health_pb2_grpc.HealthServicer):
     """Servicer handling RPCs for service statuses."""
 
     def __init__(self):
diff --git a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py
index 363b4c5f99..1bc8669dad 100644
--- a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py
+++ b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py
@@ -34,6 +34,7 @@ import grpc
 from grpc.framework.foundation import logging_pool
 from grpc_health.v1 import health
 from grpc_health.v1 import health_pb2
+from grpc_health.v1 import health_pb2_grpc
 
 from tests.unit.framework.common import test_constants
 
@@ -52,11 +53,11 @@ class HealthServicerTest(unittest.TestCase):
         server_pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY)
         self._server = grpc.server(server_pool)
         port = self._server.add_insecure_port('[::]:0')
-        health_pb2.add_HealthServicer_to_server(servicer, self._server)
+        health_pb2_grpc.add_HealthServicer_to_server(servicer, self._server)
         self._server.start()
 
         channel = grpc.insecure_channel('localhost:%d' % port)
-        self._stub = health_pb2.HealthStub(channel)
+        self._stub = health_pb2_grpc.HealthStub(channel)
 
     def test_empty_service(self):
         request = health_pb2.HealthCheckRequest()
diff --git a/src/python/grpcio_tests/tests/interop/_insecure_intraop_test.py b/src/python/grpcio_tests/tests/interop/_insecure_intraop_test.py
index 58f3b364ba..3325d54375 100644
--- a/src/python/grpcio_tests/tests/interop/_insecure_intraop_test.py
+++ b/src/python/grpcio_tests/tests/interop/_insecure_intraop_test.py
@@ -32,7 +32,7 @@ from concurrent import futures
 import unittest
 
 import grpc
-from src.proto.grpc.testing import test_pb2
+from src.proto.grpc.testing import test_pb2_grpc
 
 from tests.interop import _intraop_test_case
 from tests.interop import methods
@@ -44,11 +44,11 @@ class InsecureIntraopTest(_intraop_test_case.IntraopTestCase,
 
     def setUp(self):
         self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
-        test_pb2.add_TestServiceServicer_to_server(methods.TestService(),
-                                                   self.server)
+        test_pb2_grpc.add_TestServiceServicer_to_server(methods.TestService(),
+                                                        self.server)
         port = self.server.add_insecure_port('[::]:0')
         self.server.start()
-        self.stub = test_pb2.TestServiceStub(
+        self.stub = test_pb2_grpc.TestServiceStub(
             grpc.insecure_channel('localhost:{}'.format(port)))
 
 
diff --git a/src/python/grpcio_tests/tests/interop/_secure_intraop_test.py b/src/python/grpcio_tests/tests/interop/_secure_intraop_test.py
index 5fe929b99e..857e00efb3 100644
--- a/src/python/grpcio_tests/tests/interop/_secure_intraop_test.py
+++ b/src/python/grpcio_tests/tests/interop/_secure_intraop_test.py
@@ -32,7 +32,7 @@ from concurrent import futures
 import unittest
 
 import grpc
-from src.proto.grpc.testing import test_pb2
+from src.proto.grpc.testing import test_pb2_grpc
 
 from tests.interop import _intraop_test_case
 from tests.interop import methods
@@ -45,14 +45,14 @@ class SecureIntraopTest(_intraop_test_case.IntraopTestCase, unittest.TestCase):
 
     def setUp(self):
         self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
-        test_pb2.add_TestServiceServicer_to_server(methods.TestService(),
-                                                   self.server)
+        test_pb2_grpc.add_TestServiceServicer_to_server(methods.TestService(),
+                                                        self.server)
         port = self.server.add_secure_port(
             '[::]:0',
             grpc.ssl_server_credentials(
                 [(resources.private_key(), resources.certificate_chain())]))
         self.server.start()
-        self.stub = test_pb2.TestServiceStub(
+        self.stub = test_pb2_grpc.TestServiceStub(
             grpc.secure_channel('localhost:{}'.format(port),
                                 grpc.ssl_channel_credentials(
                                     resources.test_root_certificates()), (
diff --git a/src/python/grpcio_tests/tests/interop/methods.py b/src/python/grpcio_tests/tests/interop/methods.py
index 662ea9ce57..e1016f7c0d 100644
--- a/src/python/grpcio_tests/tests/interop/methods.py
+++ b/src/python/grpcio_tests/tests/interop/methods.py
@@ -40,7 +40,7 @@ from grpc.beta import implementations
 
 from src.proto.grpc.testing import empty_pb2
 from src.proto.grpc.testing import messages_pb2
-from src.proto.grpc.testing import test_pb2
+from src.proto.grpc.testing import test_pb2_grpc
 
 _INITIAL_METADATA_KEY = "x-grpc-test-echo-initial"
 _TRAILING_METADATA_KEY = "x-grpc-test-echo-trailing-bin"
@@ -66,7 +66,7 @@ def _maybe_echo_status_and_message(request, servicer_context):
         servicer_context.set_details(request.response_status.message)
 
 
-class TestService(test_pb2.TestServiceServicer):
+class TestService(test_pb2_grpc.TestServiceServicer):
 
     def EmptyCall(self, request, context):
         _maybe_echo_metadata(context)
diff --git a/src/python/grpcio_tests/tests/interop/server.py b/src/python/grpcio_tests/tests/interop/server.py
index 65f1604eb8..0ae2c97b42 100644
--- a/src/python/grpcio_tests/tests/interop/server.py
+++ b/src/python/grpcio_tests/tests/interop/server.py
@@ -34,7 +34,7 @@ import logging
 import time
 
 import grpc
-from src.proto.grpc.testing import test_pb2
+from src.proto.grpc.testing import test_pb2_grpc
 
 from tests.interop import methods
 from tests.interop import resources
@@ -53,7 +53,8 @@ def serve():
     args = parser.parse_args()
 
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
-    test_pb2.add_TestServiceServicer_to_server(methods.TestService(), server)
+    test_pb2_grpc.add_TestServiceServicer_to_server(methods.TestService(),
+                                                    server)
     if args.use_tls:
         private_key = resources.private_key()
         certificate_chain = resources.certificate_chain()
diff --git a/src/python/grpcio_tests/tests/qps/qps_worker.py b/src/python/grpcio_tests/tests/qps/qps_worker.py
index 025dfb9d4a..7cd53e7bd9 100644
--- a/src/python/grpcio_tests/tests/qps/qps_worker.py
+++ b/src/python/grpcio_tests/tests/qps/qps_worker.py
@@ -33,7 +33,7 @@ import time
 
 from concurrent import futures
 import grpc
-from src.proto.grpc.testing import services_pb2
+from src.proto.grpc.testing import services_pb2_grpc
 
 from tests.qps import worker_server
 
@@ -41,7 +41,7 @@ from tests.qps import worker_server
 def run_worker_server(port):
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=5))
     servicer = worker_server.WorkerServer()
-    services_pb2.add_WorkerServiceServicer_to_server(servicer, server)
+    services_pb2_grpc.add_WorkerServiceServicer_to_server(servicer, server)
     server.add_insecure_port('[::]:{}'.format(port))
     server.start()
     servicer.wait_for_quit()
diff --git a/src/python/grpcio_tests/tests/qps/worker_server.py b/src/python/grpcio_tests/tests/qps/worker_server.py
index ca1a777611..de9535f46e 100644
--- a/src/python/grpcio_tests/tests/qps/worker_server.py
+++ b/src/python/grpcio_tests/tests/qps/worker_server.py
@@ -35,7 +35,7 @@ import time
 from concurrent import futures
 import grpc
 from src.proto.grpc.testing import control_pb2
-from src.proto.grpc.testing import services_pb2
+from src.proto.grpc.testing import services_pb2_grpc
 from src.proto.grpc.testing import stats_pb2
 
 from tests.qps import benchmark_client
@@ -45,7 +45,7 @@ from tests.qps import histogram
 from tests.unit import resources
 
 
-class WorkerServer(services_pb2.WorkerServiceServicer):
+class WorkerServer(services_pb2_grpc.WorkerServiceServicer):
     """Python Worker Server implementation."""
 
     def __init__(self):
@@ -87,8 +87,8 @@ class WorkerServer(services_pb2.WorkerServiceServicer):
             futures.ThreadPoolExecutor(max_workers=server_threads))
         if config.server_type == control_pb2.ASYNC_SERVER:
             servicer = benchmark_server.BenchmarkServer()
-            services_pb2.add_BenchmarkServiceServicer_to_server(servicer,
-                                                                server)
+            services_pb2_grpc.add_BenchmarkServiceServicer_to_server(servicer,
+                                                                     server)
         elif config.server_type == control_pb2.ASYNC_GENERIC_SERVER:
             resp_size = config.payload_config.bytebuf_params.resp_size
             servicer = benchmark_server.GenericBenchmarkServer(resp_size)
diff --git a/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py b/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py
index d06ff064e2..4d73be6204 100644
--- a/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py
+++ b/src/python/grpcio_tests/tests/reflection/_reflection_servicer_test.py
@@ -34,6 +34,7 @@ import grpc
 from grpc.framework.foundation import logging_pool
 from grpc_reflection.v1alpha import reflection
 from grpc_reflection.v1alpha import reflection_pb2
+from grpc_reflection.v1alpha import reflection_pb2_grpc
 
 from google.protobuf import descriptor_pool
 from google.protobuf import descriptor_pb2
@@ -61,12 +62,12 @@ class ReflectionServicerTest(unittest.TestCase):
         server_pool = logging_pool.pool(test_constants.THREAD_CONCURRENCY)
         self._server = grpc.server(server_pool)
         port = self._server.add_insecure_port('[::]:0')
-        reflection_pb2.add_ServerReflectionServicer_to_server(servicer,
-                                                              self._server)
+        reflection_pb2_grpc.add_ServerReflectionServicer_to_server(servicer,
+                                                                   self._server)
         self._server.start()
 
         channel = grpc.insecure_channel('localhost:%d' % port)
-        self._stub = reflection_pb2.ServerReflectionStub(channel)
+        self._stub = reflection_pb2_grpc.ServerReflectionStub(channel)
 
     def testFileByName(self):
         requests = (reflection_pb2.ServerReflectionRequest(
diff --git a/src/python/grpcio_tests/tests/stress/client.py b/src/python/grpcio_tests/tests/stress/client.py
index b9dbe61d44..b7eb12bff8 100644
--- a/src/python/grpcio_tests/tests/stress/client.py
+++ b/src/python/grpcio_tests/tests/stress/client.py
@@ -34,7 +34,7 @@ import threading
 
 import grpc
 from six.moves import queue
-from src.proto.grpc.testing import metrics_pb2
+from src.proto.grpc.testing import metrics_pb2_grpc
 from src.proto.grpc.testing import test_pb2
 
 from tests.interop import methods
@@ -139,7 +139,7 @@ def run_test(args):
     runners = []
 
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=25))
-    metrics_pb2.add_MetricsServiceServicer_to_server(
+    metrics_pb2_grpc.add_MetricsServiceServicer_to_server(
         metrics_server.MetricsServer(hist), server)
     server.add_insecure_port('[::]:{}'.format(args.metrics_port))
     server.start()