Merge branch 'master' of github.com:grpc/grpc into grpclb_api

15 files changed, 602 insertions, 1041 deletions

diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c
index 93312abb00..37de3e9f68 100644
--- a/src/core/client_config/lb_policies/pick_first.c
+++ b/src/core/client_config/lb_policies/pick_first.c
@@ -42,7 +42,7 @@
 typedef struct pending_pick {
   struct pending_pick *next;
   grpc_pollset *pollset;
-  grpc_subchannel **target;
+  grpc_connected_subchannel **target;
   grpc_closure *on_complete;
 } pending_pick;
 
@@ -60,7 +60,7 @@ typedef struct {
   /** the selected channel
       TODO(ctiller): this should be atomically set so we don't
                      need to take a mutex in the common case */
-  grpc_subchannel *selected;
+  grpc_connected_subchannel *selected;
   /** have we started picking? */
   int started_picking;
   /** are we shut down? */
@@ -76,24 +76,6 @@ typedef struct {
   grpc_connectivity_state_tracker state_tracker;
 } pick_first_lb_policy;
 
-static void del_interested_parties_locked(grpc_exec_ctx *exec_ctx,
-                                          pick_first_lb_policy *p) {
-  pending_pick *pp;
-  for (pp = p->pending_picks; pp; pp = pp->next) {
-    grpc_subchannel_del_interested_party(
-        exec_ctx, p->subchannels[p->checking_subchannel], pp->pollset);
-  }
-}
-
-static void add_interested_parties_locked(grpc_exec_ctx *exec_ctx,
-                                          pick_first_lb_policy *p) {
-  pending_pick *pp;
-  for (pp = p->pending_picks; pp; pp = pp->next) {
-    grpc_subchannel_add_interested_party(
-        exec_ctx, p->subchannels[p->checking_subchannel], pp->pollset);
-  }
-}
-
 void pf_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
   size_t i;
@@ -102,7 +84,7 @@ void pf_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
     GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[i], "pick_first");
   }
   if (p->selected) {
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, p->selected, "picked_first");
+    GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, p->selected, "picked_first");
   }
   grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker);
   gpr_free(p->subchannels);
@@ -114,16 +96,26 @@ void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
   pending_pick *pp;
   gpr_mu_lock(&p->mu);
-  del_interested_parties_locked(exec_ctx, p);
   p->shutdown = 1;
   pp = p->pending_picks;
   p->pending_picks = NULL;
   grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                               GRPC_CHANNEL_FATAL_FAILURE, "shutdown");
+  /* cancel subscription */
+  if (p->selected != NULL) {
+    grpc_connected_subchannel_notify_on_state_change(
+        exec_ctx, p->selected, NULL, NULL, &p->connectivity_changed);
+  } else {
+    grpc_subchannel_notify_on_state_change(
+        exec_ctx, p->subchannels[p->checking_subchannel], NULL, NULL,
+        &p->connectivity_changed);
+  }
   gpr_mu_unlock(&p->mu);
   while (pp != NULL) {
     pending_pick *next = pp->next;
     *pp->target = NULL;
+    grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+                                 pp->pollset);
     grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
     gpr_free(pp);
     pp = next;
@@ -131,7 +123,7 @@ void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
 }
 
 static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
-                           grpc_subchannel **target) {
+                           grpc_connected_subchannel **target) {
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
   pending_pick *pp;
   gpr_mu_lock(&p->mu);
@@ -140,8 +132,8 @@ static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
   while (pp != NULL) {
     pending_pick *next = pp->next;
     if (pp->target == target) {
-      grpc_subchannel_del_interested_party(
-          exec_ctx, p->subchannels[p->checking_subchannel], pp->pollset);
+      grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+                                   pp->pollset);
       *target = NULL;
       grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 0);
       gpr_free(pp);
@@ -158,10 +150,11 @@ static void start_picking(grpc_exec_ctx *exec_ctx, pick_first_lb_policy *p) {
   p->started_picking = 1;
   p->checking_subchannel = 0;
   p->checking_connectivity = GRPC_CHANNEL_IDLE;
-  GRPC_LB_POLICY_REF(&p->base, "pick_first_connectivity");
+  GRPC_LB_POLICY_WEAK_REF(&p->base, "pick_first_connectivity");
   grpc_subchannel_notify_on_state_change(
       exec_ctx, p->subchannels[p->checking_subchannel],
-      &p->checking_connectivity, &p->connectivity_changed);
+      &p->base.interested_parties, &p->checking_connectivity,
+      &p->connectivity_changed);
 }
 
 void pf_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
@@ -174,8 +167,8 @@ void pf_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
 }
 
 int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
-            grpc_metadata_batch *initial_metadata, grpc_subchannel **target,
-            grpc_closure *on_complete) {
+            grpc_metadata_batch *initial_metadata,
+            grpc_connected_subchannel **target, grpc_closure *on_complete) {
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
   pending_pick *pp;
   gpr_mu_lock(&p->mu);
@@ -187,8 +180,8 @@ int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
     if (!p->started_picking) {
       start_picking(exec_ctx, p);
     }
-    grpc_subchannel_add_interested_party(
-        exec_ctx, p->subchannels[p->checking_subchannel], pollset);
+    grpc_pollset_set_add_pollset(exec_ctx, &p->base.interested_parties,
+                                 pollset);
     pp = gpr_malloc(sizeof(*pp));
     pp->next = p->pending_picks;
     pp->pollset = pollset;
@@ -204,25 +197,17 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
                                 int iomgr_success) {
   pick_first_lb_policy *p = arg;
   size_t i;
-  grpc_transport_op op;
   size_t num_subchannels = p->num_subchannels;
   grpc_subchannel **subchannels;
-  grpc_subchannel *exclude_subchannel;
 
   gpr_mu_lock(&p->mu);
   subchannels = p->subchannels;
   p->num_subchannels = 0;
   p->subchannels = NULL;
-  exclude_subchannel = p->selected;
   gpr_mu_unlock(&p->mu);
-  GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "destroy_subchannels");
+  GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "destroy_subchannels");
 
   for (i = 0; i < num_subchannels; i++) {
-    if (subchannels[i] != exclude_subchannel) {
-      memset(&op, 0, sizeof(op));
-      op.disconnect = 1;
-      grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], &op);
-    }
     GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "pick_first");
   }
 
@@ -232,23 +217,28 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
 static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
                                     int iomgr_success) {
   pick_first_lb_policy *p = arg;
+  grpc_subchannel *selected_subchannel;
   pending_pick *pp;
 
   gpr_mu_lock(&p->mu);
 
   if (p->shutdown) {
     gpr_mu_unlock(&p->mu);
-    GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
+    GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
     return;
   } else if (p->selected != NULL) {
+    if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) {
+      /* if the selected channel goes bad, we're done */
+      p->checking_connectivity = GRPC_CHANNEL_FATAL_FAILURE;
+    }
     grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                 p->checking_connectivity, "selected_changed");
     if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) {
-      grpc_subchannel_notify_on_state_change(exec_ctx, p->selected,
-                                             &p->checking_connectivity,
-                                             &p->connectivity_changed);
+      grpc_connected_subchannel_notify_on_state_change(
+          exec_ctx, p->selected, &p->base.interested_parties,
+          &p->checking_connectivity, &p->connectivity_changed);
     } else {
-      GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
+      GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
     }
   } else {
   loop:
@@ -256,39 +246,41 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
       case GRPC_CHANNEL_READY:
         grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                     GRPC_CHANNEL_READY, "connecting_ready");
-        p->selected = p->subchannels[p->checking_subchannel];
-        GRPC_SUBCHANNEL_REF(p->selected, "picked_first");
+        selected_subchannel = p->subchannels[p->checking_subchannel];
+        p->selected =
+            grpc_subchannel_get_connected_subchannel(selected_subchannel);
+        GPR_ASSERT(p->selected);
+        GRPC_CONNECTED_SUBCHANNEL_REF(p->selected, "picked_first");
         /* drop the pick list: we are connected now */
-        GRPC_LB_POLICY_REF(&p->base, "destroy_subchannels");
+        GRPC_LB_POLICY_WEAK_REF(&p->base, "destroy_subchannels");
         grpc_exec_ctx_enqueue(exec_ctx,
                               grpc_closure_create(destroy_subchannels, p), 1);
         /* update any calls that were waiting for a pick */
         while ((pp = p->pending_picks)) {
           p->pending_picks = pp->next;
           *pp->target = p->selected;
-          grpc_subchannel_del_interested_party(exec_ctx, p->selected,
-                                               pp->pollset);
+          grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+                                       pp->pollset);
           grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
           gpr_free(pp);
         }
-        grpc_subchannel_notify_on_state_change(exec_ctx, p->selected,
-                                               &p->checking_connectivity,
-                                               &p->connectivity_changed);
+        grpc_connected_subchannel_notify_on_state_change(
+            exec_ctx, p->selected, &p->base.interested_parties,
+            &p->checking_connectivity, &p->connectivity_changed);
         break;
       case GRPC_CHANNEL_TRANSIENT_FAILURE:
         grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                     GRPC_CHANNEL_TRANSIENT_FAILURE,
                                     "connecting_transient_failure");
-        del_interested_parties_locked(exec_ctx, p);
         p->checking_subchannel =
             (p->checking_subchannel + 1) % p->num_subchannels;
         p->checking_connectivity = grpc_subchannel_check_connectivity(
             p->subchannels[p->checking_subchannel]);
-        add_interested_parties_locked(exec_ctx, p);
         if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) {
           grpc_subchannel_notify_on_state_change(
               exec_ctx, p->subchannels[p->checking_subchannel],
-              &p->checking_connectivity, &p->connectivity_changed);
+              &p->base.interested_parties, &p->checking_connectivity,
+              &p->connectivity_changed);
         } else {
           goto loop;
         }
@@ -300,13 +292,13 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
                                     "connecting_changed");
         grpc_subchannel_notify_on_state_change(
             exec_ctx, p->subchannels[p->checking_subchannel],
-            &p->checking_connectivity, &p->connectivity_changed);
+            &p->base.interested_parties, &p->checking_connectivity,
+            &p->connectivity_changed);
         break;
       case GRPC_CHANNEL_FATAL_FAILURE:
-        del_interested_parties_locked(exec_ctx, p);
-        GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel],
-                 p->subchannels[p->num_subchannels - 1]);
         p->num_subchannels--;
+        GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel],
+                 p->subchannels[p->num_subchannels]);
         GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[p->num_subchannels],
                               "pick_first");
         if (p->num_subchannels == 0) {
@@ -319,7 +311,8 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
             grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
             gpr_free(pp);
           }
-          GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
+          GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base,
+                                    "pick_first_connectivity");
         } else {
           grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                       GRPC_CHANNEL_TRANSIENT_FAILURE,
@@ -327,7 +320,6 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
           p->checking_subchannel %= p->num_subchannels;
           p->checking_connectivity = grpc_subchannel_check_connectivity(
               p->subchannels[p->checking_subchannel]);
-          add_interested_parties_locked(exec_ctx, p);
           goto loop;
         }
     }
@@ -336,39 +328,6 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
   gpr_mu_unlock(&p->mu);
 }
 
-static void pf_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
-                         grpc_transport_op *op) {
-  pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
-  size_t i;
-  size_t n;
-  grpc_subchannel **subchannels;
-  grpc_subchannel *selected;
-
-  gpr_mu_lock(&p->mu);
-  n = p->num_subchannels;
-  subchannels = gpr_malloc(n * sizeof(*subchannels));
-  selected = p->selected;
-  if (selected) {
-    GRPC_SUBCHANNEL_REF(selected, "pf_broadcast_to_selected");
-  }
-  for (i = 0; i < n; i++) {
-    subchannels[i] = p->subchannels[i];
-    GRPC_SUBCHANNEL_REF(subchannels[i], "pf_broadcast");
-  }
-  gpr_mu_unlock(&p->mu);
-
-  for (i = 0; i < n; i++) {
-    if (selected == subchannels[i]) continue;
-    grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], op);
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "pf_broadcast");
-  }
-  if (p->selected) {
-    grpc_subchannel_process_transport_op(exec_ctx, selected, op);
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, selected, "pf_broadcast_to_selected");
-  }
-  gpr_free(subchannels);
-}
-
 static grpc_connectivity_state pf_check_connectivity(grpc_exec_ctx *exec_ctx,
                                                      grpc_lb_policy *pol) {
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
@@ -389,9 +348,21 @@ void pf_notify_on_state_change(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
   gpr_mu_unlock(&p->mu);
 }
 
+void pf_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
+                 grpc_closure *closure) {
+  pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
+  gpr_mu_lock(&p->mu);
+  if (p->selected) {
+    grpc_connected_subchannel_ping(exec_ctx, p->selected, closure);
+  } else {
+    grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+  }
+  gpr_mu_unlock(&p->mu);
+}
+
 static const grpc_lb_policy_vtable pick_first_lb_policy_vtable = {
-    pf_destroy, pf_shutdown, pf_pick, pf_cancel_pick, pf_exit_idle,
-    pf_broadcast, pf_check_connectivity, pf_notify_on_state_change};
+    pf_destroy, pf_shutdown, pf_pick, pf_cancel_pick, pf_ping_one, pf_exit_idle,
+    pf_check_connectivity, pf_notify_on_state_change};
 
 static void pick_first_factory_ref(grpc_lb_policy_factory *factory) {}
 
