diff options
Diffstat (limited to 'src/core/ext/lb_policy')
5 files changed, 81 insertions, 128 deletions
diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c index 38eebdc758..97f98df03a 100644 --- a/src/core/ext/lb_policy/grpclb/grpclb.c +++ b/src/core/ext/lb_policy/grpclb/grpclb.c @@ -117,6 +117,7 @@ #include "src/core/lib/iomgr/sockaddr.h" #include "src/core/lib/iomgr/sockaddr_utils.h" #include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/backoff.h" #include "src/core/lib/support/string.h" @@ -180,8 +181,7 @@ static void wrapped_rr_closure(grpc_exec_ctx *exec_ctx, void *arg, wrapped_rr_closure_arg *wc_arg = arg; GPR_ASSERT(wc_arg->wrapped_closure != NULL); - grpc_exec_ctx_sched(exec_ctx, wc_arg->wrapped_closure, GRPC_ERROR_REF(error), - NULL); + grpc_closure_sched(exec_ctx, wc_arg->wrapped_closure, GRPC_ERROR_REF(error)); if (wc_arg->rr_policy != NULL) { /* if *target is NULL, no pick has been made by the RR policy (eg, all @@ -248,7 +248,8 @@ static void add_pending_pick(pending_pick **root, pick_args->lb_token_mdelem_storage; pp->wrapped_on_complete_arg.free_when_done = pp; grpc_closure_init(&pp->wrapped_on_complete_arg.wrapper_closure, - wrapped_rr_closure, &pp->wrapped_on_complete_arg); + wrapped_rr_closure, &pp->wrapped_on_complete_arg, + grpc_schedule_on_exec_ctx); *root = pp; } @@ -268,7 +269,8 @@ static void add_pending_ping(pending_ping **root, grpc_closure *notify) { pping->wrapped_notify_arg.free_when_done = pping; pping->next = *root; grpc_closure_init(&pping->wrapped_notify_arg.wrapper_closure, - wrapped_rr_closure, &pping->wrapped_notify_arg); + wrapped_rr_closure, &pping->wrapped_notify_arg, + grpc_schedule_on_exec_ctx); *root = pping; } @@ -325,6 +327,9 @@ typedef struct glb_lb_policy { /* A response from the LB server has been received. Process it */ grpc_closure lb_on_response_received; + /* LB call retry timer callback. */ + grpc_closure lb_on_call_retry; + grpc_call *lb_call; /* streaming call to the LB server, */ grpc_metadata_array lb_initial_metadata_recv; /* initial MD from LB server */ @@ -385,8 +390,8 @@ static bool is_server_valid(const grpc_grpclb_server *server, size_t idx, static void *lb_token_copy(void *token) { return token == NULL ? NULL : GRPC_MDELEM_REF(token); } -static void lb_token_destroy(void *token) { - if (token != NULL) GRPC_MDELEM_UNREF(token); +static void lb_token_destroy(grpc_exec_ctx *exec_ctx, void *token) { + if (token != NULL) GRPC_MDELEM_UNREF(exec_ctx, token); } static int lb_token_cmp(void *token1, void *token2) { if (token1 > token2) return 1; @@ -420,7 +425,7 @@ static void parse_server(const grpc_grpclb_server *server, /* Returns addresses extracted from \a serverlist. */ static grpc_lb_addresses *process_serverlist_locked( - const grpc_grpclb_serverlist *serverlist) { + grpc_exec_ctx *exec_ctx, const grpc_grpclb_serverlist *serverlist) { size_t num_valid = 0; /* first pass: count how many are valid in order to allocate the necessary * memory in a single block */ @@ -456,8 +461,8 @@ static grpc_lb_addresses *process_serverlist_locked( strnlen(server->load_balance_token, lb_token_max_length); grpc_mdstr *lb_token_mdstr = grpc_mdstr_from_buffer( (uint8_t *)server->load_balance_token, lb_token_length); - user_data = grpc_mdelem_from_metadata_strings(GRPC_MDSTR_LB_TOKEN, - lb_token_mdstr); + user_data = grpc_mdelem_from_metadata_strings( + exec_ctx, GRPC_MDSTR_LB_TOKEN, lb_token_mdstr); } else { char *uri = grpc_sockaddr_to_uri(&addr); gpr_log(GPR_INFO, @@ -580,7 +585,8 @@ static grpc_lb_policy *create_rr_locked( grpc_lb_policy_args args; memset(&args, 0, sizeof(args)); args.client_channel_factory = glb_policy->cc_factory; - grpc_lb_addresses *addresses = process_serverlist_locked(serverlist); + grpc_lb_addresses *addresses = + process_serverlist_locked(exec_ctx, serverlist); // Replace the LB addresses in the channel args that we pass down to // the subchannel. @@ -592,8 +598,8 @@ static grpc_lb_policy *create_rr_locked( grpc_lb_policy *rr = grpc_lb_policy_create(exec_ctx, "round_robin", &args); GPR_ASSERT(rr != NULL); - grpc_lb_addresses_destroy(addresses); - grpc_channel_args_destroy(args.args); + grpc_lb_addresses_destroy(exec_ctx, addresses); + grpc_channel_args_destroy(exec_ctx, args.args); return rr; } @@ -667,7 +673,7 @@ static void rr_handover_locked(grpc_exec_ctx *exec_ctx, gpr_malloc(sizeof(rr_connectivity_data)); memset(rr_connectivity, 0, sizeof(rr_connectivity_data)); grpc_closure_init(&rr_connectivity->on_change, glb_rr_connectivity_changed, - rr_connectivity); + rr_connectivity, grpc_schedule_on_exec_ctx); rr_connectivity->glb_policy = glb_policy; rr_connectivity->state = new_rr_state; @@ -768,8 +774,14 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI); GPR_ASSERT(arg != NULL); GPR_ASSERT(arg->type == GRPC_ARG_STRING); - grpc_uri *uri = grpc_uri_parse(arg->value.string, 1); - glb_policy->server_name = gpr_strdup(uri->path); + grpc_uri *uri = grpc_uri_parse(arg->value.string, true); + GPR_ASSERT(uri->path[0] != '\0'); + glb_policy->server_name = + gpr_strdup(uri->path[0] == '/' ? uri->path + 1 : uri->path); + if (grpc_lb_glb_trace) { + gpr_log(GPR_INFO, "Will use '%s' as the server name for LB request.", + glb_policy->server_name); + } grpc_uri_destroy(uri); /* All input addresses in addresses come from a resolver that claims @@ -833,7 +845,7 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, glb_policy->lb_channel = grpc_client_channel_factory_create_channel( exec_ctx, glb_policy->cc_factory, target_uri_str, GRPC_CLIENT_CHANNEL_TYPE_LOAD_BALANCING, new_args); - grpc_channel_args_destroy(new_args); + grpc_channel_args_destroy(exec_ctx, new_args); gpr_free(target_uri_str); for (size_t i = 0; i < num_grpclb_addrs; i++) { @@ -859,7 +871,7 @@ static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { GPR_ASSERT(glb_policy->pending_picks == NULL); GPR_ASSERT(glb_policy->pending_pings == NULL); gpr_free((void *)glb_policy->server_name); - grpc_channel_args_destroy(glb_policy->args); + grpc_channel_args_destroy(exec_ctx, glb_policy->args); grpc_channel_destroy(glb_policy->lb_channel); glb_policy->lb_channel = NULL; grpc_connectivity_state_destroy(exec_ctx, &glb_policy->state_tracker); @@ -902,15 +914,15 @@ static void glb_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { while (pp != NULL) { pending_pick *next = pp->next; *pp->target = NULL; - grpc_exec_ctx_sched(exec_ctx, &pp->wrapped_on_complete_arg.wrapper_closure, - GRPC_ERROR_NONE, NULL); + grpc_closure_sched(exec_ctx, &pp->wrapped_on_complete_arg.wrapper_closure, + GRPC_ERROR_NONE); pp = next; } while (pping != NULL) { pending_ping *next = pping->next; - grpc_exec_ctx_sched(exec_ctx, &pping->wrapped_notify_arg.wrapper_closure, - GRPC_ERROR_NONE, NULL); + grpc_closure_sched(exec_ctx, &pping->wrapped_notify_arg.wrapper_closure, + GRPC_ERROR_NONE); pping = next; } } @@ -926,9 +938,9 @@ static void glb_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pending_pick *next = pp->next; if (pp->target == target) { *target = NULL; - grpc_exec_ctx_sched( + grpc_closure_sched( exec_ctx, &pp->wrapped_on_complete_arg.wrapper_closure, - GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1)); } else { pp->next = glb_policy->pending_picks; glb_policy->pending_picks = pp; @@ -951,9 +963,9 @@ static void glb_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pending_pick *next = pp->next; if ((pp->pick_args.initial_metadata_flags & initial_metadata_flags_mask) == initial_metadata_flags_eq) { - grpc_exec_ctx_sched( + grpc_closure_sched( exec_ctx, &pp->wrapped_on_complete_arg.wrapper_closure, - GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1)); } else { pp->next = glb_policy->pending_picks; glb_policy->pending_picks = pp; @@ -988,11 +1000,10 @@ static int glb_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_closure *on_complete) { if (pick_args->lb_token_mdelem_storage == NULL) { *target = NULL; - grpc_exec_ctx_sched( + grpc_closure_sched( exec_ctx, on_complete, GRPC_ERROR_CREATE("No mdelem storage for the LB token. Load reporting " - "won't work without it. Failing"), - NULL); + "won't work without it. Failing")); return 0; } @@ -1011,7 +1022,8 @@ static int glb_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, wrapped_rr_closure_arg *wc_arg = gpr_malloc(sizeof(wrapped_rr_closure_arg)); memset(wc_arg, 0, sizeof(wrapped_rr_closure_arg)); - grpc_closure_init(&wc_arg->wrapper_closure, wrapped_rr_closure, wc_arg); + grpc_closure_init(&wc_arg->wrapper_closure, wrapped_rr_closure, wc_arg, + grpc_schedule_on_exec_ctx); wc_arg->rr_policy = glb_policy->rr_policy; wc_arg->target = target; wc_arg->wrapped_closure = on_complete; @@ -1082,7 +1094,8 @@ static void lb_on_server_status_received(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error); static void lb_on_response_received(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error); -static void lb_call_init_locked(glb_lb_policy *glb_policy) { +static void lb_call_init_locked(grpc_exec_ctx *exec_ctx, + glb_lb_policy *glb_policy) { GPR_ASSERT(glb_policy->server_name != NULL); GPR_ASSERT(glb_policy->server_name[0] != '\0'); GPR_ASSERT(!glb_policy->shutting_down); @@ -1091,7 +1104,7 @@ static void lb_call_init_locked(glb_lb_policy *glb_policy) { * glb_policy->base.interested_parties, which is comprised of the polling * entities from \a client_channel. */ glb_policy->lb_call = grpc_channel_create_pollset_set_call( - glb_policy->lb_channel, NULL, GRPC_PROPAGATE_DEFAULTS, + exec_ctx, glb_policy->lb_channel, NULL, GRPC_PROPAGATE_DEFAULTS, glb_policy->base.interested_parties, "/grpc.lb.v1.LoadBalancer/BalanceLoad", glb_policy->server_name, glb_policy->deadline, NULL); @@ -1104,16 +1117,18 @@ static void lb_call_init_locked(glb_lb_policy *glb_policy) { grpc_slice request_payload_slice = grpc_grpclb_request_encode(request); glb_policy->lb_request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); - grpc_slice_unref(request_payload_slice); + grpc_slice_unref_internal(exec_ctx, request_payload_slice); grpc_grpclb_request_destroy(request); glb_policy->lb_call_status_details = NULL; glb_policy->lb_call_status_details_capacity = 0; grpc_closure_init(&glb_policy->lb_on_server_status_received, - lb_on_server_status_received, glb_policy); + lb_on_server_status_received, glb_policy, + grpc_schedule_on_exec_ctx); grpc_closure_init(&glb_policy->lb_on_response_received, - lb_on_response_received, glb_policy); + lb_on_response_received, glb_policy, + grpc_schedule_on_exec_ctx); gpr_backoff_init(&glb_policy->lb_call_backoff_state, GRPC_GRPCLB_INITIAL_CONNECT_BACKOFF_SECONDS, @@ -1143,7 +1158,7 @@ static void query_for_backends_locked(grpc_exec_ctx *exec_ctx, GPR_ASSERT(glb_policy->lb_channel != NULL); if (glb_policy->shutting_down) return; - lb_call_init_locked(glb_policy); + lb_call_init_locked(exec_ctx, glb_policy); if (grpc_lb_glb_trace) { gpr_log(GPR_INFO, "Query for backends (grpclb: %p, lb_call: %p)", @@ -1228,7 +1243,7 @@ static void lb_on_response_received(grpc_exec_ctx *exec_ctx, void *arg, grpc_grpclb_response_parse_serverlist(response_slice); if (serverlist != NULL) { GPR_ASSERT(glb_policy->lb_call != NULL); - grpc_slice_unref(response_slice); + grpc_slice_unref_internal(exec_ctx, response_slice); if (grpc_lb_glb_trace) { gpr_log(GPR_INFO, "Serverlist with %lu servers received", (unsigned long)serverlist->num_servers); @@ -1272,7 +1287,7 @@ static void lb_on_response_received(grpc_exec_ctx *exec_ctx, void *arg, } else { /* serverlist == NULL */ gpr_log(GPR_ERROR, "Invalid LB response received: '%s'. Ignoring.", grpc_dump_slice(response_slice, GPR_DUMP_ASCII | GPR_DUMP_HEX)); - grpc_slice_unref(response_slice); + grpc_slice_unref_internal(exec_ctx, response_slice); } if (!glb_policy->shutting_down) { @@ -1352,8 +1367,10 @@ static void lb_on_server_status_received(grpc_exec_ctx *exec_ctx, void *arg, } } GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "grpclb_retry_timer"); + grpc_closure_init(&glb_policy->lb_on_call_retry, lb_call_on_retry_timer, + glb_policy, grpc_schedule_on_exec_ctx); grpc_timer_init(exec_ctx, &glb_policy->lb_call_retry_timer, next_try, - lb_call_on_retry_timer, glb_policy, now); + &glb_policy->lb_on_call_retry, now); } gpr_mu_unlock(&glb_policy->mu); GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &glb_policy->base, diff --git a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c index afecb716fb..e352e0396f 100644 --- a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c +++ b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c @@ -1,35 +1,3 @@ -/* - * - * Copyright 2016, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ /* Automatically generated nanopb constant definitions */ /* Generated by nanopb-0.3.7-dev */ diff --git a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h index e36d0966f8..725aa7e386 100644 --- a/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h +++ b/src/core/ext/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h @@ -1,35 +1,3 @@ -/* - * - * Copyright 2016, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ /* Automatically generated nanopb header */ /* Generated by nanopb-0.3.7-dev */ diff --git a/src/core/ext/lb_policy/pick_first/pick_first.c b/src/core/ext/lb_policy/pick_first/pick_first.c index b9cfe6b5c0..821becff69 100644 --- a/src/core/ext/lb_policy/pick_first/pick_first.c +++ b/src/core/ext/lb_policy/pick_first/pick_first.c @@ -120,7 +120,7 @@ static void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { while (pp != NULL) { pending_pick *next = pp->next; *pp->target = NULL; - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE, NULL); + grpc_closure_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE); gpr_free(pp); pp = next; } @@ -138,9 +138,9 @@ static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pending_pick *next = pp->next; if (pp->target == target) { *target = NULL; - grpc_exec_ctx_sched( + grpc_closure_sched( exec_ctx, pp->on_complete, - GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1)); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -165,9 +165,9 @@ static void pf_cancel_picks(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_exec_ctx_sched( + grpc_closure_sched( exec_ctx, pp->on_complete, - GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1)); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -306,14 +306,15 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg, /* drop the pick list: we are connected now */ GRPC_LB_POLICY_WEAK_REF(&p->base, "destroy_subchannels"); gpr_atm_rel_store(&p->selected, (gpr_atm)selected); - grpc_exec_ctx_sched(exec_ctx, - grpc_closure_create(destroy_subchannels, p), - GRPC_ERROR_NONE, NULL); + grpc_closure_sched(exec_ctx, + grpc_closure_create(destroy_subchannels, p, + grpc_schedule_on_exec_ctx), + GRPC_ERROR_NONE); /* update any calls that were waiting for a pick */ while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = GRPC_CONNECTED_SUBCHANNEL_REF(selected, "picked"); - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE, NULL); + grpc_closure_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE); gpr_free(pp); } grpc_connected_subchannel_notify_on_state_change( @@ -366,8 +367,7 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg, while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = NULL; - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE, - NULL); + grpc_closure_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE); gpr_free(pp); } GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, @@ -419,8 +419,7 @@ static void pf_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, if (selected) { grpc_connected_subchannel_ping(exec_ctx, selected, closure); } else { - grpc_exec_ctx_sched(exec_ctx, closure, GRPC_ERROR_CREATE("Not connected"), - NULL); + grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_CREATE("Not connected")); } } @@ -485,7 +484,8 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx, p->num_subchannels = subchannel_idx; grpc_lb_policy_init(&p->base, &pick_first_lb_policy_vtable); - grpc_closure_init(&p->connectivity_changed, pf_connectivity_changed, p); + grpc_closure_init(&p->connectivity_changed, pf_connectivity_changed, p, + grpc_schedule_on_exec_ctx); gpr_mu_init(&p->mu); return &p->base; } diff --git a/src/core/ext/lb_policy/round_robin/round_robin.c b/src/core/ext/lb_policy/round_robin/round_robin.c index f0305473d2..cb679489c3 100644 --- a/src/core/ext/lb_policy/round_robin/round_robin.c +++ b/src/core/ext/lb_policy/round_robin/round_robin.c @@ -284,7 +284,7 @@ static void rr_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { GRPC_SUBCHANNEL_UNREF(exec_ctx, sd->subchannel, "rr_destroy"); if (sd->user_data != NULL) { GPR_ASSERT(sd->user_data_vtable != NULL); - sd->user_data_vtable->destroy(sd->user_data); + sd->user_data_vtable->destroy(exec_ctx, sd->user_data); } gpr_free(sd); } @@ -321,8 +321,8 @@ static void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = NULL; - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, - GRPC_ERROR_CREATE("Channel Shutdown"), NULL); + grpc_closure_sched(exec_ctx, pp->on_complete, + GRPC_ERROR_CREATE("Channel Shutdown")); gpr_free(pp); } grpc_connectivity_state_set( @@ -348,9 +348,9 @@ static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pending_pick *next = pp->next; if (pp->target == target) { *target = NULL; - grpc_exec_ctx_sched( + grpc_closure_sched( exec_ctx, pp->on_complete, - GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1), NULL); + GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1)); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -376,9 +376,9 @@ static void rr_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, if ((pp->initial_metadata_flags & initial_metadata_flags_mask) == initial_metadata_flags_eq) { *pp->target = NULL; - grpc_exec_ctx_sched( + grpc_closure_sched( exec_ctx, pp->on_complete, - GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1), NULL); + GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1)); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -581,7 +581,7 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg, "[RR CONN CHANGED] TARGET <-- SUBCHANNEL %p (NODE %p)", (void *)selected->subchannel, (void *)selected); } - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE, NULL); + grpc_closure_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE); gpr_free(pp); } update_lb_connectivity_status(exec_ctx, sd, error); @@ -634,7 +634,7 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg, while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = NULL; - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE, NULL); + grpc_closure_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE); gpr_free(pp); } } @@ -684,8 +684,8 @@ static void rr_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, target, "rr_picked"); } else { gpr_mu_unlock(&p->mu); - grpc_exec_ctx_sched(exec_ctx, closure, - GRPC_ERROR_CREATE("Round Robin not connected"), NULL); + grpc_closure_sched(exec_ctx, closure, + GRPC_ERROR_CREATE("Round Robin not connected")); } } @@ -749,7 +749,7 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx, } ++subchannel_idx; grpc_closure_init(&sd->connectivity_changed_closure, - rr_connectivity_changed, sd); + rr_connectivity_changed, sd, grpc_schedule_on_exec_ctx); } } if (subchannel_idx == 0) { |