Changes to resolver-lb interfaces.

- Creation of a load balancer now takes an execution context.
- load balancers are now passed addresses as arguments and are responsible for
  the creation of the corresponding subchannels, something that used to be done
  by the resolvers.

src/core/client_config/lb_policies/pick_first.c

@@ -385,19 +385,42 @@ static void pick_first_factory_ref(grpc_lb_policy_factory *factory) {}
 
 static void pick_first_factory_unref(grpc_lb_policy_factory *factory) {}
 
-static grpc_lb_policy *create_pick_first(grpc_lb_policy_factory *factory,
+static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
+                                         grpc_lb_policy_factory *factory,
                                          grpc_lb_policy_args *args) {
-  if (args->num_subchannels == 0) return NULL;
+  GPR_ASSERT(args->addresses != NULL);
+  GPR_ASSERT(args->subchannel_factory != NULL);
+
+  if (args->addresses->naddrs == 0) return NULL;
+
   pick_first_lb_policy *p = gpr_malloc(sizeof(*p));
   memset(p, 0, sizeof(*p));
-  grpc_lb_policy_init(&p->base, &pick_first_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,
-                               "pick_first");
-  memcpy(p->subchannels, args->subchannels,
-         sizeof(grpc_subchannel *) * args->num_subchannels);
+      gpr_malloc(sizeof(grpc_subchannel *) * args->addresses->naddrs);
+  memset(p->subchannels, 0, sizeof(*p->subchannels) * args->addresses->naddrs);
+  grpc_subchannel_args sc_args;
+  size_t subchannel_idx = 0;
+  for (size_t i = 0; i < args->addresses->naddrs; i++) {
+    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;
+
+    grpc_subchannel *subchannel = grpc_subchannel_factory_create_subchannel(
+        exec_ctx, args->subchannel_factory, &sc_args);
+
+    if (subchannel != NULL) {
+      p->subchannels[subchannel_idx++] = subchannel;
+    }
+  }
+  if (subchannel_idx == 0) {
+    gpr_free(p->subchannels);
+    gpr_free(p);
+    return NULL;
+  }
+  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);
   gpr_mu_init(&p->mu);
   return &p->base;

src/core/client_config/lb_policies/round_robin.c

@@ -490,30 +490,47 @@ 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,
+static grpc_lb_policy *create_round_robin(grpc_exec_ctx *exec_ctx,
+                                          grpc_lb_policy_factory *factory,
                                           grpc_lb_policy_args *args) {
-  size_t i;
+  GPR_ASSERT(args->addresses != NULL);
+  GPR_ASSERT(args->subchannel_factory != NULL);
+
   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->num_subchannels = args->num_subchannels;
-  p->subchannels = gpr_malloc(sizeof(*p->subchannels) * p->num_subchannels);
-  memset(p->subchannels, 0, sizeof(*p->subchannels) * p->num_subchannels);
-  grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE,
-                               "round_robin");
 