diff --git a/src/core/client_config/lb_policies/round_robin.c b/src/core/client_config/lb_policies/round_robin.c
index 1ffe32fff2..d487456363 100644
--- a/src/core/client_config/lb_policies/round_robin.c
+++ b/src/core/client_config/lb_policies/round_robin.c
@@ -38,6 +38,8 @@
 #include <grpc/support/alloc.h>
 #include "src/core/transport/connectivity_state.h"
 
+typedef struct round_robin_lb_policy round_robin_lb_policy;
+
 int grpc_lb_round_robin_trace = 0;
 
 /** List of entities waiting for a pick.
@@ -46,7 +48,7 @@ int grpc_lb_round_robin_trace = 0;
 typedef struct pending_pick {
   struct pending_pick *next;
   grpc_pollset *pollset;
-  grpc_subchannel **target;
+  grpc_connected_subchannel **target;
   grpc_closure *on_complete;
 } pending_pick;
 
@@ -58,22 +60,27 @@ typedef struct ready_list {
 } ready_list;
 
 typedef struct {
-  size_t subchannel_idx; /**< Index over p->subchannels */
-  void *p;               /**< round_robin_lb_policy instance */
-} connectivity_changed_cb_arg;
-
-typedef struct {
+  /** index within policy->subchannels */
+  size_t index;
+  /** backpointer to owning policy */
+  round_robin_lb_policy *policy;
+  /** subchannel itself */
+  grpc_subchannel *subchannel;
+  /** notification that connectivity has changed on subchannel */
+  grpc_closure connectivity_changed_closure;
+  /** this subchannels current position in subchannel->ready_list */
+  ready_list *ready_list_node;
+  /** last observed connectivity */
+  grpc_connectivity_state connectivity_state;
+} subchannel_data;
+
+struct round_robin_lb_policy {
   /** base policy: must be first */
   grpc_lb_policy base;
 
   /** all our subchannels */
-  grpc_subchannel **subchannels;
   size_t num_subchannels;
-
-  /** Callbacks, one per subchannel being watched, to be called when their
-   * respective connectivity changes */
-  grpc_closure *connectivity_changed_cbs;
-  connectivity_changed_cb_arg *cb_args;
+  subchannel_data **subchannels;
 
   /** mutex protecting remaining members */
   gpr_mu mu;
@@ -81,8 +88,6 @@ typedef struct {
   int started_picking;
   /** are we shutting down? */
   int shutdown;
-  /** Connectivity state of the subchannels being watched */
-  grpc_connectivity_state *subchannel_connectivity;
   /** List of picks that are waiting on connectivity */
   pending_pick *pending_picks;
 
@@ -93,13 +98,7 @@ typedef struct {
   ready_list ready_list;
   /** Last pick from the ready list. */
   ready_list *ready_list_last_pick;
-
-  /** Subchannel index to ready_list node.
-   *
-   * Kept in order to remove nodes from the ready list associated with a
-   * subchannel */
-  ready_list **subchannel_index_to_readylist_node;
-} round_robin_lb_policy;
+};
 
 /** Returns the next subchannel from the connected list or NULL if the list is
  * empty.
@@ -144,9 +143,9 @@ static void advance_last_picked_locked(round_robin_lb_policy *p) {
 /** Prepends (relative to the root at p->ready_list) the connected subchannel \a
  * csc to the list of ready subchannels. */
 static ready_list *add_connected_sc_locked(round_robin_lb_policy *p,
-                                           grpc_subchannel *csc) {
+                                           grpc_subchannel *sc) {
   ready_list *new_elem = gpr_malloc(sizeof(ready_list));
-  new_elem->subchannel = csc;
+  new_elem->subchannel = sc;
   if (p->ready_list.prev == NULL) {
     /* first element */
     new_elem->next = &p->ready_list;
@@ -160,7 +159,7 @@ static ready_list *add_connected_sc_locked(round_robin_lb_policy *p,
     p->ready_list.prev = new_elem;
   }
   if (grpc_lb_round_robin_trace) {
-    gpr_log(GPR_DEBUG, "[READYLIST] ADDING NODE %p (SC %p)", new_elem, csc);
+    gpr_log(GPR_DEBUG, "[READYLIST] ADDING NODE %p (SC %p)", new_elem, sc);
   }
   return new_elem;
 }
@@ -200,28 +199,15 @@ static void remove_disconnected_sc_locked(round_robin_lb_policy *p,
   gpr_free(node);
 }
 
-static void del_interested_parties_locked(grpc_exec_ctx *exec_ctx,
-                                          round_robin_lb_policy *p,
-                                          const size_t subchannel_idx) {
-  pending_pick *pp;
-  for (pp = p->pending_picks; pp; pp = pp->next) {
-    grpc_subchannel_del_interested_party(
-        exec_ctx, p->subchannels[subchannel_idx], pp->pollset);
-  }
-}
-
 void rr_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
   round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
   size_t i;
   ready_list *elem;
   for (i = 0; i < p->num_subchannels; i++) {
-    del_interested_parties_locked(exec_ctx, p, i);
-  }
-  for (i = 0; i < p->num_subchannels; i++) {
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[i], "round_robin");
+    subchannel_data *sd = p->subchannels[i];
+    GRPC_SUBCHANNEL_UNREF(exec_ctx, sd->subchannel, "round_robin");
+    gpr_free(sd);
   }
-  gpr_free(p->connectivity_changed_cbs);
-  gpr_free(p->subchannel_connectivity);
 
   grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker);
   gpr_free(p->subchannels);
@@ -237,20 +223,15 @@ void rr_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
     gpr_free(elem);
     elem = tmp;
   }
-  gpr_free(p->subchannel_index_to_readylist_node);
-  gpr_free(p->cb_args);
   gpr_free(p);
 }
 
 void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
-  size_t i;
   round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
   pending_pick *pp;
-  gpr_mu_lock(&p->mu);
+  size_t i;
 
-  for (i = 0; i < p->num_subchannels; i++) {
-    del_interested_parties_locked(exec_ctx, p, i);
-  }
+  gpr_mu_lock(&p->mu);
 
   p->shutdown = 1;
   while ((pp = p->pending_picks)) {
@@ -261,24 +242,26 @@ void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
   }
   grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                               GRPC_CHANNEL_FATAL_FAILURE, "shutdown");
+  for (i = 0; i < p->num_subchannels; i++) {
+    subchannel_data *sd = p->subchannels[i];
+    grpc_subchannel_notify_on_state_change(exec_ctx, sd->subchannel, NULL, NULL,
+                                           &sd->connectivity_changed_closure);
+  }
   gpr_mu_unlock(&p->mu);
 }
 
 static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
-                           grpc_subchannel **target) {
+                           grpc_connected_subchannel **target) {
   round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
   pending_pick *pp;
-  size_t i;
   gpr_mu_lock(&p->mu);
   pp = p->pending_picks;
   p->pending_picks = NULL;
   while (pp != NULL) {
     pending_pick *next = pp->next;
     if (pp->target == target) {
-      for (i = 0; i < p->num_subchannels; i++) {
-        grpc_subchannel_add_interested_party(exec_ctx, p->subchannels[i],
-                                             pp->pollset);
-      }
+      grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+                                   pp->pollset);
       *target = NULL;
       grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 0);
       gpr_free(pp);
@@ -295,12 +278,16 @@ static void start_picking(grpc_exec_ctx *exec_ctx, round_robin_lb_policy *p) {
   size_t i;
   p->started_picking = 1;
 
+  gpr_log(GPR_DEBUG, "LB_POLICY: p=%p num_subchannels=%d", p,
+          p->num_subchannels);
+
   for (i = 0; i < p->num_subchannels; i++) {
-    p->subchannel_connectivity[i] = GRPC_CHANNEL_IDLE;
-    grpc_subchannel_notify_on_state_change(exec_ctx, p->subchannels[i],
-                                           &p->subchannel_connectivity[i],
-                                           &p->connectivity_changed_cbs[i]);
-    GRPC_LB_POLICY_REF(&p->base, "round_robin_connectivity");
+    subchannel_data *sd = p->subchannels[i];
+    sd->connectivity_state = GRPC_CHANNEL_IDLE;
+    grpc_subchannel_notify_on_state_change(
+        exec_ctx, sd->subchannel, &p->base.interested_parties,
+        &sd->connectivity_state, &sd->connectivity_changed_closure);
+    GRPC_LB_POLICY_WEAK_REF(&p->base, "round_robin_connectivity");
   }
 }
 
@@ -314,18 +301,18 @@ void rr_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
 }
 
 int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
-            grpc_metadata_batch *initial_metadata, grpc_subchannel **target,
-            grpc_closure *on_complete) {
-  size_t i;
+            grpc_metadata_batch *initial_metadata,
+            grpc_connected_subchannel **target, grpc_closure *on_complete) {
   round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
   pending_pick *pp;
   ready_list *selected;
   gpr_mu_lock(&p->mu);
   if ((selected = peek_next_connected_locked(p))) {
     gpr_mu_unlock(&p->mu);
-    *target = selected->subchannel;
+    *target = grpc_subchannel_get_connected_subchannel(selected->subchannel);
     if (grpc_lb_round_robin_trace) {
-      gpr_log(GPR_DEBUG, "[RR PICK] TARGET <-- SUBCHANNEL %p (NODE %p)",
+      gpr_log(GPR_DEBUG,
+              "[RR PICK] TARGET <-- CONNECTED SUBCHANNEL %p (NODE %p)",
               selected->subchannel, selected);
     }
     /* only advance the last picked pointer if the selection was used */
@@ -335,10 +322,8 @@ int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
     if (!p->started_picking) {
       start_picking(exec_ctx, p);
     }
-    for (i = 0; i < p->num_subchannels; i++) {
-      grpc_subchannel_add_interested_party(exec_ctx, p->subchannels[i],
-                                           pollset);
-    }
+    grpc_pollset_set_add_pollset(exec_ctx, &p->base.interested_parties,
+                                 pollset);
     pp = gpr_malloc(sizeof(*pp));
     pp->next = p->pending_picks;
     pp->pollset = pollset;
@@ -352,33 +337,25 @@ int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
 
 static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
                                     int iomgr_success) {
-  connectivity_changed_cb_arg *cb_arg = arg;
-  round_robin_lb_policy *p = cb_arg->p;
-  /* index over p->subchannels of this cb's subchannel */
-  const size_t this_idx = cb_arg->subchannel_idx;
+  subchannel_data *sd = arg;
+  round_robin_lb_policy *p = sd->policy;
   pending_pick *pp;
   ready_list *selected;
 
   int unref = 0;
 
-  /* connectivity state of this cb's subchannel */
-  grpc_connectivity_state *this_connectivity;
-
   gpr_mu_lock(&p->mu);
 
-  this_connectivity = &p->subchannel_connectivity[this_idx];
-
   if (p->shutdown) {
     unref = 1;
   } else {
-    switch (*this_connectivity) {
+    switch (sd->connectivity_state) {
       case GRPC_CHANNEL_READY:
         grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                     GRPC_CHANNEL_READY, "connecting_ready");
         /* add the newly connected subchannel to the list of connected ones.
          * Note that it goes to the "end of the line". */
-        p->subchannel_index_to_readylist_node[this_idx] =
-            add_connected_sc_locked(p, p->subchannels[this_idx]);
+        sd->ready_list_node = add_connected_sc_locked(p, sd->subchannel);
         /* at this point we know there's at least one suitable subchannel. Go
          * ahead and pick one and notify the pending suitors in
          * p->pending_picks. This preemtively replicates rr_pick()'s actions. */
@@ -390,60 +367,60 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
         }
         while ((pp = p->pending_picks)) {
           p->pending_picks = pp->next;
-          *pp->target = selected->subchannel;
+          *pp->target =
+              grpc_subchannel_get_connected_subchannel(selected->subchannel);
           if (grpc_lb_round_robin_trace) {
             gpr_log(GPR_DEBUG,
                     "[RR CONN CHANGED] TARGET <-- SUBCHANNEL %p (NODE %p)",
                     selected->subchannel, selected);
           }
-          grpc_subchannel_del_interested_party(exec_ctx, selected->subchannel,
-                                               pp->pollset);
+          grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+                                       pp->pollset);
           grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
           gpr_free(pp);
         }
         grpc_subchannel_notify_on_state_change(
-            exec_ctx, p->subchannels[this_idx], this_connectivity,
-            &p->connectivity_changed_cbs[this_idx]);
+            exec_ctx, sd->subchannel, &p->base.interested_parties,
+            &sd->connectivity_state, &sd->connectivity_changed_closure);
         break;
       case GRPC_CHANNEL_CONNECTING:
       case GRPC_CHANNEL_IDLE:
         grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
-                                    *this_connectivity, "connecting_changed");
+                                    sd->connectivity_state,
+                                    "connecting_changed");
         grpc_subchannel_notify_on_state_change(
-            exec_ctx, p->subchannels[this_idx], this_connectivity,
-            &p->connectivity_changed_cbs[this_idx]);
+            exec_ctx, sd->subchannel, &p->base.interested_parties,
+            &sd->connectivity_state, &sd->connectivity_changed_closure);
         break;
       case GRPC_CHANNEL_TRANSIENT_FAILURE:
