|
@@ -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,
|
|
static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
|
|
int iomgr_success) {
|
|
int iomgr_success) {
|
|
pick_first_lb_policy *p = arg;
|
|
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_connectivity_state_set(exec_ctx, &p->state_tracker,
|
|
GRPC_CHANNEL_READY, "connecting_ready");
|
|
GRPC_CHANNEL_READY, "connecting_ready");
|
|
p->selected = p->subchannels[p->checking_subchannel];
|
|
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)) {
|
|
while ((pp = p->pending_picks)) {
|
|
p->pending_picks = pp->next;
|
|
p->pending_picks = pp->next;
|
|
*pp->target = p->selected;
|
|
*pp->target = p->selected;
|