-  gpr_mu_init(&p->mu);
-  for (i = 0; i < args->num_subchannels; i++) {
-    subchannel_data *sd = gpr_malloc(sizeof(*sd));
-    memset(sd, 0, sizeof(*sd));
-    p->subchannels[i] = sd;
-    sd->policy = p;
-    sd->index = i;
-    sd->subchannel = args->subchannels[i];
-    grpc_closure_init(&sd->connectivity_changed_closure,
-                      rr_connectivity_changed, sd);
+  p->subchannels =
+      gpr_malloc(sizeof(*p->subchannels) * args->addresses->naddrs);
+  memset(p->subchannels, 0, sizeof(*p->subchannels) * args->addresses->naddrs);
+
+  grpc_subchannel_args sc_args;
+  size_t subchannel_idx = 0;
+  for (size_t i = 0; i < args->addresses->naddrs; i++) {
+    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;
+
+    grpc_subchannel *subchannel = grpc_subchannel_factory_create_subchannel(
+        exec_ctx, args->subchannel_factory, &sc_args);
+
+    if (subchannel != NULL) {
+      subchannel_data *sd = gpr_malloc(sizeof(*sd));
+      memset(sd, 0, sizeof(*sd));
+      p->subchannels[subchannel_idx] = sd;
+      sd->policy = p;
+      sd->index = subchannel_idx;
+      sd->subchannel = subchannel;
+      ++subchannel_idx;
+      grpc_closure_init(&sd->connectivity_changed_closure,
+                        rr_connectivity_changed, sd);
+    }
+  }
+  if (subchannel_idx == 0) {
+    gpr_free(p->subchannels);
+    gpr_free(p);
+    return NULL;
   }
+  p->num_subchannels = subchannel_idx;
 
   /* The (dummy node) root of the ready list */
   p->ready_list.subchannel = NULL;
@@ -521,6 +538,10 @@ static grpc_lb_policy *create_round_robin(grpc_lb_policy_factory *factory,
   p->ready_list.next = NULL;
   p->ready_list_last_pick = &p->ready_list;
 
+  grpc_lb_policy_init(&p->base, &round_robin_lb_policy_vtable);
+  grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE,
+                               "round_robin");
+  gpr_mu_init(&p->mu);
   return &p->base;
 }
 

src/core/client_config/lb_policy_factory.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -42,7 +42,8 @@ void grpc_lb_policy_factory_unref(grpc_lb_policy_factory* factory) {
 }
 
 grpc_lb_policy* grpc_lb_policy_factory_create_lb_policy(
-    grpc_lb_policy_factory* factory, grpc_lb_policy_args* args) {
+    grpc_exec_ctx* exec_ctx, grpc_lb_policy_factory* factory,
+    grpc_lb_policy_args* args) {
   if (factory == NULL) return NULL;
-  return factory->vtable->create_lb_policy(factory, args);
+  return factory->vtable->create_lb_policy(exec_ctx, factory, args);
 }

src/core/client_config/lb_policy_factory.h

@@ -36,6 +36,10 @@
 
 #include "src/core/client_config/lb_policy.h"
 #include "src/core/client_config/subchannel.h"
+#include "src/core/client_config/subchannel_factory.h"
+
+#include "src/core/iomgr/exec_ctx.h"
+#include "src/core/iomgr/resolve_address.h"
 
 typedef struct grpc_lb_policy_factory grpc_lb_policy_factory;
 typedef struct grpc_lb_policy_factory_vtable grpc_lb_policy_factory_vtable;