-        del_interested_parties_locked(exec_ctx, p, this_idx);
         /* renew state notification */
         grpc_subchannel_notify_on_state_change(
-            exec_ctx, p->subchannels[this_idx], this_connectivity,
-            &p->connectivity_changed_cbs[this_idx]);
+            exec_ctx, sd->subchannel, &p->base.interested_parties,
+            &sd->connectivity_state, &sd->connectivity_changed_closure);
 
         /* remove from ready list if still present */
-        if (p->subchannel_index_to_readylist_node[this_idx] != NULL) {
-          remove_disconnected_sc_locked(
-              p, p->subchannel_index_to_readylist_node[this_idx]);
-          p->subchannel_index_to_readylist_node[this_idx] = NULL;
+        if (sd->ready_list_node != NULL) {
+          remove_disconnected_sc_locked(p, sd->ready_list_node);
+          sd->ready_list_node = NULL;
         }
         grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                     GRPC_CHANNEL_TRANSIENT_FAILURE,
                                     "connecting_transient_failure");
         break;
       case GRPC_CHANNEL_FATAL_FAILURE:
-        del_interested_parties_locked(exec_ctx, p, this_idx);
-        if (p->subchannel_index_to_readylist_node[this_idx] != NULL) {
-          remove_disconnected_sc_locked(
-              p, p->subchannel_index_to_readylist_node[this_idx]);
-          p->subchannel_index_to_readylist_node[this_idx] = NULL;
+        if (sd->ready_list_node != NULL) {
+          remove_disconnected_sc_locked(p, sd->ready_list_node);
+          sd->ready_list_node = NULL;
         }
 
-        GPR_SWAP(grpc_subchannel *, p->subchannels[this_idx],
-                 p->subchannels[p->num_subchannels - 1]);
         p->num_subchannels--;
-        GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[p->num_subchannels],
-                              "round_robin");
+        GPR_SWAP(subchannel_data *, p->subchannels[sd->index],
+                 p->subchannels[p->num_subchannels]);
+        GRPC_SUBCHANNEL_UNREF(exec_ctx, sd->subchannel, "round_robin");
+        p->subchannels[sd->index]->index = sd->index;
+        gpr_free(sd);
 
+        unref = 1;
         if (p->num_subchannels == 0) {
           grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                       GRPC_CHANNEL_FATAL_FAILURE,
@@ -454,7 +431,6 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
             grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
             gpr_free(pp);
           }
-          unref = 1;
         } else {
           grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                       GRPC_CHANNEL_TRANSIENT_FAILURE,
@@ -466,31 +442,8 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
   gpr_mu_unlock(&p->mu);
 
   if (unref) {
-    GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "round_robin_connectivity");
-  }
-}
-
-static void rr_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
-                         grpc_transport_op *op) {
-  round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
-  size_t i;
-  size_t n;
-  grpc_subchannel **subchannels;
-
-  gpr_mu_lock(&p->mu);
-  n = p->num_subchannels;
-  subchannels = gpr_malloc(n * sizeof(*subchannels));
-  for (i = 0; i < n; i++) {
-    subchannels[i] = p->subchannels[i];
-    GRPC_SUBCHANNEL_REF(subchannels[i], "rr_broadcast");
-  }
-  gpr_mu_unlock(&p->mu);
-
-  for (i = 0; i < n; i++) {
-    grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], op);
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "rr_broadcast");
+    GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "round_robin_connectivity");
   }
-  gpr_free(subchannels);
 }
 
 static grpc_connectivity_state rr_check_connectivity(grpc_exec_ctx *exec_ctx,
@@ -514,9 +467,25 @@ static void rr_notify_on_state_change(grpc_exec_ctx *exec_ctx,
   gpr_mu_unlock(&p->mu);
 }
 
+static void rr_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
+                        grpc_closure *closure) {
+  round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
+  ready_list *selected;
+  grpc_connected_subchannel *target;
+  gpr_mu_lock(&p->mu);
+  if ((selected = peek_next_connected_locked(p))) {
+    gpr_mu_unlock(&p->mu);
+    target = grpc_subchannel_get_connected_subchannel(selected->subchannel);
+    grpc_connected_subchannel_ping(exec_ctx, target, closure);
+  } else {
+    gpr_mu_unlock(&p->mu);
+    grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+  }
+}
+
 static const grpc_lb_policy_vtable round_robin_lb_policy_vtable = {
-    rr_destroy, rr_shutdown, rr_pick, rr_cancel_pick, rr_exit_idle,
-    rr_broadcast, rr_check_connectivity, rr_notify_on_state_change};
+    rr_destroy, rr_shutdown, rr_pick, rr_cancel_pick, rr_ping_one, rr_exit_idle,
+    rr_check_connectivity, rr_notify_on_state_change};
 
 static void round_robin_factory_ref(grpc_lb_policy_factory *factory) {}
 
@@ -529,27 +498,22 @@ static grpc_lb_policy *create_round_robin(grpc_lb_policy_factory *factory,
   GPR_ASSERT(args->num_subchannels > 0);
   memset(p, 0, sizeof(*p));
   grpc_lb_policy_init(&p->base, &round_robin_lb_policy_vtable);
-  p->subchannels =
-      gpr_malloc(sizeof(grpc_subchannel *) * args->num_subchannels);
   p->num_subchannels = args->num_subchannels;
+  p->subchannels = gpr_malloc(sizeof(*p->subchannels) * p->num_subchannels);
+  memset(p->subchannels, 0, sizeof(*p->subchannels) * p->num_subchannels);
   grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE,
                                "round_robin");
-  memcpy(p->subchannels, args->subchannels,
-         sizeof(grpc_subchannel *) * args->num_subchannels);
 
   gpr_mu_init(&p->mu);
-  p->connectivity_changed_cbs =
-      gpr_malloc(sizeof(grpc_closure) * args->num_subchannels);
-  p->subchannel_connectivity =
-      gpr_malloc(sizeof(grpc_connectivity_state) * args->num_subchannels);
-
-  p->cb_args =
-      gpr_malloc(sizeof(connectivity_changed_cb_arg) * args->num_subchannels);
   for (i = 0; i < args->num_subchannels; i++) {
-    p->cb_args[i].subchannel_idx = i;
-    p->cb_args[i].p = p;
-    grpc_closure_init(&p->connectivity_changed_cbs[i], rr_connectivity_changed,
-                      &p->cb_args[i]);
+    subchannel_data *sd = gpr_malloc(sizeof(*sd));
+    memset(sd, 0, sizeof(*sd));
+    p->subchannels[i] = sd;
+    sd->policy = p;
+    sd->index = i;
+    sd->subchannel = args->subchannels[i];
+    grpc_closure_init(&sd->connectivity_changed_closure,
+                      rr_connectivity_changed, sd);
   }
 
   /* The (dummy node) root of the ready list */
@@ -558,10 +522,6 @@ static grpc_lb_policy *create_round_robin(grpc_lb_policy_factory *factory,
   p->ready_list.next = NULL;
   p->ready_list_last_pick = &p->ready_list;
 
-  p->subchannel_index_to_readylist_node =
-      gpr_malloc(sizeof(grpc_subchannel *) * args->num_subchannels);
-  memset(p->subchannel_index_to_readylist_node, 0,
-         sizeof(grpc_subchannel *) * args->num_subchannels);
   return &p->base;
 }
 
diff --git a/src/core/client_config/lb_policy.c b/src/core/client_config/lb_policy.c
index 36a2454309..d4672f6b25 100644
--- a/src/core/client_config/lb_policy.c
+++ b/src/core/client_config/lb_policy.c
@@ -33,63 +33,94 @@
 
 #include "src/core/client_config/lb_policy.h"
 
+#define WEAK_REF_BITS 16
+
 void grpc_lb_policy_init(grpc_lb_policy *policy,
                          const grpc_lb_policy_vtable *vtable) {
   policy->vtable = vtable;
-  gpr_ref_init(&policy->refs, 1);
+  gpr_atm_no_barrier_store(&policy->ref_pair, 1 << WEAK_REF_BITS);
+  grpc_pollset_set_init(&policy->interested_parties);
 }
 
 #ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG
-void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line,
-                        const char *reason) {
-  gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "LB_POLICY:%p   ref %d -> %d %s",
-          policy, (int)policy->refs.count, (int)policy->refs.count + 1, reason);
+#define REF_FUNC_EXTRA_ARGS , const char *file, int line, const char *reason
+#define REF_MUTATE_EXTRA_ARGS REF_FUNC_EXTRA_ARGS, const char *purpose
+#define REF_FUNC_PASS_ARGS(new_reason) , file, line, new_reason
+#define REF_MUTATE_PASS_ARGS(purpose) , file, line, reason, purpose
 #else
-void grpc_lb_policy_ref(grpc_lb_policy *policy) {
+#define REF_FUNC_EXTRA_ARGS
+#define REF_MUTATE_EXTRA_ARGS
+#define REF_FUNC_PASS_ARGS(new_reason)
+#define REF_MUTATE_PASS_ARGS(x)
 #endif
-  gpr_ref(&policy->refs);
-}
 
+static gpr_atm ref_mutate(grpc_lb_policy *c, gpr_atm delta,
+                          int barrier REF_MUTATE_EXTRA_ARGS) {
+  gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta)
+                            : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta);
 #ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG
-void grpc_lb_policy_unref(grpc_lb_policy *policy,
-                          grpc_closure_list *closure_list, const char *file,
-                          int line, const char *reason) {
-  gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "LB_POLICY:%p unref %d -> %d %s",
-          policy, (int)policy->refs.count, (int)policy->refs.count - 1, reason);
-#else
-void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
+  gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
+          "LB_POLICY: %p % 12s 0x%08x -> 0x%08x [%s]", c, purpose, old_val,
+          old_val + delta, reason);
 #endif
-  if (gpr_unref(&policy->refs)) {
-    policy->vtable->destroy(exec_ctx, policy);
+  return old_val;
+}
+
+void grpc_lb_policy_ref(grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+  ref_mutate(policy, 1 << WEAK_REF_BITS, 0 REF_MUTATE_PASS_ARGS("STRONG_REF"));
+}
+
+void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx,
+                          grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+  gpr_atm old_val =
+      ref_mutate(policy, (gpr_atm)1 - (gpr_atm)(1 << WEAK_REF_BITS),
+                 1 REF_MUTATE_PASS_ARGS("STRONG_UNREF"));
+  gpr_atm mask = ~(gpr_atm)((1 << WEAK_REF_BITS) - 1);
+  gpr_atm check = 1 << WEAK_REF_BITS;
+  if ((old_val & mask) == check) {
+    policy->vtable->shutdown(exec_ctx, policy);
   }
+  grpc_lb_policy_weak_unref(exec_ctx,
+                            policy REF_FUNC_PASS_ARGS("strong-unref"));
 }
 
-void grpc_lb_policy_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
-  policy->vtable->shutdown(exec_ctx, policy);
+void grpc_lb_policy_weak_ref(grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+  ref_mutate(policy, 1, 0 REF_MUTATE_PASS_ARGS("WEAK_REF"));
+}
+
+void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx,
+                               grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+  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);
+    policy->vtable->destroy(exec_ctx, policy);
+  }
 }
 
 int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                         grpc_pollset *pollset,
                         grpc_metadata_batch *initial_metadata,
-                        grpc_subchannel **target, grpc_closure *on_complete) {
+                        grpc_connected_subchannel **target,
+                        grpc_closure *on_complete) {
   return policy->vtable->pick(exec_ctx, policy, pollset, initial_metadata,
                               target, on_complete);
 }
 
 void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                                grpc_subchannel **target) {
+                                grpc_connected_subchannel **target) {
   policy->vtable->cancel_pick(exec_ctx, policy, target);
 }
 
