Merge branch 'grpclb_resolver_changes2' into service_config

8 files changed, 206 insertions, 114 deletions

diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c
index c50585d154..2cf5490122 100644
--- a/src/core/ext/client_config/client_channel.c
+++ b/src/core/ext/client_config/client_channel.c
@@ -184,6 +184,8 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
     grpc_lb_policy_args lb_policy_args;
     lb_policy_args.addresses =
         grpc_resolver_result_get_addresses(chand->incoming_resolver_result);
+    lb_policy_args.additional_args = grpc_resolver_result_get_lb_policy_args(
+        chand->incoming_resolver_result);
     lb_policy_args.client_channel_factory = chand->client_channel_factory;
     lb_policy = grpc_lb_policy_create(
         exec_ctx,
@@ -413,13 +415,15 @@ typedef struct client_channel_call_data {
   grpc_mdstr *path;
   grpc_method_config *method_config;
 
-  grpc_transport_stream_op *waiting_ops;
+  grpc_transport_stream_op **waiting_ops;
   size_t waiting_ops_count;
   size_t waiting_ops_capacity;
 
   grpc_closure next_step;
 
   grpc_call_stack *owning_call;
+
+  grpc_linked_mdelem lb_token_mdelem;
 } call_data;
 
 static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) {
@@ -430,7 +434,7 @@ static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) {
         gpr_realloc(calld->waiting_ops,
                     calld->waiting_ops_capacity * sizeof(*calld->waiting_ops));
   }
-  calld->waiting_ops[calld->waiting_ops_count++] = *op;
+  calld->waiting_ops[calld->waiting_ops_count++] = op;
   GPR_TIMER_END("add_waiting_locked", 0);
 }
 
@@ -439,14 +443,14 @@ static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld,
   size_t i;
   for (i = 0; i < calld->waiting_ops_count; i++) {
     grpc_transport_stream_op_finish_with_failure(
-        exec_ctx, &calld->waiting_ops[i], GRPC_ERROR_REF(error));
+        exec_ctx, calld->waiting_ops[i], GRPC_ERROR_REF(error));
   }
   calld->waiting_ops_count = 0;
   GRPC_ERROR_UNREF(error);
 }
 
 typedef struct {
-  grpc_transport_stream_op *ops;
+  grpc_transport_stream_op **ops;
   size_t nops;
   grpc_subchannel_call *call;
 } retry_ops_args;
@@ -455,7 +459,7 @@ static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) {
   retry_ops_args *a = args;
   size_t i;
   for (i = 0; i < a->nops; i++) {
-    grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]);
+    grpc_subchannel_call_process_op(exec_ctx, a->call, a->ops[i]);
   }
   GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops");
   gpr_free(a->ops);
@@ -463,6 +467,10 @@ static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) {
 }
 
 static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
+  if (calld->waiting_ops_count == 0) {
+    return;
+  }
+
   retry_ops_args *a = gpr_malloc(sizeof(*a));
   a->ops = calld->waiting_ops;
   a->nops = calld->waiting_ops_count;
@@ -599,9 +607,11 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
     int r;
     GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel");
     gpr_mu_unlock(&chand->mu);
-    r = grpc_lb_policy_pick(exec_ctx, lb_policy, calld->pollent,
-                            initial_metadata, initial_metadata_flags,
-                            connected_subchannel, on_ready);
+    const grpc_lb_policy_pick_args inputs = {calld->pollent, initial_metadata,
+                                             initial_metadata_flags,
+                                             &calld->lb_token_mdelem};
+    r = grpc_lb_policy_pick(exec_ctx, lb_policy, &inputs, connected_subchannel,
+                            NULL, on_ready);
     GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick_subchannel");
     GPR_TIMER_END("pick_subchannel", 0);
     return r;
diff --git a/src/core/ext/client_config/lb_policy.c b/src/core/ext/client_config/lb_policy.c
index 8b980b2cca..903563ef6b 100644
--- a/src/core/ext/client_config/lb_policy.c
+++ b/src/core/ext/client_config/lb_policy.c
@@ -100,13 +100,11 @@ void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx,
 }
 
 int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                        grpc_polling_entity *pollent,
