Merge github.com:grpc/grpc into framing_costs

11 files changed, 282 insertions, 110 deletions

diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c
index 16be2c70e9..ce9abdad61 100644
--- a/src/core/ext/filters/client_channel/client_channel.c
+++ b/src/core/ext/filters/client_channel/client_channel.c
@@ -238,14 +238,23 @@ static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx,
                                                   grpc_connectivity_state state,
                                                   grpc_error *error,
                                                   const char *reason) {
-  if ((state == GRPC_CHANNEL_TRANSIENT_FAILURE ||
-       state == GRPC_CHANNEL_SHUTDOWN) &&
-      chand->lb_policy != NULL) {
-    /* cancel picks with wait_for_ready=false */
-    grpc_lb_policy_cancel_picks_locked(
-        exec_ctx, chand->lb_policy,
-        /* mask= */ GRPC_INITIAL_METADATA_WAIT_FOR_READY,
-        /* check= */ 0, GRPC_ERROR_REF(error));
+  /* TODO: Improve failure handling:
+   * - Make it possible for policies to return GRPC_CHANNEL_TRANSIENT_FAILURE.
+   * - Hand over pending picks from old policies during the switch that happens
+   *   when resolver provides an update. */
+  if (chand->lb_policy != NULL) {
+    if (state == GRPC_CHANNEL_TRANSIENT_FAILURE) {
+      /* cancel picks with wait_for_ready=false */
+      grpc_lb_policy_cancel_picks_locked(
+          exec_ctx, chand->lb_policy,
+          /* mask= */ GRPC_INITIAL_METADATA_WAIT_FOR_READY,
+          /* check= */ 0, GRPC_ERROR_REF(error));
+    } else if (state == GRPC_CHANNEL_SHUTDOWN) {
+      /* cancel all picks */
+      grpc_lb_policy_cancel_picks_locked(exec_ctx, chand->lb_policy,
+                                         /* mask= */ 0, /* check= */ 0,
+                                         GRPC_ERROR_REF(error));
+    }
   }
   grpc_connectivity_state_set(exec_ctx, &chand->state_tracker, state, error,
                               reason);
@@ -348,6 +357,33 @@ static void parse_retry_throttle_params(const grpc_json *field, void *arg) {
   }
 }
 
+// Wrap a closure associated with \a lb_policy. The associated callback (\a
+// wrapped_on_pick_closure_cb) is responsible for unref'ing \a lb_policy after
+// scheduling \a wrapped_closure.
+typedef struct wrapped_on_pick_closure_arg {
+  /* the closure instance using this struct as argument */
+  grpc_closure wrapper_closure;
+
+  /* the original closure. Usually a on_complete/notify cb for pick() and ping()
+   * calls against the internal RR instance, respectively. */
+  grpc_closure *wrapped_closure;
+
+  /* The policy instance related to the closure */
+  grpc_lb_policy *lb_policy;
+} wrapped_on_pick_closure_arg;
+
+// Invoke \a arg->wrapped_closure, unref \a arg->lb_policy and free \a arg.
+static void wrapped_on_pick_closure_cb(grpc_exec_ctx *exec_ctx, void *arg,
+                                       grpc_error *error) {
+  wrapped_on_pick_closure_arg *wc_arg = arg;
+  GPR_ASSERT(wc_arg != NULL);
+  GPR_ASSERT(wc_arg->wrapped_closure != NULL);
+  GPR_ASSERT(wc_arg->lb_policy != NULL);
+  grpc_closure_run(exec_ctx, wc_arg->wrapped_closure, GRPC_ERROR_REF(error));
+  GRPC_LB_POLICY_UNREF(exec_ctx, wc_arg->lb_policy, "pick_subchannel_wrapping");
+  gpr_free(wc_arg);
+}
+
 static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx,
                                               void *arg, grpc_error *error) {
   channel_data *chand = arg;
@@ -1037,11 +1073,29 @@ static bool pick_subchannel_locked(
     const grpc_lb_policy_pick_args inputs = {
         initial_metadata, initial_metadata_flags, &calld->lb_token_mdelem,
         gpr_inf_future(GPR_CLOCK_MONOTONIC)};
-    const bool result = grpc_lb_policy_pick_locked(
-        exec_ctx, lb_policy, &inputs, connected_subchannel, NULL, on_ready);
+
+    // Wrap the user-provided callback in order to hold a strong reference to
+    // the LB policy for the duration of the pick.
+    wrapped_on_pick_closure_arg *w_on_pick_arg =
+        gpr_zalloc(sizeof(*w_on_pick_arg));
+    grpc_closure_init(&w_on_pick_arg->wrapper_closure,
+                      wrapped_on_pick_closure_cb, w_on_pick_arg,
+                      grpc_schedule_on_exec_ctx);
+    w_on_pick_arg->wrapped_closure = on_ready;
+    GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel_wrapping");
+    w_on_pick_arg->lb_policy = lb_policy;
+    const bool pick_done = grpc_lb_policy_pick_locked(
+        exec_ctx, lb_policy, &inputs, connected_subchannel, NULL,
+        &w_on_pick_arg->wrapper_closure);
+    if (pick_done) {
+      /* synchronous grpc_lb_policy_pick call. Unref the LB policy. */
+      GRPC_LB_POLICY_UNREF(exec_ctx, w_on_pick_arg->lb_policy,
+                           "pick_subchannel_wrapping");
+      gpr_free(w_on_pick_arg);
+    }
     GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick_subchannel");
     GPR_TIMER_END("pick_subchannel", 0);
-    return result;
+    return pick_done;
   }
   if (chand->resolver != NULL && !chand->started_resolving) {
     chand->started_resolving = true;
diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c
index e9adf98711..fb119c7fc8 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.c
@@ -16,6 +16,12 @@ const pb_field_t grpc_lb_v1_Duration_fields[3] = {
     PB_LAST_FIELD
 };
 
+const pb_field_t grpc_lb_v1_Timestamp_fields[3] = {
+    PB_FIELD(  1, INT64   , OPTIONAL, STATIC  , FIRST, grpc_lb_v1_Timestamp, seconds, seconds, 0),
+    PB_FIELD(  2, INT32   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_Timestamp, nanos, seconds, 0),
+    PB_LAST_FIELD
+};
+
 const pb_field_t grpc_lb_v1_LoadBalanceRequest_fields[3] = {
     PB_FIELD(  1, MESSAGE , OPTIONAL, STATIC  , FIRST, grpc_lb_v1_LoadBalanceRequest, initial_request, initial_request, &grpc_lb_v1_InitialLoadBalanceRequest_fields),
     PB_FIELD(  2, MESSAGE , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_LoadBalanceRequest, client_stats, initial_request, &grpc_lb_v1_ClientStats_fields),
@@ -27,10 +33,14 @@ const pb_field_t grpc_lb_v1_InitialLoadBalanceRequest_fields[2] = {
     PB_LAST_FIELD
 };
 
-const pb_field_t grpc_lb_v1_ClientStats_fields[4] = {
-    PB_FIELD(  1, INT64   , OPTIONAL, STATIC  , FIRST, grpc_lb_v1_ClientStats, total_requests, total_requests, 0),
-    PB_FIELD(  2, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, client_rpc_errors, total_requests, 0),
-    PB_FIELD(  3, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, dropped_requests, client_rpc_errors, 0),
+const pb_field_t grpc_lb_v1_ClientStats_fields[8] = {
+    PB_FIELD(  1, MESSAGE , OPTIONAL, STATIC  , FIRST, grpc_lb_v1_ClientStats, timestamp, timestamp, &grpc_lb_v1_Timestamp_fields),
+    PB_FIELD(  2, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, num_calls_started, timestamp, 0),
+    PB_FIELD(  3, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, num_calls_finished, num_calls_started, 0),
+    PB_FIELD(  4, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_with_drop_for_rate_limiting, num_calls_finished, 0),
+    PB_FIELD(  5, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_with_drop_for_load_balancing, num_calls_finished_with_drop_for_rate_limiting, 0),
+    PB_FIELD(  6, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_with_client_failed_to_send, num_calls_finished_with_drop_for_load_balancing, 0),
+    PB_FIELD(  7, INT64   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_ClientStats, num_calls_finished_known_received, num_calls_finished_with_client_failed_to_send, 0),
     PB_LAST_FIELD
 };
 
@@ -52,11 +62,12 @@ const pb_field_t grpc_lb_v1_ServerList_fields[3] = {
     PB_LAST_FIELD
 };
 
-const pb_field_t grpc_lb_v1_Server_fields[5] = {
+const pb_field_t grpc_lb_v1_Server_fields[6] = {
     PB_FIELD(  1, BYTES   , OPTIONAL, STATIC  , FIRST, grpc_lb_v1_Server, ip_address, ip_address, 0),
     PB_FIELD(  2, INT32   , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_Server, port, ip_address, 0),
     PB_FIELD(  3, STRING  , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_Server, load_balance_token, port, 0),
-    PB_FIELD(  4, BOOL    , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_Server, drop_request, load_balance_token, 0),
+    PB_FIELD(  4, BOOL    , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_Server, drop_for_rate_limiting, load_balance_token, 0),
+    PB_FIELD(  5, BOOL    , OPTIONAL, STATIC  , OTHER, grpc_lb_v1_Server, drop_for_load_balancing, drop_for_rate_limiting, 0),
     PB_LAST_FIELD
 };
 
@@ -70,7 +81,7 @@ const pb_field_t grpc_lb_v1_Server_fields[5] = {
  * numbers or field sizes that are larger than what can fit in 8 or 16 bit
  * field descriptors.
  */
-PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 65536 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 65536 && pb_membersize(grpc_lb_v1_ServerList, servers) < 65536 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 65536), YOU_MUST_DEFINE_PB_FIELD_32BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server)
+PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 65536 && pb_membersize(grpc_lb_v1_ClientStats, timestamp) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 65536 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 65536 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 65536 && pb_membersize(grpc_lb_v1_ServerList, servers) < 65536 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 65536), YOU_MUST_DEFINE_PB_FIELD_32BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_Timestamp_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server)
 #endif
 
 #if !defined(PB_FIELD_16BIT) && !defined(PB_FIELD_32BIT)
@@ -81,7 +92,7 @@ PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request)
  * numbers or field sizes that are larger than what can fit in the default
  * 8 bit descriptors.
  */
-PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 256 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 256 && pb_membersize(grpc_lb_v1_ServerList, servers) < 256 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 256), YOU_MUST_DEFINE_PB_FIELD_16BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server)
+PB_STATIC_ASSERT((pb_membersize(grpc_lb_v1_LoadBalanceRequest, initial_request) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceRequest, client_stats) < 256 && pb_membersize(grpc_lb_v1_ClientStats, timestamp) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, initial_response) < 256 && pb_membersize(grpc_lb_v1_LoadBalanceResponse, server_list) < 256 && pb_membersize(grpc_lb_v1_InitialLoadBalanceResponse, client_stats_report_interval) < 256 && pb_membersize(grpc_lb_v1_ServerList, servers) < 256 && pb_membersize(grpc_lb_v1_ServerList, expiration_interval) < 256), YOU_MUST_DEFINE_PB_FIELD_16BIT_FOR_MESSAGES_grpc_lb_v1_Duration_grpc_lb_v1_Timestamp_grpc_lb_v1_LoadBalanceRequest_grpc_lb_v1_InitialLoadBalanceRequest_grpc_lb_v1_ClientStats_grpc_lb_v1_LoadBalanceResponse_grpc_lb_v1_InitialLoadBalanceResponse_grpc_lb_v1_ServerList_grpc_lb_v1_Server)
 #endif
 
 
diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h
index 725aa7e386..d3ae919ec2 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h
@@ -14,16 +14,6 @@ extern "C" {
 #endif
 
 /* Struct definitions */
-typedef struct _grpc_lb_v1_ClientStats {
-    bool has_total_requests;
-    int64_t total_requests;
-    bool has_client_rpc_errors;
-    int64_t client_rpc_errors;
-    bool has_dropped_requests;
-    int64_t dropped_requests;
-/* @@protoc_insertion_point(struct:grpc_lb_v1_ClientStats) */
-} grpc_lb_v1_ClientStats;
-
 typedef struct _grpc_lb_v1_Duration {
     bool has_seconds;
     int64_t seconds;
@@ -46,11 +36,39 @@ typedef struct _grpc_lb_v1_Server {
     int32_t port;
     bool has_load_balance_token;
     char load_balance_token[50];
-    bool has_drop_request;
-    bool drop_request;
+    bool has_drop_for_rate_limiting;
+    bool drop_for_rate_limiting;
+    bool has_drop_for_load_balancing;
+    bool drop_for_load_balancing;
 /* @@protoc_insertion_point(struct:grpc_lb_v1_Server) */
 } grpc_lb_v1_Server;
 
+typedef struct _grpc_lb_v1_Timestamp {
+    bool has_seconds;
+    int64_t seconds;
+    bool has_nanos;
+    int32_t nanos;
+/* @@protoc_insertion_point(struct:grpc_lb_v1_Timestamp) */
+} grpc_lb_v1_Timestamp;
+
+typedef struct _grpc_lb_v1_ClientStats {
+    bool has_timestamp;
+    grpc_lb_v1_Timestamp timestamp;
+    bool has_num_calls_started;
+    int64_t num_calls_started;
+    bool has_num_calls_finished;
+    int64_t num_calls_finished;
+    bool has_num_calls_finished_with_drop_for_rate_limiting;
+    int64_t num_calls_finished_with_drop_for_rate_limiting;
+    bool has_num_calls_finished_with_drop_for_load_balancing;
+    int64_t num_calls_finished_with_drop_for_load_balancing;
+    bool has_num_calls_finished_with_client_failed_to_send;
+    int64_t num_calls_finished_with_client_failed_to_send;
+    bool has_num_calls_finished_known_received;
+    int64_t num_calls_finished_known_received;
+/* @@protoc_insertion_point(struct:grpc_lb_v1_ClientStats) */
+} grpc_lb_v1_ClientStats;
+
 typedef struct _grpc_lb_v1_InitialLoadBalanceResponse {
     bool has_load_balancer_delegate;
     char load_balancer_delegate[64];
@@ -59,6 +77,13 @@ typedef struct _grpc_lb_v1_InitialLoadBalanceResponse {
 /* @@protoc_insertion_point(struct:grpc_lb_v1_InitialLoadBalanceResponse) */
 } grpc_lb_v1_InitialLoadBalanceResponse;
 
+typedef struct _grpc_lb_v1_ServerList {
+    pb_callback_t servers;
+    bool has_expiration_interval;
+    grpc_lb_v1_Duration expiration_interval;
+/* @@protoc_insertion_point(struct:grpc_lb_v1_ServerList) */
+} grpc_lb_v1_ServerList;
+
 typedef struct _grpc_lb_v1_LoadBalanceRequest {
     bool has_initial_request;
     grpc_lb_v1_InitialLoadBalanceRequest initial_request;
@@ -67,13 +92,6 @@ typedef struct _grpc_lb_v1_LoadBalanceRequest {
 /* @@protoc_insertion_point(struct:grpc_lb_v1_LoadBalanceRequest) */
 } grpc_lb_v1_LoadBalanceRequest;
 
-typedef struct _grpc_lb_v1_ServerList {
-    pb_callback_t servers;
-    bool has_expiration_interval;
-    grpc_lb_v1_Duration expiration_interval;
-/* @@protoc_insertion_point(struct:grpc_lb_v1_ServerList) */
-} grpc_lb_v1_ServerList;
-
 typedef struct _grpc_lb_v1_LoadBalanceResponse {
     bool has_initial_response;
     grpc_lb_v1_InitialLoadBalanceResponse initial_response;
@@ -86,61 +104,72 @@ typedef struct _grpc_lb_v1_LoadBalanceResponse {
 
 /* Initializer values for message structs */
 #define grpc_lb_v1_Duration_init_default         {false, 0, false, 0}
+#define grpc_lb_v1_Timestamp_init_default        {false, 0, false, 0}
 #define grpc_lb_v1_LoadBalanceRequest_init_default {false, grpc_lb_v1_InitialLoadBalanceRequest_init_default, false, grpc_lb_v1_ClientStats_init_default}
 #define grpc_lb_v1_InitialLoadBalanceRequest_init_default {false, ""}
-#define grpc_lb_v1_ClientStats_init_default      {false, 0, false, 0, false, 0}
+#define grpc_lb_v1_ClientStats_init_default      {false, grpc_lb_v1_Timestamp_init_default, false, 0, false, 0, false, 0, false, 0, false, 0, false, 0}
 #define grpc_lb_v1_LoadBalanceResponse_init_default {false, grpc_lb_v1_InitialLoadBalanceResponse_init_default, false, grpc_lb_v1_ServerList_init_default}
 #define grpc_lb_v1_InitialLoadBalanceResponse_init_default {false, "", false, grpc_lb_v1_Duration_init_default}
 #define grpc_lb_v1_ServerList_init_default       {{{NULL}, NULL}, false, grpc_lb_v1_Duration_init_default}
-#define grpc_lb_v1_Server_init_default           {false, {0, {0}}, false, 0, false, "", false, 0}
+#define grpc_lb_v1_Server_init_default           {false, {0, {0}}, false, 0, false, "", false, 0, false, 0}
 #define grpc_lb_v1_Duration_init_zero            {false, 0, false, 0}
+#define grpc_lb_v1_Timestamp_init_zero           {false, 0, false, 0}
 #define grpc_lb_v1_LoadBalanceRequest_init_zero  {false, grpc_lb_v1_InitialLoadBalanceRequest_init_zero, false, grpc_lb_v1_ClientStats_init_zero}
 #define grpc_lb_v1_InitialLoadBalanceRequest_init_zero {false, ""}
-#define grpc_lb_v1_ClientStats_init_zero         {false, 0, false, 0, false, 0}
+#define grpc_lb_v1_ClientStats_init_zero         {false, grpc_lb_v1_Timestamp_init_zero, false, 0, false, 0, false, 0, false, 0, false, 0, false, 0}
 #define grpc_lb_v1_LoadBalanceResponse_init_zero {false, grpc_lb_v1_InitialLoadBalanceResponse_init_zero, false, grpc_lb_v1_ServerList_init_zero}
 #define grpc_lb_v1_InitialLoadBalanceResponse_init_zero {false, "", false, grpc_lb_v1_Duration_init_zero}
 #define grpc_lb_v1_ServerList_init_zero          {{{NULL}, NULL}, false, grpc_lb_v1_Duration_init_zero}
-#define grpc_lb_v1_Server_init_zero              {false, {0, {0}}, false, 0, false, "", false, 0}
+#define grpc_lb_v1_Server_init_zero              {false, {0, {0}}, false, 0, false, "", false, 0, false, 0}
 
 /* Field tags (for use in manual encoding/decoding) */
-#define grpc_lb_v1_ClientStats_total_requests_tag 1
-#define grpc_lb_v1_ClientStats_client_rpc_errors_tag 2
-#define grpc_lb_v1_ClientStats_dropped_requests_tag 3
 #define grpc_lb_v1_Duration_seconds_tag          1
 #define grpc_lb_v1_Duration_nanos_tag            2
 #define grpc_lb_v1_InitialLoadBalanceRequest_name_tag 1
 #define grpc_lb_v1_Server_ip_address_tag         1
 #define grpc_lb_v1_Server_port_tag               2
 #define grpc_lb_v1_Server_load_balance_token_tag 3
-#define grpc_lb_v1_Server_drop_request_tag       4
+#define grpc_lb_v1_Server_drop_for_rate_limiting_tag 4
+#define grpc_lb_v1_Server_drop_for_load_balancing_tag 5
+#define grpc_lb_v1_Timestamp_seconds_tag         1
+#define grpc_lb_v1_Timestamp_nanos_tag           2
+#define grpc_lb_v1_ClientStats_timestamp_tag     1
+#define grpc_lb_v1_ClientStats_num_calls_started_tag 2
+#define grpc_lb_v1_ClientStats_num_calls_finished_tag 3
+#define grpc_lb_v1_ClientStats_num_calls_finished_with_drop_for_rate_limiting_tag 4
+#define grpc_lb_v1_ClientStats_num_calls_finished_with_drop_for_load_balancing_tag 5
+#define grpc_lb_v1_ClientStats_num_calls_finished_with_client_failed_to_send_tag 6
+#define grpc_lb_v1_ClientStats_num_calls_finished_known_received_tag 7
 #define grpc_lb_v1_InitialLoadBalanceResponse_load_balancer_delegate_tag 1
 #define grpc_lb_v1_InitialLoadBalanceResponse_client_stats_report_interval_tag 2
-#define grpc_lb_v1_LoadBalanceRequest_initial_request_tag 1
-#define grpc_lb_v1_LoadBalanceRequest_client_stats_tag 2
 #define grpc_lb_v1_ServerList_servers_tag        1
 #define grpc_lb_v1_ServerList_expiration_interval_tag 3
+#define grpc_lb_v1_LoadBalanceRequest_initial_request_tag 1
+#define grpc_lb_v1_LoadBalanceRequest_client_stats_tag 2
 #define grpc_lb_v1_LoadBalanceResponse_initial_response_tag 1
 #define grpc_lb_v1_LoadBalanceResponse_server_list_tag 2
 
 /* Struct field encoding specification for nanopb */
 extern const pb_field_t grpc_lb_v1_Duration_fields[3];
+extern const pb_field_t grpc_lb_v1_Timestamp_fields[3];
 extern const pb_field_t grpc_lb_v1_LoadBalanceRequest_fields[3];
 extern const pb_field_t grpc_lb_v1_InitialLoadBalanceRequest_fields[2];
-extern const pb_field_t grpc_lb_v1_ClientStats_fields[4];
+extern const pb_field_t grpc_lb_v1_ClientStats_fields[8];
 extern const pb_field_t grpc_lb_v1_LoadBalanceResponse_fields[3];
 extern const pb_field_t grpc_lb_v1_InitialLoadBalanceResponse_fields[3];
 extern const pb_field_t grpc_lb_v1_ServerList_fields[3];
-extern const pb_field_t grpc_lb_v1_Server_fields[5];
+extern const pb_field_t grpc_lb_v1_Server_fields[6];
 
 /* Maximum encoded size of messages (where known) */
 #define grpc_lb_v1_Duration_size                 22
-#define grpc_lb_v1_LoadBalanceRequest_size       169
+#define grpc_lb_v1_Timestamp_size                22
+#define grpc_lb_v1_LoadBalanceRequest_size       226
 #define grpc_lb_v1_InitialLoadBalanceRequest_size 131
-#define grpc_lb_v1_ClientStats_size              33
+#define grpc_lb_v1_ClientStats_size              90
 #define grpc_lb_v1_LoadBalanceResponse_size      (98 + grpc_lb_v1_ServerList_size)
 #define grpc_lb_v1_InitialLoadBalanceResponse_size 90
 /* grpc_lb_v1_ServerList_size depends on runtime parameters */
-#define grpc_lb_v1_Server_size                   83
+#define grpc_lb_v1_Server_size                   85
 
 /* Message IDs (where set with "msgid" option) */
 #ifdef PB_MSGID
diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.c b/src/core/ext/filters/client_channel/lb_policy_factory.c
index e2af216b89..89b8bf8951 100644
--- a/src/core/ext/filters/client_channel/lb_policy_factory.c
+++ b/src/core/ext/filters/client_channel/lb_policy_factory.c
@@ -36,16 +36,18 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/string_util.h>
 
+#include "src/core/lib/channel/channel_args.h"
+
 #include "src/core/ext/filters/client_channel/lb_policy_factory.h"
+#include "src/core/ext/filters/client_channel/parse_address.h"
 
 grpc_lb_addresses* grpc_lb_addresses_create(
     size_t num_addresses, const grpc_lb_user_data_vtable* user_data_vtable) {
-  grpc_lb_addresses* addresses = gpr_malloc(sizeof(grpc_lb_addresses));
+  grpc_lb_addresses* addresses = gpr_zalloc(sizeof(grpc_lb_addresses));
   addresses->num_addresses = num_addresses;
   addresses->user_data_vtable = user_data_vtable;
   const size_t addresses_size = sizeof(grpc_lb_address) * num_addresses;
-  addresses->addresses = gpr_malloc(addresses_size);
-  memset(addresses->addresses, 0, addresses_size);
+  addresses->addresses = gpr_zalloc(addresses_size);
   return addresses;
 }
 
@@ -69,7 +71,7 @@ grpc_lb_addresses* grpc_lb_addresses_copy(const grpc_lb_addresses* 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,
+                                   bool is_balancer, const char* balancer_name,
                                    void* user_data) {
   GPR_ASSERT(index < addresses->num_addresses);
   if (user_data != NULL) GPR_ASSERT(addresses->user_data_vtable != NULL);
@@ -77,10 +79,22 @@ void grpc_lb_addresses_set_address(grpc_lb_addresses* addresses, size_t index,
   memcpy(target->address.addr, address, address_len);
   target->address.len = address_len;
   target->is_balancer = is_balancer;
-  target->balancer_name = balancer_name;
+  target->balancer_name = gpr_strdup(balancer_name);
   target->user_data = user_data;
 }
 
+bool grpc_lb_addresses_set_address_from_uri(grpc_lb_addresses* addresses,
+                                            size_t index, const grpc_uri* uri,
+                                            bool is_balancer,
+                                            const char* balancer_name,
+                                            void* user_data) {
+  grpc_resolved_address address;
+  if (!grpc_parse_uri(uri, &address)) return false;
+  grpc_lb_addresses_set_address(addresses, index, address.addr, address.len,
+                                is_balancer, balancer_name, user_data);
+  return true;
+}
+
 int grpc_lb_addresses_cmp(const grpc_lb_addresses* addresses1,
                           const grpc_lb_addresses* addresses2) {
   if (addresses1->num_addresses > addresses2->num_addresses) return 1;
@@ -147,6 +161,15 @@ grpc_arg grpc_lb_addresses_create_channel_arg(
   return arg;
 }
 
+grpc_lb_addresses* grpc_lb_addresses_find_channel_arg(
+    const grpc_channel_args* channel_args) {
+  const grpc_arg* lb_addresses_arg =
+      grpc_channel_args_find(channel_args, GRPC_ARG_LB_ADDRESSES);
+  if (lb_addresses_arg == NULL || lb_addresses_arg->type != GRPC_ARG_POINTER)
+    return NULL;
+  return lb_addresses_arg->value.pointer.p;
+}
+
 void grpc_lb_policy_factory_ref(grpc_lb_policy_factory* factory) {
   factory->vtable->ref(factory);
 }
diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.h b/src/core/ext/filters/client_channel/lb_policy_factory.h
index 81ab12ec8f..9d6c0fc139 100644
--- a/src/core/ext/filters/client_channel/lb_policy_factory.h
+++ b/src/core/ext/filters/client_channel/lb_policy_factory.h
@@ -34,12 +34,13 @@
 #ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_FACTORY_H
 #define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_FACTORY_H
 
-#include "src/core/ext/filters/client_channel/client_channel_factory.h"
-#include "src/core/ext/filters/client_channel/lb_policy.h"
-
 #include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/iomgr/resolve_address.h"
 
+#include "src/core/ext/filters/client_channel/client_channel_factory.h"
+#include "src/core/ext/filters/client_channel/lb_policy.h"
+#include "src/core/ext/filters/client_channel/uri_parser.h"
+
 // Channel arg key for grpc_lb_addresses.
 #define GRPC_ARG_LB_ADDRESSES "grpc.lb_addresses"
 
@@ -88,9 +89,18 @@ grpc_lb_addresses *grpc_lb_addresses_copy(const grpc_lb_addresses *addresses);
  * 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,
+                                   bool is_balancer, const char *balancer_name,
                                    void *user_data);
 
+/** Sets the value of the address at index \a index of \a addresses from \a uri.
+ * Returns true upon success, false otherwise. Takes ownership of \a
+ * balancer_name. */
+bool grpc_lb_addresses_set_address_from_uri(grpc_lb_addresses *addresses,
+                                            size_t index, const grpc_uri *uri,
+                                            bool is_balancer,
+                                            const char *balancer_name,
+                                            void *user_data);
+
 /** Compares \a addresses1 and \a addresses2. */
 int grpc_lb_addresses_cmp(const grpc_lb_addresses *addresses1,
                           const grpc_lb_addresses *addresses2);
@@ -103,6 +113,10 @@ void grpc_lb_addresses_destroy(grpc_exec_ctx *exec_ctx,
 grpc_arg grpc_lb_addresses_create_channel_arg(
     const grpc_lb_addresses *addresses);
 
+/** Returns the \a grpc_lb_addresses instance in \a channel_args or NULL */
+grpc_lb_addresses *grpc_lb_addresses_find_channel_arg(
+    const grpc_channel_args *channel_args);
+
 /** Arguments passed to LB policies. */
 typedef struct grpc_lb_policy_args {
   grpc_client_channel_factory *client_channel_factory;
diff --git a/src/core/ext/filters/client_channel/parse_address.c b/src/core/ext/filters/client_channel/parse_address.c
index 0c97062075..edc6ce697d 100644
--- a/src/core/ext/filters/client_channel/parse_address.c
+++ b/src/core/ext/filters/client_channel/parse_address.c
@@ -48,7 +48,12 @@
 
 #ifdef GRPC_HAVE_UNIX_SOCKET
 
-int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr) {
+bool grpc_parse_unix(const grpc_uri *uri,
+                     grpc_resolved_address *resolved_addr) {
+  if (strcmp("unix", uri->scheme) != 0) {
+    gpr_log(GPR_ERROR, "Expected 'unix' scheme, got '%s'", uri->scheme);
+    return false;
+  }
   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);
@@ -61,21 +66,29 @@ int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr) {
 
 #else /* GRPC_HAVE_UNIX_SOCKET */
 
-int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr) { abort(); }
+bool grpc_parse_unix(const grpc_uri *uri,
+                     grpc_resolved_address *resolved_addr) {
+  abort();
+}
 
 #endif /* GRPC_HAVE_UNIX_SOCKET */
 
-int parse_ipv4(grpc_uri *uri, grpc_resolved_address *resolved_addr) {
+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;
   char *host;
   char *port;
   int port_num;
-  int result = 0;
+  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 0;
+    return false;
   }
 
   memset(resolved_addr, 0, sizeof(grpc_resolved_address));
@@ -98,14 +111,19 @@ int parse_ipv4(grpc_uri *uri, grpc_resolved_address *resolved_addr) {
     goto done;
   }
 
-  result = 1;
+  result = true;
 done:
   gpr_free(host);
   gpr_free(port);
   return result;
 }
 
-int parse_ipv6(grpc_uri *uri, grpc_resolved_address *resolved_addr) {
+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;
   char *host;
   char *port;
@@ -168,3 +186,15 @@ done:
   gpr_free(port);
   return result;
 }
+
+bool grpc_parse_uri(const grpc_uri *uri, grpc_resolved_address *resolved_addr) {
+  if (strcmp("unix", uri->scheme) == 0) {
+    return grpc_parse_unix(uri, resolved_addr);
+  } else if (strcmp("ipv4", uri->scheme) == 0) {
+    return grpc_parse_ipv4(uri, resolved_addr);
+  } else if (strcmp("ipv6", uri->scheme) == 0) {
+    return grpc_parse_ipv6(uri, resolved_addr);
+  }
+  gpr_log(GPR_ERROR, "Can't parse scheme '%s'", uri->scheme);
+  return false;
+}
diff --git a/src/core/ext/filters/client_channel/parse_address.h b/src/core/ext/filters/client_channel/parse_address.h
index c8d77baa00..fa7ea33a00 100644
--- a/src/core/ext/filters/client_channel/parse_address.h
+++ b/src/core/ext/filters/client_channel/parse_address.h
@@ -39,16 +39,19 @@
 #include "src/core/ext/filters/client_channel/uri_parser.h"
 #include "src/core/lib/iomgr/resolve_address.h"
 
-/** Populate \a addr and \a len from \a uri, whose path is expected to contain a
+/** Populate \a resolved_addr from \a uri, whose path is expected to contain a
  * unix socket path. Returns true upon success. */
-int parse_unix(grpc_uri *uri, grpc_resolved_address *resolved_addr);
+bool grpc_parse_unix(const grpc_uri *uri, grpc_resolved_address *resolved_addr);
 
-/** Populate /a addr and \a len from \a uri, whose path is expected to contain a
- * host:port pair. Returns true upon success. */
-int parse_ipv4(grpc_uri *uri, grpc_resolved_address *resolved_addr);
+/** Populate \a resolved_addr from \a uri, whose path is expected to contain an
+ * IPv4 host:port pair. Returns true upon success. */
+bool grpc_parse_ipv4(const grpc_uri *uri, grpc_resolved_address *resolved_addr);
 
-/** Populate /a addr and \a len from \a uri, whose path is expected to contain a
- * host:port pair. Returns true upon success. */
-int parse_ipv6(grpc_uri *uri, grpc_resolved_address *resolved_addr);
+/** Populate \a resolved_addr from \a uri, whose path is expected to contain an
+ * IPv6 host:port pair. Returns true upon success. */
+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);
 
 #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_PARSE_ADDRESS_H */
diff --git a/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c b/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c
index 54f020d691..4d7d878c23 100644
--- a/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c
+++ b/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.c
@@ -157,8 +157,8 @@ static void do_nothing(void *ignored) {}
 
 static grpc_resolver *sockaddr_create(grpc_exec_ctx *exec_ctx,
                                       grpc_resolver_args *args,
-                                      int parse(grpc_uri *uri,
-                                                grpc_resolved_address *dst)) {
+                                      bool parse(const grpc_uri *uri,
+                                                 grpc_resolved_address *dst)) {
   if (0 != strcmp(args->uri->authority, "")) {
     gpr_log(GPR_ERROR, "authority based uri's not supported by the %s scheme",
             args->uri->scheme);
@@ -209,7 +209,7 @@ static void sockaddr_factory_unref(grpc_resolver_factory *factory) {}
   static grpc_resolver *name##_factory_create_resolver(                     \
       grpc_exec_ctx *exec_ctx, grpc_resolver_factory *factory,              \
       grpc_resolver_args *args) {                                           \
-    return sockaddr_create(exec_ctx, args, parse_##name);                   \
+    return sockaddr_create(exec_ctx, args, grpc_parse_##name);              \
   }                                                                         \
   static const grpc_resolver_factory_vtable name##_factory_vtable = {       \
       sockaddr_factory_ref, sockaddr_factory_unref,                         \
diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c
index 9a7a7a0ee5..362b4a4a28 100644
--- a/src/core/ext/filters/client_channel/subchannel.c
+++ b/src/core/ext/filters/client_channel/subchannel.c
@@ -797,13 +797,7 @@ static void grpc_uri_to_sockaddr(grpc_exec_ctx *exec_ctx, const char *uri_str,
                                  grpc_resolved_address *addr) {
   grpc_uri *uri = grpc_uri_parse(exec_ctx, 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));
-  }
+  if (!grpc_parse_uri(uri, addr)) memset(addr, 0, sizeof(*addr));
   grpc_uri_destroy(uri);
 }
 
diff --git a/src/core/ext/filters/client_channel/uri_parser.c b/src/core/ext/filters/client_channel/uri_parser.c
index 01b99911aa..b233d835cb 100644
--- a/src/core/ext/filters/client_channel/uri_parser.c
+++ b/src/core/ext/filters/client_channel/uri_parser.c
@@ -50,7 +50,7 @@
 #define NOT_SET (~(size_t)0)
 
 static grpc_uri *bad_uri(const char *uri_text, size_t pos, const char *section,
-                         int suppress_errors) {
+                         bool suppress_errors) {
   char *line_prefix;
   size_t pfx_len;
 
@@ -83,6 +83,11 @@ static char *decode_and_copy_component(grpc_exec_ctx *exec_ctx, const char *src,
   return out;
 }
 
+static bool valid_hex(char c) {
+  return ((c >= 'a') && (c <= 'f')) || ((c >= 'A') && (c <= 'F')) ||
+         ((c >= '0') && (c <= '9'));
+}
+
 /** Returns how many chars to advance if \a uri_text[i] begins a valid \a pchar
  * production. If \a uri_text[i] introduces an invalid \a pchar (such as percent
  * sign not followed by two hex digits), NOT_SET is returned. */
@@ -93,27 +98,36 @@ static size_t parse_pchar(const char *uri_text, size_t i) {
    * sub-delims = "!" / "$" / "&" / "'" / "(" / ")"
    / "*" / "+" / "," / ";" / "=" */
   char c = uri_text[i];
-  if (((c >= 'A') && (c <= 'Z')) || ((c >= 'a') && (c <= 'z')) ||
-      ((c >= '0') && (c <= '9')) ||
-      (c == '-' || c == '.' || c == '_' || c == '~') || /* unreserved */
-      (c == '!' || c == '$' || c == '&' || c == '\'' || c == '$' || c == '&' ||
-       c == '(' || c == ')' || c == '*' || c == '+' || c == ',' || c == ';' ||
-       c == '=') /* sub-delims */) {
-    return 1;
-  }
-  if (c == '%') { /* pct-encoded */
-    size_t j;
-    if (uri_text[i + 1] == 0 || uri_text[i + 2] == 0) {
-      return NOT_SET;
-    }
-    for (j = i + 1; j < 2; j++) {
-      c = uri_text[j];
-      if (!(((c >= '0') && (c <= '9')) || ((c >= 'a') && (c <= 'f')) ||
-            ((c >= 'A') && (c <= 'F')))) {
-        return NOT_SET;
+  switch (c) {
+    default:
+      if (((c >= 'a') && (c <= 'z')) || ((c >= 'A') && (c <= 'Z')) ||
+          ((c >= '0') && (c <= '9'))) {
+        return 1;
       }
-    }
-    return 2;
+      break;
+    case ':':
+    case '@':
+    case '-':
+    case '.':
+    case '_':
+    case '~':
+    case '!':
+    case '$':
+    case '&':
+    case '\'':
+    case '(':
+    case ')':
+    case '*':
+    case '+':
+    case ',':
+    case ';':
+    case '=':
+      return 1;
+    case '%': /* pct-encoded */
+      if (valid_hex(uri_text[i + 1]) && valid_hex(uri_text[i + 2])) {
+        return 2;
+      }
+      return NOT_SET;
   }
   return 0;
 }
@@ -183,7 +197,7 @@ static void parse_query_parts(grpc_uri *uri) {
 }
 
 grpc_uri *grpc_uri_parse(grpc_exec_ctx *exec_ctx, const char *uri_text,
-                         int suppress_errors) {
+                         bool suppress_errors) {
   grpc_uri *uri;
   size_t scheme_begin = 0;
   size_t scheme_end = NOT_SET;
diff --git a/src/core/ext/filters/client_channel/uri_parser.h b/src/core/ext/filters/client_channel/uri_parser.h
index 2698d448d8..b889040b16 100644
--- a/src/core/ext/filters/client_channel/uri_parser.h
+++ b/src/core/ext/filters/client_channel/uri_parser.h
@@ -53,7 +53,7 @@ typedef struct {
 
 /** parse a uri, return NULL on failure */
 grpc_uri *grpc_uri_parse(grpc_exec_ctx *exec_ctx, const char *uri_text,
-                         int suppress_errors);
+                         bool suppress_errors);
 
 /** return the part of a query string after the '=' in "?key=xxx&...", or NULL
  * if key is not present */