Merge branch 'master' into uv_core_transport

11 files changed, 605 insertions, 33 deletions

diff --git a/src/core/ext/census/grpc_filter.c b/src/core/ext/census/grpc_filter.c
index 9dacc17eb4..a4cf6f37bd 100644
--- a/src/core/ext/census/grpc_filter.c
+++ b/src/core/ext/census/grpc_filter.c
@@ -133,7 +133,7 @@ static grpc_error *client_init_call_elem(grpc_exec_ctx *exec_ctx,
   call_data *d = elem->call_data;
   GPR_ASSERT(d != NULL);
   memset(d, 0, sizeof(*d));
-  d->start_ts = gpr_now(GPR_CLOCK_REALTIME);
+  d->start_ts = args->start_time;
   return GRPC_ERROR_NONE;
 }
 
@@ -152,7 +152,7 @@ static grpc_error *server_init_call_elem(grpc_exec_ctx *exec_ctx,
   call_data *d = elem->call_data;
   GPR_ASSERT(d != NULL);
   memset(d, 0, sizeof(*d));
-  d->start_ts = gpr_now(GPR_CLOCK_REALTIME);
+  d->start_ts = args->start_time;
   /* TODO(hongyu): call census_tracing_start_op here. */
   grpc_closure_init(&d->finish_recv, server_on_done_recv, elem);
   return GRPC_ERROR_NONE;
diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c
index cbf79afa17..beaa9637c3 100644
--- a/src/core/ext/client_config/client_channel.c
+++ b/src/core/ext/client_config/client_channel.c
@@ -43,6 +43,7 @@
 #include <grpc/support/useful.h>
 
 #include "src/core/ext/client_config/lb_policy_registry.h"
+#include "src/core/ext/client_config/method_config.h"
 #include "src/core/ext/client_config/subchannel.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/connected_channel.h"
@@ -53,6 +54,9 @@
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/surface/channel.h"
 #include "src/core/lib/transport/connectivity_state.h"
+#include "src/core/lib/transport/metadata.h"
+#include "src/core/lib/transport/metadata_batch.h"
+#include "src/core/lib/transport/static_metadata.h"
 
 /* Client channel implementation */
 
@@ -68,12 +72,13 @@ typedef struct client_channel_channel_data {
   /** client channel factory */
   grpc_client_channel_factory *client_channel_factory;
 
-  /** mutex protecting client configuration, including all
-      variables below in this data structure */
+  /** mutex protecting all variables below in this data structure */
   gpr_mu mu;
-  /** currently active load balancer - guarded by mu */
+  /** currently active load balancer */
   grpc_lb_policy *lb_policy;
-  /** incoming resolver result - set by resolver.next(), guarded by mu */
+  /** method config table */
+  grpc_method_config_table *method_config_table;
+  /** incoming resolver result - set by resolver.next() */
   grpc_resolver_result *resolver_result;
   /** a list of closures that are all waiting for config to come in */
   grpc_closure_list waiting_for_config_closures;
@@ -172,6 +177,7 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
   channel_data *chand = arg;
   grpc_lb_policy *lb_policy = NULL;
   grpc_lb_policy *old_lb_policy;
+  grpc_method_config_table *method_config_table = NULL;
   grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE;
   bool exit_idle = false;
   grpc_error *state_error = GRPC_ERROR_CREATE("No load balancing policy");
@@ -220,6 +226,13 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
       state =
           grpc_lb_policy_check_connectivity(exec_ctx, lb_policy, &state_error);
     }
+    const grpc_arg *channel_arg = grpc_channel_args_find(
+        lb_policy_args.additional_args, GRPC_ARG_SERVICE_CONFIG);
+    if (channel_arg != NULL) {
+      GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER);
+      method_config_table = grpc_method_config_table_ref(
+          (grpc_method_config_table *)channel_arg->value.pointer.p);
+    }
     grpc_resolver_result_unref(exec_ctx, chand->resolver_result);
     chand->resolver_result = NULL;
   }
@@ -232,6 +245,10 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
   gpr_mu_lock(&chand->mu);
   old_lb_policy = chand->lb_policy;
   chand->lb_policy = lb_policy;
+  if (chand->method_config_table != NULL) {
+    grpc_method_config_table_unref(chand->method_config_table);
+  }
+  chand->method_config_table = method_config_table;
   if (lb_policy != NULL) {
     grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures,
                                NULL);
@@ -392,6 +409,9 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
                                      chand->interested_parties);
     GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel");
   }
+  if (chand->method_config_table != NULL) {
+    grpc_method_config_table_unref(chand->method_config_table);
+  }
   grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
   grpc_pollset_set_destroy(chand->interested_parties);
   gpr_mu_destroy(&chand->mu);
