Memory fixes

src/core/channel/client_channel.c

@@ -244,7 +244,6 @@ static void cc_on_config_changed(grpc_exec_ctx *exec_ctx, void *arg,
 static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
                                   grpc_channel_element *elem,
                                   grpc_transport_op *op) {
-  grpc_lb_policy *lb_policy = NULL;
   channel_data *chand = elem->channel_data;
   grpc_resolver *destroy_resolver = NULL;
 
@@ -262,7 +261,6 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
     op->connectivity_state = NULL;
   }
 
-  lb_policy = chand->lb_policy;
   if (op->disconnect && chand->resolver != NULL) {
     grpc_connectivity_state_set(exec_ctx, &chand->state_tracker,
                                 GRPC_CHANNEL_FATAL_FAILURE, "disconnect");

src/core/channel/client_uchannel.c

@@ -85,7 +85,7 @@ static void monitor_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
                               chand->subchannel_connectivity,
                               "uchannel_monitor_subchannel");
   grpc_connected_subchannel_notify_on_state_change(
-      exec_ctx, chand->connected_subchannel, &chand->subchannel_connectivity,
+      exec_ctx, chand->connected_subchannel, NULL, &chand->subchannel_connectivity,
       &chand->connectivity_cb);
 }
 
@@ -168,9 +168,10 @@ static void cuc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
   channel_data *chand = elem->channel_data;
   /* cancel subscription */
   grpc_connected_subchannel_notify_on_state_change(
-      exec_ctx, chand->connected_subchannel, NULL, &chand->connectivity_cb);
+      exec_ctx, chand->connected_subchannel, NULL, NULL, &chand->connectivity_cb);
   grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
   gpr_mu_destroy(&chand->mu_state);
+  GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, chand->connected_subchannel, "uchannel");
 }
 
 static void cuc_set_pollset(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
@@ -190,17 +191,8 @@ grpc_connectivity_state grpc_client_uchannel_check_connectivity_state(
     grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, int try_to_connect) {
   channel_data *chand = elem->channel_data;
   grpc_connectivity_state out;
-  out = grpc_connectivity_state_check(&chand->state_tracker);
   gpr_mu_lock(&chand->mu_state);
-  if (out == GRPC_CHANNEL_IDLE && try_to_connect) {
-    grpc_connectivity_state_set(exec_ctx, &chand->state_tracker,
-                                GRPC_CHANNEL_CONNECTING,
-                                "uchannel_connecting_changed");
-    chand->subchannel_connectivity = out;
-    grpc_connected_subchannel_notify_on_state_change(
-        exec_ctx, chand->connected_subchannel, &chand->subchannel_connectivity,
-        &chand->connectivity_cb);
-  }
+  out = grpc_connectivity_state_check(&chand->state_tracker);
   gpr_mu_unlock(&chand->mu_state);
   return out;
 }
@@ -244,5 +236,6 @@ void grpc_client_uchannel_set_connected_subchannel(
   GPR_ASSERT(elem->filter == &grpc_client_uchannel_filter);
   gpr_mu_lock(&chand->mu_state);
   chand->connected_subchannel = connected_subchannel;
+  GRPC_CONNECTED_SUBCHANNEL_REF(connected_subchannel, "uchannel");
   gpr_mu_unlock(&chand->mu_state);
 }

src/core/client_config/lb_policies/pick_first.c

@@ -101,9 +101,10 @@ void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
   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, &p->connectivity_changed);
+        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,
@@ -198,13 +199,11 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
   size_t i;
   size_t num_subchannels = p->num_subchannels;
   grpc_subchannel **subchannels;
-  grpc_connected_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_WEAK_UNREF(exec_ctx, &p->base, "destroy_subchannels");
 
@@ -236,7 +235,7 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
                                 p->checking_connectivity, "selected_changed");
     if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) {
       grpc_connected_subchannel_notify_on_state_change(
-          exec_ctx, p->selected, &p->checking_connectivity,
+          exec_ctx, p->selected, &p->base.interested_parties, &p->checking_connectivity,
           &p->connectivity_changed);
     } else {
       GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
@@ -266,7 +265,7 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
           gpr_free(pp);
         }
         grpc_connected_subchannel_notify_on_state_change(
-            exec_ctx, p->selected, &p->checking_connectivity,
+            exec_ctx, p->selected, &p->base.interested_parties, &p->checking_connectivity,
             &p->connectivity_changed);
         break;
       case GRPC_CHANNEL_TRANSIENT_FAILURE:

src/core/client_config/subchannel.c

@@ -266,7 +266,7 @@ void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
                                     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 == 0) {
+  if (old_refs == 1) {
     grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(subchannel_destroy, c),
                           1);
   }
@@ -426,7 +426,7 @@ static void subchannel_on_child_state_changed(grpc_exec_ctx *exec_ctx, void *p,
                                 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),
+          exec_ctx, GET_CONNECTED_SUBCHANNEL(c, no_barrier), NULL,
           &sw->connectivity_state, &sw->closure);
       GRPC_SUBCHANNEL_WEAK_REF(c, "state_watcher");
       sw = NULL;
@@ -440,6 +440,7 @@ static void subchannel_on_child_state_changed(grpc_exec_ctx *exec_ctx, void *p,
 
 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;
@@ -447,14 +448,15 @@ static void connected_subchannel_state_op(grpc_exec_ctx *exec_ctx,
   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);
 }
 
 void grpc_connected_subchannel_notify_on_state_change(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
-    grpc_connectivity_state *state, grpc_closure *closure) {
-  connected_subchannel_state_op(exec_ctx, con, state, closure);
+    grpc_pollset_set *interested_parties, grpc_connectivity_state *state, grpc_closure *closure) {
+  connected_subchannel_state_op(exec_ctx, con, interested_parties, state, closure);
 }
 
 static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
@@ -512,7 +514,7 @@ static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
   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, &sw_subchannel->connectivity_state,
+      exec_ctx, con, &c->pollset_set, &sw_subchannel->connectivity_state,
       &sw_subchannel->closure);
 
   /* signal completion */

src/core/client_config/subchannel.h

@@ -122,7 +122,7 @@ void grpc_subchannel_notify_on_state_change(
     grpc_closure *notify);
 void grpc_connected_subchannel_notify_on_state_change(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *channel,
-    grpc_connectivity_state *state, grpc_closure *notify);
+    grpc_pollset_set *interested_parties, grpc_connectivity_state *state, grpc_closure *notify);
 
 /** retrieve the grpc_connected_subchannel - or NULL if called before
     the subchannel becomes connected */

test/core/end2end/fixtures/h2_uchannel.c

@@ -162,6 +162,9 @@ static grpc_subchannel *subchannel_factory_create_subchannel(
   s = grpc_subchannel_create(&c->base, args);
   grpc_connector_unref(exec_ctx, &c->base);
   grpc_channel_args_destroy(final_args);
+  if (*f->sniffed_subchannel) {
+    GRPC_SUBCHANNEL_UNREF(exec_ctx, *f->sniffed_subchannel, "sniffed");
+  }
   *f->sniffed_subchannel = s;
   GRPC_SUBCHANNEL_REF(s, "sniffed");
   return s;
@@ -224,6 +227,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_micro_fullstack(
   micro_fullstack_fixture_data *ffd =
       gpr_malloc(sizeof(micro_fullstack_fixture_data));
   memset(&f, 0, sizeof(f));
+  memset(ffd, 0, sizeof(*ffd));
 
   gpr_join_host_port(&ffd->localaddr, "127.0.0.1", port);
 

test/core/end2end/tests/channel_connectivity.c

@@ -153,7 +153,8 @@ static void test_connectivity(grpc_end2end_test_config config) {
   cq_verify(cqv);
   state = grpc_channel_check_connectivity_state(f.client, 0);
   GPR_ASSERT(state == GRPC_CHANNEL_TRANSIENT_FAILURE ||
-             state == GRPC_CHANNEL_CONNECTING);
+             state == GRPC_CHANNEL_CONNECTING ||
+             state == GRPC_CHANNEL_IDLE);
 
   /* cleanup server */
   grpc_server_destroy(f.server);