-                        grpc_metadata_batch *initial_metadata,
-                        uint32_t initial_metadata_flags,
-                        grpc_connected_subchannel **target,
+                        const grpc_lb_policy_pick_args *pick_args,
+                        grpc_connected_subchannel **target, void **user_data,
                         grpc_closure *on_complete) {
-  return policy->vtable->pick(exec_ctx, policy, pollent, initial_metadata,
-                              initial_metadata_flags, target, on_complete);
+  return policy->vtable->pick(exec_ctx, policy, pick_args, target, user_data,
+                              on_complete);
 }
 
 void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
diff --git a/src/core/ext/client_config/lb_policy.h b/src/core/ext/client_config/lb_policy.h
index a2f5446fc6..37c93d707c 100644
--- a/src/core/ext/client_config/lb_policy.h
+++ b/src/core/ext/client_config/lb_policy.h
@@ -53,23 +53,38 @@ struct grpc_lb_policy {
   grpc_pollset_set *interested_parties;
 };
 
+/** Extra arguments for an LB pick */
+typedef struct grpc_lb_policy_pick_args {
+  /** Parties interested in the pick's progress */
+  grpc_polling_entity *pollent;
+  /** Initial metadata associated with the picking call. */
+  grpc_metadata_batch *initial_metadata;
+  /** See \a GRPC_INITIAL_METADATA_* in grpc_types.h */
+  uint32_t initial_metadata_flags;
+  /** Storage for LB token in \a initial_metadata, or NULL if not used */
+  grpc_linked_mdelem *lb_token_mdelem_storage;
+} grpc_lb_policy_pick_args;
+
 struct grpc_lb_policy_vtable {
   void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
-
   void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
 
-  /** implement grpc_lb_policy_pick */
+  /** \see grpc_lb_policy_pick */
   int (*pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-              grpc_polling_entity *pollent,
-              grpc_metadata_batch *initial_metadata,
-              uint32_t initial_metadata_flags,
-              grpc_connected_subchannel **target, grpc_closure *on_complete);
+              const grpc_lb_policy_pick_args *pick_args,
+              grpc_connected_subchannel **target, void **user_data,
+              grpc_closure *on_complete);
+
+  /** \see grpc_lb_policy_cancel_pick */
   void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                       grpc_connected_subchannel **target);
+
+  /** \see grpc_lb_policy_cancel_picks */
   void (*cancel_picks)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                        uint32_t initial_metadata_flags_mask,
                        uint32_t initial_metadata_flags_eq);
 
+  /** \see grpc_lb_policy_ping_one */
   void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                    grpc_closure *closure);
 
@@ -83,8 +98,7 @@ struct grpc_lb_policy_vtable {
 
   /** call notify when the connectivity state of a channel changes from *state.
       Updates *state with the new state of the policy. Calling with a NULL \a
-      state cancels the subscription.
-      */
+      state cancels the subscription.  */
   void (*notify_on_state_change)(grpc_exec_ctx *exec_ctx,
                                  grpc_lb_policy *policy,
                                  grpc_connectivity_state *state,
@@ -124,26 +138,34 @@ void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
 void grpc_lb_policy_init(grpc_lb_policy *policy,
                          const grpc_lb_policy_vtable *vtable);
 
-/** Given initial metadata in \a initial_metadata, find an appropriate
-    target for this rpc, and 'return' it by calling \a on_complete after setting
-    \a target.
-    Picking can be asynchronous. Any IO should be done under \a pollent. */
+/** Find an appropriate target for this call, based on \a pick_args.
+    Picking can be synchronous or asynchronous. In the synchronous case, when a
+    pick is readily available, it'll be returned in \a target and a non-zero
+    value will be returned.
+    In the asynchronous case, zero is returned and \a on_complete will be called
+    once \a target and \a user_data have been set. Any IO should be done under
+    \a pick_args->pollent. The opaque \a user_data output argument corresponds
+    to information that may need be propagated from the LB policy. It may be
+    NULL. Errors are signaled by receiving a NULL \a *target. */
 int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
-                        grpc_polling_entity *pollent,
-                        grpc_metadata_batch *initial_metadata,
-                        uint32_t initial_metadata_flags,
-                        grpc_connected_subchannel **target,
+                        const grpc_lb_policy_pick_args *pick_args,
+                        grpc_connected_subchannel **target, void **user_data,
                         grpc_closure *on_complete);
 
+/** Perform a connected subchannel ping (see \a grpc_connected_subchannel_ping)
+    against one of the connected subchannels managed by \a policy. */
 void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                              grpc_closure *closure);
 
