diff options
Diffstat (limited to 'src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc')
| -rw-r--r-- | src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc | 194 |
1 files changed, 92 insertions, 102 deletions
diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc index 228a77d9db..0861261359 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc @@ -57,12 +57,12 @@ typedef struct { grpc_connectivity_state_tracker state_tracker; } pick_first_lb_policy; -static void pf_destroy(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol) { +static void pf_destroy(grpc_lb_policy* pol) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; GPR_ASSERT(p->subchannel_list == nullptr); GPR_ASSERT(p->latest_pending_subchannel_list == nullptr); GPR_ASSERT(p->pending_picks == nullptr); - grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker); + grpc_connectivity_state_destroy(&p->state_tracker); gpr_free(p); grpc_subchannel_index_unref(); if (grpc_lb_pick_first_trace.enabled()) { @@ -70,7 +70,7 @@ static void pf_destroy(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol) { } } -static void pf_shutdown_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol) { +static void pf_shutdown_locked(grpc_lb_policy* pol) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown"); if (grpc_lb_pick_first_trace.enabled()) { @@ -81,28 +81,27 @@ static void pf_shutdown_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol) { while ((pp = p->pending_picks) != nullptr) { p->pending_picks = pp->next; *pp->target = nullptr; - GRPC_CLOSURE_SCHED(exec_ctx, pp->on_complete, GRPC_ERROR_REF(error)); + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_REF(error)); gpr_free(pp); } - grpc_connectivity_state_set(exec_ctx, &p->state_tracker, - GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), - "shutdown"); + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_SHUTDOWN, + GRPC_ERROR_REF(error), "shutdown"); if (p->subchannel_list != nullptr) { - grpc_lb_subchannel_list_shutdown_and_unref(exec_ctx, p->subchannel_list, + grpc_lb_subchannel_list_shutdown_and_unref(p->subchannel_list, "pf_shutdown"); p->subchannel_list = nullptr; } if (p->latest_pending_subchannel_list != nullptr) { grpc_lb_subchannel_list_shutdown_and_unref( - exec_ctx, p->latest_pending_subchannel_list, "pf_shutdown"); + p->latest_pending_subchannel_list, "pf_shutdown"); p->latest_pending_subchannel_list = nullptr; } - grpc_lb_policy_try_reresolve(exec_ctx, &p->base, &grpc_lb_pick_first_trace, + grpc_lb_policy_try_reresolve(&p->base, &grpc_lb_pick_first_trace, GRPC_ERROR_CANCELLED); GRPC_ERROR_UNREF(error); } -static void pf_cancel_pick_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, +static void pf_cancel_pick_locked(grpc_lb_policy* pol, grpc_connected_subchannel** target, grpc_error* error) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; @@ -112,7 +111,7 @@ static void pf_cancel_pick_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, pending_pick* next = pp->next; if (pp->target == target) { *target = nullptr; - GRPC_CLOSURE_SCHED(exec_ctx, pp->on_complete, + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick Cancelled", &error, 1)); gpr_free(pp); @@ -125,7 +124,7 @@ static void pf_cancel_pick_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, GRPC_ERROR_UNREF(error); } -static void pf_cancel_picks_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, +static void pf_cancel_picks_locked(grpc_lb_policy* pol, uint32_t initial_metadata_flags_mask, uint32_t initial_metadata_flags_eq, grpc_error* error) { @@ -136,7 +135,7 @@ static void pf_cancel_picks_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, pending_pick* next = pp->next; if ((pp->initial_metadata_flags & initial_metadata_flags_mask) == initial_metadata_flags_eq) { - GRPC_CLOSURE_SCHED(exec_ctx, pp->on_complete, + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick Cancelled", &error, 1)); gpr_free(pp); @@ -149,8 +148,7 @@ static void pf_cancel_picks_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, GRPC_ERROR_UNREF(error); } -static void start_picking_locked(grpc_exec_ctx* exec_ctx, - pick_first_lb_policy* p) { +static void start_picking_locked(pick_first_lb_policy* p) { p->started_picking = true; if (p->subchannel_list != nullptr && p->subchannel_list->num_subchannels > 0) { @@ -160,21 +158,21 @@ static void start_picking_locked(grpc_exec_ctx* exec_ctx, grpc_lb_subchannel_list_ref_for_connectivity_watch( p->subchannel_list, "connectivity_watch+start_picking"); grpc_lb_subchannel_data_start_connectivity_watch( - exec_ctx, &p->subchannel_list->subchannels[i]); + &p->subchannel_list->subchannels[i]); break; } } } } -static void pf_exit_idle_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol) { +static void pf_exit_idle_locked(grpc_lb_policy* pol) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; if (!p->started_picking) { - start_picking_locked(exec_ctx, p); + start_picking_locked(p); } } -static int pf_pick_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, +static int pf_pick_locked(grpc_lb_policy* pol, const grpc_lb_policy_pick_args* pick_args, grpc_connected_subchannel** target, grpc_call_context_element* context, void** user_data, @@ -188,7 +186,7 @@ static int pf_pick_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, } // No subchannel selected yet, so handle asynchronously. if (!p->started_picking) { - start_picking_locked(exec_ctx, p); + start_picking_locked(p); } pending_pick* pp = (pending_pick*)gpr_malloc(sizeof(*pp)); pp->next = p->pending_picks; @@ -199,48 +197,47 @@ static int pf_pick_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, return 0; } -static void destroy_unselected_subchannels_locked(grpc_exec_ctx* exec_ctx, - pick_first_lb_policy* p) { +static void destroy_unselected_subchannels_locked(pick_first_lb_policy* p) { for (size_t i = 0; i < p->subchannel_list->num_subchannels; ++i) { grpc_lb_subchannel_data* sd = &p->subchannel_list->subchannels[i]; if (p->selected != sd) { - grpc_lb_subchannel_data_unref_subchannel(exec_ctx, sd, + grpc_lb_subchannel_data_unref_subchannel(sd, "selected_different_subchannel"); } } } static grpc_connectivity_state pf_check_connectivity_locked( - grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, grpc_error** error) { + grpc_lb_policy* pol, grpc_error** error) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; return grpc_connectivity_state_get(&p->state_tracker, error); } -static void pf_notify_on_state_change_locked(grpc_exec_ctx* exec_ctx, - grpc_lb_policy* pol, +static void pf_notify_on_state_change_locked(grpc_lb_policy* pol, grpc_connectivity_state* current, grpc_closure* notify) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; - grpc_connectivity_state_notify_on_state_change(exec_ctx, &p->state_tracker, - current, notify); + grpc_connectivity_state_notify_on_state_change(&p->state_tracker, current, + notify); } -static void pf_ping_one_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* pol, - grpc_closure* closure) { +static void pf_ping_one_locked(grpc_lb_policy* pol, grpc_closure* on_initiate, + grpc_closure* on_ack) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; if (p->selected) { - grpc_connected_subchannel_ping(exec_ctx, p->selected->connected_subchannel, - closure); + grpc_connected_subchannel_ping(p->selected->connected_subchannel, + on_initiate, on_ack); } else { - GRPC_CLOSURE_SCHED(exec_ctx, closure, + GRPC_CLOSURE_SCHED(on_initiate, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Not connected")); + GRPC_CLOSURE_SCHED(on_ack, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Not connected")); } } -static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error); +static void pf_connectivity_changed_locked(void* arg, grpc_error* error); -static void pf_update_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, +static void pf_update_locked(grpc_lb_policy* policy, const grpc_lb_policy_args* args) { pick_first_lb_policy* p = (pick_first_lb_policy*)policy; const grpc_arg* arg = @@ -249,7 +246,7 @@ static void pf_update_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, if (p->subchannel_list == nullptr) { // If we don't have a current subchannel list, go into TRANSIENT FAILURE. grpc_connectivity_state_set( - exec_ctx, &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, + &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Missing update in args"), "pf_update_missing"); } else { @@ -268,17 +265,17 @@ static void pf_update_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, (void*)p, (unsigned long)addresses->num_addresses); } grpc_lb_subchannel_list* subchannel_list = grpc_lb_subchannel_list_create( - exec_ctx, &p->base, &grpc_lb_pick_first_trace, addresses, args, + &p->base, &grpc_lb_pick_first_trace, addresses, args, pf_connectivity_changed_locked); if (subchannel_list->num_subchannels == 0) { // Empty update or no valid subchannels. Unsubscribe from all current // subchannels and put the channel in TRANSIENT_FAILURE. grpc_connectivity_state_set( - exec_ctx, &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, + &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Empty update"), "pf_update_empty"); if (p->subchannel_list != nullptr) { - grpc_lb_subchannel_list_shutdown_and_unref(exec_ctx, p->subchannel_list, + grpc_lb_subchannel_list_shutdown_and_unref(p->subchannel_list, "sl_shutdown_empty_update"); } p->subchannel_list = subchannel_list; // Empty list. @@ -289,7 +286,7 @@ static void pf_update_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, // We don't yet have a selected subchannel, so replace the current // subchannel list immediately. if (p->subchannel_list != nullptr) { - grpc_lb_subchannel_list_shutdown_and_unref(exec_ctx, p->subchannel_list, + grpc_lb_subchannel_list_shutdown_and_unref(p->subchannel_list, "pf_update_before_selected"); } p->subchannel_list = subchannel_list; @@ -314,19 +311,19 @@ static void pf_update_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, p->selected = sd; if (p->subchannel_list != nullptr) { grpc_lb_subchannel_list_shutdown_and_unref( - exec_ctx, p->subchannel_list, "pf_update_includes_selected"); + p->subchannel_list, "pf_update_includes_selected"); } p->subchannel_list = subchannel_list; - destroy_unselected_subchannels_locked(exec_ctx, p); + destroy_unselected_subchannels_locked(p); grpc_lb_subchannel_list_ref_for_connectivity_watch( subchannel_list, "connectivity_watch+replace_selected"); - grpc_lb_subchannel_data_start_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_start_connectivity_watch(sd); // If there was a previously pending update (which may or may // not have contained the currently selected subchannel), drop // it, so that it doesn't override what we've done here. if (p->latest_pending_subchannel_list != nullptr) { grpc_lb_subchannel_list_shutdown_and_unref( - exec_ctx, p->latest_pending_subchannel_list, + p->latest_pending_subchannel_list, "pf_update_includes_selected+outdated"); p->latest_pending_subchannel_list = nullptr; } @@ -346,8 +343,7 @@ static void pf_update_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, (void*)subchannel_list); } grpc_lb_subchannel_list_shutdown_and_unref( - exec_ctx, p->latest_pending_subchannel_list, - "sl_outdated_dont_smash"); + p->latest_pending_subchannel_list, "sl_outdated_dont_smash"); } p->latest_pending_subchannel_list = subchannel_list; } @@ -357,12 +353,11 @@ static void pf_update_locked(grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, grpc_lb_subchannel_list_ref_for_connectivity_watch( subchannel_list, "connectivity_watch+update"); grpc_lb_subchannel_data_start_connectivity_watch( - exec_ctx, &subchannel_list->subchannels[0]); + &subchannel_list->subchannels[0]); } } -static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void pf_connectivity_changed_locked(void* arg, grpc_error* error) { grpc_lb_subchannel_data* sd = (grpc_lb_subchannel_data*)arg; pick_first_lb_policy* p = (pick_first_lb_policy*)sd->subchannel_list->policy; if (grpc_lb_pick_first_trace.enabled()) { @@ -380,18 +375,18 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, } // If the policy is shutting down, unref and return. if (p->shutdown) { - grpc_lb_subchannel_data_stop_connectivity_watch(exec_ctx, sd); - grpc_lb_subchannel_data_unref_subchannel(exec_ctx, sd, "pf_shutdown"); - grpc_lb_subchannel_list_unref_for_connectivity_watch( - exec_ctx, sd->subchannel_list, "pf_shutdown"); + grpc_lb_subchannel_data_stop_connectivity_watch(sd); + grpc_lb_subchannel_data_unref_subchannel(sd, "pf_shutdown"); + grpc_lb_subchannel_list_unref_for_connectivity_watch(sd->subchannel_list, + "pf_shutdown"); return; } // If the subchannel list is shutting down, stop watching. if (sd->subchannel_list->shutting_down || error == GRPC_ERROR_CANCELLED) { - grpc_lb_subchannel_data_stop_connectivity_watch(exec_ctx, sd); - grpc_lb_subchannel_data_unref_subchannel(exec_ctx, sd, "pf_sl_shutdown"); - grpc_lb_subchannel_list_unref_for_connectivity_watch( - exec_ctx, sd->subchannel_list, "pf_sl_shutdown"); + grpc_lb_subchannel_data_stop_connectivity_watch(sd); + grpc_lb_subchannel_data_unref_subchannel(sd, "pf_sl_shutdown"); + grpc_lb_subchannel_list_unref_for_connectivity_watch(sd->subchannel_list, + "pf_sl_shutdown"); return; } // If we're still here, the notification must be for a subchannel in @@ -407,15 +402,15 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, if (sd->curr_connectivity_state != GRPC_CHANNEL_READY && p->latest_pending_subchannel_list != nullptr) { p->selected = nullptr; - grpc_lb_subchannel_data_stop_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_stop_connectivity_watch(sd); grpc_lb_subchannel_list_unref_for_connectivity_watch( - exec_ctx, sd->subchannel_list, "selected_not_ready+switch_to_update"); + sd->subchannel_list, "selected_not_ready+switch_to_update"); grpc_lb_subchannel_list_shutdown_and_unref( - exec_ctx, p->subchannel_list, "selected_not_ready+switch_to_update"); + p->subchannel_list, "selected_not_ready+switch_to_update"); p->subchannel_list = p->latest_pending_subchannel_list; p->latest_pending_subchannel_list = nullptr; grpc_connectivity_state_set( - exec_ctx, &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, + &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, GRPC_ERROR_REF(error), "selected_not_ready+switch_to_update"); } else { // TODO(juanlishen): we re-resolve when the selected subchannel goes to @@ -426,27 +421,26 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, if (sd->curr_connectivity_state == GRPC_CHANNEL_SHUTDOWN || sd->curr_connectivity_state == GRPC_CHANNEL_TRANSIENT_FAILURE) { // If the selected channel goes bad, request a re-resolution. - grpc_connectivity_state_set(exec_ctx, &p->state_tracker, - GRPC_CHANNEL_IDLE, GRPC_ERROR_NONE, + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_IDLE, + GRPC_ERROR_NONE, "selected_changed+reresolve"); p->started_picking = false; - grpc_lb_policy_try_reresolve( - exec_ctx, &p->base, &grpc_lb_pick_first_trace, GRPC_ERROR_NONE); + grpc_lb_policy_try_reresolve(&p->base, &grpc_lb_pick_first_trace, + GRPC_ERROR_NONE); } else { - grpc_connectivity_state_set(exec_ctx, &p->state_tracker, + grpc_connectivity_state_set(&p->state_tracker, sd->curr_connectivity_state, GRPC_ERROR_REF(error), "selected_changed"); } if (sd->curr_connectivity_state != GRPC_CHANNEL_SHUTDOWN) { // Renew notification. - grpc_lb_subchannel_data_start_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_start_connectivity_watch(sd); } else { p->selected = nullptr; - grpc_lb_subchannel_data_stop_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_stop_connectivity_watch(sd); grpc_lb_subchannel_list_unref_for_connectivity_watch( - exec_ctx, sd->subchannel_list, "pf_selected_shutdown"); - grpc_lb_subchannel_data_unref_subchannel(exec_ctx, sd, - "pf_selected_shutdown"); + sd->subchannel_list, "pf_selected_shutdown"); + grpc_lb_subchannel_data_unref_subchannel(sd, "pf_selected_shutdown"); } } return; @@ -466,15 +460,14 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, // p->subchannel_list. if (sd->subchannel_list == p->latest_pending_subchannel_list) { GPR_ASSERT(p->subchannel_list != nullptr); - grpc_lb_subchannel_list_shutdown_and_unref(exec_ctx, p->subchannel_list, + grpc_lb_subchannel_list_shutdown_and_unref(p->subchannel_list, "finish_update"); p->subchannel_list = p->latest_pending_subchannel_list; p->latest_pending_subchannel_list = nullptr; } // Cases 1 and 2. - grpc_connectivity_state_set(exec_ctx, &p->state_tracker, - GRPC_CHANNEL_READY, GRPC_ERROR_NONE, - "connecting_ready"); + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_READY, + GRPC_ERROR_NONE, "connecting_ready"); sd->connected_subchannel = GRPC_CONNECTED_SUBCHANNEL_REF( grpc_subchannel_get_connected_subchannel(sd->subchannel), "connected"); @@ -484,7 +477,7 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, (void*)sd->subchannel); } // Drop all other subchannels, since we are now connected. - destroy_unselected_subchannels_locked(exec_ctx, p); + destroy_unselected_subchannels_locked(p); // Update any calls that were waiting for a pick. pending_pick* pp; while ((pp = p->pending_picks)) { @@ -496,15 +489,15 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, "Servicing pending pick with selected subchannel %p", (void*)p->selected); } - GRPC_CLOSURE_SCHED(exec_ctx, pp->on_complete, GRPC_ERROR_NONE); + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_NONE); gpr_free(pp); } // Renew notification. - grpc_lb_subchannel_data_start_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_start_connectivity_watch(sd); break; } case GRPC_CHANNEL_TRANSIENT_FAILURE: { - grpc_lb_subchannel_data_stop_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_stop_connectivity_watch(sd); do { sd->subchannel_list->checking_subchannel = (sd->subchannel_list->checking_subchannel + 1) % @@ -517,29 +510,28 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, if (sd->subchannel_list->checking_subchannel == 0 && sd->subchannel_list == p->subchannel_list) { grpc_connectivity_state_set( - exec_ctx, &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, + &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, GRPC_ERROR_REF(error), "connecting_transient_failure"); } // Reuses the connectivity refs from the previous watch. - grpc_lb_subchannel_data_start_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_start_connectivity_watch(sd); break; } case GRPC_CHANNEL_CONNECTING: case GRPC_CHANNEL_IDLE: { // Only update connectivity state in case 1. if (sd->subchannel_list == p->subchannel_list) { - grpc_connectivity_state_set( - exec_ctx, &p->state_tracker, GRPC_CHANNEL_CONNECTING, - GRPC_ERROR_REF(error), "connecting_changed"); + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_CONNECTING, + GRPC_ERROR_REF(error), + "connecting_changed"); } // Renew notification. - grpc_lb_subchannel_data_start_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_start_connectivity_watch(sd); break; } case GRPC_CHANNEL_SHUTDOWN: { - grpc_lb_subchannel_data_stop_connectivity_watch(exec_ctx, sd); - grpc_lb_subchannel_data_unref_subchannel(exec_ctx, sd, - "pf_candidate_shutdown"); + grpc_lb_subchannel_data_stop_connectivity_watch(sd); + grpc_lb_subchannel_data_unref_subchannel(sd, "pf_candidate_shutdown"); // Advance to next subchannel and check its state. grpc_lb_subchannel_data* original_sd = sd; do { @@ -551,31 +543,30 @@ static void pf_connectivity_changed_locked(grpc_exec_ctx* exec_ctx, void* arg, } while (sd->subchannel == nullptr && sd != original_sd); if (sd == original_sd) { grpc_lb_subchannel_list_unref_for_connectivity_watch( - exec_ctx, sd->subchannel_list, "pf_exhausted_subchannels"); + sd->subchannel_list, "pf_exhausted_subchannels"); if (sd->subchannel_list == p->subchannel_list) { - grpc_connectivity_state_set(exec_ctx, &p->state_tracker, - GRPC_CHANNEL_IDLE, GRPC_ERROR_NONE, + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_IDLE, + GRPC_ERROR_NONE, "exhausted_subchannels+reresolve"); p->started_picking = false; - grpc_lb_policy_try_reresolve( - exec_ctx, &p->base, &grpc_lb_pick_first_trace, GRPC_ERROR_NONE); + grpc_lb_policy_try_reresolve(&p->base, &grpc_lb_pick_first_trace, + GRPC_ERROR_NONE); } } else { if (sd->subchannel_list == p->subchannel_list) { grpc_connectivity_state_set( - exec_ctx, &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, + &p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, GRPC_ERROR_REF(error), "subchannel_failed"); } // Reuses the connectivity refs from the previous watch. - grpc_lb_subchannel_data_start_connectivity_watch(exec_ctx, sd); + grpc_lb_subchannel_data_start_connectivity_watch(sd); } } } } static void pf_set_reresolve_closure_locked( - grpc_exec_ctx* exec_ctx, grpc_lb_policy* policy, - grpc_closure* request_reresolution) { + grpc_lb_policy* policy, grpc_closure* request_reresolution) { pick_first_lb_policy* p = (pick_first_lb_policy*)policy; GPR_ASSERT(!p->shutdown); GPR_ASSERT(policy->request_reresolution == nullptr); @@ -599,15 +590,14 @@ static void pick_first_factory_ref(grpc_lb_policy_factory* factory) {} static void pick_first_factory_unref(grpc_lb_policy_factory* factory) {} -static grpc_lb_policy* create_pick_first(grpc_exec_ctx* exec_ctx, - grpc_lb_policy_factory* factory, +static grpc_lb_policy* create_pick_first(grpc_lb_policy_factory* factory, grpc_lb_policy_args* args) { GPR_ASSERT(args->client_channel_factory != nullptr); pick_first_lb_policy* p = (pick_first_lb_policy*)gpr_zalloc(sizeof(*p)); if (grpc_lb_pick_first_trace.enabled()) { gpr_log(GPR_DEBUG, "Pick First %p created.", (void*)p); } - pf_update_locked(exec_ctx, &p->base, args); + pf_update_locked(&p->base, args); grpc_lb_policy_init(&p->base, &pick_first_lb_policy_vtable, args->combiner); grpc_subchannel_index_ref(); return &p->base; |