-void grpc_lb_policy_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                              grpc_transport_op *op) {
-  policy->vtable->broadcast(exec_ctx, policy, op);
-}
-
 void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
   policy->vtable->exit_idle(exec_ctx, policy);
 }
 
+void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
+                             grpc_closure *closure) {
+  policy->vtable->ping_one(exec_ctx, policy, closure);
+}
+
 void grpc_lb_policy_notify_on_state_change(grpc_exec_ctx *exec_ctx,
                                            grpc_lb_policy *policy,
                                            grpc_connectivity_state *state,
diff --git a/src/core/client_config/lb_policy.h b/src/core/client_config/lb_policy.h
index a696c3ce64..db5238c8ca 100644
--- a/src/core/client_config/lb_policy.h
+++ b/src/core/client_config/lb_policy.h
@@ -47,7 +47,8 @@ typedef void (*grpc_lb_completion)(void *cb_arg, grpc_subchannel *subchannel,
 
 struct grpc_lb_policy {
   const grpc_lb_policy_vtable *vtable;
-  gpr_refcount refs;
+  gpr_atm ref_pair;
+  grpc_pollset_set interested_parties;
 };
 
 struct grpc_lb_policy_vtable {
@@ -58,17 +59,16 @@ struct grpc_lb_policy_vtable {
   /** implement grpc_lb_policy_pick */
   int (*pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
               grpc_pollset *pollset, grpc_metadata_batch *initial_metadata,
-              grpc_subchannel **target, grpc_closure *on_complete);
+              grpc_connected_subchannel **target, grpc_closure *on_complete);
   void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                      grpc_subchannel **target);
+                      grpc_connected_subchannel **target);
+
+  void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
+                   grpc_closure *closure);
 
   /** try to enter a READY connectivity state */
   void (*exit_idle)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
 
-  /** broadcast a transport op to all subchannels */
-  void (*broadcast)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                    grpc_transport_op *op);
-
   /** check the current connectivity of the lb_policy */
   grpc_connectivity_state (*check_connectivity)(grpc_exec_ctx *exec_ctx,
                                                 grpc_lb_policy *policy);
@@ -81,29 +81,39 @@ struct grpc_lb_policy_vtable {
                                  grpc_closure *closure);
 };
 
+/*#define GRPC_LB_POLICY_REFCOUNT_DEBUG*/
 #ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG
 #define GRPC_LB_POLICY_REF(p, r) \
   grpc_lb_policy_ref((p), __FILE__, __LINE__, (r))
 #define GRPC_LB_POLICY_UNREF(exec_ctx, p, r) \
   grpc_lb_policy_unref((exec_ctx), (p), __FILE__, __LINE__, (r))
+#define GRPC_LB_POLICY_WEAK_REF(p, r) \
+  grpc_lb_policy_weak_ref((p), __FILE__, __LINE__, (r))
+#define GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, p, r) \
+  grpc_lb_policy_weak_unref((exec_ctx), (p), __FILE__, __LINE__, (r))
 void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line,
                         const char *reason);
 void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                           const char *file, int line, const char *reason);
+void grpc_lb_policy_weak_ref(grpc_lb_policy *policy, const char *file, int line,
+                             const char *reason);
+void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
+                               const char *file, int line, const char *reason);
 #else
 #define GRPC_LB_POLICY_REF(p, r) grpc_lb_policy_ref((p))
 #define GRPC_LB_POLICY_UNREF(cl, p, r) grpc_lb_policy_unref((cl), (p))
+#define GRPC_LB_POLICY_WEAK_REF(p, r) grpc_lb_policy_weak_ref((p))
+#define GRPC_LB_POLICY_WEAK_UNREF(cl, p, r) grpc_lb_policy_weak_unref((cl), (p))
 void grpc_lb_policy_ref(grpc_lb_policy *policy);
 void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
+void grpc_lb_policy_weak_ref(grpc_lb_policy *policy);
+void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
 #endif
 
 /** called by concrete implementations to initialize the base struct */
 void grpc_lb_policy_init(grpc_lb_policy *policy,
                          const grpc_lb_policy_vtable *vtable);
 
-/** Start shutting down (fail any pending picks) */
-void grpc_lb_policy_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
-
 /** Given initial metadata in \a initial_metadata, find an appropriate
     target for this rpc, and 'return' it by calling \a on_complete after setting
     \a target.
@@ -111,13 +121,14 @@ void grpc_lb_policy_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
 int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                         grpc_pollset *pollset,
                         grpc_metadata_batch *initial_metadata,
-                        grpc_subchannel **target, grpc_closure *on_complete);
+                        grpc_connected_subchannel **target,
+                        grpc_closure *on_complete);
 
-void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                                grpc_subchannel **target);
+void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
+                             grpc_closure *closure);
 
-void grpc_lb_policy_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                              grpc_transport_op *op);
+void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
+                                grpc_connected_subchannel **target);
 
 void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
 
diff --git a/src/core/client_config/resolver.c b/src/core/client_config/resolver.c
index 081097eb19..eda01e72ba 100644
--- a/src/core/client_config/resolver.c
+++ b/src/core/client_config/resolver.c
@@ -71,11 +71,8 @@ void grpc_resolver_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver) {
 }
 
 void grpc_resolver_channel_saw_error(grpc_exec_ctx *exec_ctx,
-                                     grpc_resolver *resolver,
-                                     struct sockaddr *failing_address,
-                                     int failing_address_len) {
-  resolver->vtable->channel_saw_error(exec_ctx, resolver, failing_address,
-                                      failing_address_len);
+                                     grpc_resolver *resolver) {
+  resolver->vtable->channel_saw_error(exec_ctx, resolver);
 }
 
 void grpc_resolver_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
diff --git a/src/core/client_config/resolver.h b/src/core/client_config/resolver.h
index 7ba0cd5bd4..e612eaf3b3 100644
--- a/src/core/client_config/resolver.h
+++ b/src/core/client_config/resolver.h
@@ -35,8 +35,8 @@
 #define GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_H
 
 #include "src/core/client_config/client_config.h"
+#include "src/core/client_config/subchannel.h"
 #include "src/core/iomgr/iomgr.h"
-#include "src/core/iomgr/sockaddr.h"
 
 typedef struct grpc_resolver grpc_resolver;
 typedef struct grpc_resolver_vtable grpc_resolver_vtable;
@@ -51,9 +51,7 @@ struct grpc_resolver {
 struct grpc_resolver_vtable {
   void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
   void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
-  void (*channel_saw_error)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
-                            struct sockaddr *failing_address,
-                            int failing_address_len);
+  void (*channel_saw_error)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
   void (*next)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
                grpc_client_config **target_config, grpc_closure *on_complete);
 };
@@ -81,9 +79,7 @@ void grpc_resolver_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
 /** Notification that the channel has seen an error on some address.
     Can be used as a hint that re-resolution is desirable soon. */
 void grpc_resolver_channel_saw_error(grpc_exec_ctx *exec_ctx,
-                                     grpc_resolver *resolver,
-                                     struct sockaddr *failing_address,
-                                     int failing_address_len);
+                                     grpc_resolver *resolver);
 
 /** Get the next client config. Called by the channel to fetch a new
     configuration. Expected to set *target_config with a new configuration,
diff --git a/src/core/client_config/resolvers/dns_resolver.c b/src/core/client_config/resolvers/dns_resolver.c
index 7f9dd2543f..28ca30b946 100644
--- a/src/core/client_config/resolvers/dns_resolver.c
+++ b/src/core/client_config/resolvers/dns_resolver.c
@@ -40,7 +40,6 @@
 #include <grpc/support/string_util.h>
 
 #include "src/core/client_config/lb_policy_registry.h"
-#include "src/core/client_config/subchannel_factory_decorators/add_channel_arg.h"
 #include "src/core/iomgr/resolve_address.h"
 #include "src/core/support/string.h"
 
@@ -81,9 +80,7 @@ static void dns_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
                                          dns_resolver *r);
 
 static void dns_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
-static void dns_channel_saw_error(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
-                                  struct sockaddr *failing_address,
-                                  int failing_address_len);
+static void dns_channel_saw_error(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
 static void dns_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
                      grpc_client_config **target_config,
                      grpc_closure *on_complete);
@@ -103,8 +100,7 @@ static void dns_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver) {
 }
 
 static void dns_channel_saw_error(grpc_exec_ctx *exec_ctx,
-                                  grpc_resolver *resolver, struct sockaddr *sa,
-                                  int len) {
+                                  grpc_resolver *resolver) {
   dns_resolver *r = (dns_resolver *)resolver;
   gpr_mu_lock(&r->mu);
   if (!r->resolving) {
diff --git a/src/core/client_config/resolvers/sockaddr_resolver.c b/src/core/client_config/resolvers/sockaddr_resolver.c
index 0b017f06c7..81d6627ecc 100644
--- a/src/core/client_config/resolvers/sockaddr_resolver.c
+++ b/src/core/client_config/resolvers/sockaddr_resolver.c
@@ -83,9 +83,7 @@ static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
 
 static void sockaddr_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
 static void sockaddr_channel_saw_error(grpc_exec_ctx *exec_ctx,
-                                       grpc_resolver *r,
-                                       struct sockaddr *failing_address,
-                                       int failing_address_len);
+                                       grpc_resolver *r);
 static void sockaddr_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
                           grpc_client_config **target_config,
                           grpc_closure *on_complete);
@@ -107,8 +105,13 @@ static void sockaddr_shutdown(grpc_exec_ctx *exec_ctx,
 }
 
 static void sockaddr_channel_saw_error(grpc_exec_ctx *exec_ctx,
-                                       grpc_resolver *resolver,
-                                       struct sockaddr *sa, int len) {}
+                                       grpc_resolver *resolver) {
+  sockaddr_resolver *r = (sockaddr_resolver *)resolver;
+  gpr_mu_lock(&r->mu);
+  r->published = 0;
+  sockaddr_maybe_finish_next_locked(exec_ctx, r);
+  gpr_mu_unlock(&r->mu);
+}
 
 static void sockaddr_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
                           grpc_client_config **target_config,
@@ -344,6 +347,9 @@ static grpc_resolver *sockaddr_create(
   gpr_slice_buffer_destroy(&path_parts);
   gpr_slice_unref(path_slice);
   if (errors_found) {
+    gpr_free(r->lb_policy_name);
+    gpr_free(r->addrs);
+    gpr_free(r->addrs_len);
     gpr_free(r);
     return NULL;
   }
diff --git a/src/core/client_config/resolvers/zookeeper_resolver.c b/src/core/client_config/resolvers/zookeeper_resolver.c
index 136197d4c6..4924ca77d6 100644
--- a/src/core/client_config/resolvers/zookeeper_resolver.c
+++ b/src/core/client_config/resolvers/zookeeper_resolver.c
@@ -96,9 +96,7 @@ static void zookeeper_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
 
 static void zookeeper_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
 static void zookeeper_channel_saw_error(grpc_exec_ctx *exec_ctx,
-                                        grpc_resolver *r,
-                                        struct sockaddr *failing_address,
-                                        int failing_address_len);
+                                        grpc_resolver *r);
 static void zookeeper_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
                            grpc_client_config **target_config,
                            grpc_closure *on_complete);
@@ -125,8 +123,7 @@ static void zookeeper_shutdown(grpc_exec_ctx *exec_ctx,
 }
 
 static void zookeeper_channel_saw_error(grpc_exec_ctx *exec_ctx,
-                                        grpc_resolver *resolver,
-                                        struct sockaddr *sa, int len) {
+                                        grpc_resolver *resolver) {
   zookeeper_resolver *r = (zookeeper_resolver *)resolver;
   gpr_mu_lock(&r->mu);
   if (r->resolving == 0) {
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index 28496ac2c9..afb1cdbd6d 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -47,39 +47,44 @@
 #include "src/core/transport/connectivity_state.h"
 #include "src/core/transport/connectivity_state.h"
 
+#define INTERNAL_REF_BITS 16
+#define STRONG_REF_MASK (~(gpr_atm)((1 << INTERNAL_REF_BITS) - 1))
+
 #define GRPC_SUBCHANNEL_MIN_CONNECT_TIMEOUT_SECONDS 20
 #define GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS 1
 #define GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER 1.6
 #define GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS 120
 #define GRPC_SUBCHANNEL_RECONNECT_JITTER 0.2
 
-typedef struct {
-  /* all fields protected by subchannel->mu */
-  /** refcount */
-  int refs;
-  /** parent subchannel */
-  grpc_subchannel *subchannel;
-} connection;
+#define GET_CONNECTED_SUBCHANNEL(subchannel, barrier)      \
+  ((grpc_connected_subchannel *)(gpr_atm_##barrier##_load( \
+      &(subchannel)->connected_subchannel)))
 
 typedef struct {
   grpc_closure closure;
-  size_t version;
   grpc_subchannel *subchannel;
   grpc_connectivity_state connectivity_state;
 } state_watcher;
 