+/** Cancel picks for \a target.
+    The \a on_complete callback of the pending picks will be invoked with \a
+    *target set to NULL. */
 void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                                 grpc_connected_subchannel **target);
 
-/** Cancel all pending picks which have:
-    (initial_metadata_flags & initial_metadata_flags_mask) ==
-         initial_metadata_flags_eq */
+/** Cancel all pending picks for which their \a initial_metadata_flags (as given
+    in the call to \a grpc_lb_policy_pick) matches \a initial_metadata_flags_eq
+    when AND'd with \a initial_metadata_flags_mask */
 void grpc_lb_policy_cancel_picks(grpc_exec_ctx *exec_ctx,
                                  grpc_lb_policy *policy,
                                  uint32_t initial_metadata_flags_mask,
diff --git a/src/core/ext/client_config/lb_policy_factory.c b/src/core/ext/client_config/lb_policy_factory.c
index 70e46ef3cf..8ce8745504 100644
--- a/src/core/ext/client_config/lb_policy_factory.c
+++ b/src/core/ext/client_config/lb_policy_factory.c
@@ -31,8 +31,64 @@
  *
  */
 
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/string_util.h>
+
 #include "src/core/ext/client_config/lb_policy_factory.h"
 
+grpc_lb_addresses* grpc_lb_addresses_create(size_t num_addresses) {
+  grpc_lb_addresses* addresses = gpr_malloc(sizeof(grpc_lb_addresses));
+  addresses->num_addresses = num_addresses;
+  const size_t addresses_size = sizeof(grpc_lb_address) * num_addresses;
+  addresses->addresses = gpr_malloc(addresses_size);
+  memset(addresses->addresses, 0, addresses_size);
+  return addresses;
+}
+
+grpc_lb_addresses* grpc_lb_addresses_copy(grpc_lb_addresses* addresses,
+                                          void* (*user_data_copy)(void*)) {
+  grpc_lb_addresses* new = grpc_lb_addresses_create(addresses->num_addresses);
+  memcpy(new->addresses, addresses->addresses,
+         sizeof(grpc_lb_address) * addresses->num_addresses);
+  for (size_t i = 0; i < addresses->num_addresses; ++i) {
+    if (new->addresses[i].balancer_name != NULL) {
+      new->addresses[i].balancer_name =
+          gpr_strdup(new->addresses[i].balancer_name);
+    }
+    if (user_data_copy != NULL) {
+      new->addresses[i].user_data = user_data_copy(new->addresses[i].user_data);
+    }
+  }
+  return new;
+}
+
+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,
+                                   void* user_data) {
+  GPR_ASSERT(index < addresses->num_addresses);
+  grpc_lb_address* target = &addresses->addresses[index];
+  memcpy(target->address.addr, address, address_len);
+  target->address.len = address_len;
+  target->is_balancer = is_balancer;
+  target->balancer_name = balancer_name;
+  target->user_data = user_data;
+}
+
+void grpc_lb_addresses_destroy(grpc_lb_addresses* addresses,
+                               void (*user_data_destroy)(void*)) {
+  for (size_t i = 0; i < addresses->num_addresses; ++i) {
+    gpr_free(addresses->addresses[i].balancer_name);
+    if (user_data_destroy != NULL) {
+      user_data_destroy(addresses->addresses[i].user_data);
+    }
+  }
+  gpr_free(addresses->addresses);
+  gpr_free(addresses);
+}
+
 void grpc_lb_policy_factory_ref(grpc_lb_policy_factory* factory) {
   factory->vtable->ref(factory);
 }
