diff options
Diffstat (limited to 'src/core/client_config')
-rw-r--r-- | src/core/client_config/lb_policies/round_robin.c | 554 | ||||
-rw-r--r-- | src/core/client_config/lb_policies/round_robin.h | 46 | ||||
-rw-r--r-- | src/core/client_config/resolvers/sockaddr_resolver.c | 26 |
3 files changed, 623 insertions, 3 deletions
diff --git a/src/core/client_config/lb_policies/round_robin.c b/src/core/client_config/lb_policies/round_robin.c new file mode 100644 index 0000000000..a346007aca --- /dev/null +++ b/src/core/client_config/lb_policies/round_robin.c @@ -0,0 +1,554 @@ +/* + * + * Copyright 2015, 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. + * + */ + +#include "src/core/client_config/lb_policies/round_robin.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include "src/core/transport/connectivity_state.h" + +int grpc_lb_round_robin_trace = 0; + +/** List of entities waiting for a pick. + * + * Once a pick is available, \a target is updated and \a on_complete called. */ +typedef struct pending_pick { + struct pending_pick *next; + grpc_pollset *pollset; + grpc_subchannel **target; + grpc_closure *on_complete; +} pending_pick; + +/** List of subchannels in a connectivity READY state */ +typedef struct ready_list { + grpc_subchannel *subchannel; + struct ready_list *next; + struct ready_list *prev; +} ready_list; + +typedef struct { + size_t subchannel_idx; /**< Index over p->subchannels */ + void *p; /**< round_robin_lb_policy instance */ +} connectivity_changed_cb_arg; + +typedef struct { + /** base policy: must be first */ + grpc_lb_policy base; + + /** all our subchannels */ + grpc_subchannel **subchannels; + size_t num_subchannels; + + /** Callbacks, one per subchannel being watched, to be called when their + * respective connectivity changes */ + grpc_closure *connectivity_changed_cbs; + connectivity_changed_cb_arg *cb_args; + + /** mutex protecting remaining members */ + gpr_mu mu; + /** have we started picking? */ + int started_picking; + /** are we shutting down? */ + int shutdown; + /** Connectivity state of the subchannels being watched */ + grpc_connectivity_state *subchannel_connectivity; + /** List of picks that are waiting on connectivity */ + pending_pick *pending_picks; + + /** our connectivity state tracker */ + grpc_connectivity_state_tracker state_tracker; + + /** (Dummy) root of the doubly linked list containing READY subchannels */ + ready_list ready_list; + /** Last pick from the ready list. */ + ready_list *ready_list_last_pick; + + /** Subchannel index to ready_list node. + * + * Kept in order to remove nodes from the ready list associated with a + * subchannel */ + ready_list **subchannel_index_to_readylist_node; +} round_robin_lb_policy; + +/** Returns the next subchannel from the connected list or NULL if the list is + * empty. + * + * Note that this function does *not* advance p->ready_list_last_pick. Use \a + * advance_last_picked_locked() for that. */ +static ready_list *peek_next_connected_locked(const round_robin_lb_policy *p) { + ready_list *selected; + selected = p->ready_list_last_pick->next; + + while (selected != NULL) { + if (selected == &p->ready_list) { + GPR_ASSERT(selected->subchannel == NULL); + /* skip dummy root */ + selected = selected->next; + } else { + GPR_ASSERT(selected->subchannel != NULL); + return selected; + } + } + return NULL; +} + +/** Advance the \a ready_list picking head. */ +static void advance_last_picked_locked(round_robin_lb_policy *p) { + if (p->ready_list_last_pick->next != NULL) { /* non-empty list */ + p->ready_list_last_pick = p->ready_list_last_pick->next; + if (p->ready_list_last_pick == &p->ready_list) { + /* skip dummy root */ + p->ready_list_last_pick = p->ready_list_last_pick->next; + } + } else { /* should be an empty list */ + GPR_ASSERT(p->ready_list_last_pick == &p->ready_list); + } + + if (grpc_lb_round_robin_trace) { + gpr_log(GPR_DEBUG, "[READYLIST] ADVANCED LAST PICK. NOW AT NODE %p (SC %p)", + p->ready_list_last_pick, p->ready_list_last_pick->subchannel); + } +} + +/** Prepends (relative to the root at p->ready_list) the connected subchannel \a + * csc to the list of ready subchannels. */ +static ready_list *add_connected_sc_locked(round_robin_lb_policy *p, + grpc_subchannel *csc) { + ready_list *new_elem = gpr_malloc(sizeof(ready_list)); + new_elem->subchannel = csc; + if (p->ready_list.prev == NULL) { + /* first element */ + new_elem->next = &p->ready_list; + new_elem->prev = &p->ready_list; + p->ready_list.next = new_elem; + p->ready_list.prev = new_elem; + } else { + new_elem->next = &p->ready_list; + new_elem->prev = p->ready_list.prev; + p->ready_list.prev->next = new_elem; + p->ready_list.prev = new_elem; + } + if (grpc_lb_round_robin_trace) { + gpr_log(GPR_DEBUG, "[READYLIST] ADDING NODE %p (SC %p)", new_elem, csc); + } + return new_elem; +} + +/** Removes \a node from the list of connected subchannels */ +static void remove_disconnected_sc_locked(round_robin_lb_policy *p, + ready_list *node) { + if (node == NULL) { + return; + } + if (node == p->ready_list_last_pick) { + /* If removing the lastly picked node, reset the last pick pointer to the + * dummy root of the list */ + p->ready_list_last_pick = &p->ready_list; + } + + /* removing last item */ + if (node->next == &p->ready_list && node->prev == &p->ready_list) { + GPR_ASSERT(p->ready_list.next == node); + GPR_ASSERT(p->ready_list.prev == node); + p->ready_list.next = NULL; + p->ready_list.prev = NULL; + } else { + node->prev->next = node->next; + node->next->prev = node->prev; + } + + if (grpc_lb_round_robin_trace) { + gpr_log(GPR_DEBUG, "[READYLIST] REMOVED NODE %p (SC %p)", node, + node->subchannel); + } + + node->next = NULL; + node->prev = NULL; + node->subchannel = NULL; + + gpr_free(node); +} + +static void del_interested_parties_locked(round_robin_lb_policy *p, + const size_t subchannel_idx, + grpc_call_list *call_list) { + pending_pick *pp; + for (pp = p->pending_picks; pp; pp = pp->next) { + grpc_subchannel_del_interested_party(p->subchannels[subchannel_idx], + pp->pollset, call_list); + } +} + +void rr_destroy(grpc_lb_policy *pol, grpc_call_list *call_list) { + round_robin_lb_policy *p = (round_robin_lb_policy *)pol; + size_t i; + ready_list *elem; + for (i = 0; i < p->num_subchannels; i++) { + del_interested_parties_locked(p, i, call_list); + } + for (i = 0; i < p->num_subchannels; i++) { + GRPC_SUBCHANNEL_UNREF(p->subchannels[i], "round_robin", call_list); + } + gpr_free(p->connectivity_changed_cbs); + gpr_free(p->subchannel_connectivity); + + grpc_connectivity_state_destroy(&p->state_tracker, call_list); + gpr_free(p->subchannels); + gpr_mu_destroy(&p->mu); + + elem = p->ready_list.next; + while (elem != NULL && elem != &p->ready_list) { + ready_list *tmp; + tmp = elem->next; + elem->next = NULL; + elem->prev = NULL; + elem->subchannel = NULL; + gpr_free(elem); + elem = tmp; + } + gpr_free(p->subchannel_index_to_readylist_node); + gpr_free(p->cb_args); + gpr_free(p); +} + +void rr_shutdown(grpc_lb_policy *pol, grpc_call_list *call_list) { + size_t i; + round_robin_lb_policy *p = (round_robin_lb_policy *)pol; + pending_pick *pp; + gpr_mu_lock(&p->mu); + + for (i = 0; i < p->num_subchannels; i++) { + del_interested_parties_locked(p, i, call_list); + } + + p->shutdown = 1; + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = NULL; + grpc_call_list_add(call_list, pp->on_complete, 0); + gpr_free(pp); + } + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_FATAL_FAILURE, + "shutdown", call_list); + gpr_mu_unlock(&p->mu); +} + +static void start_picking(round_robin_lb_policy *p, grpc_call_list *call_list) { + size_t i; + p->started_picking = 1; + + for (i = 0; i < p->num_subchannels; i++) { + p->subchannel_connectivity[i] = GRPC_CHANNEL_IDLE; + grpc_subchannel_notify_on_state_change( + p->subchannels[i], &p->subchannel_connectivity[i], + &p->connectivity_changed_cbs[i], call_list); + GRPC_LB_POLICY_REF(&p->base, "round_robin_connectivity"); + } +} + +void rr_exit_idle(grpc_lb_policy *pol, grpc_call_list *call_list) { + round_robin_lb_policy *p = (round_robin_lb_policy *)pol; + gpr_mu_lock(&p->mu); + if (!p->started_picking) { + start_picking(p, call_list); + } + gpr_mu_unlock(&p->mu); +} + +void rr_pick(grpc_lb_policy *pol, grpc_pollset *pollset, + grpc_metadata_batch *initial_metadata, grpc_subchannel **target, + grpc_closure *on_complete, grpc_call_list *call_list) { + size_t i; + round_robin_lb_policy *p = (round_robin_lb_policy *)pol; + pending_pick *pp; + ready_list *selected; + gpr_mu_lock(&p->mu); + if ((selected = peek_next_connected_locked(p))) { + gpr_mu_unlock(&p->mu); + *target = selected->subchannel; + if (grpc_lb_round_robin_trace) { + gpr_log(GPR_DEBUG, "[RR PICK] TARGET <-- SUBCHANNEL %p (NODE %p)", + selected->subchannel, selected); + } + /* only advance the last picked pointer if the selection was used */ + advance_last_picked_locked(p); + on_complete->cb(on_complete->cb_arg, 1, call_list); + } else { + if (!p->started_picking) { + start_picking(p, call_list); + } + for (i = 0; i < p->num_subchannels; i++) { + grpc_subchannel_add_interested_party(p->subchannels[i], pollset, + call_list); + } + pp = gpr_malloc(sizeof(*pp)); + pp->next = p->pending_picks; + pp->pollset = pollset; + pp->target = target; + pp->on_complete = on_complete; + p->pending_picks = pp; + gpr_mu_unlock(&p->mu); + } +} + +static void rr_connectivity_changed(void *arg, int iomgr_success, + grpc_call_list *call_list) { + connectivity_changed_cb_arg *cb_arg = arg; + round_robin_lb_policy *p = cb_arg->p; + /* index over p->subchannels of this cb's subchannel */ + const size_t this_idx = cb_arg->subchannel_idx; + pending_pick *pp; + ready_list *selected; + + int unref = 0; + + /* connectivity state of this cb's subchannel */ + grpc_connectivity_state *this_connectivity; + + gpr_mu_lock(&p->mu); + + this_connectivity = &p->subchannel_connectivity[this_idx]; + + if (p->shutdown) { + unref = 1; + } else { + switch (*this_connectivity) { + case GRPC_CHANNEL_READY: + grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_READY, + "connecting_ready", call_list); + /* add the newly connected subchannel to the list of connected ones. + * Note that it goes to the "end of the line". */ + p->subchannel_index_to_readylist_node[this_idx] = + add_connected_sc_locked(p, p->subchannels[this_idx]); + /* at this point we know there's at least one suitable subchannel. Go + * ahead and pick one and notify the pending suitors in + * p->pending_picks. This preemtively replicates rr_pick()'s actions. */ + selected = peek_next_connected_locked(p); + if (p->pending_picks != NULL) { + /* if the selected subchannel is going to be used for the pending + * picks, update the last picked pointer */ + advance_last_picked_locked(p); + } + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = selected->subchannel; + if (grpc_lb_round_robin_trace) { + gpr_log(GPR_DEBUG, + "[RR CONN CHANGED] TARGET <-- SUBCHANNEL %p (NODE %p)", + selected->subchannel, selected); + } + grpc_subchannel_del_interested_party(selected->subchannel, + pp->pollset, call_list); + grpc_call_list_add(call_list, pp->on_complete, 1); + gpr_free(pp); + } + grpc_subchannel_notify_on_state_change( + p->subchannels[this_idx], this_connectivity, + &p->connectivity_changed_cbs[this_idx], call_list); + break; + case GRPC_CHANNEL_CONNECTING: + case GRPC_CHANNEL_IDLE: + grpc_connectivity_state_set(&p->state_tracker, *this_connectivity, + "connecting_changed", call_list); + grpc_subchannel_notify_on_state_change( + p->subchannels[this_idx], this_connectivity, + &p->connectivity_changed_cbs[this_idx], call_list); + break; + case GRPC_CHANNEL_TRANSIENT_FAILURE: + del_interested_parties_locked(p, this_idx, call_list); + /* renew state notification */ + grpc_subchannel_notify_on_state_change( + p->subchannels[this_idx], this_connectivity, + &p->connectivity_changed_cbs[this_idx], call_list); + + /* remove from ready list if still present */ + if (p->subchannel_index_to_readylist_node[this_idx] != NULL) { + remove_disconnected_sc_locked( + p, p->subchannel_index_to_readylist_node[this_idx]); + p->subchannel_index_to_readylist_node[this_idx] = NULL; + } + grpc_connectivity_state_set(&p->state_tracker, + GRPC_CHANNEL_TRANSIENT_FAILURE, + "connecting_transient_failure", call_list); + break; + case GRPC_CHANNEL_FATAL_FAILURE: + del_interested_parties_locked(p, this_idx, call_list); + if (p->subchannel_index_to_readylist_node[this_idx] != NULL) { + remove_disconnected_sc_locked( + p, p->subchannel_index_to_readylist_node[this_idx]); + p->subchannel_index_to_readylist_node[this_idx] = NULL; + } + + GPR_SWAP(grpc_subchannel *, p->subchannels[this_idx], + p->subchannels[p->num_subchannels - 1]); + p->num_subchannels--; + GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "round_robin", + call_list); + + if (p->num_subchannels == 0) { + grpc_connectivity_state_set(&p->state_tracker, + GRPC_CHANNEL_FATAL_FAILURE, + "no_more_channels", call_list); + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = NULL; + grpc_call_list_add(call_list, pp->on_complete, 1); + gpr_free(pp); + } + unref = 1; + } else { + grpc_connectivity_state_set(&p->state_tracker, + GRPC_CHANNEL_TRANSIENT_FAILURE, + "subchannel_failed", call_list); + } + } /* switch */ + } /* !unref */ + + gpr_mu_unlock(&p->mu); + + if (unref) { + GRPC_LB_POLICY_UNREF(&p->base, "round_robin_connectivity", call_list); + } +} + +static void rr_broadcast(grpc_lb_policy *pol, grpc_transport_op *op, + grpc_call_list *call_list) { + round_robin_lb_policy *p = (round_robin_lb_policy *)pol; + size_t i; + size_t n; + grpc_subchannel **subchannels; + + gpr_mu_lock(&p->mu); + n = p->num_subchannels; + subchannels = gpr_malloc(n * sizeof(*subchannels)); + for (i = 0; i < n; i++) { + subchannels[i] = p->subchannels[i]; + GRPC_SUBCHANNEL_REF(subchannels[i], "rr_broadcast"); + } + gpr_mu_unlock(&p->mu); + + for (i = 0; i < n; i++) { + grpc_subchannel_process_transport_op(subchannels[i], op, call_list); + GRPC_SUBCHANNEL_UNREF(subchannels[i], "rr_broadcast", call_list); + } + gpr_free(subchannels); +} + +static grpc_connectivity_state rr_check_connectivity( + grpc_lb_policy *pol, grpc_call_list *call_list) { + round_robin_lb_policy *p = (round_robin_lb_policy *)pol; + grpc_connectivity_state st; + gpr_mu_lock(&p->mu); + st = grpc_connectivity_state_check(&p->state_tracker); + gpr_mu_unlock(&p->mu); + return st; +} + +static void rr_notify_on_state_change(grpc_lb_policy *pol, + grpc_connectivity_state *current, + grpc_closure *notify, + grpc_call_list *call_list) { + round_robin_lb_policy *p = (round_robin_lb_policy *)pol; + gpr_mu_lock(&p->mu); + grpc_connectivity_state_notify_on_state_change(&p->state_tracker, current, + notify, call_list); + gpr_mu_unlock(&p->mu); +} + +static const grpc_lb_policy_vtable round_robin_lb_policy_vtable = { + rr_destroy, + rr_shutdown, + rr_pick, + rr_exit_idle, + rr_broadcast, + rr_check_connectivity, + rr_notify_on_state_change}; + +static void round_robin_factory_ref(grpc_lb_policy_factory *factory) {} + +static void round_robin_factory_unref(grpc_lb_policy_factory *factory) {} + +static grpc_lb_policy *create_round_robin(grpc_lb_policy_factory *factory, + grpc_lb_policy_args *args) { + size_t i; + round_robin_lb_policy *p = gpr_malloc(sizeof(*p)); + GPR_ASSERT(args->num_subchannels > 0); + memset(p, 0, sizeof(*p)); + grpc_lb_policy_init(&p->base, &round_robin_lb_policy_vtable); + p->subchannels = + gpr_malloc(sizeof(grpc_subchannel *) * args->num_subchannels); + p->num_subchannels = args->num_subchannels; + grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE, + "round_robin"); + memcpy(p->subchannels, args->subchannels, + sizeof(grpc_subchannel *) * args->num_subchannels); + + gpr_mu_init(&p->mu); + p->connectivity_changed_cbs = + gpr_malloc(sizeof(grpc_closure) * args->num_subchannels); + p->subchannel_connectivity = + gpr_malloc(sizeof(grpc_connectivity_state) * args->num_subchannels); + + p->cb_args = + gpr_malloc(sizeof(connectivity_changed_cb_arg) * args->num_subchannels); + for (i = 0; i < args->num_subchannels; i++) { + p->cb_args[i].subchannel_idx = i; + p->cb_args[i].p = p; + grpc_closure_init(&p->connectivity_changed_cbs[i], rr_connectivity_changed, + &p->cb_args[i]); + } + + /* The (dummy node) root of the ready list */ + p->ready_list.subchannel = NULL; + p->ready_list.prev = NULL; + p->ready_list.next = NULL; + p->ready_list_last_pick = &p->ready_list; + + p->subchannel_index_to_readylist_node = + gpr_malloc(sizeof(grpc_subchannel *) * args->num_subchannels); + memset(p->subchannel_index_to_readylist_node, 0, + sizeof(grpc_subchannel *) * args->num_subchannels); + return &p->base; +} + +static const grpc_lb_policy_factory_vtable round_robin_factory_vtable = { + round_robin_factory_ref, round_robin_factory_unref, create_round_robin, + "round_robin"}; + +static grpc_lb_policy_factory round_robin_lb_policy_factory = { + &round_robin_factory_vtable}; + +grpc_lb_policy_factory *grpc_round_robin_lb_factory_create() { + return &round_robin_lb_policy_factory; +} diff --git a/src/core/client_config/lb_policies/round_robin.h b/src/core/client_config/lb_policies/round_robin.h new file mode 100644 index 0000000000..cf1f69c85f --- /dev/null +++ b/src/core/client_config/lb_policies/round_robin.h @@ -0,0 +1,46 @@ +/* + * + * Copyright 2015, 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. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_ROUND_ROBIN_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_ROUND_ROBIN_H + +#include "src/core/client_config/lb_policy.h" + +extern int grpc_lb_round_robin_trace; + +#include "src/core/client_config/lb_policy_factory.h" + +/** Returns a load balancing factory for the round robin policy */ +grpc_lb_policy_factory *grpc_round_robin_lb_factory_create(); + +#endif diff --git a/src/core/client_config/resolvers/sockaddr_resolver.c b/src/core/client_config/resolvers/sockaddr_resolver.c index 2729036d4f..f0e2076244 100644 --- a/src/core/client_config/resolvers/sockaddr_resolver.c +++ b/src/core/client_config/resolvers/sockaddr_resolver.c @@ -284,7 +284,7 @@ done: static void do_nothing(void *ignored) {} static grpc_resolver *sockaddr_create( - grpc_resolver_args *args, const char *lb_policy_name, + grpc_resolver_args *args, const char *default_lb_policy_name, int parse(grpc_uri *uri, struct sockaddr_storage *dst, size_t *len)) { size_t i; int errors_found = 0; /* GPR_FALSE */ @@ -293,13 +293,34 @@ static grpc_resolver *sockaddr_create( gpr_slice_buffer path_parts; if (0 != strcmp(args->uri->authority, "")) { - gpr_log(GPR_ERROR, "authority based uri's not supported"); + gpr_log(GPR_ERROR, "authority based uri's not supported by the %s scheme", + args->uri->scheme); return NULL; } r = gpr_malloc(sizeof(sockaddr_resolver)); memset(r, 0, sizeof(*r)); + r->lb_policy_name = NULL; + if (0 != strcmp(args->uri->query, "")) { + gpr_slice query_slice; + gpr_slice_buffer query_parts; + + query_slice = + gpr_slice_new(args->uri->query, strlen(args->uri->query), do_nothing); + gpr_slice_buffer_init(&query_parts); + gpr_slice_split(query_slice, "=", &query_parts); + GPR_ASSERT(query_parts.count == 2); + if (0 == gpr_slice_str_cmp(query_parts.slices[0], "lb_policy")) { + r->lb_policy_name = gpr_dump_slice(query_parts.slices[1], GPR_DUMP_ASCII); + } + gpr_slice_buffer_destroy(&query_parts); + gpr_slice_unref(query_slice); + } + if (r->lb_policy_name == NULL) { + r->lb_policy_name = gpr_strdup(default_lb_policy_name); + } + path_slice = gpr_slice_new(args->uri->path, strlen(args->uri->path), do_nothing); gpr_slice_buffer_init(&path_parts); @@ -332,7 +353,6 @@ static grpc_resolver *sockaddr_create( grpc_resolver_init(&r->base, &sockaddr_resolver_vtable); r->subchannel_factory = args->subchannel_factory; grpc_subchannel_factory_ref(r->subchannel_factory); - r->lb_policy_name = gpr_strdup(lb_policy_name); return &r->base; } |