Discard unnecessary subchannels when they're no longer required

1 files changed, 35 insertions, 0 deletions

diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c
index 28155d0fbc..18c8c0412f 100644
--- a/src/core/client_config/lb_policies/pick_first.c
+++ b/src/core/client_config/lb_policies/pick_first.c
@@ -172,6 +172,37 @@ void pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
   }
 }
 
+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;
+  exclude_subchannel = p->selected;
+  p->num_subchannels = 0;
+  p->subchannels = NULL;
+  gpr_mu_unlock(&p->mu);
+  GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "destroy_subchannels");
+
+  for (i = 0; i < num_subchannels; i++) {
+    if (subchannels[i] == exclude_subchannel) {
+      exclude_subchannel = NULL;
+      continue;
+    }
+    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");
+  }
+
+  gpr_free(subchannels);
+}
+
 static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
                                     int iomgr_success) {
   pick_first_lb_policy *p = arg;
@@ -200,6 +231,10 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
         grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
                                     GRPC_CHANNEL_READY, "connecting_ready");
         p->selected = p->subchannels[p->checking_subchannel];
+        /* drop the pick list: we are connected now */
+        GRPC_LB_POLICY_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;