Move is_balancer into a new struct in the client_config directory.

pull/7991/head
Mark D. Roth 9 years ago
parent f168234310
commit e011b1e4ca
  1. 4
      src/core/ext/client_config/lb_policy_factory.h
  2. 24
      src/core/ext/client_config/resolver_result.c
  3. 24
      src/core/ext/client_config/resolver_result.h
  4. 51
      src/core/ext/lb_policy/grpclb/grpclb.c
  5. 19
      src/core/ext/lb_policy/pick_first/pick_first.c
  6. 19
      src/core/ext/lb_policy/round_robin/round_robin.c
  7. 15
      src/core/ext/resolver/dns/native/dns_resolver.c
  8. 22
      src/core/ext/resolver/sockaddr/sockaddr_resolver.c
  9. 2
      src/core/lib/iomgr/resolve_address.h
  10. 1
      src/core/lib/iomgr/resolve_address_posix.c
  11. 1
      src/core/lib/iomgr/resolve_address_windows.c
  12. 1
      src/core/lib/iomgr/unix_sockets_posix.c

@ -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"
@ -48,7 +48,7 @@ struct grpc_lb_policy_factory {
};
typedef struct grpc_lb_policy_args {
grpc_resolved_addresses *addresses;
grpc_addresses *addresses;
grpc_client_channel_factory *client_channel_factory;
} grpc_lb_policy_args;

