Merge branch 'master' into health

38 files changed, 1720 insertions, 613 deletions

diff --git a/src/core/ext/census/trace_label.h b/src/core/ext/census/trace_label.h
new file mode 100644
index 0000000000..0e4a8d885f
--- /dev/null
+++ b/src/core/ext/census/trace_label.h
@@ -0,0 +1,61 @@
+/*
+ *
+ * Copyright 2016, 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_CENSUS_TRACE_LABEL_H
+#define GRPC_CORE_EXT_CENSUS_TRACE_LABEL_H
+
+#include "src/core/ext/census/trace_string.h"
+
+/* Trace label (key/value pair) stores a label name and the label value. The
+   value can be one of trace_string/int64_t/bool. */
+typedef struct trace_label {
+  trace_string key;
+  enum label_type {
+    /* Unknown value for debugging/error purposes */
+    LABEL_UNKNOWN = 0,
+    /* A string value */
+    LABEL_STRING = 1,
+    /* An integer value. */
+    LABEL_INT = 2,
+    /* A boolean value. */
+    LABEL_BOOL = 3,
+  } value_type;
+
+  union value {
+    trace_string label_str;
+    int64_t label_int;
+    bool label_bool;
+  } value;
+} trace_label;
+
+#endif
diff --git a/src/core/ext/census/trace_propagation.h b/src/core/ext/census/trace_propagation.h
new file mode 100644
index 0000000000..75c4ebaa39
--- /dev/null
+++ b/src/core/ext/census/trace_propagation.h
@@ -0,0 +1,63 @@
+/*
+ *
+ * Copyright 2016, 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_CENSUS_TRACE_PROPAGATION_H
+#define GRPC_CORE_EXT_CENSUS_TRACE_PROPAGATION_H
+
+#include "src/core/ext/census/tracing.h"
+
+/* Encoding and decoding functions for receiving and sending trace contexts
+   over the wire.  Only RPC libraries should be calling these
+   functions.  These functions return the number of bytes encoded/decoded
+   (0 if a failure has occurred). buf_size indicates the size of the
+   input/output buffer. trace_span_context is a struct that includes the
+   trace ID, span ID, and a set of option flags (is_sampled, etc.). */
+
+/* Converts a span context to a binary byte buffer. */
+size_t trace_span_context_to_binary(const trace_span_context *ctxt,
+                                    uint8_t *buf, size_t buf_size);
+
+/* Reads a binary byte buffer and populates a span context structure. */
+size_t binary_to_trace_span_context(const uint8_t *buf, size_t buf_size,
+                                    trace_span_context *ctxt);
+
+/* Converts a span context to an http metadata compatible string. */
+size_t trace_span_context_to_http_format(const trace_span_context *ctxt,
+                                         char *buf, size_t buf_size);
+
+/* Reads an http metadata compatible string and populates a span context
+   structure. */
+size_t http_format_to_trace_span_context(const char *buf, size_t buf_size,
+                                         trace_span_context *ctxt);
+
+#endif
diff --git a/src/core/ext/census/trace_status.h b/src/core/ext/census/trace_status.h
new file mode 100644
index 0000000000..adc0ebd0c0
--- /dev/null
+++ b/src/core/ext/census/trace_status.h
@@ -0,0 +1,45 @@
+/*
+ *
+ * Copyright 2016, 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_CENSUS_TRACE_STATUS_H
+#define GRPC_CORE_EXT_CENSUS_TRACE_STATUS_H
+
+#include "src/core/ext/census/trace_string.h"
+
+/* Stores a status code and status message for a trace. */
+typedef struct trace_status {
+  int64_t errorCode;
+  trace_string errorMessage;
+} trace_status;
+
+#endif
diff --git a/src/core/ext/census/trace_string.h b/src/core/ext/census/trace_string.h
new file mode 100644
index 0000000000..8e77ee9f7e
--- /dev/null
+++ b/src/core/ext/census/trace_string.h
@@ -0,0 +1,50 @@
+/*
+ *
+ * Copyright 2016, 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_CENSUS_TRACE_STRING_H
+#define GRPC_CORE_EXT_CENSUS_TRACE_STRING_H
+
+#include <grpc/slice.h>
+
+/* String struct for tracing messages. Since this is a C API, we do not have
+   access to a string class.  This is intended for use by higher level
+   languages which wrap around the C API, as most of them have a string class.
+   This will also be more efficient when copying, as we have an explicitly
+   specified length.  Also, grpc_slice has reference counting which allows for
+   interning. */
+typedef struct trace_string {
+  char *string;
+  size_t length;
+} trace_string;
+
+#endif
diff --git a/src/core/ext/census/tracing.c b/src/core/ext/census/tracing.c
index 9371fffc8d..8b74628f4f 100644
--- a/src/core/ext/census/tracing.c
+++ b/src/core/ext/census/tracing.c
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -31,21 +31,41 @@
  *
  */
 
-//#include "src/core/ext/census/tracing.h"
+#include "src/core/ext/census/tracing.h"
 
 #include <grpc/census.h>
-#include <stdlib.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <openssl/rand.h>
+#include "src/core/ext/census/mlog.h"
 
-/* TODO(aveitch): These are all placeholder implementations. */
+void trace_start_span(const trace_span_context *span_ctxt,
+                      const trace_string name, const start_span_options *opts,
+                      trace_span_context *new_span_ctxt,
+                      bool has_remote_parent) {
+  // Noop implementation.
+}
+
+void trace_add_span_annotation(const trace_string description,
+                               const trace_label *labels, const size_t n_labels,
+                               trace_span_context *span_ctxt) {
+  // Noop implementation.
+}
 
-int census_trace_mask(const census_context *context) {
-  abort();
-  return CENSUS_TRACE_MASK_NONE;
+void trace_add_span_network_event_annotation(const trace_string description,
+                                             const trace_label *labels,
+                                             const size_t n_labels,
+                                             const gpr_timespec timestamp,
+                                             bool sent, uint64_t id,
+                                             trace_span_context *span_ctxt) {
+  // Noop implementation.
 }
 
-void census_set_trace_mask(int trace_mask) { abort(); }
+void trace_add_span_labels(const trace_label *labels, const size_t n_labels,
+                           trace_span_context *span_ctxt) {
+  // Noop implementation.
+}
 
-void census_trace_print(census_context *context, uint32_t type,
-                        const char *buffer, size_t n) {
-  abort();
+void trace_end_span(const trace_status *status, trace_span_context *span_ctxt) {
+  // Noop implementation.
 }
diff --git a/src/core/ext/census/tracing.h b/src/core/ext/census/tracing.h
new file mode 100644
index 0000000000..c2b947ae40
--- /dev/null
+++ b/src/core/ext/census/tracing.h
@@ -0,0 +1,124 @@
+/*
+ *
+ * Copyright 2016, 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_CENSUS_TRACING_H
+#define GRPC_CORE_EXT_CENSUS_TRACING_H
+
+#include <grpc/support/time.h>
+#include <stdbool.h>
+#include "src/core/ext/census/trace_context.h"
+#include "src/core/ext/census/trace_label.h"
+#include "src/core/ext/census/trace_status.h"
+
+/* This is the low level tracing API that other languages will interface with.
+   This is not intended to be accessed by the end-user, therefore it has been
+   designed with performance in mind rather than ease of use. */
+
+/* The tracing level. */
+enum TraceLevel {
+  /* Annotations on this context will be silently discarded. */
+  NO_TRACING = 0,
+  /* Annotations will not be saved to a persistent store. They will be
+     available via local APIs only. This setting is not propagated to child
+     spans. */
+  TRANSIENT_TRACING = 1,
+  /* Annotations are recorded for the entire distributed trace and they are
+     saved to a persistent store. This setting is propagated to child spans. */
+  PERSISTENT_TRACING = 2,
+};
+
+typedef struct trace_span_context {
+  /* Trace span context stores Span ID, Trace ID, and option flags. */
+  /* Trace ID is 128 bits split into 2 64-bit chunks (hi and lo). */
+  uint64_t trace_id_hi;
+  uint64_t trace_id_lo;
+  /* Span ID is 64 bits. */
+  uint64_t span_id;
+  /* Span-options is 32-bit value which contains flag options. */
+  uint32_t span_options;
+} trace_span_context;
+
+typedef struct start_span_options {
+  /* If set, this will override the Span.local_start_time for the Span. */
+  gpr_timespec local_start_timestamp;
+
+  /* Linked spans can be used to identify spans that are linked to this span in
+     a different trace.  This can be used (for example) in batching operations,
+     where a single batch handler processes multiple requests from different
+     traces. If set, points to a list of Spans are linked to the created Span.*/
+  trace_span_context *linked_spans;
+  /* The number of linked spans. */
+  size_t n_linked_spans;
+} start_span_options;
+
+/* Create a new child Span (or root if parent is NULL), with parent being the
+   designated Span. The child span will have the provided name and starting
+   span options (optional). The bool has_remote_parent marks whether the
+   context refers to a remote parent span or not. */
+void trace_start_span(const trace_span_context *span_ctxt,
+                      const trace_string name, const start_span_options *opts,
+                      trace_span_context *new_span_ctxt,
+                      bool has_remote_parent);
+
+/* Add a new Annotation to the Span. Annotations consist of a description
+   (trace_string) and a set of n labels (trace_label).  This can be populated
+   with arbitrary user data. */
+void trace_add_span_annotation(const trace_string description,
+                               const trace_label *labels, const size_t n_labels,
+                               trace_span_context *span_ctxt);
+
+/* Add a new NetworkEvent annotation to a Span. This function is only intended
+  to be used by RPC systems (either client or server), not by higher level
+  applications. The timestamp type will be system-defined, the sent argument
+  designates whether this is a network send event (client request, server
+  reply)or receive (server request, client reply). The id argument corresponds
+  to Span.Annotation.NetworkEvent.id from the data model, and serves to uniquely
+  identify each network message. */
+void trace_add_span_network_event(const trace_string description,
+                                  const trace_label *labels,
+                                  const size_t n_labels,
+                                  const gpr_timespec timestamp, bool sent,
+                                  uint64_t id, trace_span_context *span_ctxt);
+
+/* Add a set of labels to the Span. These will correspond to the field
+Span.labels in the data model. */
+void trace_add_span_labels(const trace_label *labels, const size_t n_labels,
+                           trace_span_context *span_ctxt);
+
+/* Mark the end of Span Execution with the given status. Only the timing of the
+first EndSpan call for a given Span will be recorded, and implementations are
+free to ignore all further calls using the Span. EndSpanOptions can
+optionally be NULL. */
+void trace_end_span(const trace_status *status, trace_span_context *span_ctxt);
+
+#endif
diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c
index 4ab467329a..62fb061bcf 100644
--- a/src/core/ext/client_channel/client_channel.c
+++ b/src/core/ext/client_channel/client_channel.c
@@ -44,13 +44,14 @@
 #include <grpc/support/useful.h>
 
 #include "src/core/ext/client_channel/http_connect_handshaker.h"
-#include "src/core/ext/client_channel/http_proxy.h"
 #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/subchannel.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/connected_channel.h"
 #include "src/core/lib/channel/deadline_filter.h"
+#include "src/core/lib/iomgr/combiner.h"
 #include "src/core/lib/iomgr/iomgr.h"
 #include "src/core/lib/iomgr/polling_entity.h"
 #include "src/core/lib/profiling/timers.h"
@@ -153,10 +154,6 @@ static void *method_parameters_create_from_json(const grpc_json *json) {
  */
 
 typedef struct client_channel_channel_data {
-  /** server name */
-  char *server_name;
-  /** HTTP CONNECT proxy to use, if any */
-  char *proxy_name;
   /** resolver for this channel */
   grpc_resolver *resolver;
   /** have we started resolving this channel */
@@ -164,13 +161,10 @@ typedef struct client_channel_channel_data {
   /** client channel factory */
   grpc_client_channel_factory *client_channel_factory;
 
-  /** mutex protecting all variables below in this data structure */
-  gpr_mu mu;
+  /** combiner protecting all variables below in this data structure */
+  grpc_combiner *combiner;
   /** currently active load balancer */
-  char *lb_policy_name;
   grpc_lb_policy *lb_policy;
-  /** service config in JSON form */
-  char *service_config_json;
   /** maps method names to method_parameters structs */
   grpc_slice_hash_table *method_params_table;
   /** incoming resolver result - set by resolver.next() */
@@ -187,6 +181,13 @@ typedef struct client_channel_channel_data {
   grpc_channel_stack *owning_stack;
   /** interested parties (owned) */
   grpc_pollset_set *interested_parties;
+
+  /* the following properties are guarded by a mutex since API's require them
+     to be instantaniously available */
+  gpr_mu info_mu;
+  char *info_lb_policy_name;
+  /** service config in JSON form */
+  char *info_service_config_json;
 } channel_data;
 
 /** We create one watcher for each new lb_policy that is returned from a
@@ -222,32 +223,23 @@ static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx,
 }
 
 static void on_lb_policy_state_changed_locked(grpc_exec_ctx *exec_ctx,
-                                              lb_policy_connectivity_watcher *w,
-                                              grpc_error *error) {
+                                              void *arg, grpc_error *error) {
+  lb_policy_connectivity_watcher *w = arg;
   grpc_connectivity_state publish_state = w->state;
-  /* check if the notification is for a stale policy */
-  if (w->lb_policy != w->chand->lb_policy) return;
-
-  if (publish_state == GRPC_CHANNEL_SHUTDOWN && w->chand->resolver != NULL) {
-    publish_state = GRPC_CHANNEL_TRANSIENT_FAILURE;
-    grpc_resolver_channel_saw_error(exec_ctx, w->chand->resolver);
-    GRPC_LB_POLICY_UNREF(exec_ctx, w->chand->lb_policy, "channel");
-    w->chand->lb_policy = NULL;
-  }
-  set_channel_connectivity_state_locked(exec_ctx, w->chand, publish_state,
-                                        GRPC_ERROR_REF(error), "lb_changed");
-  if (w->state != GRPC_CHANNEL_SHUTDOWN) {
-    watch_lb_policy(exec_ctx, w->chand, w->lb_policy, w->state);
+  /* check if the notification is for the latest policy */
+  if (w->lb_policy == w->chand->lb_policy) {
+    if (publish_state == GRPC_CHANNEL_SHUTDOWN && w->chand->resolver != NULL) {
+      publish_state = GRPC_CHANNEL_TRANSIENT_FAILURE;
+      grpc_resolver_channel_saw_error(exec_ctx, w->chand->resolver);
+      GRPC_LB_POLICY_UNREF(exec_ctx, w->chand->lb_policy, "channel");
+      w->chand->lb_policy = NULL;
+    }
+    set_channel_connectivity_state_locked(exec_ctx, w->chand, publish_state,
+                                          GRPC_ERROR_REF(error), "lb_changed");
+    if (w->state != GRPC_CHANNEL_SHUTDOWN) {
+      watch_lb_policy(exec_ctx, w->chand, w->lb_policy, w->state);
+    }
   }
-}
-
-static void on_lb_policy_state_changed(grpc_exec_ctx *exec_ctx, void *arg,
-                                       grpc_error *error) {
-  lb_policy_connectivity_watcher *w = arg;
-
-  gpr_mu_lock(&w->chand->mu);
-  on_lb_policy_state_changed_locked(exec_ctx, w, error);
-  gpr_mu_unlock(&w->chand->mu);
 
   GRPC_CHANNEL_STACK_UNREF(exec_ctx, w->chand->owning_stack, "watch_lb_policy");
   gpr_free(w);
@@ -260,16 +252,16 @@ static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand,
   GRPC_CHANNEL_STACK_REF(chand->owning_stack, "watch_lb_policy");
 
   w->chand = chand;
-  grpc_closure_init(&w->on_changed, on_lb_policy_state_changed, w,
-                    grpc_schedule_on_exec_ctx);
+  grpc_closure_init(&w->on_changed, on_lb_policy_state_changed_locked, w,
+                    grpc_combiner_scheduler(chand->combiner, false));
   w->state = current_state;
   w->lb_policy = lb_policy;
   grpc_lb_policy_notify_on_state_change(exec_ctx, lb_policy, &w->state,
                                         &w->on_changed);
 }
 
-static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
-                                       grpc_error *error) {
+static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx,
+                                              void *arg, grpc_error *error) {
   channel_data *chand = arg;
   char *lb_policy_name = NULL;
   grpc_lb_policy *lb_policy = NULL;
@@ -317,17 +309,6 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
     // Use pick_first if nothing was specified and we didn't select grpclb
     // above.
     if (lb_policy_name == NULL) lb_policy_name = "pick_first";
-    // If using a proxy, add channel arg for server in HTTP CONNECT request.
-    if (chand->proxy_name != NULL) {
-      grpc_arg new_arg;
-      new_arg.key = GRPC_ARG_HTTP_CONNECT_SERVER;
-      new_arg.type = GRPC_ARG_STRING;
-      new_arg.value.string = chand->server_name;
-      grpc_channel_args *tmp_args = chand->resolver_result;
-      chand->resolver_result =
-          grpc_channel_args_copy_and_add(chand->resolver_result, &new_arg, 1);
-      grpc_channel_args_destroy(exec_ctx, tmp_args);
-    }
     // Instantiate LB policy.
     grpc_lb_policy_args lb_policy_args;
     lb_policy_args.args = chand->resolver_result;
@@ -368,17 +349,18 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
                                      chand->interested_parties);
   }
 
-  gpr_mu_lock(&chand->mu);
+  gpr_mu_lock(&chand->info_mu);
   if (lb_policy_name != NULL) {
-    gpr_free(chand->lb_policy_name);
-    chand->lb_policy_name = lb_policy_name;
+    gpr_free(chand->info_lb_policy_name);
+    chand->info_lb_policy_name = lb_policy_name;
   }
   old_lb_policy = chand->lb_policy;
   chand->lb_policy = lb_policy;
   if (service_config_json != NULL) {
-    gpr_free(chand->service_config_json);
-    chand->service_config_json = service_config_json;
+    gpr_free(chand->info_service_config_json);
+    chand->info_service_config_json = service_config_json;
   }
+  gpr_mu_unlock(&chand->info_mu);
   if (chand->method_params_table != NULL) {
     grpc_slice_hash_table_unref(exec_ctx, chand->method_params_table);
   }
@@ -406,7 +388,6 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
     GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver");
     grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result,
                        &chand->on_resolver_result_changed);
-    gpr_mu_unlock(&chand->mu);
   } else {
     if (chand->resolver != NULL) {
       grpc_resolver_shutdown(exec_ctx, chand->resolver);
@@ -419,7 +400,6 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
         GRPC_ERROR_CREATE_REFERENCING("Got config after disconnection", refs,
                                       GPR_ARRAY_SIZE(refs)),
         "resolver_gone");
-    gpr_mu_unlock(&chand->mu);
   }
 
   if (exit_idle) {
@@ -441,20 +421,12 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
   GRPC_ERROR_UNREF(state_error);
 }
 
-static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
-                                  grpc_channel_element *elem,
-                                  grpc_transport_op *op) {
+static void start_transport_op_locked(grpc_exec_ctx *exec_ctx, void *arg,
+                                      grpc_error *error_ignored) {
+  grpc_transport_op *op = arg;
+  grpc_channel_element *elem = op->transport_private.args[0];
   channel_data *chand = elem->channel_data;
 
-  grpc_closure_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE);
-
-  GPR_ASSERT(op->set_accept_stream == false);
-  if (op->bind_pollset != NULL) {
-    grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties,
-                                 op->bind_pollset);
-  }
-
-  gpr_mu_lock(&chand->mu);
   if (op->on_connectivity_state_change != NULL) {
     grpc_connectivity_state_notify_on_state_change(
         exec_ctx, &chand->state_tracker, op->connectivity_state,
@@ -497,25 +469,48 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
     }
     GRPC_ERROR_UNREF(op->disconnect_with_error);
   }