-typedef struct waiting_for_connect {
-  struct waiting_for_connect *next;
-  grpc_closure *notify;
-  grpc_pollset *pollset;
-  gpr_atm *target;
+typedef struct external_state_watcher {
   grpc_subchannel *subchannel;
-  grpc_closure continuation;
-} waiting_for_connect;
+  grpc_pollset_set *pollset_set;
+  grpc_closure *notify;
+  grpc_closure closure;
+  struct external_state_watcher *next;
+  struct external_state_watcher *prev;
+} external_state_watcher;
 
 struct grpc_subchannel {
   grpc_connector *connector;
 
+  /** refcount
+      - lower INTERNAL_REF_BITS bits are for internal references:
+        these do not keep the subchannel open.
+      - upper remaining bits are for public references: these do
+        keep the subchannel open */
+  gpr_atm ref_pair;
+
   /** non-transport related channel filters */
   const grpc_channel_filter **filters;
   size_t num_filters;
@@ -88,15 +93,9 @@ struct grpc_subchannel {
   /** address to connect to */
   struct sockaddr *addr;
   size_t addr_len;
+
   /** initial string to send to peer */
   gpr_slice initial_connect_string;
-  /** master channel - the grpc_channel instance that ultimately owns
-      this channel_data via its channel stack.
-      We occasionally use this to bump the refcount on the master channel
-      to keep ourselves alive through an asynchronous operation. */
-  grpc_channel *master;
-  /** have we seen a disconnection? */
-  int disconnected;
 
   /** set during connection */
   grpc_connect_out_args connecting_result;
@@ -105,27 +104,24 @@ struct grpc_subchannel {
   grpc_closure connected;
 
   /** pollset_set tracking who's interested in a connection
-      being setup - owned by the master channel (in particular the
-     client_channel
-      filter there-in) */
-  grpc_pollset_set *pollset_set;
+      being setup */
+  grpc_pollset_set pollset_set;
+
+  /** active connection, or null; of type grpc_connected_subchannel */
+  gpr_atm connected_subchannel;
 
   /** mutex protecting remaining elements */
   gpr_mu mu;
 
-  /** active connection */
-  connection *active;
-  /** version number for the active connection */
-  size_t active_version;
-  /** refcount */
-  int refs;
+  /** have we seen a disconnection? */
+  int disconnected;
   /** are we connecting */
   int connecting;
-  /** things waiting for a connection */
-  waiting_for_connect *waiting;
   /** connectivity state tracking */
   grpc_connectivity_state_tracker state_tracker;
 
+  external_state_watcher root_external_state_watcher;
+
   /** next connect attempt time */
   gpr_timespec next_attempt;
   /** amount to backoff each failure */
@@ -139,151 +135,141 @@ struct grpc_subchannel {
 };
 
 struct grpc_subchannel_call {
-  connection *connection;
+  grpc_connected_subchannel *connection;
 };
 
 #define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1))
-#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)((con) + 1))
+#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)(con))
 #define CALLSTACK_TO_SUBCHANNEL_CALL(callstack) \
   (((grpc_subchannel_call *)(callstack)) - 1)
 
-static grpc_subchannel_call *create_call(grpc_exec_ctx *exec_ctx,
-                                         connection *con,
-                                         grpc_pollset *pollset);
-static void connectivity_state_changed_locked(grpc_exec_ctx *exec_ctx,
-                                              grpc_subchannel *c,
-                                              const char *reason);
-static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c);
 static gpr_timespec compute_connect_deadline(grpc_subchannel *c);
 static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel,
                                  int iomgr_success);
 
-static void subchannel_ref_locked(grpc_subchannel *c
-                                      GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
-static int subchannel_unref_locked(
-    grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT;
-static void connection_ref_locked(connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
-static grpc_subchannel *connection_unref_locked(
-    grpc_exec_ctx *exec_ctx,
-    connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT;
-static void subchannel_destroy(grpc_exec_ctx *exec_ctx, grpc_subchannel *c);
-
 #ifdef GRPC_STREAM_REFCOUNT_DEBUG
-#define SUBCHANNEL_REF_LOCKED(p, r) \
-  subchannel_ref_locked((p), __FILE__, __LINE__, (r))
-#define SUBCHANNEL_UNREF_LOCKED(p, r) \
-  subchannel_unref_locked((p), __FILE__, __LINE__, (r))
-#define CONNECTION_REF_LOCKED(p, r) \
-  connection_ref_locked((p), __FILE__, __LINE__, (r))
-#define CONNECTION_UNREF_LOCKED(cl, p, r) \
-  connection_unref_locked((cl), (p), __FILE__, __LINE__, (r))
-#define REF_PASS_ARGS , file, line, reason
-#define REF_PASS_REASON , reason
+#define REF_REASON reason
 #define REF_LOG(name, p)                                                  \
   gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p   ref %d -> %d %s", \
-          (name), (p), (p)->refs, (p)->refs + 1, reason)
+          (name), (p), (p)->refs.count, (p)->refs.count + 1, reason)
 #define UNREF_LOG(name, p)                                                \
   gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p unref %d -> %d %s", \
-          (name), (p), (p)->refs, (p)->refs - 1, reason)
+          (name), (p), (p)->refs.count, (p)->refs.count - 1, reason)
+#define REF_MUTATE_EXTRA_ARGS \
+  GRPC_SUBCHANNEL_REF_EXTRA_ARGS, const char *purpose
+#define REF_MUTATE_PURPOSE(x) , file, line, reason, x
 #else
-#define SUBCHANNEL_REF_LOCKED(p, r) subchannel_ref_locked((p))
-#define SUBCHANNEL_UNREF_LOCKED(p, r) subchannel_unref_locked((p))
-#define CONNECTION_REF_LOCKED(p, r) connection_ref_locked((p))
-#define CONNECTION_UNREF_LOCKED(cl, p, r) connection_unref_locked((cl), (p))
-#define REF_PASS_ARGS
-#define REF_PASS_REASON
+#define REF_REASON ""
 #define REF_LOG(name, p) \
   do {                   \
   } while (0)
 #define UNREF_LOG(name, p) \
   do {                     \
   } while (0)
+#define REF_MUTATE_EXTRA_ARGS
+#define REF_MUTATE_PURPOSE(x)
 #endif
 
 /*
  * connection implementation
  */
 
-static void connection_destroy(grpc_exec_ctx *exec_ctx, connection *c) {
-  GPR_ASSERT(c->refs == 0);
+static void connection_destroy(grpc_exec_ctx *exec_ctx, void *arg,
+                               int success) {
+  grpc_connected_subchannel *c = arg;
   grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c));
   gpr_free(c);
 }
 
-static void connection_ref_locked(connection *c
-                                      GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-  REF_LOG("CONNECTION", c);
-  subchannel_ref_locked(c->subchannel REF_PASS_ARGS);
-  ++c->refs;
+void grpc_connected_subchannel_ref(grpc_connected_subchannel *c
+                                       GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+  GRPC_CHANNEL_STACK_REF(CHANNEL_STACK_FROM_CONNECTION(c), REF_REASON);
 }
 
-static grpc_subchannel *connection_unref_locked(
-    grpc_exec_ctx *exec_ctx, connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-  grpc_subchannel *destroy = NULL;
-  UNREF_LOG("CONNECTION", c);
-  if (subchannel_unref_locked(c->subchannel REF_PASS_ARGS)) {
-    destroy = c->subchannel;
-  }
-  if (--c->refs == 0 && c->subchannel->active != c) {
-    connection_destroy(exec_ctx, c);
-  }
-  return destroy;
+void grpc_connected_subchannel_unref(grpc_exec_ctx *exec_ctx,
+                                     grpc_connected_subchannel *c
+                                         GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+  GRPC_CHANNEL_STACK_UNREF(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c),
+                           REF_REASON);
 }
 
 /*
  * grpc_subchannel implementation
  */
 
-static void subchannel_ref_locked(grpc_subchannel *c
-                                      GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-  REF_LOG("SUBCHANNEL", c);
-  ++c->refs;
+static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
+                               int success) {
+  grpc_subchannel *c = arg;
+  gpr_free((void *)c->filters);
+  grpc_channel_args_destroy(c->args);
+  gpr_free(c->addr);
+  gpr_slice_unref(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);
+  gpr_free(c);
 }
 
-static int subchannel_unref_locked(grpc_subchannel *c
-                                       GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-  UNREF_LOG("SUBCHANNEL", c);
-  return --c->refs == 0;
+static gpr_atm ref_mutate(grpc_subchannel *c, gpr_atm delta,
+                          int barrier REF_MUTATE_EXTRA_ARGS) {
+  gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta)
+                            : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta);
+#ifdef GRPC_STREAM_REFCOUNT_DEBUG
+  gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
+          "SUBCHANNEL: %p % 12s 0x%08x -> 0x%08x [%s]", c, purpose, old_val,
+          old_val + delta, reason);
+#endif
+  return old_val;
 }
 
 void grpc_subchannel_ref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-  gpr_mu_lock(&c->mu);
-  subchannel_ref_locked(c REF_PASS_ARGS);
-  gpr_mu_unlock(&c->mu);
+  gpr_atm old_refs;
+  old_refs = ref_mutate(c, (1 << INTERNAL_REF_BITS),
+                        0 REF_MUTATE_PURPOSE("STRONG_REF"));
+  GPR_ASSERT((old_refs & STRONG_REF_MASK) != 0);
 }
 
-void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
-                           grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-  int destroy;
-  gpr_mu_lock(&c->mu);
-  destroy = subchannel_unref_locked(c REF_PASS_ARGS);
-  gpr_mu_unlock(&c->mu);
-  if (destroy) subchannel_destroy(exec_ctx, c);
+void grpc_subchannel_weak_ref(grpc_subchannel *c
+                                  GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+  gpr_atm old_refs;
+  old_refs = ref_mutate(c, 1, 0 REF_MUTATE_PURPOSE("WEAK_REF"));
+  GPR_ASSERT(old_refs != 0);
 }
 
-static void subchannel_destroy(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
-  if (c->active != NULL) {
-    connection_destroy(exec_ctx, c->active);
+static void disconnect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
+  grpc_connected_subchannel *con;
+  gpr_mu_lock(&c->mu);
+  GPR_ASSERT(!c->disconnected);
+  c->disconnected = 1;
+  grpc_connector_shutdown(exec_ctx, c->connector);
+  con = GET_CONNECTED_SUBCHANNEL(c, no_barrier);
+  if (con != NULL) {
+    GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, con, "connection");
+    gpr_atm_no_barrier_store(&c->connected_subchannel, 0xdeadbeef);
   }
-  gpr_free((void *)c->filters);
-  grpc_channel_args_destroy(c->args);
-  gpr_free(c->addr);
-  gpr_slice_unref(c->initial_connect_string);
-  grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
-  grpc_connector_unref(exec_ctx, c->connector);
-  gpr_free(c);
+  gpr_mu_unlock(&c->mu);
 }
 
-void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx,
-                                          grpc_subchannel *c,
-                                          grpc_pollset *pollset) {
-  grpc_pollset_set_add_pollset(exec_ctx, c->pollset_set, pollset);
+void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
+                           grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+  gpr_atm old_refs;
+  old_refs = ref_mutate(c, (gpr_atm)1 - (gpr_atm)(1 << INTERNAL_REF_BITS),
+                        1 REF_MUTATE_PURPOSE("STRONG_UNREF"));
+  if ((old_refs & STRONG_REF_MASK) == (1 << INTERNAL_REF_BITS)) {
+    disconnect(exec_ctx, c);
+  }
+  GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "strong-unref");
 }
 
-void grpc_subchannel_del_interested_party(grpc_exec_ctx *exec_ctx,
-                                          grpc_subchannel *c,
-                                          grpc_pollset *pollset) {
-  grpc_pollset_set_del_pollset(exec_ctx, c->pollset_set, pollset);
+void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
+                                grpc_subchannel *c
+                                    GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+  gpr_atm old_refs;
+  old_refs = ref_mutate(c, -(gpr_atm)1, 1 REF_MUTATE_PURPOSE("WEAK_UNREF"));
+  if (old_refs == 1) {
+    grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(subchannel_destroy, c),
+                          1);
+  }
 }
 
 static gpr_uint32 random_seed() {
@@ -293,10 +279,8 @@ static gpr_uint32 random_seed() {
 grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
                                         grpc_subchannel_args *args) {
   grpc_subchannel *c = gpr_malloc(sizeof(*c));
-  grpc_channel_element *parent_elem = grpc_channel_stack_last_element(
-      grpc_channel_get_channel_stack(args->master));
   memset(c, 0, sizeof(*c));
-  c->refs = 1;
+  gpr_atm_no_barrier_store(&c->ref_pair, 1 << INTERNAL_REF_BITS);
   c->connector = connector;
   grpc_connector_ref(c->connector);
   c->num_filters = args->filter_count;
@@ -305,13 +289,14 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
          sizeof(grpc_channel_filter *) * c->num_filters);
   c->addr = gpr_malloc(args->addr_len);
   memcpy(c->addr, args->addr, args->addr_len);
+  grpc_pollset_set_init(&c->pollset_set);
   c->addr_len = args->addr_len;
   grpc_set_initial_connect_string(&c->addr, &c->addr_len,
                                   &c->initial_connect_string);
   c->args = grpc_channel_args_copy(args->args);
-  c->master = args->master;
-  c->pollset_set = grpc_client_channel_get_connecting_pollset_set(parent_elem);
   c->random = random_seed();
+  c->root_external_state_watcher.next = c->root_external_state_watcher.prev =
+      &c->root_external_state_watcher;
   grpc_closure_init(&c->connected, subchannel_connected, c);
   grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE,
                                "subchannel");
