From 776490af0eae820fb8919b2ea6f245aff197a1ab Mon Sep 17 00:00:00 2001 From: Juanli Shen Date: Wed, 7 Feb 2018 14:51:20 -0800 Subject: Revert "Revert "grpclb re-resolution"" --- .../client_channel/lb_policy/grpclb/grpclb.cc | 133 +++++++++++---------- 1 file changed, 71 insertions(+), 62 deletions(-) (limited to 'src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc') diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc index 04bf798267..1c8809eabc 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc @@ -249,7 +249,7 @@ typedef struct glb_lb_policy { /** the RR policy to use of the backend servers returned by the LB server */ grpc_lb_policy* rr_policy; - grpc_closure on_rr_connectivity_changed; + /** the connectivity state of the embedded RR policy */ grpc_connectivity_state rr_connectivity_state; bool started_picking; @@ -292,6 +292,12 @@ typedef struct glb_lb_policy { /** called upon changes to the LB channel's connectivity. */ grpc_closure lb_channel_on_connectivity_changed; + /** called upon changes to the RR's connectivity. */ + grpc_closure rr_on_connectivity_changed; + + /** called upon reresolution request from the RR policy. */ + grpc_closure rr_on_reresolution_requested; + /************************************************************/ /* client data associated with the LB server communication */ /************************************************************/ @@ -590,9 +596,8 @@ static grpc_lb_addresses* extract_backend_addresses_locked( return backend_addresses; } -static void update_lb_connectivity_status_locked( - glb_lb_policy* glb_policy, grpc_connectivity_state rr_state, - grpc_error* rr_state_error) { +static void update_lb_connectivity_status_locked(glb_lb_policy* glb_policy, + grpc_error* rr_state_error) { const grpc_connectivity_state curr_glb_state = grpc_connectivity_state_check(&glb_policy->state_tracker); /* The new connectivity status is a function of the previous one and the new @@ -624,7 +629,7 @@ static void update_lb_connectivity_status_locked( * * (*) This function mustn't be called during shutting down. */ GPR_ASSERT(curr_glb_state != GRPC_CHANNEL_SHUTDOWN); - switch (rr_state) { + switch (glb_policy->rr_connectivity_state) { case GRPC_CHANNEL_TRANSIENT_FAILURE: case GRPC_CHANNEL_SHUTDOWN: GPR_ASSERT(rr_state_error != GRPC_ERROR_NONE); @@ -638,11 +643,12 @@ static void update_lb_connectivity_status_locked( gpr_log( GPR_INFO, "[grpclb %p] Setting grpclb's state to %s from new RR policy %p state.", - glb_policy, grpc_connectivity_state_name(rr_state), + glb_policy, + grpc_connectivity_state_name(glb_policy->rr_connectivity_state), glb_policy->rr_policy); } - grpc_connectivity_state_set(&glb_policy->state_tracker, rr_state, - rr_state_error, + grpc_connectivity_state_set(&glb_policy->state_tracker, + glb_policy->rr_connectivity_state, rr_state_error, "update_lb_connectivity_status_locked"); } @@ -740,11 +746,36 @@ static void lb_policy_args_destroy(grpc_lb_policy_args* args) { gpr_free(args); } -static void on_rr_connectivity_changed_locked(void* arg, grpc_error* error); +static void rr_on_reresolution_requested_locked(void* arg, grpc_error* error) { + glb_lb_policy* glb_policy = (glb_lb_policy*)arg; + if (glb_policy->shutting_down || error != GRPC_ERROR_NONE) { + GRPC_LB_POLICY_UNREF(&glb_policy->base, + "rr_on_reresolution_requested_locked"); + return; + } + if (grpc_lb_glb_trace.enabled()) { + gpr_log( + GPR_DEBUG, + "[grpclb %p] Re-resolution requested from the internal RR policy (%p).", + glb_policy, glb_policy->rr_policy); + } + // If we are talking to a balancer, we expect to get updated addresses form + // the balancer, so we can ignore the re-resolution request from the RR + // policy. Otherwise, handle the re-resolution request using glb's original + // re-resolution closure. + if (glb_policy->lb_calld == nullptr || + !glb_policy->lb_calld->seen_initial_response) { + grpc_lb_policy_try_reresolve(&glb_policy->base, &grpc_lb_glb_trace, + GRPC_ERROR_NONE); + } + // Give back the wrapper closure to the RR policy. + grpc_lb_policy_set_reresolve_closure_locked( + glb_policy->rr_policy, &glb_policy->rr_on_reresolution_requested); +} + static void create_rr_locked(glb_lb_policy* glb_policy, grpc_lb_policy_args* args) { GPR_ASSERT(glb_policy->rr_policy == nullptr); - grpc_lb_policy* new_rr_policy = grpc_lb_policy_create("round_robin", args); if (new_rr_policy == nullptr) { gpr_log(GPR_ERROR, @@ -757,29 +788,25 @@ static void create_rr_locked(glb_lb_policy* glb_policy, glb_policy->rr_policy); return; } + GRPC_LB_POLICY_REF(&glb_policy->base, "rr_on_reresolution_requested_locked"); grpc_lb_policy_set_reresolve_closure_locked( - new_rr_policy, glb_policy->base.request_reresolution); - glb_policy->base.request_reresolution = nullptr; + new_rr_policy, &glb_policy->rr_on_reresolution_requested); glb_policy->rr_policy = new_rr_policy; grpc_error* rr_state_error = nullptr; glb_policy->rr_connectivity_state = grpc_lb_policy_check_connectivity_locked( glb_policy->rr_policy, &rr_state_error); /* Connectivity state is a function of the RR policy updated/created */ - update_lb_connectivity_status_locked( - glb_policy, glb_policy->rr_connectivity_state, rr_state_error); + update_lb_connectivity_status_locked(glb_policy, rr_state_error); /* Add the gRPC LB's interested_parties pollset_set to that of the newly * created RR policy. This will make the RR policy progress upon activity on * gRPC LB, which in turn is tied to the application's call */ grpc_pollset_set_add_pollset_set(glb_policy->rr_policy->interested_parties, glb_policy->base.interested_parties); - GRPC_CLOSURE_INIT(&glb_policy->on_rr_connectivity_changed, - on_rr_connectivity_changed_locked, glb_policy, - grpc_combiner_scheduler(glb_policy->base.combiner)); /* Subscribe to changes to the connectivity of the new RR */ - GRPC_LB_POLICY_REF(&glb_policy->base, "glb_rr_connectivity_cb"); + GRPC_LB_POLICY_REF(&glb_policy->base, "rr_on_connectivity_changed_locked"); grpc_lb_policy_notify_on_state_change_locked( glb_policy->rr_policy, &glb_policy->rr_connectivity_state, - &glb_policy->on_rr_connectivity_changed); + &glb_policy->rr_on_connectivity_changed); grpc_lb_policy_exit_idle_locked(glb_policy->rr_policy); // Send pending picks to RR policy. pending_pick* pp; @@ -827,28 +854,18 @@ static void rr_handover_locked(glb_lb_policy* glb_policy) { lb_policy_args_destroy(args); } -static void on_rr_connectivity_changed_locked(void* arg, grpc_error* error) { - glb_lb_policy* glb_policy = static_cast(arg); +static void rr_on_connectivity_changed_locked(void* arg, grpc_error* error) { + glb_lb_policy* glb_policy = (glb_lb_policy*)arg; if (glb_policy->shutting_down) { - GRPC_LB_POLICY_UNREF(&glb_policy->base, "glb_rr_connectivity_cb"); - return; - } - if (glb_policy->rr_connectivity_state == GRPC_CHANNEL_SHUTDOWN) { - /* An RR policy that has transitioned into the SHUTDOWN connectivity state - * should not be considered for picks or updates: the SHUTDOWN state is a - * sink, policies can't transition back from it. .*/ - GRPC_LB_POLICY_UNREF(glb_policy->rr_policy, "rr_connectivity_shutdown"); - glb_policy->rr_policy = nullptr; - GRPC_LB_POLICY_UNREF(&glb_policy->base, "glb_rr_connectivity_cb"); + GRPC_LB_POLICY_UNREF(&glb_policy->base, + "rr_on_connectivity_changed_locked"); return; } - /* rr state != SHUTDOWN && !glb_policy->shutting down: biz as usual */ - update_lb_connectivity_status_locked( - glb_policy, glb_policy->rr_connectivity_state, GRPC_ERROR_REF(error)); - /* Resubscribe. Reuse the "glb_rr_connectivity_cb" ref. */ + update_lb_connectivity_status_locked(glb_policy, GRPC_ERROR_REF(error)); + // Resubscribe. Reuse the "rr_on_connectivity_changed_locked" ref. grpc_lb_policy_notify_on_state_change_locked( glb_policy->rr_policy, &glb_policy->rr_connectivity_state, - &glb_policy->on_rr_connectivity_changed); + &glb_policy->rr_on_connectivity_changed); } static void destroy_balancer_name(void* balancer_name) { @@ -971,8 +988,6 @@ static void glb_shutdown_locked(grpc_lb_policy* pol, if (glb_policy->rr_policy != nullptr) { grpc_lb_policy_shutdown_locked(glb_policy->rr_policy, nullptr); GRPC_LB_POLICY_UNREF(glb_policy->rr_policy, "glb_shutdown"); - } else { - grpc_lb_policy_try_reresolve(pol, &grpc_lb_glb_trace, GRPC_ERROR_CANCELLED); } // We destroy the LB channel here because // glb_lb_channel_on_connectivity_changed_cb needs a valid glb_policy @@ -984,6 +999,7 @@ static void glb_shutdown_locked(grpc_lb_policy* pol, } grpc_connectivity_state_set(&glb_policy->state_tracker, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "glb_shutdown"); + grpc_lb_policy_try_reresolve(pol, &grpc_lb_glb_trace, GRPC_ERROR_CANCELLED); // Clear pending picks. pending_pick* pp = glb_policy->pending_picks; glb_policy->pending_picks = nullptr; @@ -1623,6 +1639,8 @@ static void lb_on_server_status_received_locked(void* arg, grpc_error* error) { lb_calld, lb_calld->lb_call, grpc_error_string(error)); gpr_free(status_details); } + grpc_lb_policy_try_reresolve(&glb_policy->base, &grpc_lb_glb_trace, + GRPC_ERROR_NONE); // If this lb_calld is still in use, this call ended because of a failure so // we want to retry connecting. Otherwise, we have deliberately ended this // call and no further action is required. @@ -1651,16 +1669,15 @@ static void lb_on_fallback_timer_locked(void* arg, grpc_error* error) { glb_policy->fallback_timer_callback_pending = false; /* If we receive a serverlist after the timer fires but before this callback * actually runs, don't fall back. */ - if (glb_policy->serverlist == nullptr) { - if (!glb_policy->shutting_down && error == GRPC_ERROR_NONE) { - if (grpc_lb_glb_trace.enabled()) { - gpr_log(GPR_INFO, - "[grpclb %p] Falling back to use backends from resolver", - glb_policy); - } - GPR_ASSERT(glb_policy->fallback_backend_addresses != nullptr); - rr_handover_locked(glb_policy); + if (glb_policy->serverlist == nullptr && !glb_policy->shutting_down && + error == GRPC_ERROR_NONE) { + if (grpc_lb_glb_trace.enabled()) { + gpr_log(GPR_INFO, + "[grpclb %p] Falling back to use backends from resolver", + glb_policy); } + GPR_ASSERT(glb_policy->fallback_backend_addresses != nullptr); + rr_handover_locked(glb_policy); } GRPC_LB_POLICY_UNREF(&glb_policy->base, "grpclb_fallback_timer"); } @@ -1781,19 +1798,6 @@ static void glb_lb_channel_on_connectivity_changed_cb(void* arg, } } -static void glb_set_reresolve_closure_locked( - grpc_lb_policy* policy, grpc_closure* request_reresolution) { - glb_lb_policy* glb_policy = reinterpret_cast(policy); - GPR_ASSERT(!glb_policy->shutting_down); - GPR_ASSERT(glb_policy->base.request_reresolution == nullptr); - if (glb_policy->rr_policy != nullptr) { - grpc_lb_policy_set_reresolve_closure_locked(glb_policy->rr_policy, - request_reresolution); - } else { - glb_policy->base.request_reresolution = request_reresolution; - } -} - /* Code wiring the policy with the rest of the core */ static const grpc_lb_policy_vtable glb_lb_policy_vtable = { glb_destroy, @@ -1805,8 +1809,7 @@ static const grpc_lb_policy_vtable glb_lb_policy_vtable = { glb_exit_idle_locked, glb_check_connectivity_locked, glb_notify_on_state_change_locked, - glb_update_locked, - glb_set_reresolve_closure_locked}; + glb_update_locked}; static grpc_lb_policy* glb_create(grpc_lb_policy_factory* factory, grpc_lb_policy_args* args) { @@ -1887,6 +1890,12 @@ static grpc_lb_policy* glb_create(grpc_lb_policy_factory* factory, return nullptr; } grpc_subchannel_index_ref(); + GRPC_CLOSURE_INIT(&glb_policy->rr_on_connectivity_changed, + rr_on_connectivity_changed_locked, glb_policy, + grpc_combiner_scheduler(args->combiner)); + GRPC_CLOSURE_INIT(&glb_policy->rr_on_reresolution_requested, + rr_on_reresolution_requested_locked, glb_policy, + grpc_combiner_scheduler(args->combiner)); GRPC_CLOSURE_INIT(&glb_policy->lb_channel_on_connectivity_changed, glb_lb_channel_on_connectivity_changed_cb, glb_policy, grpc_combiner_scheduler(args->combiner)); -- cgit v1.2.3