Merge remote-tracking branch 'upstream/master' into srv_record

12 files changed, 279 insertions, 173 deletions

diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c
index 24843d52e9..f2f27b9175 100644
--- a/src/core/ext/filters/client_channel/client_channel.c
+++ b/src/core/ext/filters/client_channel/client_channel.c
@@ -914,10 +914,13 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
   grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent,
                                            chand->interested_parties);
   if (calld->connected_subchannel == NULL) {
-    gpr_atm_no_barrier_store(&calld->subchannel_call, 1);
+    gpr_atm_no_barrier_store(&calld->subchannel_call, (gpr_atm)CANCELLED_CALL);
     fail_locked(exec_ctx, calld,
-                GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
-                    "Failed to create subchannel", &error, 1));
+                error == GRPC_ERROR_NONE
+                    ? GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+                          "Call dropped by load balancing policy")
+                    : GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
+                          "Failed to create subchannel", &error, 1));
   } else if (GET_CALL(calld) == CANCELLED_CALL) {
     /* already cancelled before subchannel became ready */
     grpc_error *cancellation_error =
@@ -1180,6 +1183,15 @@ static void start_transport_stream_op_batch_locked_inner(
             &calld->next_step)) {
       calld->pick_pending = false;
       GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
+      if (calld->connected_subchannel == NULL) {
+        gpr_atm_no_barrier_store(&calld->subchannel_call,
+                                 (gpr_atm)CANCELLED_CALL);
+        grpc_error *error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+            "Call dropped by load balancing policy");
+        fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error));
+        grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, op, error);
+        return;  // Early out.
+      }
     } else {
       grpc_polling_entity_add_to_pollset_set(exec_ctx, calld->pollent,
                                              chand->interested_parties);
diff --git a/src/core/ext/filters/client_channel/lb_policy.h b/src/core/ext/filters/client_channel/lb_policy.h
index 184b2ef720..fb4aa084a6 100644
--- a/src/core/ext/filters/client_channel/lb_policy.h
+++ b/src/core/ext/filters/client_channel/lb_policy.h
@@ -149,7 +149,9 @@ void grpc_lb_policy_init(grpc_lb_policy *policy,
 
 /** Finds an appropriate subchannel for a call, based on \a pick_args.
 
-    \a target will be set to the selected subchannel, or NULL on failure.
+    \a target will be set to the selected subchannel, or NULL on failure
+    or when the LB policy decides to drop the call.
+
     Upon success, \a user_data will be set to whatever opaque information
     may need to be propagated from the LB policy, or NULL if not needed.
     \a context will be populated with context to pass to the subchannel
diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c
index b7c0e929b7..d2a2856a18 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c
@@ -327,6 +327,11 @@ typedef struct glb_lb_policy {
    * response has arrived. */
   grpc_grpclb_serverlist *serverlist;
 
+  /** Index into serverlist for next pick.
+   * If the server at this index is a drop, we return a drop.
+   * Otherwise, we delegate to the RR policy. */
+  size_t serverlist_index;
+
   /** list of picks that are waiting on RR's policy connectivity */
   pending_pick *pending_picks;
 
@@ -402,6 +407,9 @@ struct rr_connectivity_data {
 
 static bool is_server_valid(const grpc_grpclb_server *server, size_t idx,
                             bool log) {
+  if (server->drop_for_rate_limiting || server->drop_for_load_balancing) {
+    return false;
+  }
   const grpc_grpclb_ip_address *ip = &server->ip_address;
   if (server->port >> 16 != 0) {
     if (log) {
@@ -411,7 +419,6 @@ static bool is_server_valid(const grpc_grpclb_server *server, size_t idx,
     }
     return false;
   }
-
   if (ip->size != 4 && ip->size != 16) {
     if (log) {
       gpr_log(GPR_ERROR,
@@ -445,11 +452,12 @@ static const grpc_lb_user_data_vtable lb_token_vtable = {
 
 static void parse_server(const grpc_grpclb_server *server,
                          grpc_resolved_address *addr) {
+  memset(addr, 0, sizeof(*addr));
+  if (server->drop_for_rate_limiting || server->drop_for_load_balancing) return;
   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;
-  memset(addr, 0, sizeof(*addr));
   if (ip->size == 4) {
     addr->len = sizeof(struct sockaddr_in);
     struct sockaddr_in *addr4 = (struct sockaddr_in *)&addr->addr;
@@ -586,16 +594,51 @@ static bool update_lb_connectivity_status_locked(
   return true;
 }
 
-/* perform a pick over \a rr_policy. Given that a pick can return immediately
- * (ignoring its completion callback) we need to perform the cleanups this
- * callback would be otherwise resposible for */
+/* Perform a pick over \a glb_policy->rr_policy. Given that a pick can return
+ * immediately (ignoring its completion callback), we need to perform the
+ * cleanups this callback would otherwise be resposible for.
+ * If \a force_async is true, then we will manually schedule the
+ * completion callback even if the pick is available immediately. */
 static bool pick_from_internal_rr_locked(
-    grpc_exec_ctx *exec_ctx, grpc_lb_policy *rr_policy,
-    const grpc_lb_policy_pick_args *pick_args,
+    grpc_exec_ctx *exec_ctx, glb_lb_policy *glb_policy,
+    const grpc_lb_policy_pick_args *pick_args, bool force_async,
     grpc_connected_subchannel **target, wrapped_rr_closure_arg *wc_arg) {
-  GPR_ASSERT(rr_policy != NULL);
+  // Look at the index into the serverlist to see if we should drop this call.
+  grpc_grpclb_server *server =
+      glb_policy->serverlist->servers[glb_policy->serverlist_index++];
+  if (glb_policy->serverlist_index == glb_policy->serverlist->num_servers) {
+    glb_policy->serverlist_index = 0;  // Wrap-around.
+  }
+  if (server->drop_for_rate_limiting || server->drop_for_load_balancing) {
+    // Not using the RR policy, so unref it.
+    if (GRPC_TRACER_ON(grpc_lb_glb_trace)) {
+      gpr_log(GPR_INFO, "Unreffing RR for drop (0x%" PRIxPTR ")",
+              (intptr_t)wc_arg->rr_policy);
+    }
+    GRPC_LB_POLICY_UNREF(exec_ctx, wc_arg->rr_policy, "glb_pick_sync");
+    // Update client load reporting stats to indicate the number of
+    // dropped calls.  Note that we have to do this here instead of in
+    // the client_load_reporting filter, because we do not create a
+    // subchannel call (and therefore no client_load_reporting filter)
+    // for dropped calls.
+    grpc_grpclb_client_stats_add_call_started(wc_arg->client_stats);
+    grpc_grpclb_client_stats_add_call_finished(
+        server->drop_for_rate_limiting, server->drop_for_load_balancing,
+        false /* failed_to_send */, false /* known_received */,
+        wc_arg->client_stats);
+    grpc_grpclb_client_stats_unref(wc_arg->client_stats);
+    if (force_async) {
+      GPR_ASSERT(wc_arg->wrapped_closure != NULL);
+      grpc_closure_sched(exec_ctx, wc_arg->wrapped_closure, GRPC_ERROR_NONE);
+      gpr_free(wc_arg->free_when_done);
+      return false;
+    }
+    gpr_free(wc_arg->free_when_done);
+    return true;
+  }
+  // Pick via the RR policy.
   const bool pick_done = grpc_lb_policy_pick_locked(
-      exec_ctx, rr_policy, pick_args, target, wc_arg->context,
+      exec_ctx, wc_arg->rr_policy, pick_args, target, wc_arg->context,
       (void **)&wc_arg->lb_token, &wc_arg->wrapper_closure);
   if (pick_done) {
     /* synchronous grpc_lb_policy_pick call. Unref the RR policy. */
@@ -604,17 +647,20 @@ static bool pick_from_internal_rr_locked(
               (intptr_t)wc_arg->rr_policy);
     }
     GRPC_LB_POLICY_UNREF(exec_ctx, wc_arg->rr_policy, "glb_pick_sync");
-
     /* add the load reporting initial metadata */
     initial_metadata_add_lb_token(exec_ctx, pick_args->initial_metadata,
                                   pick_args->lb_token_mdelem_storage,
                                   GRPC_MDELEM_REF(wc_arg->lb_token));
-
     // Pass on client stats via context. Passes ownership of the reference.
     GPR_ASSERT(wc_arg->client_stats != NULL);
     wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].value = wc_arg->client_stats;
     wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].destroy = destroy_client_stats;
-
+    if (force_async) {
+      GPR_ASSERT(wc_arg->wrapped_closure != NULL);
+      grpc_closure_sched(exec_ctx, wc_arg->wrapped_closure, GRPC_ERROR_NONE);
+      gpr_free(wc_arg->free_when_done);
+      return false;
+    }
     gpr_free(wc_arg->free_when_done);
   }
   /* else, the pending pick will be registered and taken care of by the
@@ -744,8 +790,8 @@ static void rr_handover_locked(grpc_exec_ctx *exec_ctx,
       gpr_log(GPR_INFO, "Pending pick about to PICK from 0x%" PRIxPTR "",
               (intptr_t)glb_policy->rr_policy);
     }
-    pick_from_internal_rr_locked(exec_ctx, glb_policy->rr_policy,
-                                 &pp->pick_args, pp->target,
+    pick_from_internal_rr_locked(exec_ctx, glb_policy, &pp->pick_args,
+                                 true /* force_async */, pp->target,
                                  &pp->wrapped_on_complete_arg);
   }
 
@@ -1115,8 +1161,9 @@ static int glb_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
     wc_arg->lb_token_mdelem_storage = pick_args->lb_token_mdelem_storage;
     wc_arg->initial_metadata = pick_args->initial_metadata;
     wc_arg->free_when_done = wc_arg;
-    pick_done = pick_from_internal_rr_locked(exec_ctx, glb_policy->rr_policy,
-                                             pick_args, target, wc_arg);
+    pick_done =
+        pick_from_internal_rr_locked(exec_ctx, glb_policy, pick_args,
+                                     false /* force_async */, target, wc_arg);
   } else {
     if (GRPC_TRACER_ON(grpc_lb_glb_trace)) {
       gpr_log(GPR_DEBUG,
@@ -1517,7 +1564,7 @@ static void lb_on_response_received_locked(grpc_exec_ctx *exec_ctx, void *arg,
              * serverlist instance will be destroyed either upon the next
              * update or in glb_destroy() */
             glb_policy->serverlist = serverlist;
-
+            glb_policy->serverlist_index = 0;
             rr_handover_locked(exec_ctx, glb_policy);
           }
         } else {
diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c
index 81b6932fae..90e7c2efe5 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c
@@ -37,44 +37,39 @@
 
 #include <grpc/support/alloc.h>
 
+/* invoked once for every Server in ServerList */
+static bool count_serverlist(pb_istream_t *stream, const pb_field_t *field,
+                             void **arg) {
+  grpc_grpclb_serverlist *sl = *arg;
+  grpc_grpclb_server server;
+  if (!pb_decode(stream, grpc_lb_v1_Server_fields, &server)) {
+    gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(stream));
+    return false;
+  }
+  ++sl->num_servers;
+  return true;
+}
+
 typedef struct decode_serverlist_arg {
-  /* The first pass counts the number of servers in the server list. The second
-   * one allocates and decodes. */
-  bool first_pass;
   /* The decoding callback is invoked once per server in serverlist. Remember
    * which index of the serverlist are we currently decoding */
   size_t decoding_idx;
-  /* Populated after the first pass. Number of server in the input serverlist */
-  size_t num_servers;
   /* The decoded serverlist */
-  grpc_grpclb_server **servers;
+  grpc_grpclb_serverlist *serverlist;
 } decode_serverlist_arg;
 
 /* invoked once for every Server in ServerList */
 static bool decode_serverlist(pb_istream_t *stream, const pb_field_t *field,
                               void **arg) {
   decode_serverlist_arg *dec_arg = *arg;
-  if (dec_arg->first_pass) { /* count how many server do we have */
-    grpc_grpclb_server server;
-    if (!pb_decode(stream, grpc_lb_v1_Server_fields, &server)) {
-      gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(stream));
-      return false;
-    }
-    dec_arg->num_servers++;
-  } else { /* second pass. Actually decode. */
-    grpc_grpclb_server *server = gpr_zalloc(sizeof(grpc_grpclb_server));
-    GPR_ASSERT(dec_arg->num_servers > 0);
-    if (dec_arg->decoding_idx == 0) { /* first iteration of second pass */
-      dec_arg->servers =
-          gpr_malloc(sizeof(grpc_grpclb_server *) * dec_arg->num_servers);
-    }
-    if (!pb_decode(stream, grpc_lb_v1_Server_fields, server)) {
-      gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(stream));
-      return false;
-    }
-    dec_arg->servers[dec_arg->decoding_idx++] = server;
+  GPR_ASSERT(dec_arg->serverlist->num_servers >= dec_arg->decoding_idx);
+  grpc_grpclb_server *server = gpr_zalloc(sizeof(grpc_grpclb_server));
+  if (!pb_decode(stream, grpc_lb_v1_Server_fields, server)) {
+    gpr_free(server);
+    gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(stream));
+    return false;
   }
-
+  dec_arg->serverlist->servers[dec_arg->decoding_idx++] = server;
   return true;
 }
 
@@ -165,36 +160,38 @@ grpc_grpclb_initial_response *grpc_grpclb_initial_response_parse(
 
 grpc_grpclb_serverlist *grpc_grpclb_response_parse_serverlist(
     grpc_slice encoded_grpc_grpclb_response) {
-  bool status;
-  decode_serverlist_arg arg;
   pb_istream_t stream =
       pb_istream_from_buffer(GRPC_SLICE_START_PTR(encoded_grpc_grpclb_response),
                              GRPC_SLICE_LENGTH(encoded_grpc_grpclb_response));
   pb_istream_t stream_at_start = stream;
+  grpc_grpclb_serverlist *sl = gpr_zalloc(sizeof(grpc_grpclb_serverlist));
   grpc_grpclb_response res;
   memset(&res, 0, sizeof(grpc_grpclb_response));
-  memset(&arg, 0, sizeof(decode_serverlist_arg));
-
-  res.server_list.servers.funcs.decode = decode_serverlist;
-  res.server_list.servers.arg = &arg;
-  arg.first_pass = true;
-  status = pb_decode(&stream, grpc_lb_v1_LoadBalanceResponse_fields, &res);
+  // First pass: count number of servers.
+  res.server_list.servers.funcs.decode = count_serverlist;
+  res.server_list.servers.arg = sl;
+  bool status = pb_decode(&stream, grpc_lb_v1_LoadBalanceResponse_fields, &res);
   if (!status) {
+    gpr_free(sl);
     gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(&stream));
     return NULL;
   }
-
-  arg.first_pass = false;
-  status =
-      pb_decode(&stream_at_start, grpc_lb_v1_LoadBalanceResponse_fields, &res);
-  if (!status) {
-    gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(&stream));
-    return NULL;
+  // Second pass: populate servers.
+  if (sl->num_servers > 0) {
+    sl->servers = gpr_zalloc(sizeof(grpc_grpclb_server *) * sl->num_servers);
+    decode_serverlist_arg decode_arg;
+    memset(&decode_arg, 0, sizeof(decode_arg));
+    decode_arg.serverlist = sl;
+    res.server_list.servers.funcs.decode = decode_serverlist;
+    res.server_list.servers.arg = &decode_arg;
+    status = pb_decode(&stream_at_start, grpc_lb_v1_LoadBalanceResponse_fields,
+                       &res);
+    if (!status) {
+      grpc_grpclb_destroy_serverlist(sl);
+      gpr_log(GPR_ERROR, "nanopb error: %s", PB_GET_ERROR(&stream));
+      return NULL;
+    }
   }
-
-  grpc_grpclb_serverlist *sl = gpr_zalloc(sizeof(grpc_grpclb_serverlist));
-  sl->num_servers = arg.num_servers;
-  sl->servers = arg.servers;
   if (res.server_list.has_expiration_interval) {
     sl->expiration_interval = res.server_list.expiration_interval;
   }
@@ -228,7 +225,7 @@ grpc_grpclb_serverlist *grpc_grpclb_serverlist_copy(
 
 bool grpc_grpclb_serverlist_equals(const grpc_grpclb_serverlist *lhs,
                                    const grpc_grpclb_serverlist *rhs) {
-  if ((lhs == NULL) || (rhs == NULL)) {
+  if (lhs == NULL || rhs == NULL) {
     return false;
   }
   if (lhs->num_servers != rhs->num_servers) {
diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h
index 06873821bd..7f596ce1f1 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h
@@ -51,7 +51,7 @@ typedef grpc_lb_v1_LoadBalanceRequest grpc_grpclb_request;
 typedef grpc_lb_v1_InitialLoadBalanceResponse grpc_grpclb_initial_response;
 typedef grpc_lb_v1_Server grpc_grpclb_server;
 typedef grpc_lb_v1_Duration grpc_grpclb_duration;
-typedef struct grpc_grpclb_serverlist {
+typedef struct {
   grpc_grpclb_server **servers;
   size_t num_servers;
   grpc_grpclb_duration expiration_interval;
diff --git a/src/core/ext/filters/client_channel/parse_address.c b/src/core/ext/filters/client_channel/parse_address.c
index edc6ce697d..18381eec55 100644
--- a/src/core/ext/filters/client_channel/parse_address.c
+++ b/src/core/ext/filters/client_channel/parse_address.c
@@ -57,11 +57,11 @@ bool grpc_parse_unix(const grpc_uri *uri,
   struct sockaddr_un *un = (struct sockaddr_un *)resolved_addr->addr;
   const size_t maxlen = sizeof(un->sun_path);
   const size_t path_len = strnlen(uri->path, maxlen);
-  if (path_len == maxlen) return 0;
+  if (path_len == maxlen) return false;
   un->sun_family = AF_UNIX;
   strcpy(un->sun_path, uri->path);
   resolved_addr->len = sizeof(*un);
-  return 1;
+  return true;
 }
 
 #else /* GRPC_HAVE_UNIX_SOCKET */
@@ -73,74 +73,65 @@ bool grpc_parse_unix(const grpc_uri *uri,
 
 #endif /* GRPC_HAVE_UNIX_SOCKET */
 
-bool grpc_parse_ipv4(const grpc_uri *uri,
-                     grpc_resolved_address *resolved_addr) {
-  if (strcmp("ipv4", uri->scheme) != 0) {
-    gpr_log(GPR_ERROR, "Expected 'ipv4' scheme, got '%s'", uri->scheme);
-    return false;
-  }
-  const char *host_port = uri->path;
+bool grpc_parse_ipv4_hostport(const char *hostport, grpc_resolved_address *addr,
+                              bool log_errors) {
+  bool success = false;
+  // Split host and port.
   char *host;
   char *port;
-  int port_num;
-  bool result = false;
-  struct sockaddr_in *in = (struct sockaddr_in *)resolved_addr->addr;
-
-  if (*host_port == '/') ++host_port;
-  if (!gpr_split_host_port(host_port, &host, &port)) {
-    return false;
-  }
-
-  memset(resolved_addr, 0, sizeof(grpc_resolved_address));
-  resolved_addr->len = sizeof(struct sockaddr_in);
+  if (!gpr_split_host_port(hostport, &host, &port)) return false;
+  // Parse IP address.
+  memset(addr, 0, sizeof(*addr));
+  addr->len = sizeof(struct sockaddr_in);
+  struct sockaddr_in *in = (struct sockaddr_in *)addr->addr;
   in->sin_family = AF_INET;
   if (inet_pton(AF_INET, host, &in->sin_addr) == 0) {
-    gpr_log(GPR_ERROR, "invalid ipv4 address: '%s'", host);
+    if (log_errors) gpr_log(GPR_ERROR, "invalid ipv4 address: '%s'", host);
     goto done;
   }
-
-  if (port != NULL) {
-    if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 ||
-        port_num > 65535) {
-      gpr_log(GPR_ERROR, "invalid ipv4 port: '%s'", port);
-      goto done;
-    }
-    in->sin_port = htons((uint16_t)port_num);
-  } else {
-    gpr_log(GPR_ERROR, "no port given for ipv4 scheme");
+  // Parse port.
+  if (port == NULL) {
+    if (log_errors) gpr_log(GPR_ERROR, "no port given for ipv4 scheme");
     goto done;
   }
-
-  result = true;
+  int port_num;
+  if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 || port_num > 65535) {
+    if (log_errors) gpr_log(GPR_ERROR, "invalid ipv4 port: '%s'", port);
+    goto done;
+  }
+  in->sin_port = htons((uint16_t)port_num);
+  success = true;
 done:
   gpr_free(host);
   gpr_free(port);
-  return result;
+  return success;
 }
 
-bool grpc_parse_ipv6(const grpc_uri *uri,
+bool grpc_parse_ipv4(const grpc_uri *uri,
                      grpc_resolved_address *resolved_addr) {
-  if (strcmp("ipv6", uri->scheme) != 0) {
-    gpr_log(GPR_ERROR, "Expected 'ipv6' scheme, got '%s'", uri->scheme);
+  if (strcmp("ipv4", uri->scheme) != 0) {
+    gpr_log(GPR_ERROR, "Expected 'ipv4' scheme, got '%s'", uri->scheme);
     return false;
   }
   const char *host_port = uri->path;
-  char *host;
-  char *port;
-  int port_num;
-  int result = 0;
-  struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)resolved_addr->addr;
-
   if (*host_port == '/') ++host_port;
-  if (!gpr_split_host_port(host_port, &host, &port)) {
-    return 0;
-  }
+  return grpc_parse_ipv4_hostport(host_port, resolved_addr,
+                                  true /* log_errors */);
+}
 
-  memset(in6, 0, sizeof(*in6));
-  resolved_addr->len = sizeof(*in6);
+bool grpc_parse_ipv6_hostport(const char *hostport, grpc_resolved_address *addr,
+                              bool log_errors) {
+  bool success = false;
+  // Split host and port.
+  char *host;
+  char *port;
+  if (!gpr_split_host_port(hostport, &host, &port)) return false;
+  // Parse IP address.
+  memset(addr, 0, sizeof(*addr));
+  addr->len = sizeof(struct sockaddr_in6);
+  struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)addr->addr;
   in6->sin6_family = AF_INET6;
-
-  /* Handle the RFC6874 syntax for IPv6 zone identifiers. */
+  // Handle the RFC6874 syntax for IPv6 zone identifiers.
   char *host_end = (char *)gpr_memrchr(host, '%', strlen(host));
   if (host_end != NULL) {
     GPR_ASSERT(host_end >= host);
@@ -159,7 +150,7 @@ bool grpc_parse_ipv6(const grpc_uri *uri,
       gpr_log(GPR_ERROR, "invalid ipv6 scope id: '%s'", host_end + 1);
       goto done;
     }
-    // Handle "sin6_scope_id" being type "u_long". See grpc issue ##10027.
+    // Handle "sin6_scope_id" being type "u_long". See grpc issue #10027.
     in6->sin6_scope_id = sin6_scope_id;
   } else {
     if (inet_pton(AF_INET6, host, &in6->sin6_addr) == 0) {
@@ -167,24 +158,34 @@ bool grpc_parse_ipv6(const grpc_uri *uri,
       goto done;
     }
   }
-
-  if (port != NULL) {
-    if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 ||
-        port_num > 65535) {
-      gpr_log(GPR_ERROR, "invalid ipv6 port: '%s'", port);
-      goto done;
-    }
-    in6->sin6_port = htons((uint16_t)port_num);
-  } else {
-    gpr_log(GPR_ERROR, "no port given for ipv6 scheme");
+  // Parse port.
+  if (port == NULL) {
+    if (log_errors) gpr_log(GPR_ERROR, "no port given for ipv6 scheme");
     goto done;
   }
-
-  result = 1;
+  int port_num;
+  if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 || port_num > 65535) {
+    if (log_errors) gpr_log(GPR_ERROR, "invalid ipv6 port: '%s'", port);
+    goto done;
+  }
+  in6->sin6_port = htons((uint16_t)port_num);
+  success = true;
 done:
   gpr_free(host);
   gpr_free(port);
-  return result;
+  return success;
+}
+
+bool grpc_parse_ipv6(const grpc_uri *uri,
+                     grpc_resolved_address *resolved_addr) {
+  if (strcmp("ipv6", uri->scheme) != 0) {
+    gpr_log(GPR_ERROR, "Expected 'ipv6' scheme, got '%s'", uri->scheme);
+    return false;
+  }
+  const char *host_port = uri->path;
+  if (*host_port == '/') ++host_port;
+  return grpc_parse_ipv6_hostport(host_port, resolved_addr,
+                                  true /* log_errors */);
 }
 
 bool grpc_parse_uri(const grpc_uri *uri, grpc_resolved_address *resolved_addr) {
diff --git a/src/core/ext/filters/client_channel/parse_address.h b/src/core/ext/filters/client_channel/parse_address.h
index fa7ea33a00..1a203a3b26 100644
--- a/src/core/ext/filters/client_channel/parse_address.h
+++ b/src/core/ext/filters/client_channel/parse_address.h
@@ -54,4 +54,10 @@ bool grpc_parse_ipv6(const grpc_uri *uri, grpc_resolved_address *resolved_addr);
 /** Populate \a resolved_addr from \a uri. Returns true upon success. */
 bool grpc_parse_uri(const grpc_uri *uri, grpc_resolved_address *resolved_addr);
 
+/** Parse bare IPv4 or IPv6 "IP:port" strings. */
+bool grpc_parse_ipv4_hostport(const char *hostport, grpc_resolved_address *addr,
+                              bool log_errors);
+bool grpc_parse_ipv6_hostport(const char *hostport, grpc_resolved_address *addr,
+                              bool log_errors);
+
 #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_PARSE_ADDRESS_H */
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c
index ca7c7315b5..85986e5617 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c
@@ -61,6 +61,8 @@
 typedef struct {
   /** base class: must be first */
   grpc_resolver base;
+  /** DNS server to use (if not system default) */
+  char *dns_server;
   /** name to resolve (usually the same as target_name) */
   char *name_to_resolve;
   /** default port to use */
@@ -163,6 +165,8 @@ static void dns_ares_on_resolved_locked(grpc_exec_ctx *exec_ctx, void *arg,
     result = grpc_channel_args_copy_and_add(r->channel_args, &new_arg, 1);
     grpc_lb_addresses_destroy(exec_ctx, r->lb_addresses);
   } else {
+    const char *msg = grpc_error_string(error);
+    gpr_log(GPR_DEBUG, "dns resolution failed: %s", msg);
     gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
     gpr_timespec next_try = gpr_backoff_step(&r->backoff_state, now);
     gpr_timespec timeout = gpr_time_sub(next_try, now);
@@ -212,9 +216,10 @@ static void dns_ares_start_resolving_locked(grpc_exec_ctx *exec_ctx,
   GPR_ASSERT(!r->resolving);
   r->resolving = true;
   r->lb_addresses = NULL;
-  grpc_resolve_grpclb_address_ares(
-      exec_ctx, r->name_to_resolve, r->default_port, r->interested_parties,
-      &r->dns_ares_on_resolved_locked, &r->lb_addresses);
+  grpc_dns_lookup_ares(exec_ctx, r->dns_server, r->name_to_resolve,
+                       r->default_port, r->interested_parties,
+                       &r->dns_ares_on_resolved_locked,
+                       (void **)&r->lb_addresses, true /* is_lb_addrs_out */);
 }
 
 static void dns_ares_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
@@ -238,6 +243,7 @@ static void dns_ares_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *gr) {
     grpc_channel_args_destroy(exec_ctx, r->resolved_result);
   }
   grpc_pollset_set_destroy(exec_ctx, r->interested_parties);
+  gpr_free(r->dns_server);
   gpr_free(r->name_to_resolve);
   gpr_free(r->default_port);
   grpc_channel_args_destroy(exec_ctx, r->channel_args);
@@ -249,14 +255,13 @@ static grpc_resolver *dns_ares_create(grpc_exec_ctx *exec_ctx,
                                       const char *default_port) {
   /* Get name from args. */
   const char *path = args->uri->path;
-  if (0 != strcmp(args->uri->authority, "")) {
-    gpr_log(GPR_ERROR, "authority based dns uri's not supported");
-    return NULL;
-  }
   if (path[0] == '/') ++path;
   /* Create resolver. */
   ares_dns_resolver *r = gpr_zalloc(sizeof(ares_dns_resolver));
   grpc_resolver_init(&r->base, &dns_ares_resolver_vtable, args->combiner);
+  if (0 != strcmp(args->uri->authority, "")) {
+    r->dns_server = gpr_strdup(args->uri->authority);
+  }
   r->name_to_resolve = gpr_strdup(path);
   r->default_port = gpr_strdup(default_port);
   r->channel_args = grpc_channel_args_copy(args->args);
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c
index 55059018c8..ead078caa0 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c
@@ -49,7 +49,10 @@
 #include <grpc/support/time.h>
 #include <grpc/support/useful.h>
 #include <nameser.h>
+
+#include "src/core/ext/filters/client_channel/parse_address.h"
 #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h"
+#include "src/core/lib/iomgr/error.h"
 #include "src/core/lib/iomgr/executor.h"
 #include "src/core/lib/iomgr/iomgr_internal.h"
 #include "src/core/lib/iomgr/sockaddr_utils.h"
@@ -59,6 +62,8 @@ static gpr_once g_basic_init = GPR_ONCE_INIT;
 static gpr_mu g_init_mu;
 
 typedef struct grpc_ares_request {
+  /** indicates the DNS server to use, if specified */
+  struct ares_addr_port_node dns_server_addr;
   /** following members are set in grpc_resolve_address_ares_impl */
   /** closure to call when the request completes */
   grpc_closure *on_done;
@@ -332,11 +337,12 @@ static void on_srv_query_done_cb(void *arg, int status, int timeouts,
   grpc_exec_ctx_finish(&exec_ctx);
 }
 
-static void resolve_address_ares_impl(grpc_exec_ctx *exec_ctx, const char *name,
-                                      const char *default_port,
-                                      grpc_pollset_set *interested_parties,
-                                      grpc_closure *on_done, void **addrs,
-                                      bool is_lb_addrs_out) {
+void grpc_dns_lookup_ares(
+    grpc_exec_ctx *exec_ctx, const char *dns_server, const char *name,
+    const char *default_port, grpc_pollset_set *interested_parties,
+    grpc_closure *on_done,
+    void **addrs, bool is_lb_addrs_out) {
+  grpc_error *error = GRPC_ERROR_NONE;
   /* TODO(zyc): Enable tracing after #9603 is checked in */
   /* if (grpc_dns_trace) {
       gpr_log(GPR_DEBUG, "resolve_address (blocking): name=%s, default_port=%s",
@@ -348,28 +354,23 @@ static void resolve_address_ares_impl(grpc_exec_ctx *exec_ctx, const char *name,
   char *port;
   gpr_split_host_port(name, &host, &port);
   if (host == NULL) {
-    grpc_error *err = grpc_error_set_str(
+    error = grpc_error_set_str(
         GRPC_ERROR_CREATE_FROM_STATIC_STRING("unparseable host:port"),
         GRPC_ERROR_STR_TARGET_ADDRESS, grpc_slice_from_copied_string(name));
-    grpc_closure_sched(exec_ctx, on_done, err);
     goto error_cleanup;
   } else if (port == NULL) {
     if (default_port == NULL) {
-      grpc_error *err = grpc_error_set_str(
+      error = grpc_error_set_str(
           GRPC_ERROR_CREATE_FROM_STATIC_STRING("no port in name"),
           GRPC_ERROR_STR_TARGET_ADDRESS, grpc_slice_from_copied_string(name));
-      grpc_closure_sched(exec_ctx, on_done, err);
       goto error_cleanup;
     }
     port = gpr_strdup(default_port);
   }
 
   grpc_ares_ev_driver *ev_driver;
-  grpc_error *err = grpc_ares_ev_driver_create(&ev_driver, interested_parties);
-  if (err != GRPC_ERROR_NONE) {
-    GRPC_LOG_IF_ERROR("grpc_ares_ev_driver_create() failed", err);
-    goto error_cleanup;
-  }
+  error = grpc_ares_ev_driver_create(&ev_driver, interested_parties);
+  if (error != GRPC_ERROR_NONE) goto error_cleanup;
 
   grpc_ares_request *r = gpr_malloc(sizeof(grpc_ares_request));
   gpr_mu_init(&r->mu);
@@ -384,6 +385,40 @@ static void resolve_address_ares_impl(grpc_exec_ctx *exec_ctx, const char *name,
   r->success = false;
   r->error = GRPC_ERROR_NONE;
   ares_channel *channel = grpc_ares_ev_driver_get_channel(r->ev_driver);
+
+  // If dns_server is specified, use it.
+  if (dns_server != NULL) {
+    gpr_log(GPR_INFO, "Using DNS server %s", dns_server);
+    grpc_resolved_address addr;
+    if (grpc_parse_ipv4_hostport(dns_server, &addr, false /* log_errors */)) {
+      r->dns_server_addr.family = AF_INET;
+      memcpy(&r->dns_server_addr.addr.addr4, addr.addr, addr.len);
+      r->dns_server_addr.tcp_port = grpc_sockaddr_get_port(&addr);
+      r->dns_server_addr.udp_port = grpc_sockaddr_get_port(&addr);
+    } else if (grpc_parse_ipv6_hostport(dns_server, &addr,
+                                        false /* log_errors */)) {
+      r->dns_server_addr.family = AF_INET6;
+      memcpy(&r->dns_server_addr.addr.addr6, addr.addr, addr.len);
+      r->dns_server_addr.tcp_port = grpc_sockaddr_get_port(&addr);
+      r->dns_server_addr.udp_port = grpc_sockaddr_get_port(&addr);
+    } else {
+      error = grpc_error_set_str(
+          GRPC_ERROR_CREATE_FROM_STATIC_STRING("cannot parse authority"),
+          GRPC_ERROR_STR_TARGET_ADDRESS, grpc_slice_from_copied_string(name));
+      goto error_cleanup;
+    }
+    int status = ares_set_servers_ports(*channel, &r->dns_server_addr);
+    if (status != ARES_SUCCESS) {
+      char *error_msg;
+      gpr_asprintf(&error_msg, "C-ares status is not ARES_SUCCESS: %s",
+                   ares_strerror(status));
+      error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_msg);
+      gpr_free(error_msg);
+      goto error_cleanup;
+    }
+  }
+  // An extra reference is put here to avoid destroying the request in
+  // on_done_cb before calling grpc_ares_ev_driver_start.
   gpr_ref_init(&r->pending_queries, 1);
   if (grpc_ipv6_loopback_available()) {
     grpc_ares_hostbyname_request *hr = create_hostbyname_request(
@@ -408,6 +443,7 @@ static void resolve_address_ares_impl(grpc_exec_ctx *exec_ctx, const char *name,
   return;
 
 error_cleanup:
+  grpc_closure_sched(exec_ctx, on_done, error);
   gpr_free(host);
   gpr_free(port);
 }
@@ -417,19 +453,9 @@ void grpc_resolve_address_ares_impl(grpc_exec_ctx *exec_ctx, const char *name,
                                     grpc_pollset_set *interested_parties,
                                     grpc_closure *on_done,
                                     grpc_resolved_addresses **addrs) {
-  resolve_address_ares_impl(exec_ctx, name, default_port, interested_parties,
-                            on_done, (void **)addrs,
-                            false /* is_lb_addrs_out */);
-}
-
-void grpc_resolve_grpclb_address_ares(grpc_exec_ctx *exec_ctx, const char *name,
-                                      const char *default_port,
-                                      grpc_pollset_set *interested_parties,
-                                      grpc_closure *on_done,
-                                      grpc_lb_addresses **addrs) {
-  resolve_address_ares_impl(exec_ctx, name, default_port, interested_parties,
-                            on_done, (void **)addrs,
-                            true /* is_lb_addrs_out */);
+  grpc_dns_lookup_ares(exec_ctx, NULL /* dns_server */, name, default_port,
+                       interested_parties, on_done, (void **)addrs,
+                       false /* is_lb_addrs_out */);
 }
 
 void (*grpc_resolve_address_ares)(
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h
index 1450ef373d..47d7b146f6 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h
@@ -59,11 +59,11 @@ extern void (*grpc_resolve_address_ares)(grpc_exec_ctx *exec_ctx,
   function. \a on_done may be called directly in this function without being
   scheduled with \a exec_ctx, it must not try to acquire locks that are being
   held by the caller. */
-void grpc_resolve_grpclb_address_ares(grpc_exec_ctx *exec_ctx, const char *addr,
-                                      const char *default_port,
-                                      grpc_pollset_set *interested_parties,
-                                      grpc_closure *on_done,
-                                      grpc_lb_addresses **addresses);
+void grpc_dns_lookup_ares(grpc_exec_ctx *exec_ctx, const char *dns_server,
+                          const char *addr, const char *default_port,
+                          grpc_pollset_set *interested_parties,
+                          grpc_closure *on_done, void **addresses,
+                          bool is_lb_addrs_out);
 
 /* Initialize gRPC ares wrapper. Must be called at least once before
    grpc_resolve_address_ares(). */
diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c
index 1af3393a62..dd14bf1d02 100644
--- a/src/core/ext/filters/client_channel/subchannel.c
+++ b/src/core/ext/filters/client_channel/subchannel.c
@@ -283,6 +283,7 @@ static void disconnect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
 void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
                            grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
   gpr_atm old_refs;
+  // add a weak ref and subtract a strong ref (atomically)
   old_refs = ref_mutate(c, (gpr_atm)1 - (gpr_atm)(1 << INTERNAL_REF_BITS),
                         1 REF_MUTATE_PURPOSE("STRONG_UNREF"));
   if ((old_refs & STRONG_REF_MASK) == (1 << INTERNAL_REF_BITS)) {
@@ -656,7 +657,6 @@ static bool publish_transport_locked(grpc_exec_ctx *exec_ctx,
     gpr_free(sw_subchannel);
     grpc_channel_stack_destroy(exec_ctx, stk);
     gpr_free(con);
-    GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
     return false;
   }
 
diff --git a/src/core/ext/filters/client_channel/subchannel_index.c b/src/core/ext/filters/client_channel/subchannel_index.c
index f6ef4a845e..b25dbfcf51 100644
--- a/src/core/ext/filters/client_channel/subchannel_index.c
+++ b/src/core/ext/filters/client_channel/subchannel_index.c
@@ -183,8 +183,11 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
   enter_ctx(exec_ctx);
 
   grpc_subchannel *c = NULL;
+  bool need_to_unref_constructed;
 
   while (c == NULL) {
+    need_to_unref_constructed = false;
+
     // Compare and swap loop:
     // - take a reference to the current index
     gpr_mu_lock(&g_mu);
@@ -194,8 +197,11 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
     // - Check to see if a subchannel already exists
     c = gpr_avl_get(index, key);
     if (c != NULL) {
+      c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(c, "index_register");
+    }
+    if (c != NULL) {
       // yes -> we're done
-      GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, constructed, "index_register");
+      need_to_unref_constructed = true;
     } else {
       // no -> update the avl and compare/swap
       gpr_avl updated =
@@ -219,6 +225,10 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
 
   leave_ctx(exec_ctx);
 
+  if (need_to_unref_constructed) {
+    GRPC_SUBCHANNEL_UNREF(exec_ctx, constructed, "index_register");
+  }
+
   return c;
 }