-  gpr_mu_unlock(&chand->mu);
+  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->owning_stack, "start_transport_op");
+
+  grpc_closure_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE);
+}
+
+static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
+                                  grpc_channel_element *elem,
+                                  grpc_transport_op *op) {
+  channel_data *chand = elem->channel_data;
+
+  GPR_ASSERT(op->set_accept_stream == false);
+  if (op->bind_pollset != NULL) {
+    grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties,
+                                 op->bind_pollset);
+  }
+
+  op->transport_private.args[0] = elem;
+  GRPC_CHANNEL_STACK_REF(chand->owning_stack, "start_transport_op");
+  grpc_closure_sched(
+      exec_ctx, grpc_closure_init(
+                    &op->transport_private.closure, start_transport_op_locked,
+                    op, grpc_combiner_scheduler(chand->combiner, false)),
+      GRPC_ERROR_NONE);
 }
 
 static void cc_get_channel_info(grpc_exec_ctx *exec_ctx,
                                 grpc_channel_element *elem,
                                 const grpc_channel_info *info) {
   channel_data *chand = elem->channel_data;
-  gpr_mu_lock(&chand->mu);
+  gpr_mu_lock(&chand->info_mu);
   if (info->lb_policy_name != NULL) {
-    *info->lb_policy_name = chand->lb_policy_name == NULL
+    *info->lb_policy_name = chand->info_lb_policy_name == NULL
                                 ? NULL
-                                : gpr_strdup(chand->lb_policy_name);
+                                : gpr_strdup(chand->info_lb_policy_name);
   }
   if (info->service_config_json != NULL) {
-    *info->service_config_json = chand->service_config_json == NULL
-                                     ? NULL
-                                     : gpr_strdup(chand->service_config_json);
+    *info->service_config_json =
+        chand->info_service_config_json == NULL
+            ? NULL
+            : gpr_strdup(chand->info_service_config_json);
   }
-  gpr_mu_unlock(&chand->mu);
+  gpr_mu_unlock(&chand->info_mu);
 }
 
 /* Constructor for channel_data */
@@ -527,11 +522,12 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
   GPR_ASSERT(args->is_last);
   GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
   // Initialize data members.
-  gpr_mu_init(&chand->mu);
+  chand->combiner = grpc_combiner_create(NULL);
+  gpr_mu_init(&chand->info_mu);
   chand->owning_stack = args->channel_stack;
   grpc_closure_init(&chand->on_resolver_result_changed,
-                    on_resolver_result_changed, chand,
-                    grpc_schedule_on_exec_ctx);
+                    on_resolver_result_changed_locked, chand,
+                    grpc_combiner_scheduler(chand->combiner, false));
   chand->interested_parties = grpc_pollset_set_create();
   grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE,
                                "client_channel");
@@ -542,24 +538,21 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
   GPR_ASSERT(arg->type == GRPC_ARG_POINTER);
   grpc_client_channel_factory_ref(arg->value.pointer.p);
   chand->client_channel_factory = arg->value.pointer.p;
-  // Instantiate resolver.
+  // Get server name to resolve, using proxy mapper if needed.
   arg = grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVER_URI);
   GPR_ASSERT(arg != NULL);
   GPR_ASSERT(arg->type == GRPC_ARG_STRING);
-  grpc_uri *uri = grpc_uri_parse(arg->value.string, true);
-  if (uri == NULL) return GRPC_ERROR_CREATE("cannot parse server URI");
-  if (uri->path[0] == '\0') {
-    grpc_uri_destroy(uri);
-    return GRPC_ERROR_CREATE("server URI is missing path");
-  }
-  chand->server_name =
-      gpr_strdup(uri->path[0] == '/' ? uri->path + 1 : uri->path);
-  grpc_uri_destroy(uri);
-  chand->proxy_name = grpc_get_http_proxy_server();
-  char *name_to_resolve =
-      chand->proxy_name == NULL ? arg->value.string : chand->proxy_name;
+  char *proxy_name = NULL;
+  grpc_channel_args *new_args = NULL;
+  grpc_proxy_mappers_map_name(exec_ctx, arg->value.string, args->channel_args,
+                              &proxy_name, &new_args);
+  // Instantiate resolver.
   chand->resolver = grpc_resolver_create(
-      exec_ctx, name_to_resolve, args->channel_args, chand->interested_parties);
+      exec_ctx, proxy_name != NULL ? proxy_name : arg->value.string,
+      new_args != NULL ? new_args : args->channel_args,
+      chand->interested_parties);
+  if (proxy_name != NULL) gpr_free(proxy_name);
+  if (new_args != NULL) grpc_channel_args_destroy(exec_ctx, new_args);
   if (chand->resolver == NULL) {
     return GRPC_ERROR_CREATE("resolver creation failed");
   }
@@ -570,8 +563,6 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
 static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
                                     grpc_channel_element *elem) {
   channel_data *chand = elem->channel_data;
-  gpr_free(chand->server_name);
-  gpr_free(chand->proxy_name);
   if (chand->resolver != NULL) {
     grpc_resolver_shutdown(exec_ctx, chand->resolver);
     GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel");
@@ -585,14 +576,15 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
                                      chand->interested_parties);
     GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel");
   }
-  gpr_free(chand->lb_policy_name);
-  gpr_free(chand->service_config_json);
+  gpr_free(chand->info_lb_policy_name);
+  gpr_free(chand->info_service_config_json);
   if (chand->method_params_table != NULL) {
     grpc_slice_hash_table_unref(exec_ctx, chand->method_params_table);
   }
   grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
-  grpc_pollset_set_destroy(chand->interested_parties);
-  gpr_mu_destroy(&chand->mu);
+  grpc_pollset_set_destroy(exec_ctx, chand->interested_parties);
+  GRPC_COMBINER_UNREF(exec_ctx, chand->combiner, "client_channel");
+  gpr_mu_destroy(&chand->info_mu);
 }
 
 /*************************************************************************
@@ -635,8 +627,6 @@ typedef struct client_channel_call_data {
       grpc_subchannel_call */
   gpr_atm subchannel_call;
 
-  gpr_mu mu;
-
   subchannel_creation_phase creation_phase;
   grpc_connected_subchannel *connected_subchannel;
   grpc_polling_entity *pollent;
@@ -681,52 +671,32 @@ static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld,
   GRPC_ERROR_UNREF(error);
 }
 
-typedef struct {
-  grpc_transport_stream_op **ops;
-  size_t nops;
-  grpc_subchannel_call *call;
-} retry_ops_args;
-
-static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) {
-  retry_ops_args *a = args;
-  size_t i;
-  for (i = 0; i < a->nops; i++) {
-    grpc_subchannel_call_process_op(exec_ctx, a->call, a->ops[i]);
-  }
-  GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops");
-  gpr_free(a->ops);
-  gpr_free(a);
-}
-
 static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
   if (calld->waiting_ops_count == 0) {
     return;
   }
 
-  retry_ops_args *a = gpr_malloc(sizeof(*a));
-  a->ops = calld->waiting_ops;
-  a->nops = calld->waiting_ops_count;
-  a->call = GET_CALL(calld);
-  if (a->call == CANCELLED_CALL) {
-    gpr_free(a);
+  grpc_subchannel_call *call = GET_CALL(calld);
+  grpc_transport_stream_op **ops = calld->waiting_ops;
+  size_t nops = calld->waiting_ops_count;
+  if (call == CANCELLED_CALL) {
     fail_locked(exec_ctx, calld, GRPC_ERROR_CANCELLED);
     return;
   }
   calld->waiting_ops = NULL;
   calld->waiting_ops_count = 0;
   calld->waiting_ops_capacity = 0;
-  GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops");
-  grpc_closure_sched(
-      exec_ctx, grpc_closure_create(retry_ops, a, grpc_schedule_on_exec_ctx),
-      GRPC_ERROR_NONE);
+  for (size_t i = 0; i < nops; i++) {
+    grpc_subchannel_call_process_op(exec_ctx, call, ops[i]);
+  }
+  gpr_free(ops);
 }
 
-static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg,
-                             grpc_error *error) {
+static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
+                                    grpc_error *error) {
   grpc_call_element *elem = arg;
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
-  gpr_mu_lock(&calld->mu);
   GPR_ASSERT(calld->creation_phase ==
              GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL);
   grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent,
@@ -762,7 +732,6 @@ static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg,
                       (gpr_atm)(uintptr_t)subchannel_call);
     retry_waiting_locked(exec_ctx, calld);
   }
-  gpr_mu_unlock(&calld->mu);
   GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
 }
 
@@ -788,37 +757,35 @@ typedef struct {
 /** Return true if subchannel is available immediately (in which case on_ready
     should not be called), or false otherwise (in which case on_ready should be
     called when the subchannel is available). */
-static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-                            grpc_metadata_batch *initial_metadata,
-                            uint32_t initial_metadata_flags,
-                            grpc_connected_subchannel **connected_subchannel,
-                            grpc_closure *on_ready, grpc_error *error);
-
-static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg,
-                             grpc_error *error) {
+static bool pick_subchannel_locked(
+    grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+    grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags,
+    grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready,
+    grpc_error *error);
+
+static void continue_picking_locked(grpc_exec_ctx *exec_ctx, void *arg,
+                                    grpc_error *error) {
   continue_picking_args *cpa = arg;
   if (cpa->connected_subchannel == NULL) {
     /* cancelled, do nothing */
   } else if (error != GRPC_ERROR_NONE) {
     grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error));
   } else {
-    call_data *calld = cpa->elem->call_data;
-    gpr_mu_lock(&calld->mu);
-    if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
-                        cpa->initial_metadata_flags, cpa->connected_subchannel,
-                        cpa->on_ready, GRPC_ERROR_NONE)) {
+    if (pick_subchannel_locked(exec_ctx, cpa->elem, cpa->initial_metadata,
+                               cpa->initial_metadata_flags,
+                               cpa->connected_subchannel, cpa->on_ready,
+                               GRPC_ERROR_NONE)) {
       grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE);
     }
-    gpr_mu_unlock(&calld->mu);
   }
   gpr_free(cpa);
 }
 
-static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-                            grpc_metadata_batch *initial_metadata,
-                            uint32_t initial_metadata_flags,
-                            grpc_connected_subchannel **connected_subchannel,
-                            grpc_closure *on_ready, grpc_error *error) {
+static bool pick_subchannel_locked(
+    grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+    grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags,
+    grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready,
+    grpc_error *error) {
   GPR_TIMER_BEGIN("pick_subchannel", 0);
 
   channel_data *chand = elem->channel_data;
@@ -828,7 +795,6 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
 
   GPR_ASSERT(connected_subchannel);
 
-  gpr_mu_lock(&chand->mu);
   if (initial_metadata == NULL) {
     if (chand->lb_policy != NULL) {
       grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy,
@@ -844,7 +810,6 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
             GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1));
       }
     }
-    gpr_mu_unlock(&chand->mu);
     GPR_TIMER_END("pick_subchannel", 0);
     GRPC_ERROR_UNREF(error);
     return true;
@@ -853,7 +818,6 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
   if (chand->lb_policy != NULL) {
     grpc_lb_policy *lb_policy = chand->lb_policy;
     GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel");
-    gpr_mu_unlock(&chand->mu);
     // If the application explicitly set wait_for_ready, use that.
     // Otherwise, if the service config specified a value for this
     // method, use that.
@@ -892,88 +856,66 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
     cpa->connected_subchannel = connected_subchannel;
     cpa->on_ready = on_ready;
     cpa->elem = elem;
-    grpc_closure_init(&cpa->closure, continue_picking, cpa,
-                      grpc_schedule_on_exec_ctx);
+    grpc_closure_init(&cpa->closure, continue_picking_locked, cpa,
+                      grpc_combiner_scheduler(chand->combiner, true));
     grpc_closure_list_append(&chand->waiting_for_config_closures, &cpa->closure,
                              GRPC_ERROR_NONE);
   } else {
     grpc_closure_sched(exec_ctx, on_ready, GRPC_ERROR_CREATE("Disconnected"));
   }
-  gpr_mu_unlock(&chand->mu);
 
   GPR_TIMER_END("pick_subchannel", 0);
   return false;
 }
 
-// The logic here is fairly complicated, due to (a) the fact that we
-// need to handle the case where we receive the send op before the
-// initial metadata op, and (b) the need for efficiency, especially in
-// the streaming case.
-// TODO(ctiller): Explain this more thoroughly.
-static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
-                                         grpc_call_element *elem,
-                                         grpc_transport_stream_op *op) {
-  call_data *calld = elem->call_data;
+static void start_transport_stream_op_locked_inner(grpc_exec_ctx *exec_ctx,
+                                                   grpc_transport_stream_op *op,
+                                                   grpc_call_element *elem) {
   channel_data *chand = elem->channel_data;
-  GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
-  grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op);
-  /* try to (atomically) get the call */
-  grpc_subchannel_call *call = GET_CALL(calld);
-  GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0);
-  if (call == CANCELLED_CALL) {
-    grpc_transport_stream_op_finish_with_failure(
-        exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
-    GPR_TIMER_END("cc_start_transport_stream_op", 0);
-    return;
-  }
-  if (call != NULL) {
-    grpc_subchannel_call_process_op(exec_ctx, call, op);
-    GPR_TIMER_END("cc_start_transport_stream_op", 0);
-    return;
-  }
-  /* we failed; lock and figure out what to do */
-  gpr_mu_lock(&calld->mu);
-retry:
+  call_data *calld = elem->call_data;
+  grpc_subchannel_call *call;
+
   /* need to recheck that another thread hasn't set the call */
   call = GET_CALL(calld);
   if (call == CANCELLED_CALL) {
-    gpr_mu_unlock(&calld->mu);
     grpc_transport_stream_op_finish_with_failure(
         exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
-    GPR_TIMER_END("cc_start_transport_stream_op", 0);
+    /* early out */
     return;
   }
   if (call != NULL) {
-    gpr_mu_unlock(&calld->mu);
     grpc_subchannel_call_process_op(exec_ctx, call, op);
-    GPR_TIMER_END("cc_start_transport_stream_op", 0);
+    /* early out */
     return;
   }
   /* if this is a cancellation, then we can raise our cancelled flag */
   if (op->cancel_error != GRPC_ERROR_NONE) {
     if (!gpr_atm_rel_cas(&calld->subchannel_call, 0,
                          (gpr_atm)(uintptr_t)CANCELLED_CALL)) {
-      goto retry;
+      /* recurse to retry */
+      start_transport_stream_op_locked_inner(exec_ctx, op, elem);
+      /* early out */
+      return;
     } else {
-      // Stash a copy of cancel_error in our call data, so that we can use
-      // it for subsequent operations.  This ensures that if the call is
-      // cancelled before any ops are passed down (e.g., if the deadline
-      // is in the past when the call starts), we can return the right
-      // error to the caller when the first op does get passed down.
+      /* Stash a copy of cancel_error in our call data, so that we can use
+         it for subsequent operations.  This ensures that if the call is
+         cancelled before any ops are passed down (e.g., if the deadline
+         is in the past when the call starts), we can return the right
+         error to the caller when the first op does get passed down. */
       calld->cancel_error = GRPC_ERROR_REF(op->cancel_error);
       switch (calld->creation_phase) {
         case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING:
           fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error));
           break;
         case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL:
-          pick_subchannel(exec_ctx, elem, NULL, 0, &calld->connected_subchannel,
-                          NULL, GRPC_ERROR_REF(op->cancel_error));
+          pick_subchannel_locked(exec_ctx, elem, NULL, 0,
+                                 &calld->connected_subchannel, NULL,
+                                 GRPC_ERROR_REF(op->cancel_error));
           break;
       }
-      gpr_mu_unlock(&calld->mu);
       grpc_transport_stream_op_finish_with_failure(
           exec_ctx, op, GRPC_ERROR_REF(op->cancel_error));
-      GPR_TIMER_END("cc_start_transport_stream_op", 0);
+      /* early out */
       return;
     }
   }
