diff options
author | 2015-07-16 16:19:14 -0700 | |
---|---|---|
committer | 2015-07-16 16:19:14 -0700 | |
commit | 1ada6ad8e57e9033f9cc9a6b0082e27a413e9584 (patch) | |
tree | 282cbc4f22d5e25edf155654b9734501845f1ebf /src/core/client_config/lb_policies/pick_first.c | |
parent | 48cb07c9096250374d8d9288452195cd9b90ce16 (diff) |
Added connectivity tests, fixed bugs
Diffstat (limited to 'src/core/client_config/lb_policies/pick_first.c')
-rw-r--r-- | src/core/client_config/lb_policies/pick_first.c | 135 |
1 files changed, 80 insertions, 55 deletions
diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c index 4f4c7eb64c..8409aab14e 100644 --- a/src/core/client_config/lb_policies/pick_first.c +++ b/src/core/client_config/lb_policies/pick_first.c @@ -73,12 +73,30 @@ typedef struct { grpc_connectivity_state_tracker state_tracker; } pick_first_lb_policy; +static void del_interested_parties_locked(pick_first_lb_policy *p) { + pending_pick *pp; + for (pp = p->pending_picks; pp; pp = pp->next) { + grpc_subchannel_del_interested_party(p->subchannels[p->checking_subchannel], + pp->pollset); + } +} + +static void add_interested_parties_locked(pick_first_lb_policy *p) { + pending_pick *pp; + for (pp = p->pending_picks; pp; pp = pp->next) { + grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel], + pp->pollset); + } +} + void pf_destroy(grpc_lb_policy *pol) { pick_first_lb_policy *p = (pick_first_lb_policy *)pol; size_t i; + del_interested_parties_locked(p); for (i = 0; i < p->num_subchannels; i++) { GRPC_SUBCHANNEL_UNREF(p->subchannels[i], "pick_first"); } + grpc_connectivity_state_destroy(&p->state_tracker); gpr_free(p->subchannels); gpr_mu_destroy(&p->mu); gpr_free(p); @@ -88,6 +106,7 @@ void pf_shutdown(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(p); while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = NULL; @@ -142,77 +161,82 @@ void pf_pick(grpc_lb_policy *pol, grpc_pollset *pollset, } } -static void del_interested_parties_locked(pick_first_lb_policy *p) { - pending_pick *pp; - for (pp = p->pending_picks; pp; pp = pp->next) { - grpc_subchannel_del_interested_party(p->subchannels[p->checking_subchannel], - pp->pollset); - } -} - -static void add_interested_parties_locked(pick_first_lb_policy *p) { - pending_pick *pp; - for (pp = p->pending_picks; pp; pp = pp->next) { - grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel], - pp->pollset); - } -} - static void pf_connectivity_changed(void *arg, int iomgr_success) { pick_first_lb_policy *p = arg; pending_pick *pp; int unref = 0; gpr_mu_lock(&p->mu); -loop: - switch (p->checking_connectivity) { - case GRPC_CHANNEL_READY: - p->selected = p->subchannels[p->checking_subchannel]; - while ((pp = p->pending_picks)) { - p->pending_picks = pp->next; - *pp->target = p->selected; - grpc_subchannel_del_interested_party(p->selected, pp->pollset); - grpc_iomgr_add_delayed_callback(pp->on_complete, 1); - gpr_free(pp); - } + + if (p->selected != NULL) { + grpc_connectivity_state_set(&p->state_tracker, p->checking_connectivity); + if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) { + grpc_subchannel_notify_on_state_change(p->selected, &p->checking_connectivity, &p->connectivity_changed); + } else { unref = 1; - break; - case GRPC_CHANNEL_TRANSIENT_FAILURE: - del_interested_parties_locked(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(p); - goto loop; - case GRPC_CHANNEL_CONNECTING: - case GRPC_CHANNEL_IDLE: - grpc_subchannel_notify_on_state_change( - p->subchannels[p->checking_subchannel], &p->checking_connectivity, - &p->connectivity_changed); - break; - case GRPC_CHANNEL_FATAL_FAILURE: - del_interested_parties_locked(p); - GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel], - p->subchannels[p->num_subchannels - 1]); - p->num_subchannels--; - GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first"); - if (p->num_subchannels == 0) { + } + } else { +loop: + switch (p->checking_connectivity) { + case GRPC_CHANNEL_READY: + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_READY); + p->selected = p->subchannels[p->checking_subchannel]; while ((pp = p->pending_picks)) { p->pending_picks = pp->next; - *pp->target = NULL; + *pp->target = p->selected; + grpc_subchannel_del_interested_party(p->selected, pp->pollset); grpc_iomgr_add_delayed_callback(pp->on_complete, 1); gpr_free(pp); } - unref = 1; - } else { - p->checking_subchannel %= p->num_subchannels; + grpc_subchannel_notify_on_state_change(p->selected, &p->checking_connectivity, &p->connectivity_changed); + break; + case GRPC_CHANNEL_TRANSIENT_FAILURE: + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE); + del_interested_parties_locked(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(p); - goto loop; - } + if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) { + grpc_subchannel_notify_on_state_change(p->subchannels[p->checking_subchannel], &p->checking_connectivity, &p->connectivity_changed); + } else { + goto loop; + } + break; + case GRPC_CHANNEL_CONNECTING: + case GRPC_CHANNEL_IDLE: + grpc_connectivity_state_set(&p->state_tracker, p->checking_connectivity); + grpc_subchannel_notify_on_state_change( + p->subchannels[p->checking_subchannel], &p->checking_connectivity, + &p->connectivity_changed); + break; + case GRPC_CHANNEL_FATAL_FAILURE: + del_interested_parties_locked(p); + GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel], + p->subchannels[p->num_subchannels - 1]); + p->num_subchannels--; + GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first"); + if (p->num_subchannels == 0) { + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_FATAL_FAILURE); + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = NULL; + grpc_iomgr_add_delayed_callback(pp->on_complete, 1); + gpr_free(pp); + } + unref = 1; + } else { + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE); + p->checking_subchannel %= p->num_subchannels; + p->checking_connectivity = grpc_subchannel_check_connectivity( + p->subchannels[p->checking_subchannel]); + add_interested_parties_locked(p); + goto loop; + } + } } + gpr_mu_unlock(&p->mu); if (unref) { @@ -278,6 +302,7 @@ grpc_lb_policy *grpc_create_pick_first_lb_policy(grpc_subchannel **subchannels, grpc_lb_policy_init(&p->base, &pick_first_lb_policy_vtable); p->subchannels = gpr_malloc(sizeof(grpc_subchannel *) * num_subchannels); p->num_subchannels = num_subchannels; + grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE, "pick_first"); memcpy(p->subchannels, subchannels, sizeof(grpc_subchannel *) * num_subchannels); grpc_iomgr_closure_init(&p->connectivity_changed, pf_connectivity_changed, p); |