@@ -424,7 +444,16 @@ typedef struct client_channel_call_data {
   // stack and each has its own mutex.  If/when we have time, find a way
   // to avoid this without breaking the grpc_deadline_state abstraction.
   grpc_deadline_state deadline_state;
+
+  grpc_mdstr *path;  // Request path.
+  gpr_timespec call_start_time;
   gpr_timespec deadline;
+  enum {
+    WAIT_FOR_READY_UNSET,
+    WAIT_FOR_READY_FALSE,
+    WAIT_FOR_READY_TRUE
+  } wait_for_ready_from_service_config;
+  grpc_closure read_service_config;
 
   grpc_error *cancel_error;
 
@@ -532,10 +561,11 @@ static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg,
                 GRPC_ERROR_CREATE_REFERENCING(
                     "Cancelled before creating subchannel", &error, 1));
   } else {
+    /* Create call on subchannel. */
     grpc_subchannel_call *subchannel_call = NULL;
     grpc_error *new_error = grpc_connected_subchannel_create_call(
-        exec_ctx, calld->connected_subchannel, calld->pollent, calld->deadline,
-        &subchannel_call);
+        exec_ctx, calld->connected_subchannel, calld->pollent, calld->path,
+        calld->deadline, &subchannel_call);
     if (new_error != GRPC_ERROR_NONE) {
       new_error = grpc_error_add_child(new_error, error);
       subchannel_call = CANCELLED_CALL;
@@ -584,11 +614,15 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg,
     /* cancelled, do nothing */
   } else if (error != GRPC_ERROR_NONE) {
     grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error), NULL);
-  } else if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
-                             cpa->initial_metadata_flags,
-                             cpa->connected_subchannel, cpa->on_ready,
-                             GRPC_ERROR_NONE)) {
-    grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL);
+  } else {
+    call_data *calld = cpa->elem->call_data;
+    gpr_mu_lock(&calld->mu);
+    if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
+                        cpa->initial_metadata_flags, cpa->connected_subchannel,
+                        cpa->on_ready, GRPC_ERROR_NONE)) {
+      grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL);
+    }
+    gpr_mu_unlock(&calld->mu);
   }
   gpr_free(cpa);
 }
@@ -631,18 +665,33 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
   GPR_ASSERT(error == GRPC_ERROR_NONE);
   if (chand->lb_policy != NULL) {
     grpc_lb_policy *lb_policy = chand->lb_policy;
-    int r;
     GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel");
     gpr_mu_unlock(&chand->mu);
+    // If the application explicitly set wait_for_ready, use that.
+    // Otherwise, if the service config specified a value for this
+    // method, use that.
+    const bool wait_for_ready_set_from_api =
+        initial_metadata_flags &
+        GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET;
+    const bool wait_for_ready_set_from_service_config =
+        calld->wait_for_ready_from_service_config != WAIT_FOR_READY_UNSET;
+    if (!wait_for_ready_set_from_api &&
+        wait_for_ready_set_from_service_config) {
+      if (calld->wait_for_ready_from_service_config == WAIT_FOR_READY_TRUE) {
+        initial_metadata_flags |= GRPC_INITIAL_METADATA_WAIT_FOR_READY;
+      } else {
+        initial_metadata_flags &= ~GRPC_INITIAL_METADATA_WAIT_FOR_READY;
+      }
+    }
     // TODO(dgq): make this deadline configurable somehow.
     const grpc_lb_policy_pick_args inputs = {
         initial_metadata, initial_metadata_flags, &calld->lb_token_mdelem,
         gpr_inf_future(GPR_CLOCK_MONOTONIC)};
-    r = grpc_lb_policy_pick(exec_ctx, lb_policy, &inputs, connected_subchannel,
-                            NULL, on_ready);
+    const bool result = 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;
+    return result;
   }
   if (chand->resolver != NULL && !chand->started_resolving) {
     chand->started_resolving = true;
@@ -768,8 +817,8 @@ retry:
       calld->connected_subchannel != NULL) {
     grpc_subchannel_call *subchannel_call = NULL;
     grpc_error *error = grpc_connected_subchannel_create_call(
-        exec_ctx, calld->connected_subchannel, calld->pollent, calld->deadline,
-        &subchannel_call);
+        exec_ctx, calld->connected_subchannel, calld->pollent, calld->path,
+        calld->deadline, &subchannel_call);
     if (error != GRPC_ERROR_NONE) {
       subchannel_call = CANCELLED_CALL;
       fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error));
@@ -786,13 +835,69 @@ retry:
   GPR_TIMER_END("cc_start_transport_stream_op", 0);
 }
 