@ -37,6 +37,30 @@
#include <grpc/support/alloc.h>
grpc_addresses *grpc_addresses_create(size_t num_addresses) {
grpc_addresses *addresses = gpr_malloc(sizeof(grpc_addresses));
addresses->num_addresses = num_addresses;
const size_t addresses_size = sizeof(grpc_address) * num_addresses;
addresses->addresses = gpr_malloc(addresses_size);
memset(addresses->addresses, 0, addresses_size);
return addresses;
}
void grpc_addresses_set_address(grpc_addresses *addresses, size_t index,
void *address, size_t address_len,
bool is_balancer) {
GPR_ASSERT(index < addresses->num_addresses);
grpc_address *target = &addresses->addresses[index];
memcpy(target->address.addr, address, address_len);
target->address.len = address_len;
target->is_balancer = is_balancer;
}
void grpc_addresses_destroy(grpc_addresses *addresses) {
gpr_free(addresses->addresses);
gpr_free(addresses);
}
struct grpc_resolver_result {
gpr_refcount refs;
grpc_lb_policy *lb_policy;

@ -34,7 +34,31 @@
#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"
/** Used to represent addresses returned by the resolver. */
typedef struct grpc_address {
grpc_resolved_address address;
bool is_balancer;
} grpc_address;
typedef struct grpc_addresses {
size_t num_addresses;
grpc_address *addresses;
} grpc_addresses;
/** Returns a grpc_addresses struct with enough space for
\a num_addresses addresses. */
grpc_addresses *grpc_addresses_create(size_t num_addresses);
void grpc_addresses_set_address(grpc_addresses *addresses, size_t index,
void *address, size_t address_len,
bool is_balancer);
void grpc_addresses_destroy(grpc_addresses *addresses);
/** Results reported from a grpc_resolver. */
typedef struct grpc_resolver_result grpc_resolver_result;

@ -296,10 +296,7 @@ static grpc_lb_policy *create_rr(grpc_exec_ctx *exec_ctx,
grpc_lb_policy_args args;
args.client_channel_factory = glb_policy->cc_factory;
args.addresses = gpr_malloc(sizeof(grpc_resolved_addresses));
args.addresses->naddrs = serverlist->num_servers;
args.addresses->addrs =
gpr_malloc(sizeof(grpc_resolved_address) * args.addresses->naddrs);
args.addresses = grpc_addresses_create(serverlist->num_servers);
size_t out_addrs_idx = 0;
for (size_t i = 0; i < serverlist->num_servers; ++i) {
grpc_uri uri;
@ -307,13 +304,12 @@ static grpc_lb_policy *create_rr(grpc_exec_ctx *exec_ctx,
size_t sa_len;
uri.path = host_ports[i];
if (parse_ipv4(&uri, &sa, &sa_len)) { /* TODO(dgq): add support for ipv6 */
memcpy(args.addresses->addrs[out_addrs_idx].addr, &sa, sa_len);
args.addresses->addrs[out_addrs_idx].len = sa_len;
// These are, of course, actually balancer addresses. However, we
// want the round_robin LB policy to treat them as normal backend
// addresses, since we don't need to talk to balancers in order to
// find the balancers themselves.
args.addresses->addrs[out_addrs_idx].is_balancer = false;
/* These are, of course, actually balancer addresses. However, we
* want the round_robin LB policy to treat them as normal backend
* addresses, since we don't need to talk to balancers in order to
* find the balancers themselves, so we set is_balancer=false. */
grpc_addresses_set_address(args.addresses, out_addrs_idx, &sa, sa_len,
false /* is_balancer */);
++out_addrs_idx;
} else {
gpr_log(GPR_ERROR, "Invalid LB service address '%s', ignoring.",
@ -328,8 +324,7 @@ static grpc_lb_policy *create_rr(grpc_exec_ctx *exec_ctx,
gpr_free(host_ports[i]);
}
gpr_free(host_ports);
gpr_free(args.addresses->addrs);
gpr_free(args.addresses);
grpc_addresses_destroy(args.addresses);
return rr;
}
@ -419,17 +414,16 @@ static void 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, so there must be
// at least one balancer address. 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.
/* 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->naddrs; ++i) {
if (args->addresses->addrs[i].is_balancer) ++num_grpclb_addrs;
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;
@ -448,14 +442,15 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
* ipvX://ip1:port1,ip2:port2,...
* TODO(dgq): support mixed ip version */
char **addr_strs = gpr_malloc(sizeof(char *) * num_grpclb_addrs);
addr_strs[0] =
grpc_sockaddr_to_uri((const struct sockaddr *)&args->addresses->addrs[0]);
addr_strs[0] = grpc_sockaddr_to_uri(
(const struct sockaddr *)&args->addresses->addresses[0].address.addr);
size_t addr_index = 1;
for (size_t i = 1; i < args->addresses->naddrs; i++) {
if (args->addresses->addrs[i].is_balancer) {
for (size_t i = 1; i < args->addresses->num_addresses; i++) {
if (args->addresses->addresses[i].is_balancer) {
GPR_ASSERT(grpc_sockaddr_to_string(
&addr_strs[addr_index++],
(const struct sockaddr *)&args->addresses->addrs[i],
(const struct sockaddr *)&args->addresses->addresses[i]
.address.addr,
true) == 0);
}
}

@ -443,11 +443,11 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(args->addresses != NULL);
GPR_ASSERT(args->client_channel_factory != NULL);
// Find the number of backend addresses. We ignore balancer
// addresses, since we don't know how to handle them.
/* 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->naddrs; i++) {
if (!args->addresses->addrs[i].is_balancer) ++num_addrs;
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;
@ -458,13 +458,14 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
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->addresses->naddrs; i++) {
// Skip balancer addresses, since we only know how to handle backends.
if (args->addresses->addrs[i].is_balancer) continue;
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;
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
sc_args.addr = (struct sockaddr *)(args->addresses->addrs[i].addr);
sc_args.addr_len = (size_t)args->addresses->addrs[i].len;
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);

@ -571,11 +571,11 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(args->addresses != NULL);
GPR_ASSERT(args->client_channel_factory != NULL);
// Find the number of backend addresses. We ignore balancer
// addresses, since we don't know how to handle them.
/* 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->naddrs; i++) {
if (!args->addresses->addrs[i].is_balancer) ++num_addrs;
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;
@ -587,13 +587,14 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
grpc_subchannel_args sc_args;
size_t subchannel_idx = 0;
for (size_t i = 0; i < args->addresses->naddrs; i++) {
// Skip balancer addresses, since we only know how to handle backends.
if (args->addresses->addrs[i].is_balancer) continue;
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;
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
sc_args.addr = (struct sockaddr *)(args->addresses->addrs[i].addr);
sc_args.addr_len = (size_t)args->addresses->addrs[i].len;
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);

@ -170,20 +170,25 @@ 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.addresses = addresses;
lb_policy_args.addresses = grpc_addresses_create(r->addresses->naddrs);
for (size_t i = 0; i < r->addresses->naddrs; ++i) {
grpc_addresses_set_address(
lb_policy_args.addresses, i, &r->addresses->addrs[i].addr,
r->addresses->addrs[i].len, false /* is_balancer */);
}
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);
grpc_addresses_destroy(lb_policy_args.addresses);
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,7 +58,7 @@ typedef struct {
char *lb_policy_name;
/** the addresses that we've 'resolved' */
grpc_resolved_addresses *addresses;
grpc_addresses *addresses;
/** mutex guarding the rest of the state */
gpr_mu mu;
@ -142,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_addresses_destroy(r->addresses);
gpr_free(r->lb_policy_name);
gpr_free(r);
}
@ -216,22 +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_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)) {
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->addrs[i].is_balancer = lb_enabled;
r->addresses->addresses[i].is_balancer = lb_enabled;
if (errors_found) break;
}
@ -239,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_addresses_destroy(r->addresses);
gpr_free(r);
return NULL;
}

@ -34,7 +34,6 @@
#ifndef GRPC_CORE_LIB_IOMGR_RESOLVE_ADDRESS_H
#define GRPC_CORE_LIB_IOMGR_RESOLVE_ADDRESS_H
#include <stdbool.h>
#include <stddef.h>
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/iomgr.h"
@ -44,7 +43,6 @@
typedef struct {
char addr[GRPC_MAX_SOCKADDR_SIZE];
size_t len;
bool is_balancer;
} grpc_resolved_address;
typedef struct {

@ -132,7 +132,6 @@ static grpc_error *blocking_resolve_address_impl(
for (resp = result; resp != NULL; resp = resp->ai_next) {
memcpy(&(*addresses)->addrs[i].addr, resp->ai_addr, resp->ai_addrlen);
(*addresses)->addrs[i].len = resp->ai_addrlen;
(*addresses)->addrs[i].is_balancer = false;
i++;
}
err = GRPC_ERROR_NONE;

@ -118,7 +118,6 @@ static grpc_error *blocking_resolve_address_impl(
for (resp = result; resp != NULL; resp = resp->ai_next) {
memcpy(&(*addresses)->addrs[i].addr, resp->ai_addr, resp->ai_addrlen);
(*addresses)->addrs[i].len = resp->ai_addrlen;
(*addresses)->addrs[i].is_balancer = false;
i++;
}

@ -58,7 +58,6 @@ grpc_error *grpc_resolve_unix_domain_address(const char *name,
un->sun_family = AF_UNIX;
strcpy(un->sun_path, name);
(*addrs)->addrs->len = strlen(un->sun_path) + sizeof(un->sun_family) + 1;
(*addrs)->addrs->is_balancer = false;
return GRPC_ERROR_NONE;
}

Loading…
Cancel
Save