@@ -982,16 +924,16 @@ retry:
       calld->connected_subchannel == NULL &&
       op->send_initial_metadata != NULL) {
     calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL;
-    grpc_closure_init(&calld->next_step, subchannel_ready, elem,
-                      grpc_schedule_on_exec_ctx);
+    grpc_closure_init(&calld->next_step, subchannel_ready_locked, elem,
+                      grpc_combiner_scheduler(chand->combiner, true));
     GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel");
     /* If a subchannel is not available immediately, the polling entity from
        call_data should be provided to channel_data's interested_parties, so
        that IO of the lb_policy and resolver could be done under it. */
-    if (pick_subchannel(exec_ctx, elem, op->send_initial_metadata,
-                        op->send_initial_metadata_flags,
-                        &calld->connected_subchannel, &calld->next_step,
-                        GRPC_ERROR_NONE)) {
+    if (pick_subchannel_locked(exec_ctx, elem, op->send_initial_metadata,
+                               op->send_initial_metadata_flags,
+                               &calld->connected_subchannel, &calld->next_step,
+                               GRPC_ERROR_NONE)) {
       calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
       GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
     } else {
@@ -1014,31 +956,89 @@ retry:
     gpr_atm_rel_store(&calld->subchannel_call,
                       (gpr_atm)(uintptr_t)subchannel_call);
     retry_waiting_locked(exec_ctx, calld);
-    goto retry;
+    /* recurse to retry */
+    start_transport_stream_op_locked_inner(exec_ctx, op, elem);
+    /* early out */
+    return;
   }
   /* nothing to be done but wait */
   add_waiting_locked(calld, op);
-  gpr_mu_unlock(&calld->mu);
+}
+
+static void cc_start_transport_stream_op_locked(grpc_exec_ctx *exec_ctx,
+                                                void *arg,
+                                                grpc_error *error_ignored) {
+  GPR_TIMER_BEGIN("cc_start_transport_stream_op_locked", 0);
+
+  grpc_transport_stream_op *op = arg;
+  grpc_call_element *elem = op->handler_private.args[0];
+  call_data *calld = elem->call_data;
+
+  start_transport_stream_op_locked_inner(exec_ctx, op, elem);
+
+  GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call,
+                        "start_transport_stream_op");
+  GPR_TIMER_END("cc_start_transport_stream_op_locked", 0);
+}
+
+/* The logic here is fairly complicated, due to (a) the fact that we
+   need to handle the case where we receive the send op before the
+   initial metadata op, and (b) the need for efficiency, especially in
+   the streaming case.
+
+   We use double-checked locking to initially see if initialization has been
+   performed. If it has not, we acquire the combiner and perform initialization.
+   If it has, we proceed on the fast path. */
+static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
+                                         grpc_call_element *elem,
+                                         grpc_transport_stream_op *op) {
+  call_data *calld = elem->call_data;
+  channel_data *chand = elem->channel_data;
+  GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
+  grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op);
+  /* try to (atomically) get the call */
+  grpc_subchannel_call *call = GET_CALL(calld);
+  GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0);
+  if (call == CANCELLED_CALL) {
+    grpc_transport_stream_op_finish_with_failure(
+        exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
+    GPR_TIMER_END("cc_start_transport_stream_op", 0);
+    /* early out */
+    return;
+  }
+  if (call != NULL) {
+    grpc_subchannel_call_process_op(exec_ctx, call, op);
+    GPR_TIMER_END("cc_start_transport_stream_op", 0);
+    /* early out */
+    return;
+  }
+  /* we failed; lock and figure out what to do */
+  GRPC_CALL_STACK_REF(calld->owning_call, "start_transport_stream_op");
+  op->handler_private.args[0] = elem;
+  grpc_closure_sched(
+      exec_ctx,
+      grpc_closure_init(&op->handler_private.closure,
+                        cc_start_transport_stream_op_locked, op,
+                        grpc_combiner_scheduler(chand->combiner, false)),
+      GRPC_ERROR_NONE);
   GPR_TIMER_END("cc_start_transport_stream_op", 0);
 }
 
 // Gets data from the service config.  Invoked when the resolver returns
 // its initial result.
-static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
-                                grpc_error *error) {
+static void read_service_config_locked(grpc_exec_ctx *exec_ctx, void *arg,
+                                       grpc_error *error) {
   grpc_call_element *elem = arg;
   channel_data *chand = elem->channel_data;
   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) {
     // Get the method config table from channel data.
-    gpr_mu_lock(&chand->mu);
     grpc_slice_hash_table *method_params_table = NULL;
     if (chand->method_params_table != NULL) {
       method_params_table =
           grpc_slice_hash_table_ref(chand->method_params_table);
     }
-    gpr_mu_unlock(&chand->mu);
     // If the method config table was present, use it.
     if (method_params_table != NULL) {
       const method_parameters *method_params = grpc_method_config_table_get(
@@ -1048,7 +1048,6 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
             gpr_time_cmp(method_params->timeout, gpr_time_0(GPR_TIMESPAN)) != 0;
         if (have_method_timeout ||
             method_params->wait_for_ready != WAIT_FOR_READY_UNSET) {
-          gpr_mu_lock(&calld->mu);
           if (have_method_timeout) {
             const gpr_timespec per_method_deadline =
                 gpr_time_add(calld->call_start_time, method_params->timeout);
@@ -1062,7 +1061,6 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
             calld->wait_for_ready_from_service_config =
                 method_params->wait_for_ready;
           }
-          gpr_mu_unlock(&calld->mu);
         }
       }
       grpc_slice_hash_table_unref(exec_ctx, method_params_table);
@@ -1071,43 +1069,25 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
   GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config");
 }
 
-/* Constructor for call_data */
-static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
-                                     grpc_call_element *elem,
-                                     grpc_call_element_args *args) {
+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;
-  // 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->wait_for_ready_from_service_config = WAIT_FOR_READY_UNSET;
-  calld->cancel_error = GRPC_ERROR_NONE;
-  gpr_atm_rel_store(&calld->subchannel_call, 0);
-  gpr_mu_init(&calld->mu);
-  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;
   // 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.
   // TODO(roth): This code is almost but not quite identical to the code
   // in read_service_config() above.  It would be nice to find a way to
   // combine them, to avoid having to maintain it twice.
-  gpr_mu_lock(&chand->mu);
   if (chand->lb_policy != NULL) {
     // We already have a resolver result, so check for service config.
     if (chand->method_params_table != NULL) {
       grpc_slice_hash_table *method_params_table =
           grpc_slice_hash_table_ref(chand->method_params_table);
-      gpr_mu_unlock(&chand->mu);
       method_parameters *method_params = grpc_method_config_table_get(
-          exec_ctx, method_params_table, args->path);
+          exec_ctx, method_params_table, calld->path);
       if (method_params != NULL) {
         if (gpr_time_cmp(method_params->timeout,
                          gpr_time_0(GPR_CLOCK_MONOTONIC)) != 0) {
@@ -1121,24 +1101,53 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
         }
       }
       grpc_slice_hash_table_unref(exec_ctx, method_params_table);
-    } else {
-      gpr_mu_unlock(&chand->mu);
     }
   } 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, elem,
-                      grpc_schedule_on_exec_ctx);
+    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);
-    gpr_mu_unlock(&chand->mu);
   }
   // 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,
+                                     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->wait_for_ready_from_service_config = WAIT_FOR_READY_UNSET;
+  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);
   return GRPC_ERROR_NONE;
 }
 
@@ -1156,7 +1165,6 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
     GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call");
   }
   GPR_ASSERT(calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING);
-  gpr_mu_destroy(&calld->mu);
   GPR_ASSERT(calld->waiting_ops_count == 0);
   if (calld->connected_subchannel != NULL) {
     GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, calld->connected_subchannel,
@@ -1192,26 +1200,36 @@ const grpc_channel_filter grpc_client_channel_filter = {
     "client-channel",
 };
 
+static void try_to_connect_locked(grpc_exec_ctx *exec_ctx, void *arg,
+                                  grpc_error *error_ignored) {
+  channel_data *chand = arg;
+  if (chand->lb_policy != NULL) {
+    grpc_lb_policy_exit_idle(exec_ctx, chand->lb_policy);
+  } else {
+    chand->exit_idle_when_lb_policy_arrives = true;
+    if (!chand->started_resolving && chand->resolver != NULL) {
+      GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver");
+      chand->started_resolving = true;
+      grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result,
+                         &chand->on_resolver_result_changed);
+    }
+  }
+  GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->owning_stack, "try_to_connect");
+}
+
 grpc_connectivity_state grpc_client_channel_check_connectivity_state(
     grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, int try_to_connect) {
   channel_data *chand = elem->channel_data;
-  grpc_connectivity_state out;
-  gpr_mu_lock(&chand->mu);
-  out = grpc_connectivity_state_check(&chand->state_tracker, NULL);
+  grpc_connectivity_state out =
+      grpc_connectivity_state_check(&chand->state_tracker);
   if (out == GRPC_CHANNEL_IDLE && try_to_connect) {
-    if (chand->lb_policy != NULL) {
-      grpc_lb_policy_exit_idle(exec_ctx, chand->lb_policy);
-    } else {
-      chand->exit_idle_when_lb_policy_arrives = true;
-      if (!chand->started_resolving && chand->resolver != NULL) {
-        GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver");
-        chand->started_resolving = true;
-        grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result,
-                           &chand->on_resolver_result_changed);
-      }
-    }
+    GRPC_CHANNEL_STACK_REF(chand->owning_stack, "try_to_connect");
+    grpc_closure_sched(
+        exec_ctx,
+        grpc_closure_create(try_to_connect_locked, chand,
+                            grpc_combiner_scheduler(chand->combiner, false)),
+        GRPC_ERROR_NONE);
   }
-  gpr_mu_unlock(&chand->mu);
   return out;
 }
 
@@ -1219,6 +1237,7 @@ typedef struct {
   channel_data *chand;
   grpc_pollset *pollset;
   grpc_closure *on_complete;
+  grpc_connectivity_state *state;
   grpc_closure my_closure;
 } external_connectivity_watcher;
 
@@ -1231,7 +1250,16 @@ static void on_external_watch_complete(grpc_exec_ctx *exec_ctx, void *arg,
   GRPC_CHANNEL_STACK_UNREF(exec_ctx, w->chand->owning_stack,
                            "external_connectivity_watcher");
   gpr_free(w);
-  follow_up->cb(exec_ctx, follow_up->cb_arg, error);
+  grpc_closure_run(exec_ctx, follow_up, GRPC_ERROR_REF(error));
+}
+
+static void watch_connectivity_state_locked(grpc_exec_ctx *exec_ctx, void *arg,
+                                            grpc_error *error_ignored) {
+  external_connectivity_watcher *w = arg;
+  grpc_closure_init(&w->my_closure, on_external_watch_complete, w,
+                    grpc_schedule_on_exec_ctx);
+  grpc_connectivity_state_notify_on_state_change(
+      exec_ctx, &w->chand->state_tracker, w->state, &w->my_closure);
 }
 
 void grpc_client_channel_watch_connectivity_state(
@@ -1242,13 +1270,13 @@ void grpc_client_channel_watch_connectivity_state(
   w->chand = chand;
   w->pollset = pollset;
   w->on_complete = on_complete;
+  w->state = state;
   grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties, pollset);
-  grpc_closure_init(&w->my_closure, on_external_watch_complete, w,
-                    grpc_schedule_on_exec_ctx);
   GRPC_CHANNEL_STACK_REF(w->chand->owning_stack,
                          "external_connectivity_watcher");
-  gpr_mu_lock(&chand->mu);
-  grpc_connectivity_state_notify_on_state_change(
-      exec_ctx, &chand->state_tracker, state, &w->my_closure);
-  gpr_mu_unlock(&chand->mu);
+  grpc_closure_sched(
+      exec_ctx,
+      grpc_closure_init(&w->my_closure, watch_connectivity_state_locked, w,
+                        grpc_combiner_scheduler(chand->combiner, true)),
+      GRPC_ERROR_NONE);
 }
diff --git a/src/core/ext/client_channel/client_channel_plugin.c b/src/core/ext/client_channel/client_channel_plugin.c
index 7f75233727..6f9df3e386 100644
--- a/src/core/ext/client_channel/client_channel_plugin.c
+++ b/src/core/ext/client_channel/client_channel_plugin.c
@@ -39,6 +39,7 @@
 
 #include "src/core/ext/client_channel/client_channel.h"
 #include "src/core/ext/client_channel/http_connect_handshaker.h"
+#include "src/core/ext/client_channel/http_proxy.h"
 #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"
@@ -82,6 +83,7 @@ void grpc_client_channel_init(void) {
   grpc_lb_policy_registry_init();
   grpc_resolver_registry_init();
   grpc_proxy_mapper_registry_init();
+  grpc_register_http_proxy_mapper();
   grpc_subchannel_index_init();
   grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MIN,
                                    set_default_host_if_unset, NULL);
diff --git a/src/core/ext/client_channel/http_proxy.c b/src/core/ext/client_channel/http_proxy.c
index 9a6c818c4e..bbe4ff550c 100644
--- a/src/core/ext/client_channel/http_proxy.c
+++ b/src/core/ext/client_channel/http_proxy.c
@@ -40,10 +40,13 @@
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 
+#include "src/core/ext/client_channel/http_connect_handshaker.h"
+#include "src/core/ext/client_channel/proxy_mapper_registry.h"
 #include "src/core/ext/client_channel/uri_parser.h"
+#include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/support/env.h"
 
-char* grpc_get_http_proxy_server() {
+static char* grpc_get_http_proxy_server() {
   char* uri_str = gpr_getenv("http_proxy");
   if (uri_str == NULL) return NULL;
   grpc_uri* uri = grpc_uri_parse(uri_str, false /* suppress_errors */);
@@ -66,3 +69,55 @@ done:
   grpc_uri_destroy(uri);
   return proxy_name;
 }
+
+static bool proxy_mapper_map_name(grpc_exec_ctx* exec_ctx,
+                                  grpc_proxy_mapper* mapper,
+                                  const char* server_uri,
+                                  const grpc_channel_args* args,
+                                  char** name_to_resolve,
+                                  grpc_channel_args** new_args) {
+  *name_to_resolve = grpc_get_http_proxy_server();
+  if (*name_to_resolve == NULL) return false;
+  grpc_uri* uri = grpc_uri_parse(server_uri, false /* suppress_errors */);
+  if (uri == NULL || uri->path[0] == '\0') {
+    gpr_log(GPR_ERROR,
+            "'http_proxy' environment variable set, but cannot "
+            "parse server URI '%s' -- not using proxy",
+            server_uri);
+    if (uri != NULL) grpc_uri_destroy(uri);
+    return false;
+  }
+  if (strcmp(uri->scheme, "unix") == 0) {
+    gpr_log(GPR_INFO, "not using proxy for Unix domain socket '%s'",
+            server_uri);
+    grpc_uri_destroy(uri);
+    return false;
+  }
+  grpc_arg new_arg;
+  new_arg.key = GRPC_ARG_HTTP_CONNECT_SERVER;
+  new_arg.type = GRPC_ARG_STRING;
+  new_arg.value.string = uri->path[0] == '/' ? uri->path + 1 : uri->path;
+  *new_args = grpc_channel_args_copy_and_add(args, &new_arg, 1);
+  grpc_uri_destroy(uri);
+  return true;
+}
+
+static bool proxy_mapper_map_address(grpc_exec_ctx* exec_ctx,
+                                     grpc_proxy_mapper* mapper,
+                                     const grpc_resolved_address* address,
+                                     const grpc_channel_args* args,
+                                     grpc_resolved_address** new_address,
+                                     grpc_channel_args** new_args) {
+  return false;
+}
+
+static void proxy_mapper_destroy(grpc_proxy_mapper* mapper) {}
+
+static const grpc_proxy_mapper_vtable proxy_mapper_vtable = {
+    proxy_mapper_map_name, proxy_mapper_map_address, proxy_mapper_destroy};
+
+static grpc_proxy_mapper proxy_mapper = {&proxy_mapper_vtable};
+
+void grpc_register_http_proxy_mapper() {
+  grpc_proxy_mapper_register(true /* at_start */, &proxy_mapper);
+}
diff --git a/src/core/ext/client_channel/http_proxy.h b/src/core/ext/client_channel/http_proxy.h
index 0d77ae253b..c8882b1ef1 100644
--- a/src/core/ext/client_channel/http_proxy.h
+++ b/src/core/ext/client_channel/http_proxy.h
@@ -34,8 +34,6 @@
 #ifndef GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_PROXY_H
 #define GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_PROXY_H
 
-/// Returns the name of the proxy to use, or NULL if no proxy is configured.
-/// Caller takes ownership of result.
-char* grpc_get_http_proxy_server();
+void grpc_register_http_proxy_mapper();
 
 #endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_PROXY_H */
diff --git a/src/core/ext/client_channel/lb_policy.c b/src/core/ext/client_channel/lb_policy.c
index 45ee72e2f0..90401b586f 100644
--- a/src/core/ext/client_channel/lb_policy.c
+++ b/src/core/ext/client_channel/lb_policy.c
@@ -94,7 +94,7 @@ void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx,
   gpr_atm old_val =
       ref_mutate(policy, -(gpr_atm)1, 1 REF_MUTATE_PASS_ARGS("WEAK_UNREF"));
   if (old_val == 1) {
-    grpc_pollset_set_destroy(policy->interested_parties);
+    grpc_pollset_set_destroy(exec_ctx, policy->interested_parties);
     policy->vtable->destroy(exec_ctx, policy);
   }
 }