+// Gets data from the service config.  Invoked when the resolver returns
+// its initial result.
+static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
+                                grpc_error *error) {
+  grpc_call_element *elem = arg;
+  channel_data *chand = elem->channel_data;
+  call_data *calld = elem->call_data;
+  // If this is an error, there's no point in looking at the service config.
+  if (error == GRPC_ERROR_NONE) {
+    // Get the method config table from channel data.
+    gpr_mu_lock(&chand->mu);
+    grpc_method_config_table *method_config_table = NULL;
+    if (chand->method_config_table != NULL) {
+      method_config_table =
+          grpc_method_config_table_ref(chand->method_config_table);
+    }
+    gpr_mu_unlock(&chand->mu);
+    // If the method config table was present, use it.
+    if (method_config_table != NULL) {
+      const grpc_method_config *method_config =
+          grpc_method_config_table_get_method_config(method_config_table,
+                                                     calld->path);
+      if (method_config != NULL) {
+        const gpr_timespec *per_method_timeout =
+            grpc_method_config_get_timeout(method_config);
+        const bool *wait_for_ready =
+            grpc_method_config_get_wait_for_ready(method_config);
+        if (per_method_timeout != NULL || wait_for_ready != NULL) {
+          gpr_mu_lock(&calld->mu);
+          if (per_method_timeout != NULL) {
+            gpr_timespec per_method_deadline =
+                gpr_time_add(calld->call_start_time, *per_method_timeout);
+            if (gpr_time_cmp(per_method_deadline, calld->deadline) < 0) {
+              calld->deadline = per_method_deadline;
+              // Reset deadline timer.
+              grpc_deadline_state_reset(exec_ctx, elem, calld->deadline);
+            }
+          }
+          if (wait_for_ready != NULL) {
+            calld->wait_for_ready_from_service_config =
+                *wait_for_ready ? WAIT_FOR_READY_TRUE : WAIT_FOR_READY_FALSE;
+          }
+          gpr_mu_unlock(&calld->mu);
+        }
+      }
+      grpc_method_config_table_unref(method_config_table);
+    }
+  }
+  GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config");
+}
+
 /* Constructor for call_data */
 static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
                                      grpc_call_element *elem,
                                      grpc_call_element_args *args) {
+  channel_data *chand = elem->channel_data;
   call_data *calld = elem->call_data;
-  grpc_deadline_state_init(exec_ctx, elem, args);
-  calld->deadline = args->deadline;
+  // Initialize data members.
+  grpc_deadline_state_init(exec_ctx, elem, args->call_stack);
+  calld->path = GRPC_MDSTR_REF(args->path);
+  calld->call_start_time = args->start_time;
+  calld->deadline = gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC);
+  calld->wait_for_ready_from_service_config = WAIT_FOR_READY_UNSET;
   calld->cancel_error = GRPC_ERROR_NONE;
   gpr_atm_rel_store(&calld->subchannel_call, 0);
   gpr_mu_init(&calld->mu);
@@ -803,6 +908,55 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
   calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
   calld->owning_call = args->call_stack;
   calld->pollent = NULL;
+  // If the resolver has already returned results, then we can access
+  // the service config parameters immediately.  Otherwise, we need to
+  // defer that work until the resolver returns an initial result.
+  // TODO(roth): This code is almost but not quite identical to the code
+  // in read_service_config() above.  It would be nice to find a way to
+  // combine them, to avoid having to maintain it twice.
+  gpr_mu_lock(&chand->mu);
+  if (chand->lb_policy != NULL) {
+    // We already have a resolver result, so check for service config.
+    if (chand->method_config_table != NULL) {
+      grpc_method_config_table *method_config_table =
+          grpc_method_config_table_ref(chand->method_config_table);
+      gpr_mu_unlock(&chand->mu);
+      grpc_method_config *method_config =
+          grpc_method_config_table_get_method_config(method_config_table,
+                                                     args->path);
+      if (method_config != NULL) {
+        const gpr_timespec *per_method_timeout =
+            grpc_method_config_get_timeout(method_config);
+        if (per_method_timeout != NULL) {
+          gpr_timespec per_method_deadline =
+              gpr_time_add(calld->call_start_time, *per_method_timeout);
+          calld->deadline = gpr_time_min(calld->deadline, per_method_deadline);
+        }
+        const bool *wait_for_ready =
+            grpc_method_config_get_wait_for_ready(method_config);
+        if (wait_for_ready != NULL) {
+          calld->wait_for_ready_from_service_config =
+              *wait_for_ready ? WAIT_FOR_READY_TRUE : WAIT_FOR_READY_FALSE;
+        }
+      }
+      grpc_method_config_table_unref(method_config_table);
+    } else {
+      gpr_mu_unlock(&chand->mu);
+    }
+  } else {
+    // We don't yet have a resolver result, so register a callback to
+    // get the service config data once the resolver returns.
+    // Take a reference to the call stack to be owned by the callback.
+    GRPC_CALL_STACK_REF(calld->owning_call, "read_service_config");
+    grpc_closure_init(&calld->read_service_config, read_service_config, elem);
+    grpc_closure_list_append(&chand->waiting_for_config_closures,
+                             &calld->read_service_config, GRPC_ERROR_NONE);
+    gpr_mu_unlock(&chand->mu);
+  }
+  // Start the deadline timer with the current deadline value.  If we
+  // do not yet have service config data, then the timer may be reset
+  // later.
+  grpc_deadline_state_start(exec_ctx, elem, calld->deadline);
   return GRPC_ERROR_NONE;
 }
 
