Merge pull request #7991 from markdroth/grpclb_resolver_changes

Track whether or not each resolved address is a gRPC-LB balancer.
pull/8139/head
Mark D. Roth 9 years ago committed by GitHub
commit f9652f2b84
  1. 38
      src/core/ext/client_config/lb_policy_factory.c
  2. 32
      src/core/ext/client_config/lb_policy_factory.h
  3. 80
      src/core/ext/client_config/resolver_result.c
  4. 83
      src/core/ext/client_config/resolver_result.h
  5. 149
      src/core/ext/lb_policy/grpclb/grpclb.c
  6. 23
      src/core/ext/lb_policy/pick_first/pick_first.c
  7. 36
      src/core/ext/lb_policy/round_robin/round_robin.c
  8. 23
      src/core/ext/resolver/dns/native/dns_resolver.c
  9. 41
      src/core/ext/resolver/sockaddr/sockaddr_resolver.c

@ -31,8 +31,46 @@
*
*/
#include <string.h>
#include <grpc/support/alloc.h>
#include "src/core/ext/client_config/lb_policy_factory.h"
grpc_lb_addresses* grpc_lb_addresses_create(size_t num_addresses) {
grpc_lb_addresses* addresses = gpr_malloc(sizeof(grpc_lb_addresses));
addresses->num_addresses = num_addresses;
const size_t addresses_size = sizeof(grpc_lb_address) * num_addresses;
addresses->addresses = gpr_malloc(addresses_size);
memset(addresses->addresses, 0, addresses_size);
return addresses;
}
void grpc_lb_addresses_set_address(grpc_lb_addresses* addresses, size_t index,
void* address, size_t address_len,
bool is_balancer, char* balancer_name,
void* user_data) {
GPR_ASSERT(index < addresses->num_addresses);
grpc_lb_address* target = &addresses->addresses[index];
memcpy(target->address.addr, address, address_len);
target->address.len = address_len;
target->is_balancer = is_balancer;
target->balancer_name = balancer_name;
target->user_data = user_data;
}
void grpc_lb_addresses_destroy(grpc_lb_addresses* addresses,
void (*user_data_destroy)(void*)) {
for (size_t i = 0; i < addresses->num_addresses; ++i) {
gpr_free(addresses->addresses[i].balancer_name);
if (user_data_destroy != NULL) {
user_data_destroy(addresses->addresses[i].user_data);
}
}
gpr_free(addresses->addresses);
gpr_free(addresses);
}
void grpc_lb_policy_factory_ref(grpc_lb_policy_factory* factory) {
factory->vtable->ref(factory);
}