diff --git a/src/core/ext/client_channel/proxy_mapper.c b/src/core/ext/client_channel/proxy_mapper.c
index 6b6f328d3c..f92afe847b 100644
--- a/src/core/ext/client_channel/proxy_mapper.c
+++ b/src/core/ext/client_channel/proxy_mapper.c
@@ -38,13 +38,24 @@ void grpc_proxy_mapper_init(const grpc_proxy_mapper_vtable* vtable,
   mapper->vtable = vtable;
 }
 
-bool grpc_proxy_mapper_map(grpc_exec_ctx* exec_ctx, grpc_proxy_mapper* mapper,
-                           const grpc_resolved_address* address,
-                           const grpc_channel_args* args,
-                           grpc_resolved_address** new_address,
-                           grpc_channel_args** new_args) {
-  return mapper->vtable->map(exec_ctx, mapper, address, args, new_address,
-                             new_args);
+bool grpc_proxy_mapper_map_name(grpc_exec_ctx* exec_ctx,
+                                grpc_proxy_mapper* mapper,
+                                const char* server_uri,
+                                const grpc_channel_args* args,
+                                char** name_to_resolve,
+                                grpc_channel_args** new_args) {
+  return mapper->vtable->map_name(exec_ctx, mapper, server_uri, args,
+                                  name_to_resolve, new_args);
+}
+
+bool grpc_proxy_mapper_map_address(grpc_exec_ctx* exec_ctx,
+                                   grpc_proxy_mapper* mapper,
+                                   const grpc_resolved_address* address,
+                                   const grpc_channel_args* args,
+                                   grpc_resolved_address** new_address,
+                                   grpc_channel_args** new_args) {
+  return mapper->vtable->map_address(exec_ctx, mapper, address, args,
+                                     new_address, new_args);
 }
 
 void grpc_proxy_mapper_destroy(grpc_proxy_mapper* mapper) {
diff --git a/src/core/ext/client_channel/proxy_mapper.h b/src/core/ext/client_channel/proxy_mapper.h
index fa930379e7..6e4607fe4d 100644
--- a/src/core/ext/client_channel/proxy_mapper.h
+++ b/src/core/ext/client_channel/proxy_mapper.h
@@ -43,15 +43,22 @@
 typedef struct grpc_proxy_mapper grpc_proxy_mapper;
 
 typedef struct {
+  /// Determines the proxy name to resolve for \a server_uri.
+  /// If no proxy is needed, returns false.
+  /// Otherwise, sets \a name_to_resolve, optionally sets \a new_args,
+  /// and returns true.
+  bool (*map_name)(grpc_exec_ctx* exec_ctx, grpc_proxy_mapper* mapper,
+                   const char* server_uri, const grpc_channel_args* args,
+                   char** name_to_resolve, grpc_channel_args** new_args);
   /// Determines the proxy address to use to contact \a address.
   /// If no proxy is needed, returns false.
   /// Otherwise, sets \a new_address, optionally sets \a new_args, and
   /// returns true.
-  bool (*map)(grpc_exec_ctx* exec_ctx, grpc_proxy_mapper* mapper,
-              const grpc_resolved_address* address,
-              const grpc_channel_args* args,
-              grpc_resolved_address** new_address,
-              grpc_channel_args** new_args);
+  bool (*map_address)(grpc_exec_ctx* exec_ctx, grpc_proxy_mapper* mapper,
+                      const grpc_resolved_address* address,
+                      const grpc_channel_args* args,
+                      grpc_resolved_address** new_address,
+                      grpc_channel_args** new_args);
   /// Destroys \a mapper.
   void (*destroy)(grpc_proxy_mapper* mapper);
 } grpc_proxy_mapper_vtable;
@@ -63,11 +70,20 @@ struct grpc_proxy_mapper {
 void grpc_proxy_mapper_init(const grpc_proxy_mapper_vtable* vtable,
                             grpc_proxy_mapper* mapper);
 
-bool grpc_proxy_mapper_map(grpc_exec_ctx* exec_ctx, grpc_proxy_mapper* mapper,
-                           const grpc_resolved_address* address,
-                           const grpc_channel_args* args,
-                           grpc_resolved_address** new_address,
-                           grpc_channel_args** new_args);
+bool grpc_proxy_mapper_map_name(grpc_exec_ctx* exec_ctx,
+                                grpc_proxy_mapper* mapper,
+                                const char* server_uri,
+                                const grpc_channel_args* args,
+                                char** name_to_resolve,
+                                grpc_channel_args** new_args);
+
+bool grpc_proxy_mapper_map_address(grpc_exec_ctx* exec_ctx,
+                                   grpc_proxy_mapper* mapper,
+                                   const grpc_resolved_address* address,
+                                   const grpc_channel_args* args,
+                                   grpc_resolved_address** new_address,
+                                   grpc_channel_args** new_args);
+
 void grpc_proxy_mapper_destroy(grpc_proxy_mapper* mapper);
 
 #endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_PROXY_MAPPER_H */
diff --git a/src/core/ext/client_channel/proxy_mapper_registry.c b/src/core/ext/client_channel/proxy_mapper_registry.c
index 0a156c8b1c..2c44b9d490 100644
--- a/src/core/ext/client_channel/proxy_mapper_registry.c
+++ b/src/core/ext/client_channel/proxy_mapper_registry.c
@@ -61,15 +61,28 @@ static void grpc_proxy_mapper_list_register(grpc_proxy_mapper_list* list,
   ++list->num_mappers;
 }
 
-static bool grpc_proxy_mapper_list_map(grpc_exec_ctx* exec_ctx,
-                                       grpc_proxy_mapper_list* list,
-                                       const grpc_resolved_address* address,
-                                       const grpc_channel_args* args,
-                                       grpc_resolved_address** new_address,
-                                       grpc_channel_args** new_args) {
+static bool grpc_proxy_mapper_list_map_name(grpc_exec_ctx* exec_ctx,
+                                            grpc_proxy_mapper_list* list,
+                                            const char* server_uri,
+                                            const grpc_channel_args* args,
+                                            char** name_to_resolve,
+                                            grpc_channel_args** new_args) {
   for (size_t i = 0; i < list->num_mappers; ++i) {
-    if (grpc_proxy_mapper_map(exec_ctx, list->list[i], address, args,
-                              new_address, new_args)) {
+    if (grpc_proxy_mapper_map_name(exec_ctx, list->list[i], server_uri, args,
+                                   name_to_resolve, new_args)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+static bool grpc_proxy_mapper_list_map_address(
+    grpc_exec_ctx* exec_ctx, grpc_proxy_mapper_list* list,
+    const grpc_resolved_address* address, const grpc_channel_args* args,
+    grpc_resolved_address** new_address, grpc_channel_args** new_args) {
+  for (size_t i = 0; i < list->num_mappers; ++i) {
+    if (grpc_proxy_mapper_map_address(exec_ctx, list->list[i], address, args,
+                                      new_address, new_args)) {
       return true;
     }
   }
@@ -101,11 +114,20 @@ void grpc_proxy_mapper_register(bool at_start, grpc_proxy_mapper* mapper) {
   grpc_proxy_mapper_list_register(&g_proxy_mapper_list, at_start, mapper);
 }
 
-bool grpc_proxy_mappers_map(grpc_exec_ctx* exec_ctx,
-                            const grpc_resolved_address* address,
-                            const grpc_channel_args* args,
-                            grpc_resolved_address** new_address,
-                            grpc_channel_args** new_args) {
-  return grpc_proxy_mapper_list_map(exec_ctx, &g_proxy_mapper_list, address,
-                                    args, new_address, new_args);
+bool grpc_proxy_mappers_map_name(grpc_exec_ctx* exec_ctx,
+                                 const char* server_uri,
+                                 const grpc_channel_args* args,
+                                 char** name_to_resolve,
+                                 grpc_channel_args** new_args) {
+  return grpc_proxy_mapper_list_map_name(exec_ctx, &g_proxy_mapper_list,
+                                         server_uri, args, name_to_resolve,
+                                         new_args);
+}
+bool grpc_proxy_mappers_map_address(grpc_exec_ctx* exec_ctx,
+                                    const grpc_resolved_address* address,
+                                    const grpc_channel_args* args,
+                                    grpc_resolved_address** new_address,
+                                    grpc_channel_args** new_args) {
+  return grpc_proxy_mapper_list_map_address(
+      exec_ctx, &g_proxy_mapper_list, address, args, new_address, new_args);
 }
diff --git a/src/core/ext/client_channel/proxy_mapper_registry.h b/src/core/ext/client_channel/proxy_mapper_registry.h
index b76af8d456..742b57a2d4 100644
--- a/src/core/ext/client_channel/proxy_mapper_registry.h
+++ b/src/core/ext/client_channel/proxy_mapper_registry.h
@@ -44,10 +44,16 @@ void grpc_proxy_mapper_registry_shutdown();
 /// the list.  Otherwise, it will be added to the end.
 void grpc_proxy_mapper_register(bool at_start, grpc_proxy_mapper* mapper);
 
-bool grpc_proxy_mappers_map(grpc_exec_ctx* exec_ctx,
-                            const grpc_resolved_address* address,
-                            const grpc_channel_args* args,
-                            grpc_resolved_address** new_address,
-                            grpc_channel_args** new_args);
+bool grpc_proxy_mappers_map_name(grpc_exec_ctx* exec_ctx,
+                                 const char* server_uri,
+                                 const grpc_channel_args* args,
+                                 char** name_to_resolve,
+                                 grpc_channel_args** new_args);
+
+bool grpc_proxy_mappers_map_address(grpc_exec_ctx* exec_ctx,
+                                    const grpc_resolved_address* address,
+                                    const grpc_channel_args* args,
+                                    grpc_resolved_address** new_address,
+                                    grpc_channel_args** new_args);
 
 #endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_PROXY_MAPPER_REGISTRY_H */
diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c
index f1e4e079e2..09c68a91dd 100644
--- a/src/core/ext/client_channel/subchannel.c
+++ b/src/core/ext/client_channel/subchannel.c
@@ -217,7 +217,7 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
   grpc_slice_unref_internal(exec_ctx, c->initial_connect_string);
   grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
   grpc_connector_unref(exec_ctx, c->connector);
-  grpc_pollset_set_destroy(c->pollset_set);
+  grpc_pollset_set_destroy(exec_ctx, c->pollset_set);
   grpc_subchannel_key_destroy(exec_ctx, c->key);
   gpr_free(c);
 }
@@ -336,8 +336,8 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
   grpc_set_initial_connect_string(&addr, &c->initial_connect_string);
   grpc_resolved_address *new_address = NULL;
   grpc_channel_args *new_args = NULL;
-  if (grpc_proxy_mappers_map(exec_ctx, addr, args->args, &new_address,
-                             &new_args)) {
+  if (grpc_proxy_mappers_map_address(exec_ctx, addr, args->args, &new_address,
+                                     &new_args)) {
     GPR_ASSERT(new_address != NULL);
     gpr_free(addr);
     addr = new_address;
@@ -419,7 +419,7 @@ grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c,
                                                            grpc_error **error) {
   grpc_connectivity_state state;
   gpr_mu_lock(&c->mu);
-  state = grpc_connectivity_state_check(&c->state_tracker, error);
+  state = grpc_connectivity_state_get(&c->state_tracker, error);
   gpr_mu_unlock(&c->mu);
   return state;
 }
@@ -788,7 +788,8 @@ grpc_call_stack *grpc_subchannel_call_get_call_stack(
   return SUBCHANNEL_CALL_TO_CALL_STACK(subchannel_call);
 }
 
-static void grpc_uri_to_sockaddr(char *uri_str, grpc_resolved_address *addr) {
+static void grpc_uri_to_sockaddr(const char *uri_str,
+                                 grpc_resolved_address *addr) {
   grpc_uri *uri = grpc_uri_parse(uri_str, 0 /* suppress_errors */);
   GPR_ASSERT(uri != NULL);
   if (strcmp(uri->scheme, "ipv4") == 0) {
@@ -803,14 +804,19 @@ static void grpc_uri_to_sockaddr(char *uri_str, grpc_resolved_address *addr) {
 
 void grpc_get_subchannel_address_arg(const grpc_channel_args *args,
                                      grpc_resolved_address *addr) {
+  const char *addr_uri_str = grpc_get_subchannel_address_uri_arg(args);
+  memset(addr, 0, sizeof(*addr));
+  if (*addr_uri_str != '\0') {
+    grpc_uri_to_sockaddr(addr_uri_str, addr);
+  }
+}
+
+const char *grpc_get_subchannel_address_uri_arg(const grpc_channel_args *args) {
   const grpc_arg *addr_arg =
       grpc_channel_args_find(args, GRPC_ARG_SUBCHANNEL_ADDRESS);
   GPR_ASSERT(addr_arg != NULL);  // Should have been set by LB policy.
   GPR_ASSERT(addr_arg->type == GRPC_ARG_STRING);
-  memset(addr, 0, sizeof(*addr));
-  if (*addr_arg->value.string != '\0') {
-    grpc_uri_to_sockaddr(addr_arg->value.string, addr);
-  }
+  return addr_arg->value.string;
 }
 
 grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address *addr) {
diff --git a/src/core/ext/client_channel/subchannel.h b/src/core/ext/client_channel/subchannel.h
index 9bd35a7704..26ce954487 100644
--- a/src/core/ext/client_channel/subchannel.h
+++ b/src/core/ext/client_channel/subchannel.h
@@ -178,6 +178,9 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
 void grpc_get_subchannel_address_arg(const grpc_channel_args *args,
                                      grpc_resolved_address *addr);
 
+/// Returns the URI string for the address to connect to.
+const char *grpc_get_subchannel_address_uri_arg(const grpc_channel_args *args);
+
 /// Returns a new channel arg encoding the subchannel address as a string.
 /// Caller is responsible for freeing the string.
 grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address *addr);
diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c
index 308facb7e7..8a2af48328 100644
--- a/src/core/ext/lb_policy/grpclb/grpclb.c
+++ b/src/core/ext/lb_policy/grpclb/grpclb.c
@@ -112,11 +112,13 @@
 #include "src/core/ext/client_channel/lb_policy_registry.h"
 #include "src/core/ext/client_channel/parse_address.h"
 #include "src/core/ext/lb_policy/grpclb/grpclb.h"
+#include "src/core/ext/lb_policy/grpclb/grpclb_channel.h"
 #include "src/core/ext/lb_policy/grpclb/load_balancer_api.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/iomgr/sockaddr.h"
 #include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "src/core/lib/iomgr/timer.h"
+#include "src/core/lib/slice/slice_hash_table.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/slice/slice_string_helpers.h"
 #include "src/core/lib/support/backoff.h"
@@ -490,9 +492,8 @@ static grpc_lb_addresses *process_serverlist_locked(
 static bool update_lb_connectivity_status_locked(
     grpc_exec_ctx *exec_ctx, glb_lb_policy *glb_policy,
     grpc_connectivity_state new_rr_state, grpc_error *new_rr_state_error) {
-  grpc_error *curr_state_error;
-  const grpc_connectivity_state curr_glb_state = grpc_connectivity_state_check(
-      &glb_policy->state_tracker, &curr_state_error);
+  const grpc_connectivity_state curr_glb_state =
+      grpc_connectivity_state_check(&glb_policy->state_tracker);
 
   /* The new connectivity status is a function of the previous one and the new
    * input coming from the status of the RR policy.
@@ -751,6 +752,96 @@ static void glb_rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
   GRPC_ERROR_UNREF(error);
 }
 
+static void destroy_balancer_name(grpc_exec_ctx *exec_ctx,
+                                  void *balancer_name) {
+  gpr_free(balancer_name);
+}
+
+static void *copy_balancer_name(void *balancer_name) {
+  return gpr_strdup(balancer_name);
+}
+
+static grpc_slice_hash_table_entry targets_info_entry_create(
+    const char *address, const char *balancer_name) {
+  static const grpc_slice_hash_table_vtable vtable = {destroy_balancer_name,
+                                                      copy_balancer_name};
+  grpc_slice_hash_table_entry entry;
+  entry.key = grpc_slice_from_copied_string(address);
+  entry.value = (void *)balancer_name;
+  entry.vtable = &vtable;
+  return entry;
+}
+
+/* Returns the target URI for the LB service whose addresses are in \a
+ * addresses.  Using this URI, a bidirectional streaming channel will be created
+ * for the reception of load balancing updates.
+ *
+ * The output argument \a targets_info will be updated to contain a mapping of
+ * "LB server address" to "balancer name", as reported by the naming system.
+ * This mapping will be propagated via the channel arguments of the
+ * aforementioned LB streaming channel, to be used by the security connector for
+ * secure naming checks. The user is responsible for freeing \a targets_info. */
+static char *get_lb_uri_target_addresses(grpc_exec_ctx *exec_ctx,
+                                         const grpc_lb_addresses *addresses,
+                                         grpc_slice_hash_table **targets_info) {
+  size_t num_grpclb_addrs = 0;
+  for (size_t i = 0; i < addresses->num_addresses; ++i) {
+    if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs;
+  }
+  /* All input addresses come from a resolver that claims they are LB services.
+   * It's the resolver's responsibility to make sure this policy is only
+   * instantiated and used in that case. Otherwise, something has gone wrong. */
+  GPR_ASSERT(num_grpclb_addrs > 0);
+
+  grpc_slice_hash_table_entry *targets_info_entries =
+      gpr_malloc(sizeof(*targets_info_entries) * num_grpclb_addrs);
+
+  /* construct a target ipvX://ip1:port1,ip2:port2,... from the addresses in \a
+   * addresses */
+  /* TODO(dgq): support mixed ip version */
+  char **addr_strs = gpr_malloc(sizeof(char *) * num_grpclb_addrs);
+  size_t addr_index = 0;
+
+  for (size_t i = 0; i < addresses->num_addresses; i++) {
+    if (addresses->addresses[i].user_data != NULL) {
+      gpr_log(GPR_ERROR,
+              "This LB policy doesn't support user data. It will be ignored");
+    }
+    if (addresses->addresses[i].is_balancer) {
+      char *addr_str;
+      GPR_ASSERT(grpc_sockaddr_to_string(
+                     &addr_str, &addresses->addresses[i].address, true) > 0);
+      targets_info_entries[addr_index] = targets_info_entry_create(
+          addr_str, addresses->addresses[i].balancer_name);
+      addr_strs[addr_index++] = addr_str;
+    }
+  }
+  GPR_ASSERT(addr_index == num_grpclb_addrs);
+
+  size_t uri_path_len;
+  char *uri_path = gpr_strjoin_sep((const char **)addr_strs, num_grpclb_addrs,
+                                   ",", &uri_path_len);
+  for (size_t i = 0; i < num_grpclb_addrs; i++) gpr_free(addr_strs[i]);
+  gpr_free(addr_strs);
+
+  char *target_uri_str = NULL;
+  /* TODO(dgq): Don't assume all addresses will share the scheme of the first
+   * one */
+  gpr_asprintf(&target_uri_str, "%s:%s",
+               grpc_sockaddr_get_uri_scheme(&addresses->addresses[0].address),
+               uri_path);
+  gpr_free(uri_path);
+
+  *targets_info =
+      grpc_slice_hash_table_create(num_grpclb_addrs, targets_info_entries);
+  for (size_t i = 0; i < num_grpclb_addrs; i++) {
+    grpc_slice_unref_internal(exec_ctx, targets_info_entries[i].key);
+  }
+  gpr_free(targets_info_entries);
+
+  return target_uri_str;
+}
+
 static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
                                   grpc_lb_policy_factory *factory,
                                   grpc_lb_policy_args *args) {
@@ -788,85 +879,30 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
   }
   grpc_uri_destroy(uri);
 
-  /* All input addresses in addresses come from a resolver that claims
-   * they are LB services. It's the resolver's responsibility to make sure
-   * this policy is only instantiated and used in that case.
-   *
-   * Create a client channel over them to communicate with a LB service */
   glb_policy->cc_factory = args->client_channel_factory;
   glb_policy->args = grpc_channel_args_copy(args->args);
   GPR_ASSERT(glb_policy->cc_factory != NULL);
 
-  /* construct a target from the addresses in args, given in the form
-   * ipvX://ip1:port1,ip2:port2,...
-   * TODO(dgq): support mixed ip version */
-  char **addr_strs = gpr_malloc(sizeof(char *) * num_grpclb_addrs);
-  size_t addr_index = 0;
-  for (size_t i = 0; i < addresses->num_addresses; i++) {
-    if (addresses->addresses[i].user_data != NULL) {
-      gpr_log(GPR_ERROR,
-              "This LB policy doesn't support user data. It will be ignored");
-    }
-    if (addresses->addresses[i].is_balancer) {
-      if (addr_index == 0) {
-        addr_strs[addr_index++] =
-            grpc_sockaddr_to_uri(&addresses->addresses[i].address);
-      } else {
-        GPR_ASSERT(grpc_sockaddr_to_string(&addr_strs[addr_index++],
-                                           &addresses->addresses[i].address,
-                                           true) > 0);
-      }
-    }
-  }
-  size_t uri_path_len;
-  char *target_uri_str = gpr_strjoin_sep((const char **)addr_strs,
-                                         num_grpclb_addrs, ",", &uri_path_len);
-
-  /* Create a channel to talk to the LBs.
-   *
-   * We strip out the channel arg for the LB policy name, since we want
-   * to use the default (pick_first) in this case.
-   *
-   * We also strip out the channel arg for the resolved addresses, since
-   * that will be generated by the name resolver used in the LB channel.
-   * Note that the LB channel will use the sockaddr resolver, so this
-   * won't actually generate a query to DNS (or some other name service).
-   * However, the addresses returned by the sockaddr resolver will have
-   * is_balancer=false, whereas our own addresses have is_balancer=true.
-   * We need the LB channel to return addresses with is_balancer=false
-   * so that it does not wind up recursively using the grpclb LB policy,
-   * as per the special case logic in client_channel.c.
-   *
-   * Finally, we also strip out the channel arg for the server URI,
-   * since that will be different for the LB channel than for the parent
-   * channel.  (The client channel factory will re-add this arg with
-   * the right value.)
-   */
-  static const char *keys_to_remove[] = {
-      GRPC_ARG_LB_POLICY_NAME, GRPC_ARG_LB_ADDRESSES, GRPC_ARG_SERVER_URI};
-  grpc_channel_args *new_args = grpc_channel_args_copy_and_remove(
-      args->args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove));
-  glb_policy->lb_channel = grpc_client_channel_factory_create_channel(
-      exec_ctx, glb_policy->cc_factory, target_uri_str,
-      GRPC_CLIENT_CHANNEL_TYPE_LOAD_BALANCING, new_args);
-  grpc_channel_args_destroy(exec_ctx, new_args);
-
-  gpr_free(target_uri_str);
-  for (size_t i = 0; i < num_grpclb_addrs; i++) {
-    gpr_free(addr_strs[i]);
-  }
-  gpr_free(addr_strs);
-
+  grpc_slice_hash_table *targets_info = NULL;
+  /* Create a client channel over them to communicate with a LB service */
+  char *lb_service_target_addresses =
+      get_lb_uri_target_addresses(exec_ctx, addresses, &targets_info);
+  grpc_channel_args *lb_channel_args =
+      get_lb_channel_args(exec_ctx, targets_info, args->args);
+  glb_policy->lb_channel = grpc_lb_policy_grpclb_create_lb_channel(
+      exec_ctx, lb_service_target_addresses, args->client_channel_factory,
+      lb_channel_args);
+  grpc_slice_hash_table_unref(exec_ctx, targets_info);
+  grpc_channel_args_destroy(exec_ctx, lb_channel_args);
+  gpr_free(lb_service_target_addresses);
   if (glb_policy->lb_channel == NULL) {
     gpr_free(glb_policy);
     return NULL;
   }
-
   grpc_lb_policy_init(&glb_policy->base, &glb_lb_policy_vtable);
   gpr_mu_init(&glb_policy->mu);
   grpc_connectivity_state_init(&glb_policy->state_tracker, GRPC_CHANNEL_IDLE,
                                "grpclb");
-
   return &glb_policy->base;
 }
 
@@ -1061,8 +1097,8 @@ static grpc_connectivity_state glb_check_connectivity(
   glb_lb_policy *glb_policy = (glb_lb_policy *)pol;
   grpc_connectivity_state st;
   gpr_mu_lock(&glb_policy->mu);
-  st = grpc_connectivity_state_check(&glb_policy->state_tracker,
-                                     connectivity_error);
+  st = grpc_connectivity_state_get(&glb_policy->state_tracker,
+                                   connectivity_error);
   gpr_mu_unlock(&glb_policy->mu);
   return st;
 }
diff --git a/src/core/ext/lb_policy/grpclb/grpclb_channel.c b/src/core/ext/lb_policy/grpclb/grpclb_channel.c
new file mode 100644
index 0000000000..1b8bbab1b6
--- /dev/null
+++ b/src/core/ext/lb_policy/grpclb/grpclb_channel.c
@@ -0,0 +1,77 @@
+/*
+ *
+ * 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/alloc.h>
+#include <grpc/support/string_util.h>
+
+#include "src/core/ext/client_channel/client_channel.h"
+#include "src/core/ext/lb_policy/grpclb/grpclb_channel.h"
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/iomgr/sockaddr_utils.h"
+#include "src/core/lib/support/string.h"
+
+grpc_channel *grpc_lb_policy_grpclb_create_lb_channel(
+    grpc_exec_ctx *exec_ctx, const char *lb_service_target_addresses,
+    grpc_client_channel_factory *client_channel_factory,
+    grpc_channel_args *args) {
+  grpc_channel *lb_channel = grpc_client_channel_factory_create_channel(
+      exec_ctx, client_channel_factory, lb_service_target_addresses,
+      GRPC_CLIENT_CHANNEL_TYPE_LOAD_BALANCING, args);
+  return lb_channel;
+}
+
+grpc_channel_args *get_lb_channel_args(grpc_exec_ctx *exec_ctx,
+                                       grpc_slice_hash_table *targets_info,
+                                       const grpc_channel_args *args) {
+  /* We strip out the channel arg for the LB policy name, since we want
+   * to use the default (pick_first) in this case.
+   *
+   * We also strip out the channel arg for the resolved addresses, since
+   * that will be generated by the name resolver used in the LB channel.
+   * Note that the LB channel will use the sockaddr resolver, so this
+   * won't actually generate a query to DNS (or some other name service).
+   * However, the addresses returned by the sockaddr resolver will have
+   * is_balancer=false, whereas our own addresses have is_balancer=true.
+   * We need the LB channel to return addresses with is_balancer=false
+   * so that it does not wind up recursively using the grpclb LB policy,
+   * as per the special case logic in client_channel.c.
+   *
+   * Lastly, we also strip out the channel arg for the server URI,
+   * since that will be different for the LB channel than for the parent
+   * channel (the client channel factory will re-add this arg with
+   * the right value). */
+  static const char *keys_to_remove[] = {
+      GRPC_ARG_LB_POLICY_NAME, GRPC_ARG_LB_ADDRESSES, GRPC_ARG_SERVER_URI};
+  return grpc_channel_args_copy_and_remove(args, keys_to_remove,
+                                           GPR_ARRAY_SIZE(keys_to_remove));
+}
diff --git a/src/core/ext/lb_policy/grpclb/grpclb_channel.h b/src/core/ext/lb_policy/grpclb/grpclb_channel.h
new file mode 100644
index 0000000000..f66082d78e
--- /dev/null
+++ b/src/core/ext/lb_policy/grpclb/grpclb_channel.h
@@ -0,0 +1,56 @@
+/*
+ *
+ * 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_LB_POLICY_GRPCLB_GRPCLB_CHANNEL_H
+#define GRPC_CORE_EXT_LB_POLICY_GRPCLB_GRPCLB_CHANNEL_H
+
+#include "src/core/ext/client_channel/lb_policy_factory.h"
+#include "src/core/lib/slice/slice_hash_table.h"
+
+/** Create the channel used for communicating with an LB service.
+ * Note that an LB *service* may be comprised of several LB *servers*.
+ *
+ * \a lb_service_target_addresses is the target URI containing the addresses
+ * from resolving the LB service's name (eg, ipv4:10.0.0.1:1234,10.2.3.4:9876).
+ * \a client_channel_factory will be used for the creation of the LB channel,
+ * alongside the channel args passed in \a args. */
+grpc_channel *grpc_lb_policy_grpclb_create_lb_channel(
+    grpc_exec_ctx *exec_ctx, const char *lb_service_target_addresses,
+    grpc_client_channel_factory *client_channel_factory,
+    grpc_channel_args *args);
+
+grpc_channel_args *get_lb_channel_args(grpc_exec_ctx *exec_ctx,
+                                       grpc_slice_hash_table *targets_info,
+                                       const grpc_channel_args *args);
+
+#endif /* GRPC_CORE_EXT_LB_POLICY_GRPCLB_GRPCLB_CHANNEL_H */
diff --git a/src/core/ext/lb_policy/grpclb/grpclb_channel_secure.c b/src/core/ext/lb_policy/grpclb/grpclb_channel_secure.c
new file mode 100644
index 0000000000..2fee5f1b8e
--- /dev/null
+++ b/src/core/ext/lb_policy/grpclb/grpclb_channel_secure.c
@@ -0,0 +1,107 @@
+/*
+ *
+ * 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/alloc.h>
+#include <grpc/support/string_util.h>
+
+#include "src/core/ext/client_channel/client_channel.h"
+#include "src/core/ext/lb_policy/grpclb/grpclb_channel.h"
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/iomgr/sockaddr_utils.h"
+#include "src/core/lib/security/credentials/credentials.h"
+#include "src/core/lib/security/transport/lb_targets_info.h"
+#include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/support/string.h"
+
+grpc_channel *grpc_lb_policy_grpclb_create_lb_channel(
+    grpc_exec_ctx *exec_ctx, const char *lb_service_target_addresses,
+    grpc_client_channel_factory *client_channel_factory,
+    grpc_channel_args *args) {
+  grpc_channel_args *new_args = args;
+  grpc_channel_credentials *channel_credentials =
+      grpc_channel_credentials_find_in_args(args);
+  if (channel_credentials != NULL) {
+    /* Substitute the channel credentials with a version without call
+     * credentials: the load balancer is not necessarily trusted to handle
+     * bearer token credentials */
+    static const char *keys_to_remove[] = {GRPC_ARG_CHANNEL_CREDENTIALS};
+    grpc_channel_credentials *creds_sans_call_creds =
+        grpc_channel_credentials_duplicate_without_call_credentials(
+            channel_credentials);
+    GPR_ASSERT(creds_sans_call_creds != NULL);
+    grpc_arg args_to_add[] = {
+        grpc_channel_credentials_to_arg(creds_sans_call_creds)};
+    /* Create the new set of channel args */
+    new_args = grpc_channel_args_copy_and_add_and_remove(
+        args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), args_to_add,
+        GPR_ARRAY_SIZE(args_to_add));
+    grpc_channel_credentials_unref(exec_ctx, creds_sans_call_creds);
+  }
+  grpc_channel *lb_channel = grpc_client_channel_factory_create_channel(
+      exec_ctx, client_channel_factory, lb_service_target_addresses,
+      GRPC_CLIENT_CHANNEL_TYPE_LOAD_BALANCING, new_args);
+  if (channel_credentials != NULL) {
+    grpc_channel_args_destroy(exec_ctx, new_args);
+  }
+  return lb_channel;
+}
+
+grpc_channel_args *get_lb_channel_args(grpc_exec_ctx *exec_ctx,
+                                       grpc_slice_hash_table *targets_info,
+                                       const grpc_channel_args *args) {
+  const grpc_arg targets_info_arg =
+      grpc_lb_targets_info_create_channel_arg(targets_info);
+  /* We strip out the channel arg for the LB policy name, since we want
+   * to use the default (pick_first) in this case.
+   *
+   * We also strip out the channel arg for the resolved addresses, since
+   * that will be generated by the name resolver used in the LB channel.
+   * Note that the LB channel will use the sockaddr resolver, so this
+   * won't actually generate a query to DNS (or some other name service).
+   * However, the addresses returned by the sockaddr resolver will have
+   * is_balancer=false, whereas our own addresses have is_balancer=true.
+   * We need the LB channel to return addresses with is_balancer=false
+   * so that it does not wind up recursively using the grpclb LB policy,
+   * as per the special case logic in client_channel.c.
+   *
+   * Lastly, we also strip out the channel arg for the server URI,
+   * since that will be different for the LB channel than for the parent
+   * channel (the client channel factory will re-add this arg with
+   * the right value). */
+  static const char *keys_to_remove[] = {
+      GRPC_ARG_LB_POLICY_NAME, GRPC_ARG_LB_ADDRESSES, GRPC_ARG_SERVER_URI};
+  /* Add the targets info table to be used for secure naming */
+  return grpc_channel_args_copy_and_add_and_remove(
+      args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), &targets_info_arg,
+      1);
+}
diff --git a/src/core/ext/lb_policy/pick_first/pick_first.c b/src/core/ext/lb_policy/pick_first/pick_first.c
index 9f2aa461be..1b965183f6 100644
--- a/src/core/ext/lb_policy/pick_first/pick_first.c
+++ b/src/core/ext/lb_policy/pick_first/pick_first.c
@@ -398,7 +398,7 @@ static grpc_connectivity_state pf_check_connectivity(grpc_exec_ctx *exec_ctx,
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
   grpc_connectivity_state st;
   gpr_mu_lock(&p->mu);
-  st = grpc_connectivity_state_check(&p->state_tracker, error);
+  st = grpc_connectivity_state_get(&p->state_tracker, error);
   gpr_mu_unlock(&p->mu);
   return st;
 }
diff --git a/src/core/ext/lb_policy/round_robin/round_robin.c b/src/core/ext/lb_policy/round_robin/round_robin.c
index d17d8fa057..63e3d033ad 100644
--- a/src/core/ext/lb_policy/round_robin/round_robin.c
+++ b/src/core/ext/lb_policy/round_robin/round_robin.c
@@ -655,7 +655,7 @@ static grpc_connectivity_state rr_check_connectivity(grpc_exec_ctx *exec_ctx,
   round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
   grpc_connectivity_state st;
   gpr_mu_lock(&p->mu);
-  st = grpc_connectivity_state_check(&p->state_tracker, error);
+  st = grpc_connectivity_state_get(&p->state_tracker, error);
   gpr_mu_unlock(&p->mu);
   return st;
 }
@@ -739,6 +739,13 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
     sc_args.args = new_args;
     grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel(
         exec_ctx, args->client_channel_factory, &sc_args);
+    if (grpc_lb_round_robin_trace) {
+      char *address_uri =
+          grpc_sockaddr_to_uri(&addresses->addresses[i].address);
+      gpr_log(GPR_DEBUG, "Created subchannel %p for address uri %s",
+              (void *)subchannel, address_uri);
+      gpr_free(address_uri);
+    }
     grpc_channel_args_destroy(exec_ctx, new_args);
 
     if (subchannel != NULL) {
diff --git a/src/core/ext/load_reporting/load_reporting.c b/src/core/ext/load_reporting/load_reporting.c
index 37b06a737f..942aea4fd1 100644
--- a/src/core/ext/load_reporting/load_reporting.c
+++ b/src/core/ext/load_reporting/load_reporting.c
@@ -34,14 +34,34 @@
 #include <limits.h>
 #include <string.h>
 
+#include <grpc/load_reporting.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/sync.h>
 
 #include "src/core/ext/load_reporting/load_reporting.h"
 #include "src/core/ext/load_reporting/load_reporting_filter.h"
 #include "src/core/lib/channel/channel_stack_builder.h"
+#include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/surface/call.h"
 #include "src/core/lib/surface/channel_init.h"
 
+static void destroy_lr_cost_context(void *c) {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_load_reporting_cost_context *cost_ctx = c;
+  for (size_t i = 0; i < cost_ctx->values_count; ++i) {
+    grpc_slice_unref_internal(&exec_ctx, cost_ctx->values[i]);
+  }
+  grpc_exec_ctx_finish(&exec_ctx);
+  gpr_free(cost_ctx->values);
+  gpr_free(cost_ctx);
+}
+
+void grpc_call_set_load_reporting_cost_context(
+    grpc_call *call, grpc_load_reporting_cost_context *ctx) {
+  grpc_call_context_set(call, GRPC_CONTEXT_LR_COST, ctx,
+                        destroy_lr_cost_context);
+}
+
 static bool is_load_reporting_enabled(const grpc_channel_args *a) {
   if (a == NULL) return false;
   for (size_t i = 0; i < a->num_args; i++) {
diff --git a/src/core/ext/load_reporting/load_reporting.h b/src/core/ext/load_reporting/load_reporting.h
index a157844734..22859a599a 100644
--- a/src/core/ext/load_reporting/load_reporting.h
+++ b/src/core/ext/load_reporting/load_reporting.h
@@ -35,23 +35,8 @@
 #define GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_H
 
 #include <grpc/impl/codegen/grpc_types.h>
-#include "src/core/lib/channel/channel_stack.h"
-
-/** Metadata key for the gRPC LB load balancer token.
- *
- * The value corresponding to this key is an opaque token that is given to the
- * frontend as part of each pick; the frontend sends this token to the backend
- * in each request it sends when using that pick. The token is used by the
- * backend to verify the request and to allow the backend to report load to the
- * gRPC LB system. */
-#define GRPC_LB_TOKEN_MD_KEY "lb-token"
 
-/** Metadata key for gRPC LB cost reporting.
- *
- * The value corresponding to this key is an opaque binary blob reported by the
- * backend as part of its trailing metadata containing cost information for the
- * call. */
-#define GRPC_LB_COST_MD_KEY "lb-cost-bin"
+#include "src/core/lib/channel/channel_stack.h"
 
 /** Identifiers for the invocation point of the users LR callback */
 typedef enum grpc_load_reporting_source {
diff --git a/src/core/ext/load_reporting/load_reporting_filter.c b/src/core/ext/load_reporting/load_reporting_filter.c
index 8af6191c3b..7fc88db603 100644
--- a/src/core/ext/load_reporting/load_reporting_filter.c
+++ b/src/core/ext/load_reporting/load_reporting_filter.c
@@ -31,11 +31,13 @@
  *
  */
 
+#include <string.h>
+
+#include <grpc/load_reporting.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include <grpc/support/sync.h>
-#include <string.h>
 
 #include "src/core/ext/load_reporting/load_reporting.h"
 #include "src/core/ext/load_reporting/load_reporting_filter.h"
@@ -46,8 +48,6 @@
 
 typedef struct call_data {
   intptr_t id; /**< an id unique to the call */
-  bool have_trailing_md_string;
-  grpc_slice trailing_md_string;
   bool have_initial_md_string;
   grpc_slice initial_md_string;
   bool have_service_method;
@@ -142,9 +142,6 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
   if (calld->have_initial_md_string) {
     grpc_slice_unref_internal(exec_ctx, calld->initial_md_string);
   }
-  if (calld->have_trailing_md_string) {
-    grpc_slice_unref_internal(exec_ctx, calld->trailing_md_string);
-  }
   if (calld->have_service_method) {
     grpc_slice_unref_internal(exec_ctx, calld->service_method);
   }
@@ -201,15 +198,6 @@ static void lr_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
     /* substitute our callback for the higher callback */
     calld->ops_recv_initial_metadata_ready = op->recv_initial_metadata_ready;
     op->recv_initial_metadata_ready = &calld->on_initial_md_ready;
-  } else if (op->send_trailing_metadata) {
-    if (op->send_trailing_metadata->idx.named.lb_cost_bin != NULL) {
-      calld->trailing_md_string = grpc_slice_ref_internal(
-          GRPC_MDVALUE(op->send_trailing_metadata->idx.named.lb_cost_bin->md));
-      calld->have_trailing_md_string = true;
-      grpc_metadata_batch_remove(
-          exec_ctx, op->send_trailing_metadata,
-          op->send_trailing_metadata->idx.named.lb_cost_bin);
-    }
   }
   grpc_call_next_op(exec_ctx, elem, op);
 
diff --git a/src/core/ext/resolver/dns/native/dns_resolver.c b/src/core/ext/resolver/dns/native/dns_resolver.c
index 2c9623211b..c08b53ea04 100644
--- a/src/core/ext/resolver/dns/native/dns_resolver.c
+++ b/src/core/ext/resolver/dns/native/dns_resolver.c
@@ -244,7 +244,7 @@ static void dns_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *gr) {
   if (r->resolved_result != NULL) {
     grpc_channel_args_destroy(exec_ctx, r->resolved_result);
   }
-  grpc_pollset_set_destroy(r->interested_parties);
+  grpc_pollset_set_destroy(exec_ctx, r->interested_parties);
   gpr_free(r->name_to_resolve);
   gpr_free(r->default_port);
   grpc_channel_args_destroy(exec_ctx, r->channel_args);
diff --git a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
index d3e53984f2..d8c18eb122 100644
--- a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
+++ b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
@@ -40,10 +40,15 @@
 
 #include "src/core/ext/client_channel/client_channel.h"
 #include "src/core/ext/client_channel/resolver_registry.h"
+#include "src/core/ext/client_channel/uri_parser.h"
 #include "src/core/ext/transport/chttp2/client/chttp2_connector.h"
 #include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "src/core/lib/security/credentials/credentials.h"
+#include "src/core/lib/security/transport/lb_targets_info.h"
 #include "src/core/lib/security/transport/security_connector.h"
+#include "src/core/lib/slice/slice_hash_table.h"
+#include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/surface/api_trace.h"
 #include "src/core/lib/surface/channel.h"
 
@@ -53,12 +58,114 @@ static void client_channel_factory_ref(
 static void client_channel_factory_unref(
     grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory) {}
 
+static grpc_subchannel_args *get_secure_naming_subchannel_args(
+    grpc_exec_ctx *exec_ctx, const grpc_subchannel_args *args) {
+  grpc_channel_credentials *channel_credentials =
+      grpc_channel_credentials_find_in_args(args->args);
+  if (channel_credentials == NULL) {
+    gpr_log(GPR_ERROR,
+            "Can't create subchannel: channel credentials missing for secure "
+            "channel.");
+    return NULL;
+  }
+  // Make sure security connector does not already exist in args.
+  if (grpc_security_connector_find_in_args(args->args) != NULL) {
+    gpr_log(GPR_ERROR,
+            "Can't create subchannel: security connector already present in "
+            "channel args.");
+    return NULL;
+  }
+  // To which address are we connecting? By default, use the server URI.
+  const grpc_arg *server_uri_arg =
+      grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI);
+  GPR_ASSERT(server_uri_arg != NULL);
+  GPR_ASSERT(server_uri_arg->type == GRPC_ARG_STRING);
+  const char *server_uri_str = server_uri_arg->value.string;
+  GPR_ASSERT(server_uri_str != NULL);
+  grpc_uri *server_uri =
+      grpc_uri_parse(server_uri_str, true /* supress errors */);
+  GPR_ASSERT(server_uri != NULL);
+  const char *server_uri_path;
+  server_uri_path =
+      server_uri->path[0] == '/' ? server_uri->path + 1 : server_uri->path;
+  const grpc_slice_hash_table *targets_info =
+      grpc_lb_targets_info_find_in_args(args->args);
+  char *target_name_to_check = NULL;
+  if (targets_info != NULL) {  // LB channel
+    // Find the balancer name for the target.
+    const char *target_uri_str =
+        grpc_get_subchannel_address_uri_arg(args->args);
+    grpc_uri *target_uri =
+        grpc_uri_parse(target_uri_str, false /* suppress errors */);
+    GPR_ASSERT(target_uri != NULL);
+    if (target_uri->path[0] != '\0') {  // "path" may be empty
+      const grpc_slice key = grpc_slice_from_static_string(
+          target_uri->path[0] == '/' ? target_uri->path + 1 : target_uri->path);
+      const char *value = grpc_slice_hash_table_get(targets_info, key);
+      if (value != NULL) target_name_to_check = gpr_strdup(value);
+      grpc_slice_unref_internal(exec_ctx, key);
+    }
+    if (target_name_to_check == NULL) {
+      // If the target name to check hasn't already been set, fall back to using
+      // SERVER_URI
+      target_name_to_check = gpr_strdup(server_uri_path);
+    }
+    grpc_uri_destroy(target_uri);
+  } else {  // regular channel: the secure name is the original server URI.
+    target_name_to_check = gpr_strdup(server_uri_path);
+  }
+  grpc_uri_destroy(server_uri);
+  GPR_ASSERT(target_name_to_check != NULL);
+  grpc_channel_security_connector *subchannel_security_connector = NULL;
+  // Create the security connector using the credentials and target name.
+  grpc_channel_args *new_args_from_connector = NULL;
+  const grpc_security_status security_status =
+      grpc_channel_credentials_create_security_connector(
+          exec_ctx, channel_credentials, target_name_to_check, args->args,
+          &subchannel_security_connector, &new_args_from_connector);
+  if (security_status != GRPC_SECURITY_OK) {
+    gpr_log(GPR_ERROR,
+            "Failed to create secure subchannel for secure name '%s'",
+            target_name_to_check);
+    gpr_free(target_name_to_check);
+    return NULL;
+  }
+  gpr_free(target_name_to_check);
+  grpc_arg new_security_connector_arg =
+      grpc_security_connector_to_arg(&subchannel_security_connector->base);
+
+  grpc_channel_args *new_args = grpc_channel_args_copy_and_add(
+      new_args_from_connector != NULL ? new_args_from_connector : args->args,
+      &new_security_connector_arg, 1);
+  GRPC_SECURITY_CONNECTOR_UNREF(exec_ctx, &subchannel_security_connector->base,
+                                "lb_channel_create");
+  if (new_args_from_connector != NULL) {
+    grpc_channel_args_destroy(exec_ctx, new_args_from_connector);
+  }
+  grpc_subchannel_args *final_sc_args = gpr_malloc(sizeof(*final_sc_args));
+  memcpy(final_sc_args, args, sizeof(*args));
+  final_sc_args->args = new_args;
+  return final_sc_args;
+}
+
 static grpc_subchannel *client_channel_factory_create_subchannel(
     grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
     const grpc_subchannel_args *args) {
+  grpc_subchannel_args *subchannel_args =
+      get_secure_naming_subchannel_args(exec_ctx, args);
+  if (subchannel_args == NULL) {
+    gpr_log(
+        GPR_ERROR,
+        "Failed to create subchannel arguments during subchannel creation.");
+    return NULL;
+  }
   grpc_connector *connector = grpc_chttp2_connector_create();
-  grpc_subchannel *s = grpc_subchannel_create(exec_ctx, connector, args);
+  grpc_subchannel *s =
+      grpc_subchannel_create(exec_ctx, connector, subchannel_args);
   grpc_connector_unref(exec_ctx, connector);
+  grpc_channel_args_destroy(exec_ctx,
+                            (grpc_channel_args *)subchannel_args->args);
+  gpr_free(subchannel_args);
   return s;
 }
 
@@ -91,10 +198,10 @@ static const grpc_client_channel_factory_vtable client_channel_factory_vtable =
 static grpc_client_channel_factory client_channel_factory = {
     &client_channel_factory_vtable};
 
-/* Create a secure client channel:
-   Asynchronously: - resolve target
-                   - connect to it (trying alternatives as presented)
-                   - perform handshakes */
+// Create a secure client channel:
+//   Asynchronously: - resolve target
+//                   - connect to it (trying alternatives as presented)
+//                   - perform handshakes
 grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds,
                                          const char *target,
                                          const grpc_channel_args *args,
@@ -103,47 +210,25 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds,
   GRPC_API_TRACE(
       "grpc_secure_channel_create(creds=%p, target=%s, args=%p, "
       "reserved=%p)",
-      4, (creds, target, args, reserved));
+      4, ((void *)creds, target, (void *)args, (void *)reserved));
   GPR_ASSERT(reserved == NULL);
-  // Make sure security connector does not already exist in args.
-  if (grpc_find_security_connector_in_args(args) != NULL) {
-    gpr_log(GPR_ERROR, "Cannot set security context in channel args.");
+  grpc_channel *channel = NULL;
+  if (creds != NULL) {
+    // Add channel args containing the client channel factory and channel
+    // credentials.
+    grpc_arg args_to_add[] = {
+        grpc_client_channel_factory_create_channel_arg(&client_channel_factory),
+        grpc_channel_credentials_to_arg(creds)};
+    grpc_channel_args *new_args = grpc_channel_args_copy_and_add(
+        args, args_to_add, GPR_ARRAY_SIZE(args_to_add));
+    // Create channel.
+    channel = client_channel_factory_create_channel(
+        &exec_ctx, &client_channel_factory, target,
+        GRPC_CLIENT_CHANNEL_TYPE_REGULAR, new_args);
+    // Clean up.
+    grpc_channel_args_destroy(&exec_ctx, new_args);
     grpc_exec_ctx_finish(&exec_ctx);
-    return grpc_lame_client_channel_create(
-        target, GRPC_STATUS_INTERNAL,
-        "Security connector exists in channel args.");
-  }
-  // Create security connector and construct new channel args.
-  grpc_channel_security_connector *security_connector;
-  grpc_channel_args *new_args_from_connector;
-  if (grpc_channel_credentials_create_security_connector(
-          &exec_ctx, creds, target, args, &security_connector,
-          &new_args_from_connector) != GRPC_SECURITY_OK) {
-    grpc_exec_ctx_finish(&exec_ctx);
-    return grpc_lame_client_channel_create(
-        target, GRPC_STATUS_INTERNAL, "Failed to create security connector.");
-  }
-  // Add channel args containing the client channel factory and security
-  // connector.
-  grpc_arg args_to_add[2];
-  args_to_add[0] =
-      grpc_client_channel_factory_create_channel_arg(&client_channel_factory);
-  args_to_add[1] = grpc_security_connector_to_arg(&security_connector->base);
-  grpc_channel_args *new_args = grpc_channel_args_copy_and_add(
-      new_args_from_connector != NULL ? new_args_from_connector : args,
-      args_to_add, GPR_ARRAY_SIZE(args_to_add));
-  if (new_args_from_connector != NULL) {
-    grpc_channel_args_destroy(&exec_ctx, new_args_from_connector);
   }
-  // Create channel.
-  grpc_channel *channel = client_channel_factory_create_channel(
-      &exec_ctx, &client_channel_factory, target,
-      GRPC_CLIENT_CHANNEL_TYPE_REGULAR, new_args);
-  // Clean up.
-  GRPC_SECURITY_CONNECTOR_UNREF(&exec_ctx, &security_connector->base,
-                                "secure_client_channel_factory_create_channel");
-  grpc_channel_args_destroy(&exec_ctx, new_args);
-  grpc_exec_ctx_finish(&exec_ctx);
   return channel != NULL ? channel
                          : grpc_lame_client_channel_create(
                                target, GRPC_STATUS_INTERNAL,
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index fa18f5a725..d1fab25478 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -168,7 +168,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
   grpc_chttp2_stream_map_destroy(&t->stream_map);
   grpc_connectivity_state_destroy(exec_ctx, &t->channel_callback.state_tracker);
 
-  grpc_combiner_destroy(exec_ctx, t->combiner);
+  GRPC_COMBINER_UNREF(exec_ctx, t->combiner, "chttp2_transport");
 
   cancel_pings(exec_ctx, t, GRPC_ERROR_CREATE("Transport destroyed"));
 
@@ -265,7 +265,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                                  .gain_d = 0,
                                  .initial_control_value = log2(DEFAULT_WINDOW),
                                  .min_control_value = -1,
-                                 .max_control_value = 22,
+                                 .max_control_value = 25,
                                  .integral_range = 10});
 
   grpc_chttp2_goaway_parser_init(&t->goaway_parser);
@@ -292,6 +292,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   t->force_send_settings = 1 << GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
   t->sent_local_settings = 0;
   t->write_buffer_size = DEFAULT_WINDOW;
+  t->enable_bdp_probe = true;
 
   if (is_client) {
     grpc_slice_buffer_add(&t->outbuf, grpc_slice_from_copied_string(
@@ -358,6 +359,10 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
         t->write_buffer_size = (uint32_t)grpc_channel_arg_get_integer(
             &channel_args->args[i],
             (grpc_integer_options){0, 0, MAX_WRITE_BUFFER_SIZE});
+      } else if (0 ==
+                 strcmp(channel_args->args[i].key, GRPC_ARG_HTTP2_BDP_PROBE)) {
+        t->enable_bdp_probe = grpc_channel_arg_get_integer(
+            &channel_args->args[i], (grpc_integer_options){1, 0, 1});
       } else {
         static const struct {
           const char *channel_arg_name;
@@ -564,6 +569,14 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
   GRPC_ERROR_UNREF(s->read_closed_error);
   GRPC_ERROR_UNREF(s->write_closed_error);
 
+  if (s->incoming_window_delta > 0) {
+    GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA(
+        "destroy", t, s, s->incoming_window_delta);
+  } else if (s->incoming_window_delta < 0) {
+    GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA(
+        "destroy", t, s, -s->incoming_window_delta);
+  }
+
   GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "stream");
 
   GPR_TIMER_END("destroy_stream", 0);
@@ -1028,8 +1041,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
   GPR_TIMER_BEGIN("perform_stream_op_locked", 0);
 
   grpc_transport_stream_op *op = stream_op;
-  grpc_chttp2_transport *t = op->transport_private.args[0];
-  grpc_chttp2_stream *s = op->transport_private.args[1];
+  grpc_chttp2_transport *t = op->handler_private.args[0];
+  grpc_chttp2_stream *s = op->handler_private.args[1];
 
   if (grpc_http_trace) {
     char *str = grpc_transport_stream_op_string(op);
@@ -1250,13 +1263,13 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
     gpr_free(str);
   }
 
-  op->transport_private.args[0] = gt;
-  op->transport_private.args[1] = gs;
+  op->handler_private.args[0] = gt;
+  op->handler_private.args[1] = gs;
   GRPC_CHTTP2_STREAM_REF(s, "perform_stream_op");
   grpc_closure_sched(
       exec_ctx,
       grpc_closure_init(
-          &op->transport_private.closure, perform_stream_op_locked, op,
+          &op->handler_private.closure, perform_stream_op_locked, op,
           grpc_combiner_scheduler(t->combiner, op->covered_by_poller)),
       GRPC_ERROR_NONE);
   GPR_TIMER_END("perform_stream_op", 0);
@@ -1796,13 +1809,13 @@ static void update_bdp(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   if (delta == 0 || (bdp != 0 && delta > -1024 && delta < 1024)) {
     return;
   }
+  if (grpc_bdp_estimator_trace) {
+    gpr_log(GPR_DEBUG, "%s: update initial window size to %d", t->peer_string,
+            (int)bdp);
+  }
   push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, bdp);
 }
 
-/*******************************************************************************
- * INPUT PROCESSING - PARSING
- */
-
 static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx,
                                     grpc_chttp2_transport *t) {
   grpc_http_parser parser;
@@ -1908,33 +1921,36 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
     grpc_endpoint_read(exec_ctx, t->ep, &t->read_buffer,
                        &t->read_action_locked);
 
-    if (need_bdp_ping) {
-      GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping");
-      grpc_bdp_estimator_schedule_ping(&t->bdp_estimator);
-      send_ping_locked(exec_ctx, t,
-                       GRPC_CHTTP2_PING_BEFORE_TRANSPORT_WINDOW_UPDATE,
-                       &t->start_bdp_ping_locked, &t->finish_bdp_ping_locked);
-    }
-
-    int64_t estimate = -1;
-    if (grpc_bdp_estimator_get_estimate(&t->bdp_estimator, &estimate)) {
-      double target = 1 + log2((double)estimate);
-      double memory_pressure = grpc_resource_quota_get_memory_pressure(
-          grpc_resource_user_quota(grpc_endpoint_get_resource_user(t->ep)));
-      if (memory_pressure > 0.8) {
-        target *= 1 - GPR_MIN(1, (memory_pressure - 0.8) / 0.1);
+    if (t->enable_bdp_probe) {
+      if (need_bdp_ping) {
+        GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping");
+        grpc_bdp_estimator_schedule_ping(&t->bdp_estimator);
+        send_ping_locked(exec_ctx, t,
+                         GRPC_CHTTP2_PING_BEFORE_TRANSPORT_WINDOW_UPDATE,
+                         &t->start_bdp_ping_locked, &t->finish_bdp_ping_locked);
       }
-      double bdp_error = target - grpc_pid_controller_last(&t->pid_controller);
-      gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
-      gpr_timespec dt_timespec = gpr_time_sub(now, t->last_pid_update);
-      double dt = (double)dt_timespec.tv_sec + dt_timespec.tv_nsec * 1e-9;
-      if (dt > 0.1) {
-        dt = 0.1;
+
+      int64_t estimate = -1;
+      if (grpc_bdp_estimator_get_estimate(&t->bdp_estimator, &estimate)) {
+        double target = 1 + log2((double)estimate);
+        double memory_pressure = grpc_resource_quota_get_memory_pressure(
+            grpc_resource_user_quota(grpc_endpoint_get_resource_user(t->ep)));
+        if (memory_pressure > 0.8) {
+          target *= 1 - GPR_MIN(1, (memory_pressure - 0.8) / 0.1);
+        }
+        double bdp_error =
+            target - grpc_pid_controller_last(&t->pid_controller);
+        gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
+        gpr_timespec dt_timespec = gpr_time_sub(now, t->last_pid_update);
+        double dt = (double)dt_timespec.tv_sec + dt_timespec.tv_nsec * 1e-9;
+        if (dt > 0.1) {
+          dt = 0.1;
+        }
+        double log2_bdp_guess =
+            grpc_pid_controller_update(&t->pid_controller, bdp_error, dt);
+        update_bdp(exec_ctx, t, pow(2, log2_bdp_guess));
+        t->last_pid_update = now;
       }
-      double log2_bdp_guess =
-          grpc_pid_controller_update(&t->pid_controller, bdp_error, dt);
-      update_bdp(exec_ctx, t, pow(2, log2_bdp_guess));
-      t->last_pid_update = now;
     }
     GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "keep_reading");
   } else {
@@ -2046,8 +2062,8 @@ static void incoming_byte_stream_update_flow_control(grpc_exec_ctx *exec_ctx,
         (int64_t)have_already) {
       write_type = GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED;
     }
-    GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, incoming_window_delta,
-                                   add_max_recv_bytes);
+    GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA("op", t, s,
+                                                         add_max_recv_bytes);
     GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, announce_window,
                                    add_max_recv_bytes);
     if ((int64_t)s->incoming_window_delta + (int64_t)initial_window_size -
diff --git a/src/core/ext/transport/chttp2/transport/frame_rst_stream.c b/src/core/ext/transport/chttp2/transport/frame_rst_stream.c
index 7d5beed09d..cb017e75f0 100644
--- a/src/core/ext/transport/chttp2/transport/frame_rst_stream.c
+++ b/src/core/ext/transport/chttp2/transport/frame_rst_stream.c
@@ -109,8 +109,13 @@ grpc_error *grpc_chttp2_rst_stream_parser_parse(grpc_exec_ctx *exec_ctx,
                       (((uint32_t)p->reason_bytes[3]));
     grpc_error *error = GRPC_ERROR_NONE;
     if (reason != GRPC_HTTP2_NO_ERROR || s->header_frames_received < 2) {
-      error = grpc_error_set_int(GRPC_ERROR_CREATE("RST_STREAM"),
-                                 GRPC_ERROR_INT_HTTP2_ERROR, (intptr_t)reason);
+      char *message;
+      gpr_asprintf(&message, "Received RST_STREAM with error code %d", reason);
+      error = grpc_error_set_int(
+          grpc_error_set_str(GRPC_ERROR_CREATE("RST_STREAM"),
+                             GRPC_ERROR_STR_GRPC_MESSAGE, message),
+          GRPC_ERROR_INT_HTTP2_ERROR, (intptr_t)reason);
+      gpr_free(message);
     }
     grpc_chttp2_mark_stream_closed(exec_ctx, t, s, true, true, error);
   }
diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.h b/src/core/ext/transport/chttp2/transport/frame_settings.h
index a29dc82106..44137798c0 100644
--- a/src/core/ext/transport/chttp2/transport/frame_settings.h
+++ b/src/core/ext/transport/chttp2/transport/frame_settings.h
@@ -87,7 +87,7 @@ extern const grpc_chttp2_setting_parameters
     grpc_chttp2_settings_parameters[GRPC_CHTTP2_NUM_SETTINGS];
 
 /* Create a settings frame by diffing old & new, and updating old to be new */
-grpc_slice grpc_chttp2_settings_create(uint32_t *old, const uint32_t *new,
+grpc_slice grpc_chttp2_settings_create(uint32_t *old, const uint32_t *newval,
                                        uint32_t force_mask, size_t count);
 /* Create an ack settings frame */
 grpc_slice grpc_chttp2_settings_ack_create(void);
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index 1dabf9edba..5d41f4bfda 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -227,6 +227,9 @@ struct grpc_chttp2_transport {
   /** is there a read request to the endpoint outstanding? */
   uint8_t endpoint_reading;
 
+  /** should we probe bdp? */
+  bool enable_bdp_probe;
+
   /** various lists of streams */
   grpc_chttp2_stream_list lists[STREAM_LIST_COUNT];
 
@@ -268,10 +271,6 @@ struct grpc_chttp2_transport {
   /** data to write next write */
   grpc_slice_buffer qbuf;
 
-  /** window available to announce to peer */
-  int64_t announce_incoming_window;
-  /** how much window would we like to have for incoming_window */
-  uint32_t connection_window_target;
   /** how much data are we willing to buffer when the WRITE_BUFFER_HINT is set?
    */
   uint32_t write_buffer_size;
@@ -325,6 +324,16 @@ struct grpc_chttp2_transport {
 
   /** window available for peer to send to us */
   int64_t incoming_window;
+  /** calculating what we should give for incoming window:
+      we track the total amount of flow control over initial window size
+      across all streams: this is data that we want to receive right now (it
+      has an outstanding read)
+      and the total amount of flow control under initial window size across all
+      streams: this is data we've read early
+      we want to adjust incoming_window such that:
+      incoming_window = total_over - max(bdp - total_under, 0) */
+  int64_t stream_total_over_incoming_window;
+  int64_t stream_total_under_incoming_window;
 
   /* deframing */
   grpc_chttp2_deframe_transport_state deframe_state;
@@ -631,6 +640,44 @@ typedef enum {
   GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, dst_context, 0, dst_context, dst_var,  \
                                  amount)
 
+#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE( \
+    phase, transport, dst_context)                               \
+  if (dst_context->incoming_window_delta < 0) {                  \
+    transport->stream_total_under_incoming_window +=             \
+        dst_context->incoming_window_delta;                      \
+  } else if (dst_context->incoming_window_delta > 0) {           \
+    transport->stream_total_over_incoming_window -=              \
+        dst_context->incoming_window_delta;                      \
+  }
+
+#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE( \
+    phase, transport, dst_context)                                \
+  if (dst_context->incoming_window_delta < 0) {                   \
+    transport->stream_total_under_incoming_window -=              \
+        dst_context->incoming_window_delta;                       \
+  } else if (dst_context->incoming_window_delta > 0) {            \
+    transport->stream_total_over_incoming_window +=               \
+        dst_context->incoming_window_delta;                       \
+  }
+
+#define GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA(                 \
+    phase, transport, dst_context, amount)                                   \
+  GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport,  \
+                                                          dst_context);      \
+  GRPC_CHTTP2_FLOW_DEBIT_STREAM(phase, transport, dst_context,               \
+                                incoming_window_delta, amount);              \
+  GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \
+                                                           dst_context);
+
+#define GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA(                \
+    phase, transport, dst_context, amount)                                   \
+  GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport,  \
+                                                          dst_context);      \
+  GRPC_CHTTP2_FLOW_CREDIT_STREAM(phase, transport, dst_context,              \
+                                 incoming_window_delta, amount);             \
+  GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \
+                                                           dst_context);
+
 #define GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, transport, id, dst_context,       \
                                       dst_var, amount)                         \
   do {                                                                         \
@@ -749,4 +796,6 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
                                      grpc_chttp2_transport *t,
                                      grpc_chttp2_stream *s, grpc_error *error);
 
+uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t);
+
 #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */
diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c
index 24bd93067b..7ed00522c3 100644
--- a/src/core/ext/transport/chttp2/transport/parsing.c
+++ b/src/core/ext/transport/chttp2/transport/parsing.c
@@ -376,15 +376,6 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
     return err;
   }
 
-  uint32_t target_incoming_window = GPR_MAX(
-      t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
-      1024);
-  GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", t, incoming_window,
-                                   incoming_frame_size);
-  if (t->incoming_window <= target_incoming_window / 2) {
-    grpc_chttp2_initiate_write(exec_ctx, t, false, "flow_control");
-  }
-
   if (s != NULL) {
     if (incoming_frame_size >
         s->incoming_window_delta +
@@ -402,8 +393,8 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
       return err;
     }
 
-    GRPC_CHTTP2_FLOW_DEBIT_STREAM("parse", t, s, incoming_window_delta,
-                                  incoming_frame_size);
+    GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA("parse", t, s,
+                                                        incoming_frame_size);
     if ((int64_t)t->settings[GRPC_SENT_SETTINGS]
                             [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] +
             (int64_t)s->incoming_window_delta - (int64_t)s->announce_window <=
@@ -417,6 +408,13 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
     s->received_bytes += incoming_frame_size;
   }
 
+  uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t);
+  GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", t, incoming_window,
+                                   incoming_frame_size);
+  if (t->incoming_window <= target_incoming_window / 2) {
+    grpc_chttp2_initiate_write(exec_ctx, t, false, "flow_control");
+  }
+
   return GRPC_ERROR_NONE;
 }
 
diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c
index 05e6f59947..2b9d93cae7 100644
--- a/src/core/ext/transport/chttp2/transport/writing.c
+++ b/src/core/ext/transport/chttp2/transport/writing.c
@@ -152,6 +152,17 @@ static bool stream_ref_if_not_destroyed(gpr_refcount *r) {
   return true;
 }
 
+uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t) {
+  return (uint32_t)GPR_MAX(
+      (int64_t)((1u << 31) - 1),
+      t->stream_total_over_incoming_window +
+          (int64_t)GPR_MAX(
+              t->settings[GRPC_SENT_SETTINGS]
+                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] -
+                  t->stream_total_under_incoming_window,
+              0));
+}
+
 bool grpc_chttp2_begin_write(grpc_exec_ctx *exec_ctx,
                              grpc_chttp2_transport *t) {
   grpc_chttp2_stream *s;
@@ -310,13 +321,12 @@ bool grpc_chttp2_begin_write(grpc_exec_ctx *exec_ctx,
 
   /* if the grpc_chttp2_transport is ready to send a window update, do so here
      also; 3/4 is a magic number that will likely get tuned soon */
-  uint32_t target_incoming_window = GPR_MAX(
-      t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
-      1024);
+  uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t);
   uint32_t threshold_to_send_transport_window_update =
       t->outbuf.count > 0 ? 3 * target_incoming_window / 4
                           : target_incoming_window / 2;
-  if (t->incoming_window <= threshold_to_send_transport_window_update) {
+  if (t->incoming_window <= threshold_to_send_transport_window_update &&
+      t->incoming_window != target_incoming_window) {
     maybe_initiate_ping(exec_ctx, t,
                         GRPC_CHTTP2_PING_BEFORE_TRANSPORT_WINDOW_UPDATE);
     uint32_t announced = (uint32_t)GPR_CLAMP(
diff --git a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
index 477cf07f45..b6e9e845df 100644
--- a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
+++ b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
@@ -39,6 +39,7 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 
+#include "src/core/ext/transport/cronet/transport/cronet_transport.h"
 #include "src/core/lib/surface/channel.h"
 #include "src/core/lib/transport/transport_impl.h"
 
@@ -54,16 +55,14 @@ extern grpc_transport_vtable grpc_cronet_vtable;
 GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
     void *engine, const char *target, const grpc_channel_args *args,
     void *reserved) {
-  cronet_transport *ct = gpr_malloc(sizeof(cronet_transport));
-  ct->base.vtable = &grpc_cronet_vtable;
-  ct->engine = engine;
-  ct->host = gpr_malloc(strlen(target) + 1);
-  strcpy(ct->host, target);
   gpr_log(GPR_DEBUG,
           "grpc_create_cronet_transport: stream_engine = %p, target=%s", engine,
-          ct->host);
+          target);
+
+  grpc_transport *ct =
+      grpc_create_cronet_transport(engine, target, args, reserved);
 
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   return grpc_channel_create(&exec_ctx, target, args,
-                             GRPC_CLIENT_DIRECT_CHANNEL, (grpc_transport *)ct);
+                             GRPC_CLIENT_DIRECT_CHANNEL, ct);
 }
diff --git a/src/core/ext/transport/cronet/transport/cronet_api_dummy.c b/src/core/ext/transport/cronet/transport/cronet_api_dummy.c
index da6c0b4fbc..0dc6a5152f 100644
--- a/src/core/ext/transport/cronet/transport/cronet_api_dummy.c
+++ b/src/core/ext/transport/cronet/transport/cronet_api_dummy.c
@@ -80,4 +80,16 @@ void bidirectional_stream_cancel(bidirectional_stream* stream) {
   GPR_ASSERT(0);
 }
 
+void bidirectional_stream_disable_auto_flush(bidirectional_stream* stream,
+                                             bool disable_auto_flush) {
+  GPR_ASSERT(0);
+}
+
+void bidirectional_stream_delay_request_headers_until_flush(
+    bidirectional_stream* stream, bool delay_headers_until_flush) {
+  GPR_ASSERT(0);
+}
+
+void bidirectional_stream_flush(bidirectional_stream* stream) { GPR_ASSERT(0); }
+
 #endif /* GRPC_COMPILE_WITH_CRONET */
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index d755b1f147..01a03533da 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -88,7 +88,7 @@ enum e_op_id {
 
 /* Cronet callbacks. See cronet_c_for_grpc.h for documentation for each. */
 
-static void on_request_headers_sent(bidirectional_stream *);
+static void on_stream_ready(bidirectional_stream *);
 static void on_response_headers_received(
     bidirectional_stream *, const bidirectional_stream_header_array *,
     const char *);
@@ -100,7 +100,7 @@ static void on_succeeded(bidirectional_stream *);
 static void on_failed(bidirectional_stream *, int);
 static void on_canceled(bidirectional_stream *);
 static bidirectional_stream_callback cronet_callbacks = {
-    on_request_headers_sent,
+    on_stream_ready,
     on_response_headers_received,
     on_read_completed,
     on_write_completed,
@@ -114,6 +114,7 @@ struct grpc_cronet_transport {
   grpc_transport base; /* must be first element in this structure */
   stream_engine *engine;
   char *host;
+  bool use_packet_coalescing;
 };
 typedef struct grpc_cronet_transport grpc_cronet_transport;
 
@@ -152,6 +153,9 @@ struct op_state {
   bool state_callback_received[OP_NUM_OPS];
   bool fail_state;
   bool flush_read;
+  bool flush_cronet_when_ready;
+  bool pending_write_for_trailer;
+  bool unprocessed_send_message;
   grpc_error *cancel_error;
   /* data structure for storing data coming from server */
   struct read_state rs;
@@ -175,7 +179,7 @@ struct op_storage {
 struct stream_obj {
   struct op_and_state *oas;
   grpc_transport_stream_op *curr_op;
-  grpc_cronet_transport curr_ct;
+  grpc_cronet_transport *curr_ct;
   grpc_stream *curr_gs;
   bidirectional_stream *cbs;
   bidirectional_stream_header_array header_array;
@@ -274,6 +278,9 @@ static void add_to_storage(struct stream_obj *s, grpc_transport_stream_op *op) {
   new_op->next = storage->head;
   storage->head = new_op;
   storage->num_pending_ops++;
+  if (op->send_message) {
+    s->state.unprocessed_send_message = true;
+  }
   CRONET_LOG(GPR_DEBUG, "adding new op %p. %d in the queue.", new_op,
              storage->num_pending_ops);
   gpr_mu_unlock(&s->mu);
@@ -406,9 +413,10 @@ static void on_succeeded(bidirectional_stream *stream) {
 /*
   Cronet callback
 */
-static void on_request_headers_sent(bidirectional_stream *stream) {
-  CRONET_LOG(GPR_DEBUG, "W: on_request_headers_sent(%p)", stream);
+static void on_stream_ready(bidirectional_stream *stream) {
+  CRONET_LOG(GPR_DEBUG, "W: on_stream_ready(%p)", stream);
   stream_obj *s = (stream_obj *)stream->annotation;
+  grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
   gpr_mu_lock(&s->mu);
   s->state.state_op_done[OP_SEND_INITIAL_METADATA] = true;
   s->state.state_callback_received[OP_SEND_INITIAL_METADATA] = true;
@@ -417,6 +425,14 @@ static void on_request_headers_sent(bidirectional_stream *stream) {
     gpr_free(s->header_array.headers);
     s->header_array.headers = NULL;
   }
+  /* Send the initial metadata on wire if there is no SEND_MESSAGE or
+   * SEND_TRAILING_METADATA ops pending */
+  if (t->use_packet_coalescing) {
+    if (s->state.flush_cronet_when_ready) {
+      CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
+      bidirectional_stream_flush(stream);
+    }
+  }
   gpr_mu_unlock(&s->mu);
   execute_from_storage(s);
 }
@@ -528,6 +544,7 @@ static void on_response_trailers_received(
   CRONET_LOG(GPR_DEBUG, "R: on_response_trailers_received(%p,%p)", stream,
              trailers);
   stream_obj *s = (stream_obj *)stream->annotation;
+  grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
   gpr_mu_lock(&s->mu);
   memset(&s->state.rs.trailing_metadata, 0,
          sizeof(s->state.rs.trailing_metadata));
@@ -558,6 +575,10 @@ static void on_response_trailers_received(
     CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
     s->state.state_callback_received[OP_SEND_MESSAGE] = false;
     bidirectional_stream_write(s->cbs, "", 0, true);
+    if (t->use_packet_coalescing) {
+      CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+      bidirectional_stream_flush(s->cbs);
+    }
     s->state.state_op_done[OP_SEND_TRAILING_METADATA] = true;
 
     gpr_mu_unlock(&s->mu);
@@ -607,7 +628,7 @@ static void convert_metadata_to_cronet_headers(
     curr = curr->next;
     num_headers_available++;
   }
-  /* Allocate enough memory. It is freed in the on_request_headers_sent callback
+  /* Allocate enough memory. It is freed in the on_stream_ready callback
    */
   bidirectional_stream_header *headers =
       (bidirectional_stream_header *)gpr_malloc(
@@ -687,8 +708,10 @@ static bool header_has_authority(grpc_linked_mdelem *head) {
   executed. This is the heart of the state machine.
 */
 static bool op_can_be_run(grpc_transport_stream_op *curr_op,
-                          struct op_state *stream_state,
-                          struct op_state *op_state, enum e_op_id op_id) {
+                          struct stream_obj *s, struct op_state *op_state,
+                          enum e_op_id op_id) {
+  struct op_state *stream_state = &s->state;
+  grpc_cronet_transport *t = s->curr_ct;
   bool result = true;
   /* When call is canceled, every op can be run, except under following
   conditions
@@ -755,12 +778,14 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
     else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
       result = false;
     /* we haven't sent message yet */
-    else if (curr_op->send_message &&
+    else if (stream_state->unprocessed_send_message &&
              !stream_state->state_op_done[OP_SEND_MESSAGE])
       result = false;
     /* we haven't got on_write_completed for the send yet */
     else if (stream_state->state_op_done[OP_SEND_MESSAGE] &&
-             !stream_state->state_callback_received[OP_SEND_MESSAGE])
+             !stream_state->state_callback_received[OP_SEND_MESSAGE] &&
+             !(t->use_packet_coalescing &&
+               stream_state->pending_write_for_trailer))
       result = false;
   } else if (op_id == OP_CANCEL_ERROR) {
     /* already executed */
@@ -833,24 +858,28 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
                                           struct op_and_state *oas) {
   grpc_transport_stream_op *stream_op = &oas->op;
   struct stream_obj *s = oas->s;
+  grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
   struct op_state *stream_state = &s->state;
   enum e_op_result result = NO_ACTION_POSSIBLE;
   if (stream_op->send_initial_metadata &&
-      op_can_be_run(stream_op, stream_state, &oas->state,
-                    OP_SEND_INITIAL_METADATA)) {
+      op_can_be_run(stream_op, s, &oas->state, OP_SEND_INITIAL_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_INITIAL_METADATA", oas);
     /* Start new cronet stream. It is destroyed in on_succeeded, on_canceled,
      * on_failed */
     GPR_ASSERT(s->cbs == NULL);
     GPR_ASSERT(!stream_state->state_op_done[OP_SEND_INITIAL_METADATA]);
-    s->cbs = bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
-                                         &cronet_callbacks);
+    s->cbs =
+        bidirectional_stream_create(t->engine, s->curr_gs, &cronet_callbacks);
     CRONET_LOG(GPR_DEBUG, "%p = bidirectional_stream_create()", s->cbs);
+    if (t->use_packet_coalescing) {
+      bidirectional_stream_disable_auto_flush(s->cbs, true);
+      bidirectional_stream_delay_request_headers_until_flush(s->cbs, true);
+    }
     char *url = NULL;
     const char *method = "POST";
     s->header_array.headers = NULL;
     convert_metadata_to_cronet_headers(
-        stream_op->send_initial_metadata->list.head, s->curr_ct.host, &url,
+        stream_op->send_initial_metadata->list.head, t->host, &url,
         &s->header_array.headers, &s->header_array.count, &method);
     s->header_array.capacity = s->header_array.count;
     CRONET_LOG(GPR_DEBUG, "bidirectional_stream_start(%p, %s)", s->cbs, url);
@@ -862,30 +891,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
       gpr_free((void *)s->header_array.headers[header_index].value);
     }
     stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true;
-    result = ACTION_TAKEN_WITH_CALLBACK;
-  } else if (stream_op->recv_initial_metadata &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_RECV_INITIAL_METADATA)) {
-    CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_INITIAL_METADATA", oas);
-    if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
-      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                         GRPC_ERROR_NONE);
-    } else if (stream_state->state_callback_received[OP_FAILED]) {
-      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                         GRPC_ERROR_NONE);
-    } else {
-      grpc_chttp2_incoming_metadata_buffer_publish(
-          exec_ctx, &oas->s->state.rs.initial_metadata,
-          stream_op->recv_initial_metadata);
-      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                         GRPC_ERROR_NONE);
+    if (t->use_packet_coalescing) {
+      if (!stream_op->send_message && !stream_op->send_trailing_metadata) {
+        s->state.flush_cronet_when_ready = true;
+      }
     }
-    stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
-    result = ACTION_TAKEN_NO_CALLBACK;
+    result = ACTION_TAKEN_WITH_CALLBACK;
   } else if (stream_op->send_message &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_SEND_MESSAGE)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_SEND_MESSAGE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_MESSAGE", oas);
+    stream_state->unprocessed_send_message = false;
     if (stream_state->state_callback_received[OP_FAILED]) {
       result = NO_ACTION_POSSIBLE;
       CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
@@ -916,16 +931,63 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
         stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
         bidirectional_stream_write(s->cbs, stream_state->ws.write_buffer,
                                    (int)write_buffer_size, false);
-        result = ACTION_TAKEN_WITH_CALLBACK;
+        if (t->use_packet_coalescing) {
+          if (!stream_op->send_trailing_metadata) {
+            CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+            bidirectional_stream_flush(s->cbs);
+            result = ACTION_TAKEN_WITH_CALLBACK;
+          } else {
+            stream_state->pending_write_for_trailer = true;
+            result = ACTION_TAKEN_NO_CALLBACK;
+          }
+        } else {
+          result = ACTION_TAKEN_WITH_CALLBACK;
+        }
       } else {
         result = NO_ACTION_POSSIBLE;
       }
     }
     stream_state->state_op_done[OP_SEND_MESSAGE] = true;
     oas->state.state_op_done[OP_SEND_MESSAGE] = true;
+  } else if (stream_op->send_trailing_metadata &&
+             op_can_be_run(stream_op, s, &oas->state,
+                           OP_SEND_TRAILING_METADATA)) {
+    CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_TRAILING_METADATA", oas);
+    if (stream_state->state_callback_received[OP_FAILED]) {
+      result = NO_ACTION_POSSIBLE;
+      CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
+    } else {
+      CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
+      stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
+      bidirectional_stream_write(s->cbs, "", 0, true);
+      if (t->use_packet_coalescing) {
+        CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+        bidirectional_stream_flush(s->cbs);
+      }
+      result = ACTION_TAKEN_WITH_CALLBACK;
+    }
+    stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
+  } else if (stream_op->recv_initial_metadata &&
+             op_can_be_run(stream_op, s, &oas->state,
+                           OP_RECV_INITIAL_METADATA)) {
+    CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_INITIAL_METADATA", oas);
+    if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
+      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                         GRPC_ERROR_NONE);
+    } else if (stream_state->state_callback_received[OP_FAILED]) {
+      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                         GRPC_ERROR_NONE);
+    } else {
+      grpc_chttp2_incoming_metadata_buffer_publish(
+          exec_ctx, &oas->s->state.rs.initial_metadata,
+          stream_op->recv_initial_metadata);
+      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                         GRPC_ERROR_NONE);
+    }
+    stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
+    result = ACTION_TAKEN_NO_CALLBACK;
   } else if (stream_op->recv_message &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_RECV_MESSAGE)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_RECV_MESSAGE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_MESSAGE", oas);
     if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
       CRONET_LOG(GPR_DEBUG, "Stream is cancelled.");
@@ -980,6 +1042,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
                              GRPC_ERROR_NONE);
           stream_state->state_op_done[OP_RECV_MESSAGE] = true;
           oas->state.state_op_done[OP_RECV_MESSAGE] = true;
+
+          /* Extra read to trigger on_succeed */
+          stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes;
+          stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
+          stream_state->rs.received_bytes = 0;
+          CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs);
+          stream_state->state_op_done[OP_READ_REQ_MADE] =
+              true; /* Indicates that at least one read request has been made */
+          bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer,
+                                    stream_state->rs.remaining_bytes);
           result = ACTION_TAKEN_NO_CALLBACK;
         }
       } else if (stream_state->rs.remaining_bytes == 0) {
@@ -1027,7 +1099,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
       result = ACTION_TAKEN_NO_CALLBACK;
     }
   } else if (stream_op->recv_trailing_metadata &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
+             op_can_be_run(stream_op, s, &oas->state,
                            OP_RECV_TRAILING_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_TRAILING_METADATA", oas);
     if (oas->s->state.rs.trailing_metadata_valid) {
@@ -1038,23 +1110,8 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
     }
     stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true;
     result = ACTION_TAKEN_NO_CALLBACK;
-  } else if (stream_op->send_trailing_metadata &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_SEND_TRAILING_METADATA)) {
-    CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_TRAILING_METADATA", oas);
-    if (stream_state->state_callback_received[OP_FAILED]) {
-      result = NO_ACTION_POSSIBLE;
-      CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
-    } else {
-      CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
-      stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
-      bidirectional_stream_write(s->cbs, "", 0, true);
-      result = ACTION_TAKEN_WITH_CALLBACK;
-    }
-    stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
   } else if (stream_op->cancel_error &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_CANCEL_ERROR)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_CANCEL_ERROR)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_CANCEL_ERROR", oas);
     CRONET_LOG(GPR_DEBUG, "W: bidirectional_stream_cancel(%p)", s->cbs);
     if (s->cbs) {
@@ -1068,8 +1125,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
       stream_state->cancel_error = GRPC_ERROR_REF(stream_op->cancel_error);
     }
   } else if (stream_op->on_complete &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_ON_COMPLETE)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_ON_COMPLETE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_ON_COMPLETE", oas);
     if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
       grpc_closure_sched(exec_ctx, stream_op->on_complete,
@@ -1133,6 +1189,12 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
          sizeof(s->state.state_callback_received));
   s->state.fail_state = s->state.flush_read = false;
   s->state.cancel_error = NULL;