@@ -319,70 +304,18 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
   return c;
 }
 
-static void cancel_waiting_calls(grpc_exec_ctx *exec_ctx,
-                                 grpc_subchannel *subchannel,
-                                 int iomgr_success) {
-  waiting_for_connect *w4c;
-  gpr_mu_lock(&subchannel->mu);
-  w4c = subchannel->waiting;
-  subchannel->waiting = NULL;
-  gpr_mu_unlock(&subchannel->mu);
-  while (w4c != NULL) {
-    waiting_for_connect *next = w4c->next;
-    grpc_subchannel_del_interested_party(exec_ctx, w4c->subchannel,
-                                         w4c->pollset);
-    if (w4c->notify) {
-      w4c->notify->cb(exec_ctx, w4c->notify->cb_arg, iomgr_success);
-    }
-
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, w4c->subchannel, "waiting_for_connect");
-    gpr_free(w4c);
-
-    w4c = next;
-  }
-}
-
-void grpc_subchannel_cancel_create_call(grpc_exec_ctx *exec_ctx,
-                                        grpc_subchannel *subchannel,
-                                        gpr_atm *target) {
-  waiting_for_connect *w4c;
-  int unref_count = 0;
-  gpr_mu_lock(&subchannel->mu);
-  w4c = subchannel->waiting;
-  subchannel->waiting = NULL;
-  while (w4c != NULL) {
-    waiting_for_connect *next = w4c->next;
-    if (w4c->target == target) {
-      grpc_subchannel_del_interested_party(exec_ctx, w4c->subchannel,
-                                           w4c->pollset);
-      grpc_exec_ctx_enqueue(exec_ctx, w4c->notify, 0);
-
-      unref_count++;
-      gpr_free(w4c);
-    } else {
-      w4c->next = subchannel->waiting;
-      subchannel->waiting = w4c;
-    }
-
-    w4c = next;
-  }
-  gpr_mu_unlock(&subchannel->mu);
-
-  while (unref_count-- > 0) {
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannel, "waiting_for_connect");
-  }
-}
-
 static void continue_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
   grpc_connect_in_args args;
 
-  args.interested_parties = c->pollset_set;
+  args.interested_parties = &c->pollset_set;
   args.addr = c->addr;
   args.addr_len = c->addr_len;
   args.deadline = compute_connect_deadline(c);
   args.channel_args = c->args;
   args.initial_connect_string = c->initial_connect_string;
 
+  grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
+                              GRPC_CHANNEL_CONNECTING, "state_change");
   grpc_connector_connect(exec_ctx, c->connector, &args, &c->connecting_result,
                          &c->connected);
 }
@@ -395,66 +328,6 @@ static void start_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
   continue_connect(exec_ctx, c);
 }
 
-static void continue_creating_call(grpc_exec_ctx *exec_ctx, void *arg,
-                                   int iomgr_success) {
-  int call_creation_finished_ok;
-  waiting_for_connect *w4c = arg;
-  grpc_subchannel_del_interested_party(exec_ctx, w4c->subchannel, w4c->pollset);
-  call_creation_finished_ok = grpc_subchannel_create_call(
-      exec_ctx, w4c->subchannel, w4c->pollset, w4c->target, w4c->notify);
-  GPR_ASSERT(call_creation_finished_ok == 1);
-  w4c->notify->cb(exec_ctx, w4c->notify->cb_arg, iomgr_success);
-  GRPC_SUBCHANNEL_UNREF(exec_ctx, w4c->subchannel, "waiting_for_connect");
-  gpr_free(w4c);
-}
-
-int grpc_subchannel_create_call(grpc_exec_ctx *exec_ctx, grpc_subchannel *c,
-                                grpc_pollset *pollset, gpr_atm *target,
-                                grpc_closure *notify) {
-  connection *con;
-  grpc_subchannel_call *call;
-  GPR_TIMER_BEGIN("grpc_subchannel_create_call", 0);
-  gpr_mu_lock(&c->mu);
-  if (c->active != NULL) {
-    con = c->active;
-    CONNECTION_REF_LOCKED(con, "call");
-    gpr_mu_unlock(&c->mu);
-
-    call = create_call(exec_ctx, con, pollset);
-    if (!gpr_atm_rel_cas(target, 0, (gpr_atm)(gpr_uintptr)call)) {
-      GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "failed to set");
-    }
-    GPR_TIMER_END("grpc_subchannel_create_call", 0);
-    return 1;
-  } else {
-    waiting_for_connect *w4c = gpr_malloc(sizeof(*w4c));
-    w4c->next = c->waiting;
-    w4c->notify = notify;
-    w4c->pollset = pollset;
-    w4c->target = target;
-    w4c->subchannel = c;
-    /* released when clearing w4c */
-    SUBCHANNEL_REF_LOCKED(c, "waiting_for_connect");
-    grpc_closure_init(&w4c->continuation, continue_creating_call, w4c);
-    c->waiting = w4c;
-    grpc_subchannel_add_interested_party(exec_ctx, c, pollset);
-    if (!c->connecting) {
-      c->connecting = 1;
-      connectivity_state_changed_locked(exec_ctx, c, "create_call");
-      /* released by connection */
-      SUBCHANNEL_REF_LOCKED(c, "connecting");
-      GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting");
-      gpr_mu_unlock(&c->mu);
-
-      start_connect(exec_ctx, c);
-    } else {
-      gpr_mu_unlock(&c->mu);
-    }
-    GPR_TIMER_END("grpc_subchannel_create_call", 0);
-    return 0;
-  }
-}
-
 grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
   grpc_connectivity_state state;
   gpr_mu_lock(&c->mu);
@@ -463,153 +336,149 @@ grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
   return state;
 }
 