@@ -47,8 +51,8 @@ struct grpc_lb_policy_factory {
 };
 
 typedef struct grpc_lb_policy_args {
-  grpc_subchannel **subchannels;
-  size_t num_subchannels;
+  grpc_resolved_addresses *addresses;
+  grpc_subchannel_factory *subchannel_factory;
 } grpc_lb_policy_args;
 
 struct grpc_lb_policy_factory_vtable {
@@ -56,7 +60,8 @@ struct grpc_lb_policy_factory_vtable {
   void (*unref)(grpc_lb_policy_factory *factory);
 
   /** Implementation of grpc_lb_policy_factory_create_lb_policy */
-  grpc_lb_policy *(*create_lb_policy)(grpc_lb_policy_factory *factory,
+  grpc_lb_policy *(*create_lb_policy)(grpc_exec_ctx *exec_ctx,
+                                      grpc_lb_policy_factory *factory,
                                       grpc_lb_policy_args *args);
 
   /** Name for the LB policy this factory implements */
@@ -68,6 +73,7 @@ void grpc_lb_policy_factory_unref(grpc_lb_policy_factory *factory);
 
 /** Create a lb_policy instance. */
 grpc_lb_policy *grpc_lb_policy_factory_create_lb_policy(
-    grpc_lb_policy_factory *factory, grpc_lb_policy_args *args);
+    grpc_exec_ctx *exec_ctx, grpc_lb_policy_factory *factory,
+    grpc_lb_policy_args *args);
 
 #endif /* GRPC_CORE_CLIENT_CONFIG_LB_POLICY_FACTORY_H */

src/core/client_config/lb_policy_registry.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -79,10 +79,10 @@ static grpc_lb_policy_factory *lookup_factory(const char *name) {
   return NULL;
 }
 
-grpc_lb_policy *grpc_lb_policy_create(const char *name,
+grpc_lb_policy *grpc_lb_policy_create(grpc_exec_ctx *exec_ctx, const char *name,
                                       grpc_lb_policy_args *args) {
   grpc_lb_policy_factory *factory = lookup_factory(name);
   grpc_lb_policy *lb_policy =
-      grpc_lb_policy_factory_create_lb_policy(factory, args);
+      grpc_lb_policy_factory_create_lb_policy(exec_ctx, factory, args);
   return lb_policy;
 }

src/core/client_config/lb_policy_registry.h

@@ -35,6 +35,7 @@
 #define GRPC_CORE_CLIENT_CONFIG_LB_POLICY_REGISTRY_H
 
 #include "src/core/client_config/lb_policy_factory.h"
+#include "src/core/iomgr/exec_ctx.h"
 
 /** Initialize the registry and set \a default_factory as the factory to be
  * returned when no name is provided in a lookup */
@@ -48,7 +49,7 @@ void grpc_register_lb_policy(grpc_lb_policy_factory *factory);
  *
  * If \a name is NULL, the default factory from \a grpc_lb_policy_registry_init
  * will be returned. */
-grpc_lb_policy *grpc_lb_policy_create(const char *name,
+grpc_lb_policy *grpc_lb_policy_create(grpc_exec_ctx *exec_ctx, const char *name,
                                       grpc_lb_policy_args *args);
 
 #endif /* GRPC_CORE_CLIENT_CONFIG_LB_POLICY_REGISTRY_H */

src/core/client_config/resolvers/dns_resolver.c

@@ -162,38 +162,23 @@ static void dns_on_resolved(grpc_exec_ctx *exec_ctx, void *arg,
                             grpc_resolved_addresses *addresses) {
   dns_resolver *r = arg;
   grpc_client_config *config = NULL;
-  grpc_subchannel **subchannels;
-  grpc_subchannel_args args;
   grpc_lb_policy *lb_policy;
-  size_t i;
   gpr_mu_lock(&r->mu);
   GPR_ASSERT(r->resolving);
   r->resolving = 0;
   if (addresses != NULL) {
     grpc_lb_policy_args lb_policy_args;
     config = grpc_client_config_create();
-    subchannels = gpr_malloc(sizeof(grpc_subchannel *) * addresses->naddrs);
-    size_t naddrs = 0;
-    for (i = 0; i < addresses->naddrs; i++) {
-      memset(&args, 0, sizeof(args));
-      args.addr = (struct sockaddr *)(addresses->addrs[i].addr);
-      args.addr_len = (size_t)addresses->addrs[i].len;
-      grpc_subchannel *subchannel = grpc_subchannel_factory_create_subchannel(
-          exec_ctx, r->subchannel_factory, &args);
-      if (subchannel != NULL) {
-        subchannels[naddrs++] = subchannel;
-      }
-    }
     memset(&lb_policy_args, 0, sizeof(lb_policy_args));
-    lb_policy_args.subchannels = subchannels;
-    lb_policy_args.num_subchannels = naddrs;
-    lb_policy = grpc_lb_policy_create(r->lb_policy_name, &lb_policy_args);
+    lb_policy_args.addresses = addresses;
+    lb_policy_args.subchannel_factory = r->subchannel_factory;
+    lb_policy =
+        grpc_lb_policy_create(exec_ctx, r->lb_policy_name, &lb_policy_args);
     if (lb_policy != NULL) {
       grpc_client_config_set_lb_policy(config, lb_policy);
       GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "construction");
     }
     grpc_resolved_addresses_destroy(addresses);
-    gpr_free(subchannels);
   } else {
     gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
     gpr_timespec next_try = gpr_backoff_step(&r->backoff_state, now);

src/core/client_config/resolvers/sockaddr_resolver.c

@@ -58,11 +58,7 @@ typedef struct {
   char *lb_policy_name;
 
   /** the addresses that we've 'resolved' */
-  struct sockaddr_storage *addrs;
-  /** the corresponding length of the addresses */
-  size_t *addrs_len;
-  /** how many elements in \a addrs */
-  size_t num_addrs;
+  grpc_resolved_addresses *addresses;
 
   /** mutex guarding the rest of the state */
   gpr_mu mu;
@@ -125,28 +121,14 @@ static void sockaddr_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
 
 static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
                                               sockaddr_resolver *r) {
-  grpc_client_config *cfg;
-  grpc_lb_policy *lb_policy;
-  grpc_lb_policy_args lb_policy_args;
-  grpc_subchannel **subchannels;
-  grpc_subchannel_args args;
-
   if (r->next_completion != NULL && !r->published) {
-    size_t i;
-    cfg = grpc_client_config_create();
-    subchannels = gpr_malloc(sizeof(grpc_subchannel *) * r->num_addrs);
-    for (i = 0; i < r->num_addrs; i++) {
-      memset(&args, 0, sizeof(args));
-      args.addr = (struct sockaddr *)&r->addrs[i];
-      args.addr_len = r->addrs_len[i];
-      subchannels[i] = grpc_subchannel_factory_create_subchannel(
-          exec_ctx, r->subchannel_factory, &args);
-    }
+    grpc_client_config *cfg = grpc_client_config_create();
+    grpc_lb_policy_args lb_policy_args;
     memset(&lb_policy_args, 0, sizeof(lb_policy_args));
-    lb_policy_args.subchannels = subchannels;
-    lb_policy_args.num_subchannels = r->num_addrs;
-    lb_policy = grpc_lb_policy_create(r->lb_policy_name, &lb_policy_args);
-    gpr_free(subchannels);
+    lb_policy_args.addresses = r->addresses;
+    lb_policy_args.subchannel_factory = r->subchannel_factory;
+    grpc_lb_policy *lb_policy =
+        grpc_lb_policy_create(exec_ctx, r->lb_policy_name, &lb_policy_args);
     grpc_client_config_set_lb_policy(cfg, lb_policy);
     GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "sockaddr");
     r->published = 1;
@@ -160,8 +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_subchannel_factory_unref(exec_ctx, r->subchannel_factory);
-  gpr_free(r->addrs);
-  gpr_free(r->addrs_len);
+  grpc_resolved_addresses_destroy(r->addresses);
   gpr_free(r->lb_policy_name);
   gpr_free(r);
 }
@@ -269,7 +250,6 @@ 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)) {
-  size_t i;
   int errors_found = 0; /* GPR_FALSE */
   sockaddr_resolver *r;
   gpr_slice path_slice;
@@ -309,15 +289,18 @@ static grpc_resolver *sockaddr_create(
   gpr_slice_buffer_init(&path_parts);
 
   gpr_slice_split(path_slice, ",", &path_parts);
-  r->num_addrs = path_parts.count;
-  r->addrs = gpr_malloc(sizeof(struct sockaddr_storage) * r->num_addrs);
-  r->addrs_len = gpr_malloc(sizeof(*r->addrs_len) * r->num_addrs);
+  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 (i = 0; i < r->num_addrs; i++) {
+  for (size_t i = 0; i < r->addresses->naddrs; 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, &r->addrs[i], &r->addrs_len[i])) {
+    if (!parse(&ith_uri,
+               (struct sockaddr_storage *)(&r->addresses->addrs[i].addr),
+               &r->addresses->addrs[i].len)) {
       errors_found = 1; /* GPR_TRUE */
     }
     gpr_free(part_str);
@@ -328,8 +311,7 @@ static grpc_resolver *sockaddr_create(
   gpr_slice_unref(path_slice);
   if (errors_found) {
     gpr_free(r->lb_policy_name);
-    gpr_free(r->addrs);
-    gpr_free(r->addrs_len);
+    grpc_resolved_addresses_destroy(r->addresses);
     gpr_free(r);
     return NULL;
   }

src/core/client_config/resolvers/zookeeper_resolver.c

@@ -184,28 +184,22 @@ static void zookeeper_on_resolved(grpc_exec_ctx *exec_ctx, void *arg,
                                   grpc_resolved_addresses *addresses) {
   zookeeper_resolver *r = arg;
   grpc_client_config *config = NULL;
-  grpc_subchannel **subchannels;
-  grpc_subchannel_args args;
   grpc_lb_policy *lb_policy;
-  size_t i;
+
   if (addresses != NULL) {
     grpc_lb_policy_args lb_policy_args;
     config = grpc_client_config_create();
-    subchannels = gpr_malloc(sizeof(grpc_subchannel *) * addresses->naddrs);
-    for (i = 0; i < addresses->naddrs; i++) {
-      memset(&args, 0, sizeof(args));
-      args.addr = (struct sockaddr *)(addresses->addrs[i].addr);
-      args.addr_len = addresses->addrs[i].len;
-      subchannels[i] = grpc_subchannel_factory_create_subchannel(
-          exec_ctx, r->subchannel_factory, &args);
+
+    lb_policy_args.addresses = addresses;
+    lb_policy_args.subchannel_factory = r->subchannel_factory;
+    lb_policy =
+        grpc_lb_policy_create(exec_ctx, r->lb_policy_name, &lb_policy_args);
+
+    if (lb_policy != NULL) {
+      grpc_client_config_set_lb_policy(config, lb_policy);
+      GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "construction");
     }
-    lb_policy_args.subchannels = subchannels;
-    lb_policy_args.num_subchannels = addresses->naddrs;
-    lb_policy = grpc_lb_policy_create(r->lb_policy_name, &lb_policy_args);
-    grpc_client_config_set_lb_policy(config, lb_policy);
-    GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "construction");
     grpc_resolved_addresses_destroy(addresses);
-    gpr_free(subchannels);
   }
   gpr_mu_lock(&r->mu);
   GPR_ASSERT(r->resolving == 1);

test/core/client_config/lb_policies_test.c

@@ -870,6 +870,7 @@ static void verify_rebirth_round_robin(const servers_fixture *f,
 }
 
 int main(int argc, char **argv) {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   test_spec *spec;
   size_t i;
   const size_t NUM_ITERS = 10;
@@ -879,9 +880,9 @@ int main(int argc, char **argv) {
   grpc_init();
   grpc_lb_round_robin_trace = 1;
 
-  GPR_ASSERT(grpc_lb_policy_create("this-lb-policy-does-not-exist", NULL) ==
-             NULL);
-  GPR_ASSERT(grpc_lb_policy_create(NULL, NULL) == NULL);
+  GPR_ASSERT(grpc_lb_policy_create(&exec_ctx, "this-lb-policy-does-not-exist",
+                                   NULL) == NULL);
+  GPR_ASSERT(grpc_lb_policy_create(&exec_ctx, NULL, NULL) == NULL);
 
   spec = test_spec_create(NUM_ITERS, NUM_SERVERS);
   /* everything is fine, all servers stay up the whole time and life's peachy */
@@ -933,6 +934,7 @@ int main(int argc, char **argv) {
   test_pending_calls(4);
   test_ping();
 
+  grpc_exec_ctx_finish(&exec_ctx);
   grpc_shutdown();
   return 0;
 }