Merge github.com:grpc/grpc into bwest

6 files changed, 82 insertions, 48 deletions

diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c
index 2f25fef9a7..74350d9fee 100644
--- a/src/core/ext/client_channel/client_channel.c
+++ b/src/core/ext/client_channel/client_channel.c
@@ -52,6 +52,7 @@
 #include "src/core/lib/iomgr/iomgr.h"
 #include "src/core/lib/iomgr/polling_entity.h"
 #include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/surface/channel.h"
 #include "src/core/lib/transport/connectivity_state.h"
@@ -87,7 +88,7 @@ static void method_parameters_free(grpc_exec_ctx *exec_ctx, void *p) {
   gpr_free(p);
 }
 
-static const grpc_mdstr_hash_table_vtable method_parameters_vtable = {
+static const grpc_slice_hash_table_vtable method_parameters_vtable = {
     method_parameters_free, method_parameters_copy};
 
 static void *method_parameters_create_from_json(const grpc_json *json) {
@@ -165,7 +166,7 @@ typedef struct client_channel_channel_data {
   /** service config in JSON form */
   char *service_config_json;
   /** maps method names to method_parameters structs */
-  grpc_mdstr_hash_table *method_params_table;
+  grpc_slice_hash_table *method_params_table;
   /** incoming resolver result - set by resolver.next() */
   grpc_channel_args *resolver_result;
   /** a list of closures that are all waiting for config to come in */
@@ -267,7 +268,7 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
   char *lb_policy_name = NULL;
   grpc_lb_policy *lb_policy = NULL;
   grpc_lb_policy *old_lb_policy;
-  grpc_mdstr_hash_table *method_params_table = NULL;
+  grpc_slice_hash_table *method_params_table = NULL;
   grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE;
   bool exit_idle = false;
   grpc_error *state_error = GRPC_ERROR_CREATE("No load balancing policy");
@@ -362,7 +363,7 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
     chand->service_config_json = service_config_json;
   }
   if (chand->method_params_table != NULL) {
-    grpc_mdstr_hash_table_unref(exec_ctx, chand->method_params_table);
+    grpc_slice_hash_table_unref(exec_ctx, chand->method_params_table);
   }
   chand->method_params_table = method_params_table;
   if (lb_policy != NULL) {
@@ -558,7 +559,7 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
   gpr_free(chand->lb_policy_name);
   gpr_free(chand->service_config_json);
   if (chand->method_params_table != NULL) {
-    grpc_mdstr_hash_table_unref(exec_ctx, chand->method_params_table);
+    grpc_slice_hash_table_unref(exec_ctx, chand->method_params_table);
   }
   grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
   grpc_pollset_set_destroy(chand->interested_parties);
@@ -593,7 +594,7 @@ typedef struct client_channel_call_data {
   // to avoid this without breaking the grpc_deadline_state abstraction.
   grpc_deadline_state deadline_state;
 
-  grpc_mdstr *path;  // Request path.
+  grpc_slice path;  // Request path.
   gpr_timespec call_start_time;
   gpr_timespec deadline;
   wait_for_ready_value wait_for_ready_from_service_config;
@@ -997,10 +998,10 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
   if (error == GRPC_ERROR_NONE) {
     // Get the method config table from channel data.
     gpr_mu_lock(&chand->mu);
-    grpc_mdstr_hash_table *method_params_table = NULL;
+    grpc_slice_hash_table *method_params_table = NULL;
     if (chand->method_params_table != NULL) {
       method_params_table =
-          grpc_mdstr_hash_table_ref(chand->method_params_table);
+          grpc_slice_hash_table_ref(chand->method_params_table);
     }
     gpr_mu_unlock(&chand->mu);
     // If the method config table was present, use it.
@@ -1029,7 +1030,7 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
           gpr_mu_unlock(&calld->mu);
         }
       }
-      grpc_mdstr_hash_table_unref(exec_ctx, method_params_table);
+      grpc_slice_hash_table_unref(exec_ctx, method_params_table);
     }
   }
   GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config");
@@ -1043,7 +1044,7 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
   call_data *calld = elem->call_data;
   // Initialize data members.
   grpc_deadline_state_init(exec_ctx, elem, args->call_stack);
-  calld->path = GRPC_MDSTR_REF(args->path);
+  calld->path = grpc_slice_ref_internal(args->path);
   calld->call_start_time = args->start_time;
   calld->deadline = gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC);
   calld->wait_for_ready_from_service_config = WAIT_FOR_READY_UNSET;