@ -36,7 +36,7 @@
#include "src/core/ext/client_config/client_channel_factory.h"
#include "src/core/ext/client_config/lb_policy.h"
#include "src/core/lib/iomgr/resolve_address.h"
#include "src/core/ext/client_config/resolver_result.h"
#include "src/core/lib/iomgr/exec_ctx.h"
@ -53,13 +53,37 @@ struct grpc_lb_policy_factory {
* Those who don't will simply ignore it and will correspondingly return NULL in
* their namesake pick() output argument. */
typedef struct grpc_lb_address {
grpc_resolved_address *resolved_address;
grpc_resolved_address address;
bool is_balancer;
char *balancer_name; /* For secure naming. */
void *user_data;
} grpc_lb_address;
typedef struct grpc_lb_policy_args {
grpc_lb_address *addresses;
typedef struct grpc_lb_addresses {
size_t num_addresses;
grpc_lb_address *addresses;
} grpc_lb_addresses;
/** Returns a grpc_addresses struct with enough space for
* \a num_addresses addresses. */
grpc_lb_addresses *grpc_lb_addresses_create(size_t num_addresses);
/** Sets the value of the address at index \a index of \a addresses.
* \a address is a socket address of length \a address_len.
* Takes ownership of \a balancer_name. */
void grpc_lb_addresses_set_address(grpc_lb_addresses *addresses, size_t index,
void *address, size_t address_len,
bool is_balancer, char *balancer_name,
void *user_data);
/** Destroys \a addresses. If \a user_data_destroy is not NULL, it will
* be invoked to destroy the \a user_data field of each address. */
void grpc_lb_addresses_destroy(grpc_lb_addresses *addresses,
void (*user_data_destroy)(void *));
/** Arguments passed to LB policies. */
typedef struct grpc_lb_policy_args {
grpc_lb_addresses *addresses;
grpc_client_channel_factory *client_channel_factory;
} grpc_lb_policy_args;

@ -1,35 +1,33 @@
/*
*
* 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.
*
*/
//
// 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/ext/client_config/resolver_result.h"
@ -39,20 +37,20 @@
struct grpc_resolver_result {
gpr_refcount refs;
grpc_lb_policy *lb_policy;
grpc_lb_policy* lb_policy;
};
grpc_resolver_result *grpc_resolver_result_create() {
grpc_resolver_result *c = gpr_malloc(sizeof(*c));
grpc_resolver_result* grpc_resolver_result_create() {
grpc_resolver_result* c = gpr_malloc(sizeof(*c));
memset(c, 0, sizeof(*c));
gpr_ref_init(&c->refs, 1);
return c;
}
void grpc_resolver_result_ref(grpc_resolver_result *c) { gpr_ref(&c->refs); }
void grpc_resolver_result_ref(grpc_resolver_result* c) { gpr_ref(&c->refs); }
void grpc_resolver_result_unref(grpc_exec_ctx *exec_ctx,
grpc_resolver_result *c) {
void grpc_resolver_result_unref(grpc_exec_ctx* exec_ctx,
grpc_resolver_result* c) {
if (gpr_unref(&c->refs)) {
if (c->lb_policy != NULL) {
GRPC_LB_POLICY_UNREF(exec_ctx, c->lb_policy, "resolver_result");
@ -61,8 +59,8 @@ void grpc_resolver_result_unref(grpc_exec_ctx *exec_ctx,
}
}
void grpc_resolver_result_set_lb_policy(grpc_resolver_result *c,
grpc_lb_policy *lb_policy) {
void grpc_resolver_result_set_lb_policy(grpc_resolver_result* c,
grpc_lb_policy* lb_policy) {
GPR_ASSERT(c->lb_policy == NULL);
if (lb_policy) {
GRPC_LB_POLICY_REF(lb_policy, "resolver_result");
@ -70,6 +68,6 @@ void grpc_resolver_result_set_lb_policy(grpc_resolver_result *c,
c->lb_policy = lb_policy;
}
grpc_lb_policy *grpc_resolver_result_get_lb_policy(grpc_resolver_result *c) {
grpc_lb_policy* grpc_resolver_result_get_lb_policy(grpc_resolver_result* c) {
return c->lb_policy;
}

@ -1,52 +1,53 @@
/*
*
* 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.
*
*/
//
// 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_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H
#define GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H
#include <stdbool.h>
#include "src/core/ext/client_config/lb_policy.h"
#include "src/core/lib/iomgr/resolve_address.h"
/** Results reported from a grpc_resolver. */
/// Results reported from a grpc_resolver.
typedef struct grpc_resolver_result grpc_resolver_result;
grpc_resolver_result *grpc_resolver_result_create();
void grpc_resolver_result_ref(grpc_resolver_result *client_config);
void grpc_resolver_result_unref(grpc_exec_ctx *exec_ctx,
grpc_resolver_result *client_config);
grpc_resolver_result* grpc_resolver_result_create();
void grpc_resolver_result_ref(grpc_resolver_result* result);
void grpc_resolver_result_unref(grpc_exec_ctx* exec_ctx,
grpc_resolver_result* result);
void grpc_resolver_result_set_lb_policy(grpc_resolver_result *client_config,
grpc_lb_policy *lb_policy);
grpc_lb_policy *grpc_resolver_result_get_lb_policy(
grpc_resolver_result *client_config);
void grpc_resolver_result_set_lb_policy(grpc_resolver_result* result,
grpc_lb_policy* lb_policy);
grpc_lb_policy* grpc_resolver_result_get_lb_policy(
grpc_resolver_result* result);
#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H */

@ -107,6 +107,7 @@
#include <grpc/support/string_util.h>
#include "src/core/ext/client_config/client_channel_factory.h"
#include "src/core/ext/client_config/lb_policy_factory.h"
#include "src/core/ext/client_config/lb_policy_registry.h"
#include "src/core/ext/client_config/parse_address.h"
#include "src/core/ext/lb_policy/grpclb/grpclb.h"
@ -120,18 +121,6 @@
int grpc_lb_glb_trace = 0;
static void lb_addrs_destroy(grpc_lb_address *lb_addresses,
size_t num_addresses) {
/* free "resolved" addresses memblock */
gpr_free(lb_addresses->resolved_address);
for (size_t i = 0; i < num_addresses; ++i) {
if (lb_addresses[i].user_data != NULL) {
GRPC_MDELEM_UNREF(lb_addresses[i].user_data);
}
}
gpr_free(lb_addresses);
}
/* add lb_token of selected subchannel (address) to the call's initial
* metadata */
static void initial_metadata_add_lb_token(
@ -311,11 +300,8 @@ typedef struct glb_lb_policy {
* response has arrived. */
grpc_grpclb_serverlist *serverlist;
/** total number of valid addresses received in \a serverlist */
size_t num_ok_serverlist_addresses;
/** LB addresses from \a serverlist, \a num_ok_serverlist_addresses of them */
grpc_lb_address *lb_addresses;
/** addresses from \a serverlist */
grpc_lb_addresses *addresses;
/** list of picks that are waiting on RR's policy connectivity */
pending_pick *pending_picks;
@ -368,26 +354,18 @@ static bool is_server_valid(const grpc_grpclb_server *server, size_t idx,
return true;
}
/* populate \a addresses according to \a serverlist. Returns the number of
* addresses successfully parsed and added to \a addresses */
static size_t process_serverlist(const grpc_grpclb_serverlist *serverlist,
grpc_lb_address **lb_addresses) {
/* Returns addresses extracted from \a serverlist. */
static grpc_lb_addresses *process_serverlist(
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 */
for (size_t i = 0; i < serverlist->num_servers; ++i) {
if (is_server_valid(serverlist->servers[i], i, true)) ++num_valid;
}
if (num_valid == 0) {
return 0;
}
if (num_valid == 0) return NULL;
/* allocate the memory block for the "resolved" addresses. */
grpc_resolved_address *r_addrs_memblock =
gpr_malloc(sizeof(grpc_resolved_address) * num_valid);
memset(r_addrs_memblock, 0, sizeof(grpc_resolved_address) * num_valid);
grpc_lb_address *lb_addrs = gpr_malloc(sizeof(grpc_lb_address) * num_valid);
memset(lb_addrs, 0, sizeof(grpc_lb_address) * num_valid);
grpc_lb_addresses *lb_addresses = grpc_lb_addresses_create(num_valid);
/* second pass: actually populate the addresses and LB tokens (aka user data
* to the outside world) to be read by the RR policy during its creation.
@ -399,56 +377,58 @@ static size_t process_serverlist(const grpc_grpclb_serverlist *serverlist,
GPR_ASSERT(addr_idx < num_valid);
const grpc_grpclb_server *server = serverlist->servers[sl_idx];
if (!is_server_valid(serverlist->servers[sl_idx], sl_idx, false)) continue;
grpc_lb_address *const lb_addr = &lb_addrs[addr_idx];
/* address processing */
const uint16_t netorder_port = htons((uint16_t)server->port);
/* the addresses are given in binary format (a in(6)_addr struct) in
* server->ip_address.bytes. */
const grpc_grpclb_ip_address *ip = &server->ip_address;
lb_addr->resolved_address = &r_addrs_memblock[addr_idx];
struct sockaddr_storage *sa =
(struct sockaddr_storage *)lb_addr->resolved_address->addr;
size_t *sa_len = &lb_addr->resolved_address->len;
*sa_len = 0;
grpc_resolved_address addr;
memset(&addr, 0, sizeof(addr));
if (ip->size == 4) {
struct sockaddr_in *addr4 = (struct sockaddr_in *)sa;
*sa_len = sizeof(struct sockaddr_in);
memset(addr4, 0, *sa_len);
addr.len = sizeof(struct sockaddr_in);
struct sockaddr_in *addr4 = (struct sockaddr_in *)&addr.addr;
addr4->sin_family = AF_INET;
memcpy(&addr4->sin_addr, ip->bytes, ip->size);
addr4->sin_port = netorder_port;
} else if (ip->size == 16) {
struct sockaddr_in6 *addr6 = (struct sockaddr_in6 *)sa;
*sa_len = sizeof(struct sockaddr_in6);
memset(addr6, 0, *sa_len);
addr.len = sizeof(struct sockaddr_in6);
struct sockaddr_in6 *addr6 = (struct sockaddr_in6 *)&addr.addr;
addr6->sin6_family = AF_INET;
memcpy(&addr6->sin6_addr, ip->bytes, ip->size);
addr6->sin6_port = netorder_port;
}
GPR_ASSERT(*sa_len > 0);
/* lb token processing */
void *user_data;
if (server->has_load_balance_token) {
const size_t lb_token_size =
GPR_ARRAY_SIZE(server->load_balance_token) - 1;
grpc_mdstr *lb_token_mdstr = grpc_mdstr_from_buffer(
(uint8_t *)server->load_balance_token, lb_token_size);
lb_addr->user_data = grpc_mdelem_from_metadata_strings(
user_data = grpc_mdelem_from_metadata_strings(
GRPC_MDSTR_LOAD_REPORTING_INITIAL, lb_token_mdstr);
} else {
gpr_log(GPR_ERROR,
"Missing LB token for backend address '%s'. The empty token will "
"be used instead",
grpc_sockaddr_to_uri((struct sockaddr *)sa));
lb_addr->user_data = GRPC_MDELEM_LOAD_REPORTING_INITIAL_EMPTY;
grpc_sockaddr_to_uri((struct sockaddr *)&addr.addr));
user_data = GRPC_MDELEM_LOAD_REPORTING_INITIAL_EMPTY;
}
grpc_lb_addresses_set_address(lb_addresses, addr_idx, &addr.addr, addr.len,
false /* is_balancer */,
NULL /* balancer_name */, user_data);
++addr_idx;
}
GPR_ASSERT(addr_idx == num_valid);
*lb_addresses = lb_addrs;
return num_valid;
return lb_addresses;
}
/* A plugin for grpc_lb_addresses_destroy that unrefs the LB token metadata. */
static void lb_token_destroy(void *token) {
if (token != NULL) GRPC_MDELEM_UNREF(token);
}
static grpc_lb_policy *create_rr(grpc_exec_ctx *exec_ctx,
@ -459,19 +439,15 @@ static grpc_lb_policy *create_rr(grpc_exec_ctx *exec_ctx,
grpc_lb_policy_args args;
memset(&args, 0, sizeof(args));
args.client_channel_factory = glb_policy->cc_factory;
const size_t num_ok_addresses =
process_serverlist(serverlist, &args.addresses);
args.num_addresses = num_ok_addresses;
args.addresses = process_serverlist(serverlist);
grpc_lb_policy *rr = grpc_lb_policy_create(exec_ctx, "round_robin", &args);
if (glb_policy->lb_addresses != NULL) {
if (glb_policy->addresses != NULL) {
/* dispose of the previous version */
lb_addrs_destroy(glb_policy->lb_addresses,
glb_policy->num_ok_serverlist_addresses);
grpc_lb_addresses_destroy(glb_policy->addresses, lb_token_destroy);
}
glb_policy->num_ok_serverlist_addresses = num_ok_addresses;
glb_policy->lb_addresses = args.addresses;
glb_policy->addresses = args.addresses;
return rr;
}
@ -565,6 +541,19 @@ static void glb_rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
grpc_lb_policy_factory *factory,
grpc_lb_policy_args *args) {
/* Count the number of gRPC-LB addresses. There must be at least one.
* TODO(roth): For now, we ignore non-balancer addresses, but in the
* future, we may change the behavior such that we fall back to using
* the non-balancer addresses if we cannot reach any balancers. At that
* time, this should be changed to allow a list with no balancer addresses,
* since the resolver might fail to return a balancer address even when
* this is the right LB policy to use. */
size_t num_grpclb_addrs = 0;
for (size_t i = 0; i < args->addresses->num_addresses; ++i) {
if (args->addresses->addresses[i].is_balancer) ++num_grpclb_addrs;
}
if (num_grpclb_addrs == 0) return NULL;
glb_lb_policy *glb_policy = gpr_malloc(sizeof(*glb_policy));
memset(glb_policy, 0, sizeof(*glb_policy));
@ -576,36 +565,34 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
* Create a client channel over them to communicate with a LB service */
glb_policy->cc_factory = args->client_channel_factory;
GPR_ASSERT(glb_policy->cc_factory != NULL);
if (args->num_addresses == 0) {
return NULL;
}
if (args->addresses[0].user_data != NULL) {
gpr_log(GPR_ERROR,
"This LB policy doesn't support user data. It will be ignored");
}
/* construct a target from the addresses in args, given in the form
* ipvX://ip1:port1,ip2:port2,...
* TODO(dgq): support mixed ip version */
char **addr_strs = gpr_malloc(sizeof(char *) * args->num_addresses);
addr_strs[0] = grpc_sockaddr_to_uri(
(const struct sockaddr *)&args->addresses[0].resolved_address->addr);
for (size_t i = 1; i < args->num_addresses; i++) {
if (args->addresses[i].user_data != NULL) {
char **addr_strs = gpr_malloc(sizeof(char *) * num_grpclb_addrs);
size_t addr_index = 0;
for (size_t i = 0; i < args->addresses->num_addresses; i++) {
if (args->addresses->addresses[i].user_data != NULL) {
gpr_log(GPR_ERROR,
"This LB policy doesn't support user data. It will be ignored");
}
GPR_ASSERT(
grpc_sockaddr_to_string(
&addr_strs[i],
(const struct sockaddr *)&args->addresses[i].resolved_address->addr,
true) == 0);
if (args->addresses->addresses[i].is_balancer) {
if (addr_index == 0) {
addr_strs[addr_index++] = grpc_sockaddr_to_uri(
(const struct sockaddr *)&args->addresses->addresses[i]
.address.addr);
} else {
GPR_ASSERT(grpc_sockaddr_to_string(
&addr_strs[addr_index++],
(const struct sockaddr *)&args->addresses->addresses[i]
.address.addr,
true) == 0);
}
}
}
size_t uri_path_len;
char *target_uri_str = gpr_strjoin_sep(
(const char **)addr_strs, args->num_addresses, ",", &uri_path_len);
char *target_uri_str = gpr_strjoin_sep((const char **)addr_strs,
num_grpclb_addrs, ",", &uri_path_len);
/* will pick using pick_first */
glb_policy->lb_channel = grpc_client_channel_factory_create_channel(
@ -613,7 +600,7 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
GRPC_CLIENT_CHANNEL_TYPE_LOAD_BALANCING, NULL);
gpr_free(target_uri_str);
for (size_t i = 0; i < args->num_addresses; i++) {
for (size_t i = 0; i < num_grpclb_addrs; i++) {
gpr_free(addr_strs[i]);
}
gpr_free(addr_strs);
@ -649,9 +636,7 @@ static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
grpc_grpclb_destroy_serverlist(glb_policy->serverlist);
}
gpr_mu_destroy(&glb_policy->mu);
lb_addrs_destroy(glb_policy->lb_addresses,
glb_policy->num_ok_serverlist_addresses);
grpc_lb_addresses_destroy(glb_policy->addresses, lb_token_destroy);
gpr_free(glb_policy);
}

@ -441,25 +441,34 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(args->addresses != NULL);
GPR_ASSERT(args->client_channel_factory != NULL);
if (args->num_addresses == 0) return NULL;
/* Find the number of backend addresses. We ignore balancer
* addresses, since we don't know how to handle them. */
size_t num_addrs = 0;
for (size_t i = 0; i < args->addresses->num_addresses; i++) {
if (!args->addresses->addresses[i].is_balancer) ++num_addrs;
}
if (num_addrs == 0) return NULL;
pick_first_lb_policy *p = gpr_malloc(sizeof(*p));
memset(p, 0, sizeof(*p));
p->subchannels = gpr_malloc(sizeof(grpc_subchannel *) * args->num_addresses);
memset(p->subchannels, 0, sizeof(*p->subchannels) * args->num_addresses);
p->subchannels = gpr_malloc(sizeof(grpc_subchannel *) * num_addrs);
memset(p->subchannels, 0, sizeof(*p->subchannels) * num_addrs);
grpc_subchannel_args sc_args;
size_t subchannel_idx = 0;
for (size_t i = 0; i < args->num_addresses; i++) {
if (args->addresses[i].user_data != NULL) {
for (size_t i = 0; i < args->addresses->num_addresses; i++) {
/* Skip balancer addresses, since we only know how to handle backends. */
if (args->addresses->addresses[i].is_balancer) continue;
if (args->addresses->addresses[i].user_data != NULL) {
gpr_log(GPR_ERROR,
"This LB policy doesn't support user data. It will be ignored");
}
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
sc_args.addr =
(struct sockaddr *)(args->addresses[i].resolved_address->addr);
sc_args.addr_len = (size_t)args->addresses[i].resolved_address->len;
(struct sockaddr *)(&args->addresses->addresses[i].address.addr);
sc_args.addr_len = args->addresses->addresses[i].address.len;
grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel(
exec_ctx, args->client_channel_factory, &sc_args);

@ -130,10 +130,6 @@ struct round_robin_lb_policy {
/** total number of addresses received at creation time */
size_t num_addresses;
/** array holding the borrowed and opaque pointers to incoming user data, one
* per incoming address. These individual pointers will be returned as-is in
* successful picks. */
void **user_data_pointers;
/** all our subchannels */
size_t num_subchannels;
@ -282,7 +278,6 @@ static void rr_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
elem = tmp;
}
gpr_free(p->user_data_pointers);
gpr_free(p);
}
@ -611,25 +606,32 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
grpc_lb_policy_args *args) {
GPR_ASSERT(args->addresses != NULL);
GPR_ASSERT(args->client_channel_factory != NULL);
if (args->num_addresses == 0) return NULL;
/* Find the number of backend addresses. We ignore balancer
* addresses, since we don't know how to handle them. */
size_t num_addrs = 0;
for (size_t i = 0; i < args->addresses->num_addresses; i++) {
if (!args->addresses->addresses[i].is_balancer) ++num_addrs;
}
if (num_addrs == 0) return NULL;
round_robin_lb_policy *p = gpr_malloc(sizeof(*p));
memset(p, 0, sizeof(*p));
p->num_addresses = args->num_addresses;
p->subchannels = gpr_malloc(sizeof(subchannel_data) * p->num_addresses);
memset(p->subchannels, 0, sizeof(*p->subchannels) * p->num_addresses);
p->user_data_pointers = gpr_malloc(sizeof(void *) * p->num_addresses);
memset(p->user_data_pointers, 0, sizeof(void *) * p->num_addresses);
p->num_addresses = num_addrs;
p->subchannels = gpr_malloc(sizeof(*p->subchannels) * num_addrs);
memset(p->subchannels, 0, sizeof(*p->subchannels) * num_addrs);
grpc_subchannel_args sc_args;
size_t subchannel_idx = 0;
for (size_t i = 0; i < p->num_addresses; i++) {
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
sc_args.addr = (struct sockaddr *)args->addresses[i].resolved_address->addr;
sc_args.addr_len = args->addresses[i].resolved_address->len;
for (size_t i = 0; i < args->addresses->num_addresses; i++) {
/* Skip balancer addresses, since we only know how to handle backends. */
if (args->addresses->addresses[i].is_balancer) continue;
p->user_data_pointers[i] = args->addresses[i].user_data;
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
sc_args.addr =
(struct sockaddr *)(&args->addresses->addresses[i].address.addr);
sc_args.addr_len = args->addresses->addresses[i].address.len;
grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel(
exec_ctx, args->client_channel_factory, &sc_args);
@ -641,7 +643,7 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
sd->policy = p;
sd->index = subchannel_idx;
sd->subchannel = subchannel;
sd->user_data = p->user_data_pointers[i];
sd->user_data = args->addresses->addresses[i].user_data;
++subchannel_idx;
grpc_closure_init(&sd->connectivity_changed_closure,
rr_connectivity_changed, sd);

@ -170,28 +170,27 @@ static void dns_on_resolved(grpc_exec_ctx *exec_ctx, void *arg,
gpr_mu_lock(&r->mu);
GPR_ASSERT(r->resolving);
r->resolving = 0;
grpc_resolved_addresses *addresses = r->addresses;
if (addresses != NULL) {
if (r->addresses != NULL) {
grpc_lb_policy_args lb_policy_args;
result = grpc_resolver_result_create();
memset(&lb_policy_args, 0, sizeof(lb_policy_args));
lb_policy_args.num_addresses = addresses->naddrs;
lb_policy_args.addresses =
gpr_malloc(sizeof(grpc_lb_address) * lb_policy_args.num_addresses);
memset(lb_policy_args.addresses, 0,
sizeof(grpc_lb_address) * lb_policy_args.num_addresses);
for (size_t i = 0; i < addresses->naddrs; ++i) {
lb_policy_args.addresses[i].resolved_address = &r->addresses->addrs[i];
lb_policy_args.addresses = grpc_lb_addresses_create(r->addresses->naddrs);
for (size_t i = 0; i < r->addresses->naddrs; ++i) {
grpc_lb_addresses_set_address(
lb_policy_args.addresses, i, &r->addresses->addrs[i].addr,
r->addresses->addrs[i].len, false /* is_balancer */,
NULL /* balancer_name */, NULL /* user_data */);
}
grpc_resolved_addresses_destroy(r->addresses);
lb_policy_args.client_channel_factory = r->client_channel_factory;
lb_policy =
grpc_lb_policy_create(exec_ctx, r->lb_policy_name, &lb_policy_args);
gpr_free(lb_policy_args.addresses);
grpc_lb_addresses_destroy(lb_policy_args.addresses,
NULL /* user_data_destroy */);
result = grpc_resolver_result_create();
if (lb_policy != NULL) {
grpc_resolver_result_set_lb_policy(result, lb_policy);
GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "construction");
}
grpc_resolved_addresses_destroy(addresses);
} else {
gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
gpr_timespec next_try = gpr_backoff_step(&r->backoff_state, now);

@ -58,12 +58,12 @@ typedef struct {
char *lb_policy_name;
/** the addresses that we've 'resolved' */
grpc_resolved_addresses *addresses;
grpc_lb_addresses *addresses;
/** mutex guarding the rest of the state */
gpr_mu mu;
/** have we published? */
int published;
bool published;
/** pending next completion, or NULL */
grpc_closure *next_completion;
/** target result address for next completion */
@ -102,7 +102,7 @@ static void sockaddr_channel_saw_error(grpc_exec_ctx *exec_ctx,
grpc_resolver *resolver) {
sockaddr_resolver *r = (sockaddr_resolver *)resolver;
gpr_mu_lock(&r->mu);
r->published = 0;
r->published = false;
sockaddr_maybe_finish_next_locked(exec_ctx, r);
gpr_mu_unlock(&r->mu);
}
@ -125,21 +125,13 @@ static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
grpc_resolver_result *result = grpc_resolver_result_create();
grpc_lb_policy_args lb_policy_args;
memset(&lb_policy_args, 0, sizeof(lb_policy_args));
lb_policy_args.num_addresses = r->addresses->naddrs;
lb_policy_args.addresses =
gpr_malloc(sizeof(grpc_lb_address) * lb_policy_args.num_addresses);
memset(lb_policy_args.addresses, 0,
sizeof(grpc_lb_address) * lb_policy_args.num_addresses);
for (size_t i = 0; i < lb_policy_args.num_addresses; ++i) {
lb_policy_args.addresses[i].resolved_address = &r->addresses->addrs[i];
}
lb_policy_args.addresses = r->addresses;
lb_policy_args.client_channel_factory = r->client_channel_factory;
grpc_lb_policy *lb_policy =
grpc_lb_policy_create(exec_ctx, r->lb_policy_name, &lb_policy_args);
gpr_free(lb_policy_args.addresses);
grpc_resolver_result_set_lb_policy(result, lb_policy);
GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "sockaddr");
r->published = 1;
r->published = true;
*r->target_result = result;
grpc_exec_ctx_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE, NULL);
r->next_completion = NULL;
@ -150,7 +142,7 @@ static void sockaddr_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *gr) {
sockaddr_resolver *r = (sockaddr_resolver *)gr;
gpr_mu_destroy(&r->mu);
grpc_client_channel_factory_unref(exec_ctx, r->client_channel_factory);
grpc_resolved_addresses_destroy(r->addresses);
grpc_lb_addresses_destroy(r->addresses, NULL /* user_data_destroy */);
gpr_free(r->lb_policy_name);
gpr_free(r);
}
@ -183,7 +175,7 @@ static void do_nothing(void *ignored) {}
static grpc_resolver *sockaddr_create(
grpc_resolver_args *args, const char *default_lb_policy_name,
int parse(grpc_uri *uri, struct sockaddr_storage *dst, size_t *len)) {
int errors_found = 0; /* GPR_FALSE */
bool errors_found = false;
sockaddr_resolver *r;
gpr_slice path_slice;
gpr_slice_buffer path_parts;
@ -224,21 +216,18 @@ static grpc_resolver *sockaddr_create(
gpr_slice_buffer_init(&path_parts);
gpr_slice_split(path_slice, ",", &path_parts);
r->addresses = gpr_malloc(sizeof(grpc_resolved_addresses));
r->addresses->naddrs = path_parts.count;
r->addresses->addrs =
gpr_malloc(sizeof(grpc_resolved_address) * r->addresses->naddrs);
for (size_t i = 0; i < r->addresses->naddrs; i++) {
r->addresses = grpc_lb_addresses_create(path_parts.count);
for (size_t i = 0; i < r->addresses->num_addresses; i++) {
grpc_uri ith_uri = *args->uri;
char *part_str = gpr_dump_slice(path_parts.slices[i], GPR_DUMP_ASCII);
ith_uri.path = part_str;
if (!parse(&ith_uri,
(struct sockaddr_storage *)(&r->addresses->addrs[i].addr),
&r->addresses->addrs[i].len)) {
errors_found = 1; /* GPR_TRUE */
if (!parse(&ith_uri, (struct sockaddr_storage *)(&r->addresses->addresses[i]
.address.addr),
&r->addresses->addresses[i].address.len)) {
errors_found = true;
}
gpr_free(part_str);
r->addresses->addresses[i].is_balancer = lb_enabled;
if (errors_found) break;
}
@ -246,7 +235,7 @@ static grpc_resolver *sockaddr_create(
gpr_slice_unref(path_slice);
if (errors_found) {
gpr_free(r->lb_policy_name);
grpc_resolved_addresses_destroy(r->addresses);
grpc_lb_addresses_destroy(r->addresses, NULL /* user_data_destroy */);
gpr_free(r);
return NULL;
}

Loading…
Cancel
Save