+  s->state.flush_cronet_when_ready = s->state.pending_write_for_trailer = false;
+  s->state.unprocessed_send_message = false;
+
+  s->curr_gs = gs;
+  s->curr_ct = (grpc_cronet_transport *)gt;
+
   gpr_mu_init(&s->mu);
   return 0;
 }
@@ -1148,8 +1210,6 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                               grpc_stream *gs, grpc_transport_stream_op *op) {
   CRONET_LOG(GPR_DEBUG, "perform_stream_op");
   stream_obj *s = (stream_obj *)gs;
-  s->curr_gs = gs;
-  memcpy(&s->curr_ct, gt, sizeof(grpc_cronet_transport));
   add_to_storage(s, op);
   if (op->send_initial_metadata &&
       header_has_authority(op->send_initial_metadata->list.head)) {
@@ -1197,14 +1257,58 @@ static grpc_endpoint *get_endpoint(grpc_exec_ctx *exec_ctx,
 static void perform_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                        grpc_transport_op *op) {}
 
-const grpc_transport_vtable grpc_cronet_vtable = {sizeof(stream_obj),
-                                                  "cronet_http",
-                                                  init_stream,
-                                                  set_pollset_do_nothing,
-                                                  set_pollset_set_do_nothing,
-                                                  perform_stream_op,
-                                                  perform_op,
-                                                  destroy_stream,
-                                                  destroy_transport,
-                                                  get_peer,
-                                                  get_endpoint};
+static const grpc_transport_vtable grpc_cronet_vtable = {
+    sizeof(stream_obj),
+    "cronet_http",
+    init_stream,
+    set_pollset_do_nothing,
+    set_pollset_set_do_nothing,
+    perform_stream_op,
+    perform_op,
+    destroy_stream,
+    destroy_transport,
+    get_peer,
+    get_endpoint};
+
+grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
+                                             const grpc_channel_args *args,
+                                             void *reserved) {
+  grpc_cronet_transport *ct = gpr_malloc(sizeof(grpc_cronet_transport));
+  if (!ct) {
+    goto error;
+  }
+  ct->base.vtable = &grpc_cronet_vtable;
+  ct->engine = engine;
+  ct->host = gpr_malloc(strlen(target) + 1);
+  if (!ct->host) {
+    goto error;
+  }
+  strcpy(ct->host, target);
+
+  ct->use_packet_coalescing = true;
+  if (args) {
+    for (size_t i = 0; i < args->num_args; i++) {
+      if (0 ==
+          strcmp(args->args[i].key, GRPC_ARG_USE_CRONET_PACKET_COALESCING)) {
+        if (args->args[i].type != GRPC_ARG_INTEGER) {
+          gpr_log(GPR_ERROR, "%s ignored: it must be an integer",
+                  GRPC_ARG_USE_CRONET_PACKET_COALESCING);
+        } else {
+          ct->use_packet_coalescing = (args->args[i].value.integer != 0);
+        }
+      }
+    }
+  }
+
+  return &ct->base;
+
+error:
+  if (ct) {
+    if (ct->host) {
+      gpr_free(ct->host);
+    }
+    gpr_free(ct);
+  }
+
+  return NULL;
+}
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.h b/src/core/ext/transport/cronet/transport/cronet_transport.h
new file mode 100644
index 0000000000..169ce31fd7
--- /dev/null
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.h
@@ -0,0 +1,43 @@
+/*
+ *
+ * Copyright 2016, 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_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
+#define GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
+
+#include "src/core/lib/transport/transport.h"
+
+grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
+                                             const grpc_channel_args *args,
+                                             void *reserved);
+
+#endif /* GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H */