diff --git a/src/core/ext/client_config/lb_policy_factory.h b/src/core/ext/client_config/lb_policy_factory.h
index 6919b1eb98..c9499d4f87 100644
--- a/src/core/ext/client_config/lb_policy_factory.h
+++ b/src/core/ext/client_config/lb_policy_factory.h
@@ -36,9 +36,9 @@
 
 #include "src/core/ext/client_config/client_channel_factory.h"
 #include "src/core/ext/client_config/lb_policy.h"
-#include "src/core/ext/client_config/resolver_result.h"
 
 #include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/iomgr/resolve_address.h"
 
 typedef struct grpc_lb_policy_factory grpc_lb_policy_factory;
 typedef struct grpc_lb_policy_factory_vtable grpc_lb_policy_factory_vtable;
@@ -47,9 +47,52 @@ struct grpc_lb_policy_factory {
   const grpc_lb_policy_factory_vtable *vtable;
 };
 
+/** A resolved address alongside any LB related information associated with it.
+ * \a user_data, if not NULL, contains opaque data meant to be consumed by the
+ * gRPC LB policy. Note that no all LB policies support \a user_data as input.
+ * Those who don't will simply ignore it and will correspondingly return NULL in
+ * their namesake pick() output argument. */
+typedef struct grpc_lb_address {
+  grpc_resolved_address address;
+  bool is_balancer;
+  char *balancer_name; /* For secure naming. */
+  void *user_data;
+} grpc_lb_address;
+
+typedef struct grpc_lb_addresses {
+  size_t num_addresses;
+  grpc_lb_address *addresses;
+} grpc_lb_addresses;
+
+/** Returns a grpc_addresses struct with enough space for
+ * \a num_addresses addresses. */
+grpc_lb_addresses *grpc_lb_addresses_create(size_t num_addresses);
+
+/** Creates a copy of \a addresses.  If \a user_data_copy is not NULL,
+ * it will be invoked to copy the \a user_data field of each address. */
+grpc_lb_addresses *grpc_lb_addresses_copy(grpc_lb_addresses *addresses,
+                                          void *(*user_data_copy)(void *));
+
+/** Sets the value of the address at index \a index of \a addresses.
+ * \a address is a socket address of length \a address_len.
+ * 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,
+                                   void *user_data);
+
+/** Destroys \a addresses.  If \a user_data_destroy is not NULL, it will
+ * be invoked to destroy the \a user_data field of each address. */
+void grpc_lb_addresses_destroy(grpc_lb_addresses *addresses,
+                               void (*user_data_destroy)(void *));
+
+/** Arguments passed to LB policies. */
+/* TODO(roth, ctiller): Consider replacing this struct with
+   grpc_channel_args.  See comment in resolver_result.h for details. */
 typedef struct grpc_lb_policy_args {
-  grpc_addresses *addresses;
+  grpc_lb_addresses *addresses;
   grpc_client_channel_factory *client_channel_factory;
+  grpc_channel_args *additional_args;
 } grpc_lb_policy_args;
 
 struct grpc_lb_policy_factory_vtable {
diff --git a/src/core/ext/client_config/resolver_result.c b/src/core/ext/client_config/resolver_result.c
index c70c8d3ffa..c730551fe5 100644
--- a/src/core/ext/client_config/resolver_result.c
+++ b/src/core/ext/client_config/resolver_result.c
@@ -37,41 +37,7 @@
 #include <grpc/support/string_util.h>
 
 #include "src/core/lib/transport/metadata.h"
-
-//
-// grpc_addresses
-//
-
-grpc_addresses* grpc_addresses_create(size_t num_addresses) {
-  grpc_addresses* addresses = gpr_malloc(sizeof(grpc_addresses));
-  addresses->num_addresses = num_addresses;
-  const size_t addresses_size = sizeof(grpc_address) * num_addresses;
-  addresses->addresses = gpr_malloc(addresses_size);
-  memset(addresses->addresses, 0, addresses_size);
-  return addresses;
-}
-
-grpc_addresses* grpc_addresses_copy(grpc_addresses* addresses) {
-  grpc_addresses* new = grpc_addresses_create(addresses->num_addresses);
-  memcpy(new->addresses, addresses->addresses,
-         sizeof(grpc_address) * addresses->num_addresses);
-  return new;
-}
-
-void grpc_addresses_set_address(grpc_addresses* addresses, size_t index,
-                                void* address, size_t address_len,
-                                bool is_balancer) {
-  GPR_ASSERT(index < addresses->num_addresses);
-  grpc_address* target = &addresses->addresses[index];
-  memcpy(target->address.addr, address, address_len);
-  target->address.len = address_len;
-  target->is_balancer = is_balancer;
-}
-
-void grpc_addresses_destroy(grpc_addresses* addresses) {
-  gpr_free(addresses->addresses);
-  gpr_free(addresses);
-}
+#include "src/core/lib/channel/channel_args.h"
 
 //
 // grpc_method_config
@@ -214,18 +180,21 @@ static grpc_method_config* method_config_table_get(method_config_table* table,
 
 struct grpc_resolver_result {
   gpr_refcount refs;
-  grpc_addresses* addresses;
+  grpc_lb_addresses* addresses;
   char* lb_policy_name;
+  grpc_channel_args* lb_policy_args;
   method_config_table method_configs;
 };
 
-grpc_resolver_result* grpc_resolver_result_create(grpc_addresses* addresses,
-                                                  const char* lb_policy_name) {
+grpc_resolver_result* grpc_resolver_result_create(
+    grpc_lb_addresses* addresses, const char* lb_policy_name,
+    grpc_channel_args* lb_policy_args) {
   grpc_resolver_result* result = gpr_malloc(sizeof(*result));
   memset(result, 0, sizeof(*result));
   gpr_ref_init(&result->refs, 1);
   result->addresses = addresses;
   result->lb_policy_name = gpr_strdup(lb_policy_name);
+  result->lb_policy_args = lb_policy_args;
   method_config_table_init(&result->method_configs);
   return result;
 }
@@ -237,14 +206,15 @@ void grpc_resolver_result_ref(grpc_resolver_result* result) {
 void grpc_resolver_result_unref(grpc_exec_ctx* exec_ctx,
                                 grpc_resolver_result* result) {
   if (gpr_unref(&result->refs)) {
-    grpc_addresses_destroy(result->addresses);
+    grpc_lb_addresses_destroy(result->addresses, NULL /* user_data_destroy */);
     gpr_free(result->lb_policy_name);
+    grpc_channel_args_destroy(result->lb_policy_args);
     method_config_table_destroy(&result->method_configs);
     gpr_free(result);
   }
 }
 
-grpc_addresses* grpc_resolver_result_get_addresses(
+grpc_lb_addresses* grpc_resolver_result_get_addresses(
     grpc_resolver_result* result) {
   return result->addresses;
 }
@@ -254,6 +224,11 @@ const char* grpc_resolver_result_get_lb_policy_name(
   return result->lb_policy_name;
 }
 
+grpc_channel_args* grpc_resolver_result_get_lb_policy_args(
+    grpc_resolver_result* result) {
+  return result->lb_policy_args;
+}
+
 void grpc_resolver_result_add_method_config(grpc_resolver_result* result,
                                             grpc_mdstr** paths,
                                             size_t num_paths,
diff --git a/src/core/ext/client_config/resolver_result.h b/src/core/ext/client_config/resolver_result.h
index 8402bb581d..d911bfbe52 100644
--- a/src/core/ext/client_config/resolver_result.h
+++ b/src/core/ext/client_config/resolver_result.h
@@ -32,33 +32,18 @@
 #ifndef GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H
 #define GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H
 
-#include <stdbool.h>
-
-#include "src/core/ext/client_config/lb_policy.h"
+#include "src/core/ext/client_config/lb_policy_factory.h"
 #include "src/core/lib/iomgr/resolve_address.h"
 
-/// Used to represent addresses returned by the resolver.
-typedef struct grpc_address {
-  grpc_resolved_address address;
-  bool is_balancer;
-} grpc_address;
-
-typedef struct grpc_addresses {
-  size_t num_addresses;
-  grpc_address* addresses;
-} grpc_addresses;
-
-/// Returns a grpc_addresses struct with enough space for
-/// \a num_addresses addresses.
-grpc_addresses* grpc_addresses_create(size_t num_addresses);
-
-grpc_addresses* grpc_addresses_copy(grpc_addresses* addresses);
-
-void grpc_addresses_set_address(grpc_addresses* addresses, size_t index,
-                                void* address, size_t address_len,
-                                bool is_balancer);
-
-void grpc_addresses_destroy(grpc_addresses* addresses);
+// TODO(roth, ctiller): In the long term, we are considering replacing
+// the resolver_result data structure with grpc_channel_args.  The idea is
+// that the resolver will return a set of channel args that contains the
+// information that is currently in the resolver_result struct.  For
+// example, there will be specific args indicating the set of addresses
+// and the name of the LB policy to instantiate.  Note that if we did
+// this, we would probably want to change the data structure of
+// grpc_channel_args such to a hash table or AVL or some other data
+// structure that does not require linear search to find keys.
 
 /// Per-method configuration.
 
@@ -84,21 +69,26 @@ int32_t* grpc_method_config_get_max_response_message_bytes(
 /// Results reported from a grpc_resolver.
 typedef struct grpc_resolver_result grpc_resolver_result;
 
-/// Takes ownership of \a addresses.
-grpc_resolver_result* grpc_resolver_result_create(grpc_addresses* addresses,
-                                                  const char* lb_policy_name);
+/// Takes ownership of \a addresses and \a lb_policy_args.
+grpc_resolver_result* grpc_resolver_result_create(
+    grpc_lb_addresses* addresses, const char* lb_policy_name,
+    grpc_channel_args* lb_policy_args);
 void grpc_resolver_result_ref(grpc_resolver_result* result);
 void grpc_resolver_result_unref(grpc_exec_ctx* exec_ctx,
                                 grpc_resolver_result* result);
 
 /// Caller does NOT take ownership of result.
-grpc_addresses* grpc_resolver_result_get_addresses(
+grpc_lb_addresses* grpc_resolver_result_get_addresses(
     grpc_resolver_result* result);
 
 /// Caller does NOT take ownership of result.
 const char* grpc_resolver_result_get_lb_policy_name(
     grpc_resolver_result* result);
 
+/// Caller does NOT take ownership of result.
+grpc_channel_args* grpc_resolver_result_get_lb_policy_args(
+    grpc_resolver_result* result);
+
 /// Adds a method config.  \a paths indicates the set of path names
 /// for which this config applies.  Each name is of one of the following
 /// forms:
diff --git a/src/core/ext/client_config/subchannel.c b/src/core/ext/client_config/subchannel.c
index df35904b85..2c4364b259 100644
--- a/src/core/ext/client_config/subchannel.c
+++ b/src/core/ext/client_config/subchannel.c
@@ -504,14 +504,13 @@ static void connected_subchannel_state_op(grpc_exec_ctx *exec_ctx,
                                           grpc_pollset_set *interested_parties,
                                           grpc_connectivity_state *state,
                                           grpc_closure *closure) {
-  grpc_transport_op op;
+  grpc_transport_op *op = grpc_make_transport_op(NULL);
   grpc_channel_element *elem;
-  memset(&op, 0, sizeof(op));
-  op.connectivity_state = state;
-  op.on_connectivity_state_change = closure;
-  op.bind_pollset_set = interested_parties;
+  op->connectivity_state = state;
+  op->on_connectivity_state_change = closure;
+  op->bind_pollset_set = interested_parties;
   elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
-  elem->filter->start_transport_op(exec_ctx, elem, &op);
+  elem->filter->start_transport_op(exec_ctx, elem, op);
 }
 
 void grpc_connected_subchannel_notify_on_state_change(
@@ -525,12 +524,11 @@ void grpc_connected_subchannel_notify_on_state_change(
 void grpc_connected_subchannel_ping(grpc_exec_ctx *exec_ctx,
                                     grpc_connected_subchannel *con,
                                     grpc_closure *closure) {
-  grpc_transport_op op;
+  grpc_transport_op *op = grpc_make_transport_op(NULL);
   grpc_channel_element *elem;
-  memset(&op, 0, sizeof(op));
-  op.send_ping = closure;
+  op->send_ping = closure;
   elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
-  elem->filter->start_transport_op(exec_ctx, elem, &op);
+  elem->filter->start_transport_op(exec_ctx, elem, op);
 }
 
 static void publish_transport_locked(grpc_exec_ctx *exec_ctx,