@@ -813,6 +967,7 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
                                  void *and_free_memory) {
   call_data *calld = elem->call_data;
   grpc_deadline_state_destroy(exec_ctx, elem);
+  GRPC_MDSTR_UNREF(calld->path);
   GRPC_ERROR_UNREF(calld->cancel_error);
   grpc_subchannel_call *call = GET_CALL(calld);
   if (call != NULL && call != CANCELLED_CALL) {
diff --git a/src/core/ext/client_config/method_config.c b/src/core/ext/client_config/method_config.c
new file mode 100644
index 0000000000..f8a82323e7
--- /dev/null
+++ b/src/core/ext/client_config/method_config.c
@@ -0,0 +1,296 @@
+//
+// Copyright 2015, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//     * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+
+#include "src/core/ext/client_config/method_config.h"
+
+#include <string.h>
+
+#include <grpc/impl/codegen/grpc_types.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/transport/mdstr_hash_table.h"
+#include "src/core/lib/transport/metadata.h"
+
+//
+// grpc_method_config
+//
+
+// bool vtable
+
+static void* bool_copy(void* valuep) {
+  bool value = *(bool*)valuep;
+  bool* new_value = gpr_malloc(sizeof(bool));
+  *new_value = value;
+  return new_value;
+}
+
+static int bool_cmp(void* v1, void* v2) {
+  bool b1 = *(bool*)v1;
+  bool b2 = *(bool*)v2;
+  if (!b1 && b2) return -1;
+  if (b1 && !b2) return 1;
+  return 0;
+}
+
+static grpc_mdstr_hash_table_vtable bool_vtable = {gpr_free, bool_copy,
+                                                   bool_cmp};
+
+// timespec vtable
+
+static void* timespec_copy(void* valuep) {
+  gpr_timespec value = *(gpr_timespec*)valuep;
+  gpr_timespec* new_value = gpr_malloc(sizeof(gpr_timespec));
+  *new_value = value;
+  return new_value;
+}
+
+static int timespec_cmp(void* v1, void* v2) {
+  return gpr_time_cmp(*(gpr_timespec*)v1, *(gpr_timespec*)v2);
+}
+
+static grpc_mdstr_hash_table_vtable timespec_vtable = {gpr_free, timespec_copy,
+                                                       timespec_cmp};
+
+// int32 vtable
+
+static void* int32_copy(void* valuep) {
+  int32_t value = *(int32_t*)valuep;
+  int32_t* new_value = gpr_malloc(sizeof(int32_t));
+  *new_value = value;
+  return new_value;
+}
+
+static int int32_cmp(void* v1, void* v2) {
+  int32_t i1 = *(int32_t*)v1;
+  int32_t i2 = *(int32_t*)v2;
+  if (i1 < i2) return -1;
+  if (i1 > i2) return 1;
+  return 0;
+}
+
+static grpc_mdstr_hash_table_vtable int32_vtable = {gpr_free, int32_copy,
+                                                    int32_cmp};
+
+// Hash table keys.
+#define GRPC_METHOD_CONFIG_WAIT_FOR_READY "grpc.wait_for_ready"  // bool
+#define GRPC_METHOD_CONFIG_TIMEOUT "grpc.timeout"                // gpr_timespec
+#define GRPC_METHOD_CONFIG_MAX_REQUEST_MESSAGE_BYTES \
+  "grpc.max_request_message_bytes"  // int32
+#define GRPC_METHOD_CONFIG_MAX_RESPONSE_MESSAGE_BYTES \
+  "grpc.max_response_message_bytes"  // int32
+
+struct grpc_method_config {
+  grpc_mdstr_hash_table* table;
+  grpc_mdstr* wait_for_ready_key;
+  grpc_mdstr* timeout_key;
+  grpc_mdstr* max_request_message_bytes_key;
+  grpc_mdstr* max_response_message_bytes_key;
+};
+
+grpc_method_config* grpc_method_config_create(
+    bool* wait_for_ready, gpr_timespec* timeout,
+    int32_t* max_request_message_bytes, int32_t* max_response_message_bytes) {
+  grpc_method_config* method_config = gpr_malloc(sizeof(grpc_method_config));
+  memset(method_config, 0, sizeof(grpc_method_config));
+  method_config->wait_for_ready_key =
+      grpc_mdstr_from_string(GRPC_METHOD_CONFIG_WAIT_FOR_READY);
+  method_config->timeout_key =
+      grpc_mdstr_from_string(GRPC_METHOD_CONFIG_TIMEOUT);
+  method_config->max_request_message_bytes_key =
+      grpc_mdstr_from_string(GRPC_METHOD_CONFIG_MAX_REQUEST_MESSAGE_BYTES);
+  method_config->max_response_message_bytes_key =
+      grpc_mdstr_from_string(GRPC_METHOD_CONFIG_MAX_RESPONSE_MESSAGE_BYTES);
+  grpc_mdstr_hash_table_entry entries[4];
+  size_t num_entries = 0;
+  if (wait_for_ready != NULL) {
+    entries[num_entries].key = method_config->wait_for_ready_key;
+    entries[num_entries].value = wait_for_ready;
+    entries[num_entries].vtable = &bool_vtable;
+    ++num_entries;
+  }
+  if (timeout != NULL) {
+    entries[num_entries].key = method_config->timeout_key;
+    entries[num_entries].value = timeout;
+    entries[num_entries].vtable = &timespec_vtable;
+    ++num_entries;
+  }
+  if (max_request_message_bytes != NULL) {
+    entries[num_entries].key = method_config->max_request_message_bytes_key;
+    entries[num_entries].value = max_request_message_bytes;
+    entries[num_entries].vtable = &int32_vtable;
+    ++num_entries;
+  }
+  if (max_response_message_bytes != NULL) {
+    entries[num_entries].key = method_config->max_response_message_bytes_key;
+    entries[num_entries].value = max_response_message_bytes;
+    entries[num_entries].vtable = &int32_vtable;
+    ++num_entries;
+  }
+  method_config->table = grpc_mdstr_hash_table_create(num_entries, entries);
+  return method_config;
+}
+
+grpc_method_config* grpc_method_config_ref(grpc_method_config* method_config) {
+  grpc_mdstr_hash_table_ref(method_config->table);
+  return method_config;
+}
+
+void grpc_method_config_unref(grpc_method_config* method_config) {
+  if (grpc_mdstr_hash_table_unref(method_config->table)) {
+    GRPC_MDSTR_UNREF(method_config->wait_for_ready_key);
+    GRPC_MDSTR_UNREF(method_config->timeout_key);
+    GRPC_MDSTR_UNREF(method_config->max_request_message_bytes_key);
+    GRPC_MDSTR_UNREF(method_config->max_response_message_bytes_key);
+    gpr_free(method_config);
+  }
+}
+
+int grpc_method_config_cmp(const grpc_method_config* method_config1,
+                           const grpc_method_config* method_config2) {
+  return grpc_mdstr_hash_table_cmp(method_config1->table,
+                                   method_config2->table);
+}
+
+const bool* grpc_method_config_get_wait_for_ready(
+    const grpc_method_config* method_config) {
+  return grpc_mdstr_hash_table_get(method_config->table,
+                                   method_config->wait_for_ready_key);
+}
+
+const gpr_timespec* grpc_method_config_get_timeout(
+    const grpc_method_config* method_config) {
+  return grpc_mdstr_hash_table_get(method_config->table,
+                                   method_config->timeout_key);
+}
+
+const int32_t* grpc_method_config_get_max_request_message_bytes(
+    const grpc_method_config* method_config) {
+  return grpc_mdstr_hash_table_get(
+      method_config->table, method_config->max_request_message_bytes_key);
+}
+
+const int32_t* grpc_method_config_get_max_response_message_bytes(
+    const grpc_method_config* method_config) {
+  return grpc_mdstr_hash_table_get(
+      method_config->table, method_config->max_response_message_bytes_key);
+}
+
+//
+// grpc_method_config_table
+//
+
+static void method_config_unref(void* valuep) {
+  grpc_method_config_unref(valuep);
+}
+
+static void* method_config_ref(void* valuep) {
+  return grpc_method_config_ref(valuep);
+}
+
+static int method_config_cmp(void* valuep1, void* valuep2) {
+  return grpc_method_config_cmp(valuep1, valuep2);
+}
+
+static const grpc_mdstr_hash_table_vtable method_config_table_vtable = {
+    method_config_unref, method_config_ref, method_config_cmp};
+
+grpc_method_config_table* grpc_method_config_table_create(
+    size_t num_entries, grpc_method_config_table_entry* entries) {
+  grpc_mdstr_hash_table_entry* hash_table_entries =
+      gpr_malloc(sizeof(grpc_mdstr_hash_table_entry) * num_entries);
+  for (size_t i = 0; i < num_entries; ++i) {
+    hash_table_entries[i].key = entries[i].method_name;
+    hash_table_entries[i].value = entries[i].method_config;
+    hash_table_entries[i].vtable = &method_config_table_vtable;
+  }
+  grpc_method_config_table* method_config_table =
+      grpc_mdstr_hash_table_create(num_entries, hash_table_entries);
+  gpr_free(hash_table_entries);
+  return method_config_table;
+}
+
+grpc_method_config_table* grpc_method_config_table_ref(
+    grpc_method_config_table* table) {
+  return grpc_mdstr_hash_table_ref(table);
+}
+
+void grpc_method_config_table_unref(grpc_method_config_table* table) {
+  grpc_mdstr_hash_table_unref(table);
+}
+
+int grpc_method_config_table_cmp(const grpc_method_config_table* table1,
+                                 const grpc_method_config_table* table2) {
+  return grpc_mdstr_hash_table_cmp(table1, table2);
+}
+
+grpc_method_config* grpc_method_config_table_get_method_config(
+    const grpc_method_config_table* table, const grpc_mdstr* path) {
+  grpc_method_config* method_config = grpc_mdstr_hash_table_get(table, path);
+  // If we didn't find a match for the path, try looking for a wildcard
+  // entry (i.e., change "/service/method" to "/service/*").
+  if (method_config == NULL) {
+    const char* path_str = grpc_mdstr_as_c_string(path);
+    const char* sep = strrchr(path_str, '/') + 1;
+    const size_t len = (size_t)(sep - path_str);
+    char* buf = gpr_malloc(len + 2);  // '*' and NUL
+    memcpy(buf, path_str, len);
+    buf[len] = '*';
+    buf[len + 1] = '\0';
+    grpc_mdstr* wildcard_path = grpc_mdstr_from_string(buf);
+    gpr_free(buf);
+    method_config = grpc_mdstr_hash_table_get(table, wildcard_path);
+    GRPC_MDSTR_UNREF(wildcard_path);
+  }
+  return method_config;
+}
+
+static void* copy_arg(void* p) { return grpc_method_config_table_ref(p); }
+
+static void destroy_arg(void* p) { grpc_method_config_table_unref(p); }
+
+static int cmp_arg(void* p1, void* p2) {
+  return grpc_method_config_table_cmp(p1, p2);
+}
+
+static grpc_arg_pointer_vtable arg_vtable = {copy_arg, destroy_arg, cmp_arg};
+
+grpc_arg grpc_method_config_table_create_channel_arg(
+    grpc_method_config_table* table) {
+  grpc_arg arg;
+  arg.type = GRPC_ARG_POINTER;
+  arg.key = GRPC_ARG_SERVICE_CONFIG;
+  arg.value.pointer.p = table;
+  arg.value.pointer.vtable = &arg_vtable;
+  return arg;
+}
diff --git a/src/core/ext/client_config/method_config.h b/src/core/ext/client_config/method_config.h
new file mode 100644
index 0000000000..1302ac425d
--- /dev/null
+++ b/src/core/ext/client_config/method_config.h
@@ -0,0 +1,116 @@
+//
+// Copyright 2016, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//     * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+
+#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_METHOD_CONFIG_H
+#define GRPC_CORE_EXT_CLIENT_CONFIG_METHOD_CONFIG_H
+
+#include <stdbool.h>
+
+#include <grpc/impl/codegen/gpr_types.h>
+#include <grpc/impl/codegen/grpc_types.h>
+
+#include "src/core/lib/transport/mdstr_hash_table.h"
+#include "src/core/lib/transport/metadata.h"
+
+/// Per-method configuration.
+typedef struct grpc_method_config grpc_method_config;
+
+/// Creates a grpc_method_config with the specified parameters.
+/// Any parameter may be NULL to indicate that the value is unset.
+///
+/// \a wait_for_ready indicates whether the client should wait until the
+/// request deadline for the channel to become ready, even if there is a
+/// temporary failure before the deadline while attempting to connect.
+///
+/// \a timeout indicates the timeout for calls.
+///
+/// \a max_request_message_bytes and \a max_response_message_bytes
+/// indicate the maximum sizes of the request (checked when sending) and
+/// response (checked when receiving) messages.
+grpc_method_config* grpc_method_config_create(
+    bool* wait_for_ready, gpr_timespec* timeout,
+    int32_t* max_request_message_bytes, int32_t* max_response_message_bytes);
+
+grpc_method_config* grpc_method_config_ref(grpc_method_config* method_config);
+void grpc_method_config_unref(grpc_method_config* method_config);
+
+/// Compares two grpc_method_configs.
+/// The sort order is stable but undefined.
+int grpc_method_config_cmp(const grpc_method_config* method_config1,
+                           const grpc_method_config* method_config2);
+
+/// These methods return NULL if the requested field is unset.
+/// The caller does NOT take ownership of the result.
+const bool* grpc_method_config_get_wait_for_ready(
+    const grpc_method_config* method_config);
+const gpr_timespec* grpc_method_config_get_timeout(
+    const grpc_method_config* method_config);
+const int32_t* grpc_method_config_get_max_request_message_bytes(
+    const grpc_method_config* method_config);
+const int32_t* grpc_method_config_get_max_response_message_bytes(
+    const grpc_method_config* method_config);
+
+/// A table of method configs.
+typedef grpc_mdstr_hash_table grpc_method_config_table;
+
+typedef struct grpc_method_config_table_entry {
+  /// The name is of one of the following forms:
+  ///   service/method -- specifies exact service and method name
+  ///   service/*      -- matches all methods for the specified service
+  grpc_mdstr* method_name;
+  grpc_method_config* method_config;
+} grpc_method_config_table_entry;
+
+/// Takes new references to all keys and values in \a entries.
+grpc_method_config_table* grpc_method_config_table_create(
+    size_t num_entries, grpc_method_config_table_entry* entries);
+
+grpc_method_config_table* grpc_method_config_table_ref(
+    grpc_method_config_table* table);
+void grpc_method_config_table_unref(grpc_method_config_table* table);
+
+/// Compares two grpc_method_config_tables.
+/// The sort order is stable but undefined.
+int grpc_method_config_table_cmp(const grpc_method_config_table* table1,
+                                 const grpc_method_config_table* table2);
+
+/// Gets the method config for the specified \a path, which should be of
+/// the form "/service/method".
+/// Returns NULL if the method has no config.
+/// Caller does NOT own a reference to the result.
+grpc_method_config* grpc_method_config_table_get_method_config(
+    const grpc_method_config_table* table, const grpc_mdstr* path);
+
+/// Returns a channel arg containing \a table.
+grpc_arg grpc_method_config_table_create_channel_arg(
+    grpc_method_config_table* table);
+
+#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_METHOD_CONFIG_H */
diff --git a/src/core/ext/client_config/resolver_result.h b/src/core/ext/client_config/resolver_result.h
index 414c2e2482..a7ea7c0f4b 100644
--- a/src/core/ext/client_config/resolver_result.h
+++ b/src/core/ext/client_config/resolver_result.h
@@ -52,22 +52,17 @@ typedef struct grpc_resolver_result grpc_resolver_result;
 grpc_resolver_result* grpc_resolver_result_create(
     const char* server_name, 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.
+/// Accessors.  Caller does NOT take ownership of results.
 const char* grpc_resolver_result_get_server_name(grpc_resolver_result* result);
-
-/// Caller does NOT take ownership of result.
 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);
 
diff --git a/src/core/ext/client_config/subchannel.c b/src/core/ext/client_config/subchannel.c
index e52e5a2a32..b66a6556a2 100644
--- a/src/core/ext/client_config/subchannel.c
+++ b/src/core/ext/client_config/subchannel.c
@@ -701,7 +701,7 @@ grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
 
 grpc_error *grpc_connected_subchannel_create_call(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
-    grpc_polling_entity *pollent, gpr_timespec deadline,
+    grpc_polling_entity *pollent, grpc_mdstr *path, gpr_timespec deadline,
     grpc_subchannel_call **call) {
   grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
   *call = gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
@@ -709,7 +709,7 @@ grpc_error *grpc_connected_subchannel_create_call(
   (*call)->connection = con;  // Ref is added below.
   grpc_error *error =
       grpc_call_stack_init(exec_ctx, chanstk, 1, subchannel_call_destroy, *call,
-                           NULL, NULL, deadline, callstk);
+                           NULL, NULL, path, deadline, callstk);
   if (error != GRPC_ERROR_NONE) {
     const char *error_string = grpc_error_string(error);
     gpr_log(GPR_ERROR, "error: %s", error_string);
diff --git a/src/core/ext/client_config/subchannel.h b/src/core/ext/client_config/subchannel.h
index 5b1910213c..7061ef17b8 100644
--- a/src/core/ext/client_config/subchannel.h
+++ b/src/core/ext/client_config/subchannel.h
@@ -38,6 +38,7 @@
 #include "src/core/lib/channel/channel_stack.h"
 #include "src/core/lib/iomgr/polling_entity.h"
 #include "src/core/lib/transport/connectivity_state.h"
+#include "src/core/lib/transport/metadata.h"
 
 /** A (sub-)channel that knows how to connect to exactly one target
     address. Provides a target for load balancing. */
@@ -110,7 +111,7 @@ void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
 /** construct a subchannel call */
 grpc_error *grpc_connected_subchannel_create_call(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *connected_subchannel,
-    grpc_polling_entity *pollent, gpr_timespec deadline,
+    grpc_polling_entity *pollent, grpc_mdstr *path, gpr_timespec deadline,
     grpc_subchannel_call **subchannel_call);
 
 /** process a transport level op */
diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c
index ddfacd9ec9..0180076a44 100644
--- a/src/core/ext/lb_policy/grpclb/grpclb.c
+++ b/src/core/ext/lb_policy/grpclb/grpclb.c
@@ -115,6 +115,7 @@
 #include "src/core/ext/client_config/parse_address.h"
 #include "src/core/ext/lb_policy/grpclb/grpclb.h"
 #include "src/core/ext/lb_policy/grpclb/load_balancer_api.h"
+#include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/surface/call.h"
@@ -271,6 +272,7 @@ typedef struct glb_lb_policy {
   /** who the client is trying to communicate with */
   const char *server_name;
   grpc_client_channel_factory *cc_factory;
+  grpc_channel_args *args;
 
   /** deadline for the LB's call */
   gpr_timespec deadline;
@@ -458,6 +460,7 @@ static grpc_lb_policy *create_rr_locked(
   args.server_name = glb_policy->server_name;
   args.client_channel_factory = glb_policy->cc_factory;
   args.addresses = process_serverlist(serverlist);
+  args.additional_args = glb_policy->args;
 
   grpc_lb_policy *rr = grpc_lb_policy_create(exec_ctx, "round_robin", &args);
 
@@ -585,6 +588,7 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
    * Create a client channel over them to communicate with a LB service */
   glb_policy->server_name = gpr_strdup(args->server_name);
   glb_policy->cc_factory = args->client_channel_factory;
+  glb_policy->args = grpc_channel_args_copy(args->additional_args);
   GPR_ASSERT(glb_policy->cc_factory != NULL);
 
   /* construct a target from the addresses in args, given in the form
@@ -648,6 +652,7 @@ static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
   GPR_ASSERT(glb_policy->pending_picks == NULL);
   GPR_ASSERT(glb_policy->pending_pings == NULL);
   gpr_free((void *)glb_policy->server_name);
+  grpc_channel_args_destroy(glb_policy->args);
   grpc_channel_destroy(glb_policy->lb_channel);
   glb_policy->lb_channel = NULL;
   grpc_connectivity_state_destroy(exec_ctx, &glb_policy->state_tracker);
diff --git a/src/core/ext/lb_policy/pick_first/pick_first.c b/src/core/ext/lb_policy/pick_first/pick_first.c
index d12179f5f9..cffe4bdcc6 100644
--- a/src/core/ext/lb_policy/pick_first/pick_first.c
+++ b/src/core/ext/lb_policy/pick_first/pick_first.c
@@ -464,6 +464,7 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
      * the copying of args->server_name (a borrowed pointer) OK. */
     sc_args.server_name = args->server_name;
     sc_args.addr = &args->addresses->addresses[i].address;
+    sc_args.args = args->additional_args;
 
     grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel(
         exec_ctx, args->client_channel_factory, &sc_args);
diff --git a/src/core/ext/lb_policy/round_robin/round_robin.c b/src/core/ext/lb_policy/round_robin/round_robin.c
index 3a9250fd61..c26851653e 100644
--- a/src/core/ext/lb_policy/round_robin/round_robin.c
+++ b/src/core/ext/lb_policy/round_robin/round_robin.c
@@ -627,6 +627,7 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
      * the copying of args->server_name (a borrowed pointer) OK. */
     sc_args.server_name = args->server_name;
     sc_args.addr = &args->addresses->addresses[i].address;
+    sc_args.args = args->additional_args;
 
     grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel(
         exec_ctx, args->client_channel_factory, &sc_args);
diff --git a/src/core/ext/resolver/sockaddr/sockaddr_resolver.c b/src/core/ext/resolver/sockaddr/sockaddr_resolver.c
index ed44863a12..6c6c2481ab 100644
--- a/src/core/ext/resolver/sockaddr/sockaddr_resolver.c
+++ b/src/core/ext/resolver/sockaddr/sockaddr_resolver.c
@@ -33,6 +33,7 @@
 
 #include <stdbool.h>
 #include <stdio.h>
+#include <stdlib.h>
 #include <string.h>
 
 #include <grpc/support/alloc.h>
@@ -42,6 +43,7 @@
 
 #include "src/core/ext/client_config/parse_address.h"
 #include "src/core/ext/client_config/resolver_registry.h"
+#include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/iomgr/resolve_address.h"
 #include "src/core/lib/iomgr/unix_sockets_posix.h"
 #include "src/core/lib/support/string.h"
@@ -119,7 +121,7 @@ static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
     *r->target_result = grpc_resolver_result_create(
         r->target_name,
         grpc_lb_addresses_copy(r->addresses, NULL /* user_data_copy */),
-        NULL /* lb_policy_name */, NULL);
+        NULL /* lb_policy_name */, NULL /* lb_policy_args */);
     grpc_exec_ctx_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE, NULL);
     r->next_completion = NULL;
   }
@@ -128,8 +130,8 @@ static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
 static void sockaddr_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *gr) {
   sockaddr_resolver *r = (sockaddr_resolver *)gr;
   gpr_mu_destroy(&r->mu);
-  grpc_lb_addresses_destroy(r->addresses, NULL /* user_data_destroy */);
   gpr_free(r->target_name);
+  grpc_lb_addresses_destroy(r->addresses, NULL /* user_data_destroy */);
   gpr_free(r);
 }