@@ -1067,8 +1068,8 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
   if (chand->lb_policy != NULL) {
     // We already have a resolver result, so check for service config.
     if (chand->method_params_table != NULL) {
-      grpc_mdstr_hash_table *method_params_table =
-          grpc_mdstr_hash_table_ref(chand->method_params_table);
+      grpc_slice_hash_table *method_params_table =
+          grpc_slice_hash_table_ref(chand->method_params_table);
       gpr_mu_unlock(&chand->mu);
       method_parameters *method_params = grpc_method_config_table_get(
           exec_ctx, method_params_table, args->path);
@@ -1084,7 +1085,7 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
               method_params->wait_for_ready;
         }
       }
-      grpc_mdstr_hash_table_unref(exec_ctx, method_params_table);
+      grpc_slice_hash_table_unref(exec_ctx, method_params_table);
     } else {
       gpr_mu_unlock(&chand->mu);
     }
@@ -1113,7 +1114,7 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                  void *and_free_memory) {
   call_data *calld = elem->call_data;
   grpc_deadline_state_destroy(exec_ctx, elem);
-  GRPC_MDSTR_UNREF(exec_ctx, calld->path);
+  grpc_slice_unref_internal(exec_ctx, calld->path);
   GRPC_ERROR_UNREF(calld->cancel_error);
   grpc_subchannel_call *call = GET_CALL(calld);
   if (call != NULL && call != CANCELLED_CALL) {
diff --git a/src/core/ext/client_channel/connector.h b/src/core/ext/client_channel/connector.h
index 3de061620e..395f89b3b2 100644
--- a/src/core/ext/client_channel/connector.h
+++ b/src/core/ext/client_channel/connector.h
@@ -48,9 +48,6 @@ struct grpc_connector {
 typedef struct {
   /** set of pollsets interested in this connection */
   grpc_pollset_set *interested_parties;
-  /** address to connect to */
-  const grpc_resolved_address *addr;
-  size_t addr_len;
   /** initial connect string to send */
   grpc_slice initial_connect_string;
   /** deadline for connection */
diff --git a/src/core/ext/client_channel/resolver_registry.h b/src/core/ext/client_channel/resolver_registry.h
index 4fb16131db..a4606463eb 100644
--- a/src/core/ext/client_channel/resolver_registry.h
+++ b/src/core/ext/client_channel/resolver_registry.h
@@ -60,7 +60,9 @@ void grpc_register_resolver_type(grpc_resolver_factory *factory);
     return it.
     If a resolver factory was not found, return NULL.
     \a args is a set of channel arguments to be included in the result
-    (typically the set of arguments passed in from the client API). */
+    (typically the set of arguments passed in from the client API).
+    \a pollset_set is used to drive IO in the name resolution process, it
+    should not be NULL. */
 grpc_resolver *grpc_resolver_create(grpc_exec_ctx *exec_ctx, const char *target,
                                     const grpc_channel_args *args,
                                     grpc_pollset_set *pollset_set);
diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c
index 1bac82b451..b7379b30b3 100644
--- a/src/core/ext/client_channel/subchannel.c
+++ b/src/core/ext/client_channel/subchannel.c
@@ -38,12 +38,16 @@
 
 #include <grpc/support/alloc.h>
 #include <grpc/support/avl.h>
+#include <grpc/support/string_util.h>
 
 #include "src/core/ext/client_channel/client_channel.h"
 #include "src/core/ext/client_channel/initial_connect_string.h"
+#include "src/core/ext/client_channel/parse_address.h"
 #include "src/core/ext/client_channel/subchannel_index.h"
+#include "src/core/ext/client_channel/uri_parser.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/connected_channel.h"
+#include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "src/core/lib/iomgr/timer.h"
 #include "src/core/lib/profiling/timers.h"
 #include "src/core/lib/slice/slice_internal.h"
@@ -95,8 +99,6 @@ struct grpc_subchannel {
   size_t num_filters;
   /** channel arguments */
   grpc_channel_args *args;
-  /** address to connect to */
-  grpc_resolved_address *addr;
 
   grpc_subchannel_key *key;
 
@@ -211,7 +213,6 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
   grpc_subchannel *c = arg;
   gpr_free((void *)c->filters);
   grpc_channel_args_destroy(exec_ctx, c->args);
-  gpr_free(c->addr);
   grpc_slice_unref_internal(exec_ctx, c->initial_connect_string);
   grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
   grpc_connector_unref(exec_ctx, c->connector);
@@ -327,12 +328,17 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
   } else {
     c->filters = NULL;
   }
-  c->addr = gpr_malloc(sizeof(grpc_resolved_address));
-  if (args->addr->len)
-    memcpy(c->addr, args->addr, sizeof(grpc_resolved_address));
   c->pollset_set = grpc_pollset_set_create();
-  grpc_set_initial_connect_string(&c->addr, &c->initial_connect_string);
-  c->args = grpc_channel_args_copy(args->args);
+  grpc_resolved_address *addr = gpr_malloc(sizeof(*addr));
+  grpc_get_subchannel_address_arg(args->args, addr);
+  grpc_set_initial_connect_string(&addr, &c->initial_connect_string);
+  static const char *keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS};
+  grpc_arg new_arg = grpc_create_subchannel_address_arg(addr);
+  gpr_free(addr);
+  c->args = grpc_channel_args_copy_and_add_and_remove(
+      args->args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), &new_arg, 1);
+  gpr_free(new_arg.value.string);
+
   c->root_external_state_watcher.next = c->root_external_state_watcher.prev =
       &c->root_external_state_watcher;
   grpc_closure_init(&c->connected, subchannel_connected, c,
@@ -385,7 +391,6 @@ static void continue_connect_locked(grpc_exec_ctx *exec_ctx,
   grpc_connect_in_args args;
 
   args.interested_parties = c->pollset_set;
-  args.addr = c->addr;
   args.deadline = c->next_attempt;
   args.channel_args = c->args;
   args.initial_connect_string = c->initial_connect_string;
@@ -620,9 +625,8 @@ static void publish_transport_locked(grpc_exec_ctx *exec_ctx,
   grpc_error *error = grpc_channel_stack_builder_finish(
       exec_ctx, builder, 0, 1, connection_destroy, NULL, (void **)&con);
   if (error != GRPC_ERROR_NONE) {
-    const char *msg = grpc_error_string(error);
-    gpr_log(GPR_ERROR, "error initializing subchannel stack: %s", msg);
-    grpc_error_free_string(msg);
+    gpr_log(GPR_ERROR, "error initializing subchannel stack: %s",
+            grpc_error_string(error));
     GRPC_ERROR_UNREF(error);
     abort(); /* TODO(ctiller): what to do here? */
   }
@@ -687,7 +691,6 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg,
 
     const char *errmsg = grpc_error_string(error);
     gpr_log(GPR_INFO, "Connect failed: %s", errmsg);
-    grpc_error_free_string(errmsg);
 
     maybe_start_connecting_locked(exec_ctx, c);
     GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
@@ -746,7 +749,7 @@ grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
 
 grpc_error *grpc_connected_subchannel_create_call(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
-    grpc_polling_entity *pollent, grpc_mdstr *path, gpr_timespec start_time,
+    grpc_polling_entity *pollent, grpc_slice path, gpr_timespec start_time,
     gpr_timespec deadline, grpc_subchannel_call **call) {
   grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
   *call = gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
@@ -758,7 +761,7 @@ grpc_error *grpc_connected_subchannel_create_call(
   if (error != GRPC_ERROR_NONE) {
     const char *error_string = grpc_error_string(error);
     gpr_log(GPR_ERROR, "error: %s", error_string);
-    grpc_error_free_string(error_string);
+
     gpr_free(*call);
     return error;
   }
@@ -771,3 +774,37 @@ grpc_call_stack *grpc_subchannel_call_get_call_stack(
     grpc_subchannel_call *subchannel_call) {
   return SUBCHANNEL_CALL_TO_CALL_STACK(subchannel_call);
 }
+
+static void grpc_uri_to_sockaddr(char *uri_str, grpc_resolved_address *addr) {
+  grpc_uri *uri = grpc_uri_parse(uri_str, 0 /* suppress_errors */);
+  GPR_ASSERT(uri != NULL);
+  if (strcmp(uri->scheme, "ipv4") == 0) {
+    GPR_ASSERT(parse_ipv4(uri, addr));
+  } else if (strcmp(uri->scheme, "ipv6") == 0) {
+    GPR_ASSERT(parse_ipv6(uri, addr));
+  } else {
+    GPR_ASSERT(parse_unix(uri, addr));
+  }
+  grpc_uri_destroy(uri);
+}
+
+void grpc_get_subchannel_address_arg(const grpc_channel_args *args,
+                                     grpc_resolved_address *addr) {
+  const grpc_arg *addr_arg =
+      grpc_channel_args_find(args, GRPC_ARG_SUBCHANNEL_ADDRESS);
+  GPR_ASSERT(addr_arg != NULL);  // Should have been set by LB policy.
+  GPR_ASSERT(addr_arg->type == GRPC_ARG_STRING);
+  memset(addr, 0, sizeof(*addr));
+  if (*addr_arg->value.string != '\0') {
+    grpc_uri_to_sockaddr(addr_arg->value.string, addr);
+  }
+}
+
+grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address *addr) {
+  grpc_arg new_arg;
+  new_arg.key = GRPC_ARG_SUBCHANNEL_ADDRESS;
+  new_arg.type = GRPC_ARG_STRING;
+  new_arg.value.string =
+      addr->len > 0 ? grpc_sockaddr_to_uri(addr) : gpr_strdup("");
+  return new_arg;
+}
diff --git a/src/core/ext/client_channel/subchannel.h b/src/core/ext/client_channel/subchannel.h
index 24aa9f73dc..9bd35a7704 100644
--- a/src/core/ext/client_channel/subchannel.h
+++ b/src/core/ext/client_channel/subchannel.h
@@ -40,6 +40,9 @@
 #include "src/core/lib/transport/connectivity_state.h"
 #include "src/core/lib/transport/metadata.h"
 
+// Channel arg containing a grpc_resolved_address to connect to.
+#define GRPC_ARG_SUBCHANNEL_ADDRESS "grpc.subchannel_address"
+
 /** A (sub-)channel that knows how to connect to exactly one target
     address. Provides a target for load balancing. */
 typedef struct grpc_subchannel grpc_subchannel;
@@ -111,7 +114,7 @@ void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
 /** construct a subchannel call */
 grpc_error *grpc_connected_subchannel_create_call(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *connected_subchannel,
-    grpc_polling_entity *pollent, grpc_mdstr *path, gpr_timespec start_time,
+    grpc_polling_entity *pollent, grpc_slice path, gpr_timespec start_time,
     gpr_timespec deadline, grpc_subchannel_call **subchannel_call);
 
 /** process a transport level op */
@@ -164,8 +167,6 @@ struct grpc_subchannel_args {
   size_t filter_count;
   /** Channel arguments to be supplied to the newly created channel */
   const grpc_channel_args *args;
-  /** Address to connect to */
-  grpc_resolved_address *addr;
 };
 
 /** create a subchannel given a connector */
@@ -173,4 +174,12 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
                                         grpc_connector *connector,
                                         const grpc_subchannel_args *args);
 
+/// Sets \a addr from \a args.
+void grpc_get_subchannel_address_arg(const grpc_channel_args *args,
+                                     grpc_resolved_address *addr);
+
+/// Returns a new channel arg encoding the subchannel address as a string.
+/// Caller is responsible for freeing the string.
+grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address *addr);
+
 #endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_SUBCHANNEL_H */
diff --git a/src/core/ext/client_channel/subchannel_index.c b/src/core/ext/client_channel/subchannel_index.c
index 1ebe03ef11..11889300a2 100644
--- a/src/core/ext/client_channel/subchannel_index.c
+++ b/src/core/ext/client_channel/subchannel_index.c
@@ -86,11 +86,6 @@ static grpc_subchannel_key *create_key(
   } else {
     k->args.filters = NULL;
   }
-  k->args.addr = gpr_malloc(sizeof(grpc_resolved_address));
-  k->args.addr->len = args->addr->len;
-  if (k->args.addr->len > 0) {
-    memcpy(k->args.addr, args->addr, sizeof(grpc_resolved_address));
-  }
   k->args.args = copy_channel_args(args->args);
   return k;
 }
@@ -108,14 +103,8 @@ static int subchannel_key_compare(grpc_subchannel_key *a,
                                   grpc_subchannel_key *b) {
   int c = GPR_ICMP(a->connector, b->connector);
   if (c != 0) return c;
-  c = GPR_ICMP(a->args.addr->len, b->args.addr->len);
-  if (c != 0) return c;
   c = GPR_ICMP(a->args.filter_count, b->args.filter_count);
   if (c != 0) return c;
-  if (a->args.addr->len) {
-    c = memcmp(a->args.addr->addr, b->args.addr->addr, a->args.addr->len);
-    if (c != 0) return c;
-  }
   if (a->args.filter_count > 0) {
     c = memcmp(a->args.filters, b->args.filters,
                a->args.filter_count * sizeof(*a->args.filters));
@@ -129,7 +118,6 @@ void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx,
   grpc_connector_unref(exec_ctx, k->connector);
   gpr_free((grpc_channel_args *)k->args.filters);
   grpc_channel_args_destroy(exec_ctx, (grpc_channel_args *)k->args.args);
-  gpr_free(k->args.addr);
   gpr_free(k);
 }