-void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
-                                            grpc_subchannel *c,
-                                            grpc_connectivity_state *state,
-                                            grpc_closure *notify) {
+static void on_external_state_watcher_done(grpc_exec_ctx *exec_ctx, void *arg,
+                                           int success) {
+  external_state_watcher *w = arg;
+  grpc_closure *follow_up = w->notify;
+  if (w->pollset_set != NULL) {
+    grpc_pollset_set_del_pollset_set(exec_ctx, &w->subchannel->pollset_set,
+                                     w->pollset_set);
+  }
+  gpr_mu_lock(&w->subchannel->mu);
+  w->next->prev = w->prev;
+  w->prev->next = w->next;
+  gpr_mu_unlock(&w->subchannel->mu);
+  GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, w->subchannel, "external_state_watcher");
+  gpr_free(w);
+  follow_up->cb(exec_ctx, follow_up->cb_arg, success);
+}
+
+void grpc_subchannel_notify_on_state_change(
+    grpc_exec_ctx *exec_ctx, grpc_subchannel *c,
+    grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
+    grpc_closure *notify) {
   int do_connect = 0;
-  gpr_mu_lock(&c->mu);
-  if (grpc_connectivity_state_notify_on_state_change(
-          exec_ctx, &c->state_tracker, state, notify)) {
-    do_connect = 1;
-    c->connecting = 1;
-    /* released by connection */
-    SUBCHANNEL_REF_LOCKED(c, "connecting");
-    GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting");
-    connectivity_state_changed_locked(exec_ctx, c, "state_change");
+  external_state_watcher *w;
+
+  if (state == NULL) {
+    gpr_mu_lock(&c->mu);
+    for (w = c->root_external_state_watcher.next;
+         w != &c->root_external_state_watcher; w = w->next) {
+      if (w->notify == notify) {
+        grpc_connectivity_state_notify_on_state_change(
+            exec_ctx, &c->state_tracker, NULL, &w->closure);
+      }
+    }
+    gpr_mu_unlock(&c->mu);
+  } else {
+    w = gpr_malloc(sizeof(*w));
+    w->subchannel = c;
+    w->pollset_set = interested_parties;
+    w->notify = notify;
+    grpc_closure_init(&w->closure, on_external_state_watcher_done, w);
+    if (interested_parties != NULL) {
+      grpc_pollset_set_add_pollset_set(exec_ctx, &c->pollset_set,
+                                       interested_parties);
+    }
+    GRPC_SUBCHANNEL_WEAK_REF(c, "external_state_watcher");
+    gpr_mu_lock(&c->mu);
+    w->next = &c->root_external_state_watcher;
+    w->prev = w->next->prev;
+    w->next->prev = w->prev->next = w;
+    if (grpc_connectivity_state_notify_on_state_change(
+            exec_ctx, &c->state_tracker, state, &w->closure)) {
+      do_connect = 1;
+      c->connecting = 1;
+      /* released by connection */
+      GRPC_SUBCHANNEL_WEAK_REF(c, "connecting");
+    }
+    gpr_mu_unlock(&c->mu);
   }
-  gpr_mu_unlock(&c->mu);
 
   if (do_connect) {
     start_connect(exec_ctx, c);
   }
 }
 
-int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
-                                             grpc_subchannel *c,
-                                             grpc_closure *subscribed_notify) {
-  int success;
-  gpr_mu_lock(&c->mu);
-  success = grpc_connectivity_state_change_unsubscribe(
-      exec_ctx, &c->state_tracker, subscribed_notify);
-  gpr_mu_unlock(&c->mu);
-  return success;
+void grpc_connected_subchannel_process_transport_op(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
+    grpc_transport_op *op) {
+  grpc_channel_stack *channel_stack = CHANNEL_STACK_FROM_CONNECTION(con);
+  grpc_channel_element *top_elem = grpc_channel_stack_element(channel_stack, 0);
+  top_elem->filter->start_transport_op(exec_ctx, top_elem, op);
 }
 
-void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
-                                          grpc_subchannel *c,
-                                          grpc_transport_op *op) {
-  connection *con = NULL;
-  grpc_subchannel *destroy;
-  int cancel_alarm = 0;
-  gpr_mu_lock(&c->mu);
-  if (c->active != NULL) {
-    con = c->active;
-    CONNECTION_REF_LOCKED(con, "transport-op");
-  }
-  if (op->disconnect) {
-    c->disconnected = 1;
-    connectivity_state_changed_locked(exec_ctx, c, "disconnect");
-    if (c->have_alarm) {
-      cancel_alarm = 1;
-    }
-  }
-  gpr_mu_unlock(&c->mu);
+static void subchannel_on_child_state_changed(grpc_exec_ctx *exec_ctx, void *p,
+                                              int iomgr_success) {
+  state_watcher *sw = p;
+  grpc_subchannel *c = sw->subchannel;
+  gpr_mu *mu = &c->mu;
 
-  if (con != NULL) {
-    grpc_channel_stack *channel_stack = CHANNEL_STACK_FROM_CONNECTION(con);
-    grpc_channel_element *top_elem =
-        grpc_channel_stack_element(channel_stack, 0);
-    top_elem->filter->start_transport_op(exec_ctx, top_elem, op);
+  gpr_mu_lock(mu);
 
-    gpr_mu_lock(&c->mu);
-    destroy = CONNECTION_UNREF_LOCKED(exec_ctx, con, "transport-op");
-    gpr_mu_unlock(&c->mu);
-    if (destroy) {
-      subchannel_destroy(exec_ctx, destroy);
+  /* if we failed just leave this closure */
+  if (iomgr_success) {
+    if (sw->connectivity_state == GRPC_CHANNEL_TRANSIENT_FAILURE) {
+      /* any errors on a subchannel ==> we're done, create a new one */
+      sw->connectivity_state = GRPC_CHANNEL_FATAL_FAILURE;
+    }
+    grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
+                                sw->connectivity_state, "reflect_child");
+    if (sw->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) {
+      grpc_connected_subchannel_notify_on_state_change(
+          exec_ctx, GET_CONNECTED_SUBCHANNEL(c, no_barrier), NULL,
+          &sw->connectivity_state, &sw->closure);
+      GRPC_SUBCHANNEL_WEAK_REF(c, "state_watcher");
+      sw = NULL;
     }
   }
 
-  if (cancel_alarm) {
-    grpc_timer_cancel(exec_ctx, &c->alarm);
-  }
-
-  if (op->disconnect) {
-    grpc_connector_shutdown(exec_ctx, c->connector);
-  }
+  gpr_mu_unlock(mu);
+  GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "state_watcher");
+  gpr_free(sw);
 }
 
-static void on_state_changed(grpc_exec_ctx *exec_ctx, void *p,
-                             int iomgr_success) {
-  state_watcher *sw = p;
-  grpc_subchannel *c = sw->subchannel;
-  gpr_mu *mu = &c->mu;
-  int destroy;
+static void connected_subchannel_state_op(grpc_exec_ctx *exec_ctx,
+                                          grpc_connected_subchannel *con,
+                                          grpc_pollset_set *interested_parties,
+                                          grpc_connectivity_state *state,
+                                          grpc_closure *closure) {
   grpc_transport_op op;
   grpc_channel_element *elem;
-  connection *destroy_connection = NULL;
-
-  gpr_mu_lock(mu);
-
-  /* if we failed or there is a version number mismatch, just leave
-     this closure */
-  if (!iomgr_success || sw->subchannel->active_version != sw->version) {
-    goto done;
-  }
+  memset(&op, 0, sizeof(op));
+  op.connectivity_state = state;
+  op.on_connectivity_state_change = closure;
+  op.bind_pollset_set = interested_parties;
+  elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
+  elem->filter->start_transport_op(exec_ctx, elem, &op);
+}
 
-  switch (sw->connectivity_state) {
-    case GRPC_CHANNEL_CONNECTING:
-    case GRPC_CHANNEL_READY:
-    case GRPC_CHANNEL_IDLE:
-      /* all is still good: keep watching */
-      memset(&op, 0, sizeof(op));
-      op.connectivity_state = &sw->connectivity_state;
-      op.on_connectivity_state_change = &sw->closure;
-      elem = grpc_channel_stack_element(
-          CHANNEL_STACK_FROM_CONNECTION(c->active), 0);
-      elem->filter->start_transport_op(exec_ctx, elem, &op);
-      /* early out */
-      gpr_mu_unlock(mu);
-      return;
-    case GRPC_CHANNEL_FATAL_FAILURE:
-    case GRPC_CHANNEL_TRANSIENT_FAILURE:
-      /* things have gone wrong, deactivate and enter idle */
-      if (sw->subchannel->active->refs == 0) {
-        destroy_connection = sw->subchannel->active;
-      }
-      sw->subchannel->active = NULL;
-      grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
-                                  c->disconnected
-                                      ? GRPC_CHANNEL_FATAL_FAILURE
-                                      : GRPC_CHANNEL_TRANSIENT_FAILURE,
-                                  "connection_failed");
-      break;
-  }
+void grpc_connected_subchannel_notify_on_state_change(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
+    grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
+    grpc_closure *closure) {
+  connected_subchannel_state_op(exec_ctx, con, interested_parties, state,
+                                closure);
+}
 
-done:
-  connectivity_state_changed_locked(exec_ctx, c, "transport_state_changed");
-  destroy = SUBCHANNEL_UNREF_LOCKED(c, "state_watcher");
-  gpr_free(sw);
-  gpr_mu_unlock(mu);
-  if (destroy) {
-    subchannel_destroy(exec_ctx, c);
-  }
-  if (destroy_connection != NULL) {
-    connection_destroy(exec_ctx, destroy_connection);
-  }
+void grpc_connected_subchannel_ping(grpc_exec_ctx *exec_ctx,
+                                    grpc_connected_subchannel *con,
+                                    grpc_closure *closure) {
+  grpc_transport_op op;
+  grpc_channel_element *elem;
+  memset(&op, 0, sizeof(op));
+  op.send_ping = closure;
+  elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
+  elem->filter->start_transport_op(exec_ctx, elem, &op);
 }
 
 static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
   size_t channel_stack_size;
-  connection *con;
+  grpc_connected_subchannel *con;
   grpc_channel_stack *stk;
   size_t num_filters;
   const grpc_channel_filter **filters;
-  waiting_for_connect *w4c;
-  grpc_transport_op op;
-  state_watcher *sw;
-  connection *destroy_connection = NULL;
-  grpc_channel_element *elem;
+  state_watcher *sw_subchannel;
 
   /* build final filter list */
   num_filters = c->num_filters + c->connecting_result.num_filters + 1;
@@ -621,74 +490,52 @@ static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
 
   /* construct channel stack */
   channel_stack_size = grpc_channel_stack_size(filters, num_filters);
-  con = gpr_malloc(sizeof(connection) + channel_stack_size);
-  stk = (grpc_channel_stack *)(con + 1);
-  con->refs = 0;
-  con->subchannel = c;
-  grpc_channel_stack_init(exec_ctx, filters, num_filters, c->master, c->args,
-                          stk);
+  con = gpr_malloc(channel_stack_size);
+  stk = CHANNEL_STACK_FROM_CONNECTION(con);
+  grpc_channel_stack_init(exec_ctx, 1, connection_destroy, con, filters,
+                          num_filters, c->args, "CONNECTED_SUBCHANNEL", stk);
   grpc_connected_channel_bind_transport(stk, c->connecting_result.transport);
   gpr_free((void *)c->connecting_result.filters);
   memset(&c->connecting_result, 0, sizeof(c->connecting_result));
 
   /* initialize state watcher */
-  sw = gpr_malloc(sizeof(*sw));
-  grpc_closure_init(&sw->closure, on_state_changed, sw);
-  sw->subchannel = c;
-  sw->connectivity_state = GRPC_CHANNEL_READY;
+  sw_subchannel = gpr_malloc(sizeof(*sw_subchannel));
+  sw_subchannel->subchannel = c;
+  sw_subchannel->connectivity_state = GRPC_CHANNEL_READY;
+  grpc_closure_init(&sw_subchannel->closure, subchannel_on_child_state_changed,
+                    sw_subchannel);
 
   gpr_mu_lock(&c->mu);
 
   if (c->disconnected) {
     gpr_mu_unlock(&c->mu);
-    gpr_free(sw);
+    gpr_free(sw_subchannel);
     gpr_free((void *)filters);
     grpc_channel_stack_destroy(exec_ctx, stk);
-    GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "connecting");
+    gpr_free(con);
+    GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
     return;
   }
 
   /* publish */
-  if (c->active != NULL && c->active->refs == 0) {
-    destroy_connection = c->active;
-  }
-  c->active = con;
-  c->active_version++;
-  sw->version = c->active_version;
+  GPR_ASSERT(gpr_atm_no_barrier_cas(&c->connected_subchannel, 0, (gpr_atm)con));
   c->connecting = 0;
 
-  /* watch for changes; subchannel ref for connecting is donated
+  /* setup subchannel watching connected subchannel for changes; subchannel ref
+     for connecting is donated
      to the state watcher */
-  memset(&op, 0, sizeof(op));
-  op.connectivity_state = &sw->connectivity_state;
-  op.on_connectivity_state_change = &sw->closure;
-  op.bind_pollset_set = c->pollset_set;
-  SUBCHANNEL_REF_LOCKED(c, "state_watcher");
-  GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
-  GPR_ASSERT(!SUBCHANNEL_UNREF_LOCKED(c, "connecting"));
-  elem =
-      grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(c->active), 0);
-  elem->filter->start_transport_op(exec_ctx, elem, &op);
+  GRPC_SUBCHANNEL_WEAK_REF(c, "state_watcher");
+  GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
+  grpc_connected_subchannel_notify_on_state_change(
+      exec_ctx, con, &c->pollset_set, &sw_subchannel->connectivity_state,
+      &sw_subchannel->closure);
 
   /* signal completion */
-  connectivity_state_changed_locked(exec_ctx, c, "connected");
-  w4c = c->waiting;
-  c->waiting = NULL;
+  grpc_connectivity_state_set(exec_ctx, &c->state_tracker, GRPC_CHANNEL_READY,
+                              "connected");
 
   gpr_mu_unlock(&c->mu);
-
-  while (w4c != NULL) {
-    waiting_for_connect *next = w4c->next;
-    grpc_exec_ctx_enqueue(exec_ctx, &w4c->continuation, 1);
-    w4c = next;
-  }
-
   gpr_free((void *)filters);
-
-  if (destroy_connection != NULL) {
-    connection_destroy(exec_ctx, destroy_connection);
-  }
 }
 
 /* Generate a random number between 0 and 1. */
@@ -742,29 +589,31 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, int iomgr_success) {
   if (c->disconnected) {
     iomgr_success = 0;
   }
-  connectivity_state_changed_locked(exec_ctx, c, "alarm");
   gpr_mu_unlock(&c->mu);
   if (iomgr_success) {
     update_reconnect_parameters(c);
     continue_connect(exec_ctx, c);
   } else {
-    cancel_waiting_calls(exec_ctx, c, iomgr_success);
-    GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
-    GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "connecting");
+    GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
   }
 }
 
 static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg,
                                  int iomgr_success) {
   grpc_subchannel *c = arg;
+
   if (c->connecting_result.transport != NULL) {
     publish_transport(exec_ctx, c);
+  } else if (c->disconnected) {
+    GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
   } else {
     gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
     gpr_mu_lock(&c->mu);
     GPR_ASSERT(!c->have_alarm);
     c->have_alarm = 1;
-    connectivity_state_changed_locked(exec_ctx, c, "connect_failed");
+    grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
+                                GRPC_CHANNEL_TRANSIENT_FAILURE,
+                                "connect_failed");
     grpc_timer_init(exec_ctx, &c->alarm, c->next_attempt, on_alarm, c, now);
     gpr_mu_unlock(&c->mu);
   }
@@ -781,29 +630,6 @@ static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
                                                           : min_deadline;
 }
 
-static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) {
-  if (c->disconnected) {
-    return GRPC_CHANNEL_FATAL_FAILURE;
-  }
-  if (c->connecting) {
-    if (c->have_alarm) {
-      return GRPC_CHANNEL_TRANSIENT_FAILURE;
-    }
-    return GRPC_CHANNEL_CONNECTING;
-  }
-  if (c->active) {
-    return GRPC_CHANNEL_READY;
-  }
-  return GRPC_CHANNEL_IDLE;
-}
-
-static void connectivity_state_changed_locked(grpc_exec_ctx *exec_ctx,
-                                              grpc_subchannel *c,
-                                              const char *reason) {
-  grpc_connectivity_state current = compute_connectivity_locked(c);
-  grpc_connectivity_state_set(exec_ctx, &c->state_tracker, current, reason);
-}
-
 /*
  * grpc_subchannel_call implementation
  */
@@ -811,37 +637,22 @@ static void connectivity_state_changed_locked(grpc_exec_ctx *exec_ctx,
 static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call,
                                     int success) {
   grpc_subchannel_call *c = call;
-  gpr_mu *mu = &c->connection->subchannel->mu;
-  grpc_subchannel *destroy;
   GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0);
   grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c));
-  gpr_mu_lock(mu);
-  destroy = CONNECTION_UNREF_LOCKED(exec_ctx, c->connection, "call");
-  gpr_mu_unlock(mu);
+  GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, c->connection, "subchannel_call");
   gpr_free(c);
-  if (destroy != NULL) {
-    subchannel_destroy(exec_ctx, destroy);
-  }
   GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0);
 }
 
 void grpc_subchannel_call_ref(grpc_subchannel_call *c
                                   GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-#ifdef GRPC_STREAM_REFCOUNT_DEBUG
-  grpc_call_stack_ref(SUBCHANNEL_CALL_TO_CALL_STACK(c), reason);
-#else
-  grpc_call_stack_ref(SUBCHANNEL_CALL_TO_CALL_STACK(c));
-#endif
+  GRPC_CALL_STACK_REF(SUBCHANNEL_CALL_TO_CALL_STACK(c), REF_REASON);
 }
 
 void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
                                 grpc_subchannel_call *c
                                     GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-#ifdef GRPC_STREAM_REFCOUNT_DEBUG
-  grpc_call_stack_unref(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), reason);
-#else
-  grpc_call_stack_unref(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c));
-#endif
+  GRPC_CALL_STACK_UNREF(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), REF_REASON);
 }
 
 char *grpc_subchannel_call_get_peer(grpc_exec_ctx *exec_ctx,
@@ -859,24 +670,26 @@ void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx,
   top_elem->filter->start_transport_stream_op(exec_ctx, top_elem, op);
 }
 
-static grpc_subchannel_call *create_call(grpc_exec_ctx *exec_ctx,
-                                         connection *con,
-                                         grpc_pollset *pollset) {
+grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
+    grpc_subchannel *c) {
+  return GET_CONNECTED_SUBCHANNEL(c, acq);
+}
+
+grpc_subchannel_call *grpc_connected_subchannel_create_call(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
+    grpc_pollset *pollset) {
   grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
   grpc_subchannel_call *call =
       gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
   grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call);
   call->connection = con;
+  GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call");
   grpc_call_stack_init(exec_ctx, chanstk, 1, subchannel_call_destroy, call,
                        NULL, NULL, callstk);
   grpc_call_stack_set_pollset(exec_ctx, callstk, pollset);
   return call;
 }
 
-grpc_channel *grpc_subchannel_get_master(grpc_subchannel *subchannel) {
-  return subchannel->master;
-}
-
 grpc_call_stack *grpc_subchannel_call_get_call_stack(
     grpc_subchannel_call *subchannel_call) {
   return SUBCHANNEL_CALL_TO_CALL_STACK(subchannel_call);
diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h
index 85ea3739e4..57c7c9dc67 100644
--- a/src/core/client_config/subchannel.h
+++ b/src/core/client_config/subchannel.h
@@ -41,6 +41,7 @@
 /** A (sub-)channel that knows how to connect to exactly one target
     address. Provides a target for load balancing. */
 typedef struct grpc_subchannel grpc_subchannel;
+typedef struct grpc_connected_subchannel grpc_connected_subchannel;
 typedef struct grpc_subchannel_call grpc_subchannel_call;
 typedef struct grpc_subchannel_args grpc_subchannel_args;
 
@@ -49,6 +50,14 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
   grpc_subchannel_ref((p), __FILE__, __LINE__, (r))
 #define GRPC_SUBCHANNEL_UNREF(cl, p, r) \
   grpc_subchannel_unref((cl), (p), __FILE__, __LINE__, (r))
+#define GRPC_SUBCHANNEL_WEAK_REF(p, r) \
+  grpc_subchannel_weak_ref((p), __FILE__, __LINE__, (r))
+#define GRPC_SUBCHANNEL_WEAK_UNREF(cl, p, r) \
+  grpc_subchannel_weak_unref((cl), (p), __FILE__, __LINE__, (r))
+#define GRPC_CONNECTED_SUBCHANNEL_REF(p, r) \
+  grpc_connected_subchannel_ref((p), __FILE__, __LINE__, (r))
+#define GRPC_CONNECTED_SUBCHANNEL_UNREF(cl, p, r) \
+  grpc_connected_subchannel_unref((cl), (p), __FILE__, __LINE__, (r))
 #define GRPC_SUBCHANNEL_CALL_REF(p, r) \
   grpc_subchannel_call_ref((p), __FILE__, __LINE__, (r))
 #define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \
@@ -58,6 +67,12 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
 #else
 #define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p))
 #define GRPC_SUBCHANNEL_UNREF(cl, p, r) grpc_subchannel_unref((cl), (p))
+#define GRPC_SUBCHANNEL_WEAK_REF(p, r) grpc_subchannel_weak_ref((p))
+#define GRPC_SUBCHANNEL_WEAK_UNREF(cl, p, r) \
+  grpc_subchannel_weak_unref((cl), (p))
+#define GRPC_CONNECTED_SUBCHANNEL_REF(p, r) grpc_connected_subchannel_ref((p))
+#define GRPC_CONNECTED_SUBCHANNEL_UNREF(cl, p, r) \
+  grpc_connected_subchannel_unref((cl), (p))
 #define GRPC_SUBCHANNEL_CALL_REF(p, r) grpc_subchannel_call_ref((p))
 #define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \
   grpc_subchannel_call_unref((cl), (p))
@@ -69,33 +84,31 @@ void grpc_subchannel_ref(grpc_subchannel *channel
 void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
                            grpc_subchannel *channel
                                GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_subchannel_weak_ref(grpc_subchannel *channel
+                                  GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
+                                grpc_subchannel *channel
+                                    GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_connected_subchannel_ref(grpc_connected_subchannel *channel
+                                       GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_connected_subchannel_unref(grpc_exec_ctx *exec_ctx,
+                                     grpc_connected_subchannel *channel
+                                         GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 void grpc_subchannel_call_ref(grpc_subchannel_call *call
                                   GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
                                 grpc_subchannel_call *call
                                     GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 
-/** construct a subchannel call (possibly asynchronously).
- *
- * If the returned status is 1, the call will return immediately and \a target
- * will point to a connected \a subchannel_call instance. Note that \a notify
- * will \em not be invoked in this case.
- * Otherwise, if the returned status is 0, the subchannel call will be created
- * asynchronously, invoking the \a notify callback upon completion. */
-int grpc_subchannel_create_call(grpc_exec_ctx *exec_ctx,
-                                grpc_subchannel *subchannel,
-                                grpc_pollset *pollset, gpr_atm *target,
-                                grpc_closure *notify);
-
-/** cancel \a call in the waiting state. */
-void grpc_subchannel_cancel_create_call(grpc_exec_ctx *exec_ctx,
-                                        grpc_subchannel *subchannel,
-                                        gpr_atm *target);
+/** construct a subchannel call */
+grpc_subchannel_call *grpc_connected_subchannel_create_call(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *connected_subchannel,
+    grpc_pollset *pollset);
 
 /** process a transport level op */
-void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
-                                          grpc_subchannel *subchannel,
-                                          grpc_transport_op *op);
+void grpc_connected_subchannel_process_transport_op(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *subchannel,
+    grpc_transport_op *op);
 
 /** poll the current connectivity state of a channel */
 grpc_connectivity_state grpc_subchannel_check_connectivity(
@@ -103,26 +116,22 @@ grpc_connectivity_state grpc_subchannel_check_connectivity(
 
 /** call notify when the connectivity state of a channel changes from *state.
     Updates *state with the new state of the channel */
-void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
-                                            grpc_subchannel *channel,
-                                            grpc_connectivity_state *state,
-                                            grpc_closure *notify);
-
-/** Remove \a subscribed_notify from the list of closures to be called on a
- * state change if present, returning 1. Otherwise, nothing is done and return
- * 0. */
-int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
-                                             grpc_subchannel *channel,
-                                             grpc_closure *subscribed_notify);
-
-/** express interest in \a channel's activities through \a pollset. */
-void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx,
-                                          grpc_subchannel *channel,
-                                          grpc_pollset *pollset);
-/** stop following \a channel's activity through \a pollset. */
-void grpc_subchannel_del_interested_party(grpc_exec_ctx *exec_ctx,
-                                          grpc_subchannel *channel,
-                                          grpc_pollset *pollset);
+void grpc_subchannel_notify_on_state_change(
+    grpc_exec_ctx *exec_ctx, grpc_subchannel *channel,
+    grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
+    grpc_closure *notify);
+void grpc_connected_subchannel_notify_on_state_change(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *channel,
+    grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
+    grpc_closure *notify);
+void grpc_connected_subchannel_ping(grpc_exec_ctx *exec_ctx,
+                                    grpc_connected_subchannel *channel,
+                                    grpc_closure *notify);
+
+/** retrieve the grpc_connected_subchannel - or NULL if called before
+    the subchannel becomes connected */
+grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
+    grpc_subchannel *subchannel);
 
 /** continue processing a transport op */
 void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx,
@@ -147,15 +156,10 @@ struct grpc_subchannel_args {
   /** Address to connect to */
   struct sockaddr *addr;
   size_t addr_len;
-  /** master channel */
-  grpc_channel *master;
 };
 
 /** create a subchannel given a connector */
 grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
                                         grpc_subchannel_args *args);
 
-/** Return the master channel associated with the subchannel */
-grpc_channel *grpc_subchannel_get_master(grpc_subchannel *subchannel);
-
 #endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_H */
diff --git a/src/core/client_config/subchannel_factory_decorators/add_channel_arg.c b/src/core/client_config/subchannel_factory_decorators/add_channel_arg.c
deleted file mode 100644
index 585e465fa4..0000000000
--- a/src/core/client_config/subchannel_factory_decorators/add_channel_arg.c
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- *     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- *     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "src/core/client_config/subchannel_factory_decorators/add_channel_arg.h"
-#include "src/core/client_config/subchannel_factory_decorators/merge_channel_args.h"
-
-grpc_subchannel_factory *grpc_subchannel_factory_add_channel_arg(
-    grpc_subchannel_factory *input, const grpc_arg *arg) {
-  grpc_channel_args args;
-  args.num_args = 1;
-  args.args = (grpc_arg *)arg;
-  return grpc_subchannel_factory_merge_channel_args(input, &args);
-}
diff --git a/src/core/client_config/subchannel_factory_decorators/add_channel_arg.h b/src/core/client_config/subchannel_factory_decorators/add_channel_arg.h
deleted file mode 100644
index 76a535ebed..0000000000
--- a/src/core/client_config/subchannel_factory_decorators/add_channel_arg.h
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- *     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- *     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_ADD_CHANNEL_ARG_H
-#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_ADD_CHANNEL_ARG_H
-
-#include "src/core/client_config/subchannel_factory.h"
-
-/** Takes a subchannel factory, returns a new one that mutates incoming
-    channel_args by adding a new argument; ownership of input, arg is retained
-    by the caller. */
-grpc_subchannel_factory *grpc_subchannel_factory_add_channel_arg(
-    grpc_subchannel_factory *input, const grpc_arg *arg);
-
-#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_ADD_CHANNEL_ARG_H \
-        */
diff --git a/src/core/client_config/subchannel_factory_decorators/merge_channel_args.c b/src/core/client_config/subchannel_factory_decorators/merge_channel_args.c
deleted file mode 100644
index cd25fdcf0f..0000000000
--- a/src/core/client_config/subchannel_factory_decorators/merge_channel_args.c
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- *     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- *     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "src/core/client_config/subchannel_factory_decorators/merge_channel_args.h"
-#include <grpc/support/alloc.h>
-#include "src/core/channel/channel_args.h"
-
-typedef struct {
-  grpc_subchannel_factory base;
-  gpr_refcount refs;
-  grpc_subchannel_factory *wrapped;
-  grpc_channel_args *merge_args;
-} merge_args_factory;
-
-static void merge_args_factory_ref(grpc_subchannel_factory *scf) {
-  merge_args_factory *f = (merge_args_factory *)scf;
-  gpr_ref(&f->refs);
-}
-
-static void merge_args_factory_unref(grpc_exec_ctx *exec_ctx,
-                                     grpc_subchannel_factory *scf) {
-  merge_args_factory *f = (merge_args_factory *)scf;
-  if (gpr_unref(&f->refs)) {
-    grpc_subchannel_factory_unref(exec_ctx, f->wrapped);
-    grpc_channel_args_destroy(f->merge_args);
-    gpr_free(f);
-  }
-}
-
-static grpc_subchannel *merge_args_factory_create_subchannel(
-    grpc_exec_ctx *exec_ctx, grpc_subchannel_factory *scf,
-    grpc_subchannel_args *args) {
-  merge_args_factory *f = (merge_args_factory *)scf;
-  grpc_channel_args *final_args =
-      grpc_channel_args_merge(args->args, f->merge_args);
-  grpc_subchannel *s;
-  args->args = final_args;
-  s = grpc_subchannel_factory_create_subchannel(exec_ctx, f->wrapped, args);
-  grpc_channel_args_destroy(final_args);
-  return s;
-}
-
-static const grpc_subchannel_factory_vtable merge_args_factory_vtable = {
-    merge_args_factory_ref, merge_args_factory_unref,
-    merge_args_factory_create_subchannel};
-
-grpc_subchannel_factory *grpc_subchannel_factory_merge_channel_args(
-    grpc_subchannel_factory *input, const grpc_channel_args *args) {
-  merge_args_factory *f = gpr_malloc(sizeof(*f));
-  f->base.vtable = &merge_args_factory_vtable;
-  gpr_ref_init(&f->refs, 1);
-  grpc_subchannel_factory_ref(input);
-  f->wrapped = input;
-  f->merge_args = grpc_channel_args_copy(args);
-  return &f->base;
-}
diff --git a/src/core/client_config/subchannel_factory_decorators/merge_channel_args.h b/src/core/client_config/subchannel_factory_decorators/merge_channel_args.h
deleted file mode 100644
index a9e1691871..0000000000
--- a/src/core/client_config/subchannel_factory_decorators/merge_channel_args.h
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- *     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- *     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_MERGE_CHANNEL_ARGS_H
-#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_MERGE_CHANNEL_ARGS_H
-
-#include "src/core/client_config/subchannel_factory.h"
-
-/** Takes a subchannel factory, returns a new one that mutates incoming
-    channel_args by adding a new argument; ownership of input, args is retained
-    by the caller. */
-grpc_subchannel_factory *grpc_subchannel_factory_merge_channel_args(
-    grpc_subchannel_factory *input, const grpc_channel_args *args);
-
-#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_MERGE_CHANNEL_ARGS_H \
-        */