Merge branch 'master' into uv_core_transport

9 files changed, 356 insertions, 177 deletions

diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c
index b61a0fb24b..55bb877576 100644
--- a/src/core/ext/client_channel/client_channel.c
+++ b/src/core/ext/client_channel/client_channel.c
@@ -61,6 +61,54 @@
 /* Client channel implementation */
 
 /*************************************************************************
+ * METHOD-CONFIG TABLE
+ */
+
+typedef enum {
+  WAIT_FOR_READY_UNSET,
+  WAIT_FOR_READY_FALSE,
+  WAIT_FOR_READY_TRUE
+} wait_for_ready_value;
+
+typedef struct method_parameters {
+  gpr_timespec timeout;
+  wait_for_ready_value wait_for_ready;
+} method_parameters;
+
+static void *method_parameters_copy(void *value) {
+  void *new_value = gpr_malloc(sizeof(method_parameters));
+  memcpy(new_value, value, sizeof(method_parameters));
+  return new_value;
+}
+
+static int method_parameters_cmp(void *value1, void *value2) {
+  const method_parameters *v1 = value1;
+  const method_parameters *v2 = value2;
+  const int retval = gpr_time_cmp(v1->timeout, v2->timeout);
+  if (retval != 0) return retval;
+  if (v1->wait_for_ready > v2->wait_for_ready) return 1;
+  if (v1->wait_for_ready < v2->wait_for_ready) return -1;
+  return 0;
+}
+
+static const grpc_mdstr_hash_table_vtable method_parameters_vtable = {
+    gpr_free, method_parameters_copy, method_parameters_cmp};
+
+static void *method_config_convert_value(
+    const grpc_method_config *method_config) {
+  method_parameters *value = gpr_malloc(sizeof(method_parameters));
+  const gpr_timespec *timeout = grpc_method_config_get_timeout(method_config);
+  value->timeout = timeout != NULL ? *timeout : gpr_time_0(GPR_TIMESPAN);
+  const bool *wait_for_ready =
+      grpc_method_config_get_wait_for_ready(method_config);
+  value->wait_for_ready =
+      wait_for_ready == NULL
+          ? WAIT_FOR_READY_UNSET
+          : (wait_for_ready ? WAIT_FOR_READY_TRUE : WAIT_FOR_READY_FALSE);
+  return value;
+}
+
+/*************************************************************************
  * CHANNEL-WIDE FUNCTIONS
  */
 
@@ -76,8 +124,8 @@ typedef struct client_channel_channel_data {
   gpr_mu mu;
   /** currently active load balancer */
   grpc_lb_policy *lb_policy;
-  /** method config table */
-  grpc_method_config_table *method_config_table;
+  /** maps method names to method_parameters structs */
+  grpc_mdstr_hash_table *method_params_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 */
@@ -177,7 +225,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_mdstr_hash_table *method_params_table = NULL;
   grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE;
   bool exit_idle = false;
   grpc_error *state_error = GRPC_ERROR_CREATE("No load balancing policy");
@@ -230,8 +278,9 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
         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);
+      method_params_table = grpc_method_config_table_convert(
+          (grpc_method_config_table *)channel_arg->value.pointer.p,
+          method_config_convert_value, &method_parameters_vtable);
     }
     grpc_resolver_result_unref(exec_ctx, chand->resolver_result);
     chand->resolver_result = NULL;
@@ -245,10 +294,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);
+  if (chand->method_params_table != NULL) {
+    grpc_mdstr_hash_table_unref(chand->method_params_table);
   }
-  chand->method_config_table = method_config_table;
+  chand->method_params_table = method_params_table;
   if (lb_policy != NULL) {
     grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures,
                                NULL);
@@ -409,8 +458,8 @@ 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);
+  if (chand->method_params_table != NULL) {
+    grpc_mdstr_hash_table_unref(chand->method_params_table);
   }
   grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
   grpc_pollset_set_destroy(chand->interested_parties);
@@ -448,11 +497,7 @@ typedef struct client_channel_call_data {
   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;
+  wait_for_ready_value wait_for_ready_from_service_config;
   grpc_closure read_service_config;
 
   grpc_error *cancel_error;
@@ -846,41 +891,39 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
   if (error == GRPC_ERROR_NONE) {
     // Get the method config table from channel data.
     gpr_mu_lock(&chand->mu);
-    grpc_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);
+    grpc_mdstr_hash_table *method_params_table = NULL;
+    if (chand->method_params_table != NULL) {
+      method_params_table =
+          grpc_mdstr_hash_table_ref(chand->method_params_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) {
+    if (method_params_table != NULL) {
+      const method_parameters *method_params =
+          grpc_method_config_table_get(method_params_table, calld->path);
+      if (method_params != NULL) {
+        const bool have_method_timeout =
+            gpr_time_cmp(method_params->timeout, gpr_time_0(GPR_TIMESPAN)) != 0;
+        if (have_method_timeout ||
+            method_params->wait_for_ready != WAIT_FOR_READY_UNSET) {
           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 (have_method_timeout) {
+            const gpr_timespec per_method_deadline =
+                gpr_time_add(calld->call_start_time, method_params->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) {
+          if (method_params->wait_for_ready != WAIT_FOR_READY_UNSET) {
             calld->wait_for_ready_from_service_config =
-                *wait_for_ready ? WAIT_FOR_READY_TRUE : WAIT_FOR_READY_FALSE;
+                method_params->wait_for_ready;
           }
           gpr_mu_unlock(&calld->mu);
         }
       }
-      grpc_method_config_table_unref(method_config_table);
+      grpc_mdstr_hash_table_unref(method_params_table);
     }
   }
   GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config");
@@ -917,29 +960,25 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
   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);
+    if (chand->method_params_table != NULL) {
+      grpc_mdstr_hash_table *method_params_table =
+          grpc_mdstr_hash_table_ref(chand->method_params_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) {
+      method_parameters *method_params =
+          grpc_method_config_table_get(method_params_table, args->path);
+      if (method_params != NULL) {
+        if (gpr_time_cmp(method_params->timeout,
+                         gpr_time_0(GPR_CLOCK_MONOTONIC)) != 0) {
           gpr_timespec per_method_deadline =
-              gpr_time_add(calld->call_start_time, *per_method_timeout);
+              gpr_time_add(calld->call_start_time, method_params->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) {
+        if (method_params->wait_for_ready != WAIT_FOR_READY_UNSET) {
           calld->wait_for_ready_from_service_config =
-              *wait_for_ready ? WAIT_FOR_READY_TRUE : WAIT_FOR_READY_FALSE;
+              method_params->wait_for_ready;
         }
       }
-      grpc_method_config_table_unref(method_config_table);
+      grpc_mdstr_hash_table_unref(method_params_table);
     } else {
       gpr_mu_unlock(&chand->mu);
     }
diff --git a/src/core/ext/client_channel/method_config.c b/src/core/ext/client_channel/method_config.c
index 49d1ad344e..4313ad5e0e 100644
--- a/src/core/ext/client_channel/method_config.c
+++ b/src/core/ext/client_channel/method_config.c
@@ -254,12 +254,12 @@ int grpc_method_config_table_cmp(const grpc_method_config_table* table1,
   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);
+void* grpc_method_config_table_get(const grpc_mdstr_hash_table* table,
+                                   const grpc_mdstr* path) {
+  void* value = 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) {
+  if (value == 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);
@@ -269,10 +269,10 @@ grpc_method_config* grpc_method_config_table_get_method_config(
     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);
+    value = grpc_mdstr_hash_table_get(table, wildcard_path);
     GRPC_MDSTR_UNREF(wildcard_path);
   }
-  return method_config;
+  return value;
 }
 
 static void* copy_arg(void* p) { return grpc_method_config_table_ref(p); }
@@ -294,3 +294,47 @@ grpc_arg grpc_method_config_table_create_channel_arg(
   arg.value.pointer.vtable = &arg_vtable;
   return arg;
 }
+
+// State used by convert_entry() below.
+typedef struct conversion_state {
+  void* (*convert_value)(const grpc_method_config* method_config);
+  const grpc_mdstr_hash_table_vtable* vtable;
+  size_t num_entries;
+  grpc_mdstr_hash_table_entry* entries;
+} conversion_state;
+
+// A function to be passed to grpc_mdstr_hash_table_iterate() to create
+// a copy of the entries.
+static void convert_entry(const grpc_mdstr_hash_table_entry* entry,
+                          void* user_data) {
+  conversion_state* state = user_data;
+  state->entries[state->num_entries].key = GRPC_MDSTR_REF(entry->key);
+  state->entries[state->num_entries].value = state->convert_value(entry->value);
+  state->entries[state->num_entries].vtable = state->vtable;
+  ++state->num_entries;
+}
+
+grpc_mdstr_hash_table* grpc_method_config_table_convert(
+    const grpc_method_config_table* table,
+    void* (*convert_value)(const grpc_method_config* method_config),
+    const grpc_mdstr_hash_table_vtable* vtable) {
+  // Create an array of the entries in the table with converted values.
+  conversion_state state;
+  state.convert_value = convert_value;
+  state.vtable = vtable;
+  state.num_entries = 0;
+  state.entries = gpr_malloc(sizeof(grpc_mdstr_hash_table_entry) *
+                             grpc_mdstr_hash_table_num_entries(table));
+  grpc_mdstr_hash_table_iterate(table, convert_entry, &state);
+  // Create a new table based on the array we just constructed.
+  grpc_mdstr_hash_table* new_table =
+      grpc_mdstr_hash_table_create(state.num_entries, state.entries);
+  // Clean up the array.
+  for (size_t i = 0; i < state.num_entries; ++i) {
+    GRPC_MDSTR_UNREF(state.entries[i].key);
+    vtable->destroy_value(state.entries[i].value);
+  }
+  gpr_free(state.entries);
+  // Return the new table.
+  return new_table;
+}
diff --git a/src/core/ext/client_channel/method_config.h b/src/core/ext/client_channel/method_config.h
index 75b32fcb17..4cbeee5625 100644
--- a/src/core/ext/client_channel/method_config.h
+++ b/src/core/ext/client_channel/method_config.h
@@ -106,11 +106,31 @@ int grpc_method_config_table_cmp(const grpc_method_config_table* table1,
 /// 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);
+///
+/// Note: This returns a void* instead of a grpc_method_config* so that
+/// it can also be used for tables constructed via
+/// grpc_method_config_table_convert().
+void* grpc_method_config_table_get(const grpc_mdstr_hash_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);
 
+/// Generates a new table from \a table whose values are converted to a
+/// new form via the \a convert_value function.  The new table will use
+/// \a vtable for its values.
+///
+/// This is generally used to convert the table's value type from
+/// grpc_method_config to a simple struct containing only the parameters
+/// relevant to a particular filter, thus avoiding the need for a hash
+/// table lookup on the fast path.  In that scenario, \a convert_value
+/// will return a new instance of the struct containing the values from
+/// the grpc_method_config, and \a vtable provides the methods for
+/// operating on the struct type.
+grpc_mdstr_hash_table* grpc_method_config_table_convert(
+    const grpc_method_config_table* table,
+    void* (*convert_value)(const grpc_method_config* method_config),
+    const grpc_mdstr_hash_table_vtable* vtable);
+
 #endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_METHOD_CONFIG_H */
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 97780d90f2..ecf3aea870 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -853,53 +853,51 @@ static bool contains_non_ok_status(grpc_metadata_batch *batch) {
 
 static void add_fetched_slice_locked(grpc_exec_ctx *exec_ctx,
                                      grpc_chttp2_transport *t,
-                                     grpc_chttp2_stream *s);
+                                     grpc_chttp2_stream *s) {
+  s->fetched_send_message_length +=
+      (uint32_t)GPR_SLICE_LENGTH(s->fetching_slice);
+  gpr_slice_buffer_add(&s->flow_controlled_buffer, s->fetching_slice);
+  if (s->id != 0) {
+    grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
+  }
+}
 
 static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx,
                                           grpc_chttp2_transport *t,
                                           grpc_chttp2_stream *s) {
-  if (s->fetching_send_message == NULL) {
-    /* Stream was cancelled before message fetch completed */
-    abort(); /* TODO(ctiller): what cleanup here? */
-    return;
-  }
-  if (s->fetched_send_message_length == s->fetching_send_message->length) {
-    int64_t notify_offset = s->next_message_end_offset;
-    if (notify_offset <= s->flow_controlled_bytes_written) {
-      grpc_chttp2_complete_closure_step(
-          exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_NONE,
-          "fetching_send_message_finished");
-    } else {
-      grpc_chttp2_write_cb *cb = t->write_cb_pool;
-      if (cb == NULL) {
-        cb = gpr_malloc(sizeof(*cb));
+  for (;;) {
+    if (s->fetching_send_message == NULL) {
+      /* Stream was cancelled before message fetch completed */
+      abort(); /* TODO(ctiller): what cleanup here? */
+      return;  /* early out */
+    }
+    if (s->fetched_send_message_length == s->fetching_send_message->length) {
+      int64_t notify_offset = s->next_message_end_offset;
+      if (notify_offset <= s->flow_controlled_bytes_written) {
+        grpc_chttp2_complete_closure_step(
+            exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_NONE,
+            "fetching_send_message_finished");
       } else {
-        t->write_cb_pool = cb->next;
+        grpc_chttp2_write_cb *cb = t->write_cb_pool;
+        if (cb == NULL) {
+          cb = gpr_malloc(sizeof(*cb));
+        } else {
+          t->write_cb_pool = cb->next;
+        }
+        cb->call_at_byte = notify_offset;
+        cb->closure = s->fetching_send_message_finished;
+        s->fetching_send_message_finished = NULL;
+        cb->next = s->on_write_finished_cbs;
+        s->on_write_finished_cbs = cb;
       }
-      cb->call_at_byte = notify_offset;
-      cb->closure = s->fetching_send_message_finished;
-      s->fetching_send_message_finished = NULL;
-      cb->next = s->on_write_finished_cbs;
-      s->on_write_finished_cbs = cb;
+      s->fetching_send_message = NULL;
+      return; /* early out */
+    } else if (grpc_byte_stream_next(exec_ctx, s->fetching_send_message,
+                                     &s->fetching_slice, UINT32_MAX,
+                                     &s->complete_fetch)) {
+      add_fetched_slice_locked(exec_ctx, t, s);
     }
-    s->fetching_send_message = NULL;
-  } else if (grpc_byte_stream_next(exec_ctx, s->fetching_send_message,
-                                   &s->fetching_slice, UINT32_MAX,
-                                   &s->complete_fetch)) {
-    add_fetched_slice_locked(exec_ctx, t, s);
-  }
-}
-
-static void add_fetched_slice_locked(grpc_exec_ctx *exec_ctx,
-                                     grpc_chttp2_transport *t,
-                                     grpc_chttp2_stream *s) {
-  s->fetched_send_message_length +=
-      (uint32_t)GPR_SLICE_LENGTH(s->fetching_slice);
-  gpr_slice_buffer_add(&s->flow_controlled_buffer, s->fetching_slice);
-  if (s->id != 0) {
-    grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
   }
-  continue_fetching_send_locked(exec_ctx, t, s);
 }
 
 static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs,
@@ -908,6 +906,7 @@ static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs,
   grpc_chttp2_transport *t = s->t;
   if (error == GRPC_ERROR_NONE) {
     add_fetched_slice_locked(exec_ctx, t, s);
+    continue_fetching_send_locked(exec_ctx, t, s);
   } else {
     /* TODO(ctiller): what to do here */
     abort();
diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c
index 69fb3819ff..b8b2546035 100644
--- a/src/core/lib/channel/message_size_filter.c
+++ b/src/core/lib/channel/message_size_filter.c
@@ -45,6 +45,44 @@
 // The protobuf library will (by default) start warning at 100 megs.
 #define DEFAULT_MAX_RECV_MESSAGE_LENGTH (4 * 1024 * 1024)
 
+typedef struct message_size_limits {
+  int max_send_size;
+  int max_recv_size;
+} message_size_limits;
+
+static void* message_size_limits_copy(void* value) {
+  void* new_value = gpr_malloc(sizeof(message_size_limits));
+  memcpy(new_value, value, sizeof(message_size_limits));
+  return new_value;
+}
+
+static int message_size_limits_cmp(void* value1, void* value2) {
+  const message_size_limits* v1 = value1;
+  const message_size_limits* v2 = value2;
+  if (v1->max_send_size > v2->max_send_size) return 1;
+  if (v1->max_send_size < v2->max_send_size) return -1;
+  if (v1->max_recv_size > v2->max_recv_size) return 1;
+  if (v1->max_recv_size < v2->max_recv_size) return -1;
+  return 0;
+}
+
+static const grpc_mdstr_hash_table_vtable message_size_limits_vtable = {
+    gpr_free, message_size_limits_copy, message_size_limits_cmp};
+
+static void* method_config_convert_value(
+    const grpc_method_config* method_config) {
+  message_size_limits* value = gpr_malloc(sizeof(message_size_limits));
+  const int32_t* max_request_message_bytes =
+      grpc_method_config_get_max_request_message_bytes(method_config);
+  value->max_send_size =
+      max_request_message_bytes != NULL ? *max_request_message_bytes : -1;
+  const int32_t* max_response_message_bytes =
+      grpc_method_config_get_max_response_message_bytes(method_config);
+  value->max_recv_size =
+      max_response_message_bytes != NULL ? *max_response_message_bytes : -1;
+  return value;
+}
+
 typedef struct call_data {
   int max_send_size;
   int max_recv_size;
@@ -61,8 +99,8 @@ typedef struct call_data {
 typedef struct channel_data {
   int max_send_size;
   int max_recv_size;
-  // Method config table.
-  grpc_method_config_table* method_config_table;
+  // Maps path names to message_size_limits structs.
+  grpc_mdstr_hash_table* method_limit_table;
 } channel_data;
 
 // Callback invoked when we receive a message.  Here we check the max
@@ -132,24 +170,19 @@ static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
   // size to the receive limit.
   calld->max_send_size = chand->max_send_size;
   calld->max_recv_size = chand->max_recv_size;
-  if (chand->method_config_table != NULL) {
-    grpc_method_config* method_config =
-        grpc_method_config_table_get_method_config(chand->method_config_table,
-                                                   args->path);
-    if (method_config != NULL) {
-      const int32_t* max_request_message_bytes =
-          grpc_method_config_get_max_request_message_bytes(method_config);
-      if (max_request_message_bytes != NULL &&
-          (*max_request_message_bytes < calld->max_send_size ||
+  if (chand->method_limit_table != NULL) {
+    message_size_limits* limits =
+        grpc_method_config_table_get(chand->method_limit_table, args->path);
+    if (limits != NULL) {
+      if (limits->max_send_size >= 0 &&
+          (limits->max_send_size < calld->max_send_size ||
            calld->max_send_size < 0)) {
-        calld->max_send_size = *max_request_message_bytes;
+        calld->max_send_size = limits->max_send_size;
       }
-      const int32_t* max_response_message_bytes =
-          grpc_method_config_get_max_response_message_bytes(method_config);
-      if (max_response_message_bytes != NULL &&
-          (*max_response_message_bytes < calld->max_recv_size ||
+      if (limits->max_recv_size >= 0 &&
+          (limits->max_recv_size < calld->max_recv_size ||
            calld->max_recv_size < 0)) {
-        calld->max_recv_size = *max_response_message_bytes;
+        calld->max_recv_size = limits->max_recv_size;
       }
     }
   }
@@ -191,8 +224,9 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx,
       grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVICE_CONFIG);
   if (channel_arg != NULL) {
     GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER);
-    chand->method_config_table = grpc_method_config_table_ref(
-        (grpc_method_config_table*)channel_arg->value.pointer.p);
+    chand->method_limit_table = grpc_method_config_table_convert(
+        (grpc_method_config_table*)channel_arg->value.pointer.p,
+        method_config_convert_value, &message_size_limits_vtable);
   }
 }
 
@@ -200,7 +234,7 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx,
 static void destroy_channel_elem(grpc_exec_ctx* exec_ctx,
                                  grpc_channel_element* elem) {
   channel_data* chand = elem->channel_data;
-  grpc_method_config_table_unref(chand->method_config_table);
+  grpc_mdstr_hash_table_unref(chand->method_limit_table);
 }
 
 const grpc_channel_filter grpc_message_size_filter = {
diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c
index 0d31b255e9..a465c2815a 100644
--- a/src/core/lib/iomgr/udp_server.c
+++ b/src/core/lib/iomgr/udp_server.c
@@ -65,12 +65,12 @@
 #include "src/core/lib/iomgr/sockaddr.h"
 #include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "src/core/lib/iomgr/socket_utils_posix.h"
+#include "src/core/lib/iomgr/unix_sockets_posix.h"
 #include "src/core/lib/support/string.h"
 
-#define INIT_PORT_CAP 2
-
 /* one listening port */
-typedef struct {
+typedef struct grpc_udp_listener grpc_udp_listener;
+struct grpc_udp_listener {
   int fd;
   grpc_fd *emfd;
   grpc_udp_server *server;
@@ -79,12 +79,13 @@ typedef struct {
   grpc_closure destroyed_closure;
   grpc_udp_server_read_cb read_cb;
   grpc_udp_server_orphan_cb orphan_cb;
-} server_port;
+
+  struct grpc_udp_listener *next;
+};
 
 /* the overall server */
 struct grpc_udp_server {
   gpr_mu mu;
-  gpr_cv cv;
 
   /* active port count: how many ports are actually still listening */
   size_t active_ports;
@@ -94,10 +95,10 @@ struct grpc_udp_server {
   /* is this server shutting down? (boolean) */
   int shutdown;
 
-  /* all listening ports */
-  server_port *ports;
-  size_t nports;
-  size_t port_capacity;
+  /* linked list of server ports */
+  grpc_udp_listener *head;
+  grpc_udp_listener *tail;
+  unsigned nports;
 
   /* shutdown callback */
   grpc_closure *shutdown_complete;
@@ -113,24 +114,29 @@ struct grpc_udp_server {
 grpc_udp_server *grpc_udp_server_create(void) {
   grpc_udp_server *s = gpr_malloc(sizeof(grpc_udp_server));
   gpr_mu_init(&s->mu);
-  gpr_cv_init(&s->cv);
   s->active_ports = 0;
   s->destroyed_ports = 0;
   s->shutdown = 0;
-  s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP);
+  s->head = NULL;
+  s->tail = NULL;
   s->nports = 0;
-  s->port_capacity = INIT_PORT_CAP;
 
   return s;
 }
 
 static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
-  grpc_exec_ctx_sched(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE, NULL);
+  if (s->shutdown_complete != NULL) {
+    grpc_exec_ctx_sched(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE, NULL);
+  }
 
   gpr_mu_destroy(&s->mu);
-  gpr_cv_destroy(&s->cv);
 
-  gpr_free(s->ports);
+  while (s->head) {
+    grpc_udp_listener *sp = s->head;
+    s->head = sp->next;
+    gpr_free(sp);
+  }
+
   gpr_free(s);
 }
 
@@ -151,8 +157,6 @@ static void destroyed_port(grpc_exec_ctx *exec_ctx, void *server,
    events will be received on them - at this point it's safe to destroy
    things */
 static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
-  size_t i;
-
   /* delete ALL the things */
   gpr_mu_lock(&s->mu);
 
@@ -161,9 +165,11 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
     return;
   }
 
-  if (s->nports) {
-    for (i = 0; i < s->nports; i++) {
-      server_port *sp = &s->ports[i];
+  if (s->head) {
+    grpc_udp_listener *sp;
+    for (sp = s->head; sp; sp = sp->next) {
+      grpc_unlink_if_unix_domain_socket(&sp->addr.sockaddr);
+
       sp->destroyed_closure.cb = destroyed_port;
       sp->destroyed_closure.cb_arg = s;
 
@@ -184,7 +190,7 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
 
 void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
                              grpc_closure *on_done) {
-  size_t i;
+  grpc_udp_listener *sp;
   gpr_mu_lock(&s->mu);
 
   GPR_ASSERT(!s->shutdown);
@@ -194,14 +200,10 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
 
   /* shutdown all fd's */
   if (s->active_ports) {
-    for (i = 0; i < s->nports; i++) {
-      server_port *sp = &s->ports[i];
-      /* Call the orphan_cb to signal that the FD is about to be closed and
-       * should no longer be used. */
+    for (sp = s->head; sp; sp = sp->next) {
       GPR_ASSERT(sp->orphan_cb);
       sp->orphan_cb(sp->emfd);
-
-      grpc_fd_shutdown(exec_ctx, s->ports[i].emfd);
+      grpc_fd_shutdown(exec_ctx, sp->emfd);
     }
     gpr_mu_unlock(&s->mu);
   } else {
@@ -279,10 +281,10 @@ error:
 
 /* event manager callback when reads are ready */
 static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-  server_port *sp = arg;
+  grpc_udp_listener *sp = arg;
 
+  gpr_mu_lock(&sp->server->mu);
   if (error != GRPC_ERROR_NONE) {
-    gpr_mu_lock(&sp->server->mu);
     if (0 == --sp->server->active_ports) {
       gpr_mu_unlock(&sp->server->mu);
       deactivated_all_ports(exec_ctx, sp->server);
@@ -298,13 +300,14 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
 
   /* Re-arm the notification event so we get another chance to read. */
   grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure);
+  gpr_mu_unlock(&sp->server->mu);
 }
 
 static int add_socket_to_server(grpc_udp_server *s, int fd,
                                 const grpc_resolved_address *addr,
                                 grpc_udp_server_read_cb read_cb,
                                 grpc_udp_server_orphan_cb orphan_cb) {
-  server_port *sp;
+  grpc_udp_listener *sp;
   int port;
   char *addr_str;
   char *name;
@@ -315,12 +318,15 @@ static int add_socket_to_server(grpc_udp_server *s, int fd,
     gpr_asprintf(&name, "udp-server-listener:%s", addr_str);
     gpr_free(addr_str);
     gpr_mu_lock(&s->mu);
-    /* append it to the list under a lock */
-    if (s->nports == s->port_capacity) {
-      s->port_capacity *= 2;
-      s->ports = gpr_realloc(s->ports, sizeof(server_port) * s->port_capacity);
+    s->nports++;
+    sp = gpr_malloc(sizeof(grpc_udp_listener));
+    sp->next = NULL;
+    if (s->head == NULL) {
+      s->head = sp;
+    } else {
+      s->tail->next = sp;
     }
-    sp = &s->ports[s->nports++];
+    s->tail = sp;
     sp->server = s;
     sp->fd = fd;
     sp->emfd = grpc_fd_create(fd, name);
@@ -339,9 +345,9 @@ int grpc_udp_server_add_port(grpc_udp_server *s,
                              const grpc_resolved_address *addr,
                              grpc_udp_server_read_cb read_cb,
                              grpc_udp_server_orphan_cb orphan_cb) {
+  grpc_udp_listener *sp;
   int allocated_port1 = -1;
   int allocated_port2 = -1;
-  unsigned i;
   int fd;
   grpc_dualstack_mode dsmode;
   grpc_resolved_address addr6_v4mapped;
@@ -355,9 +361,9 @@ int grpc_udp_server_add_port(grpc_udp_server *s,
   /* Check if this is a wildcard port, and if so, try to keep the port the same
      as some previously created listener. */
   if (grpc_sockaddr_get_port(addr) == 0) {
-    for (i = 0; i < s->nports; i++) {
+    for (sp = s->head; sp; sp = sp->next) {
       sockname_temp.len = sizeof(struct sockaddr_storage);
-      if (0 == getsockname(s->ports[i].fd,
+      if (0 == getsockname(sp->fd,
                            (struct sockaddr *)sockname_temp.addr,
                            (socklen_t *)&sockname_temp.len)) {
         port = grpc_sockaddr_get_port(&sockname_temp);
@@ -413,27 +419,40 @@ done:
 }
 
 int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index) {
-  return (port_index < s->nports) ? s->ports[port_index].fd : -1;
+  grpc_udp_listener *sp;
+  if (port_index >= s->nports) {
+    return -1;
+  }
+
+  for (sp = s->head; sp && port_index != 0; sp = sp->next) {
+    --port_index;
+  }
+  return sp->fd;
 }
 
 void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
                            grpc_pollset **pollsets, size_t pollset_count,
                            grpc_server *server) {
-  size_t i, j;
+  size_t i;
   gpr_mu_lock(&s->mu);
+  grpc_udp_listener *sp;
   GPR_ASSERT(s->active_ports == 0);
   s->pollsets = pollsets;
   s->grpc_server = server;
-  for (i = 0; i < s->nports; i++) {
-    for (j = 0; j < pollset_count; j++) {
-      grpc_pollset_add_fd(exec_ctx, pollsets[j], s->ports[i].emfd);
+
+  sp = s->head;
+  while (sp != NULL) {
+    for (i = 0; i < pollset_count; i++) {
+      grpc_pollset_add_fd(exec_ctx, pollsets[i], sp->emfd);
     }
-    s->ports[i].read_closure.cb = on_read;
-    s->ports[i].read_closure.cb_arg = &s->ports[i];
-    grpc_fd_notify_on_read(exec_ctx, s->ports[i].emfd,
-                           &s->ports[i].read_closure);
+    sp->read_closure.cb = on_read;
+    sp->read_closure.cb_arg = sp;
+    grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure);
+
     s->active_ports++;
+    sp = sp->next;
   }
+
   gpr_mu_unlock(&s->mu);
 }
 
diff --git a/src/core/lib/iomgr/udp_server.h b/src/core/lib/iomgr/udp_server.h
index a95086e9ae..f3c466a031 100644
--- a/src/core/lib/iomgr/udp_server.h
+++ b/src/core/lib/iomgr/udp_server.h
@@ -60,7 +60,7 @@ void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *udp_server,
                            grpc_pollset **pollsets, size_t pollset_count,
                            struct grpc_server *server);
 
-int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned index);
+int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index);
 
 /* Add a port to the server, returning port number on success, or negative
    on failure.
diff --git a/src/core/lib/transport/mdstr_hash_table.c b/src/core/lib/transport/mdstr_hash_table.c
index 4be0536dd7..8e914c420b 100644
--- a/src/core/lib/transport/mdstr_hash_table.c
+++ b/src/core/lib/transport/mdstr_hash_table.c
@@ -42,6 +42,7 @@
 struct grpc_mdstr_hash_table {
   gpr_refcount refs;
   size_t num_entries;
+  size_t size;
   grpc_mdstr_hash_table_entry* entries;
 };
 
@@ -50,13 +51,12 @@ struct grpc_mdstr_hash_table {
 static size_t grpc_mdstr_hash_table_find_index(
     const grpc_mdstr_hash_table* table, const grpc_mdstr* key,
     bool find_empty) {
-  for (size_t i = 0; i < table->num_entries; ++i) {
-    const size_t idx = (key->hash + i * i) % table->num_entries;
-    if (table->entries[idx].key == NULL)
-      return find_empty ? idx : table->num_entries;
+  for (size_t i = 0; i < table->size; ++i) {
+    const size_t idx = (key->hash + i * i) % table->size;
+    if (table->entries[idx].key == NULL) return find_empty ? idx : table->size;
     if (table->entries[idx].key == key) return idx;
   }
-  return table->num_entries;  // Not found.
+  return table->size;  // Not found.
 }
 
 static void grpc_mdstr_hash_table_add(
@@ -65,7 +65,7 @@ static void grpc_mdstr_hash_table_add(
   GPR_ASSERT(value != NULL);
   const size_t idx =
       grpc_mdstr_hash_table_find_index(table, key, true /* find_empty */);
-  GPR_ASSERT(idx != table->num_entries);  // Table should never be full.
+  GPR_ASSERT(idx != table->size);  // Table should never be full.
   grpc_mdstr_hash_table_entry* entry = &table->entries[idx];
   entry->key = GRPC_MDSTR_REF(key);
   entry->value = vtable->copy_value(value);
@@ -77,11 +77,11 @@ grpc_mdstr_hash_table* grpc_mdstr_hash_table_create(
   grpc_mdstr_hash_table* table = gpr_malloc(sizeof(*table));
   memset(table, 0, sizeof(*table));
   gpr_ref_init(&table->refs, 1);
+  table->num_entries = num_entries;
   // Quadratic probing gets best performance when the table is no more
   // than half full.
-  table->num_entries = num_entries * 2;
-  const size_t entry_size =
-      sizeof(grpc_mdstr_hash_table_entry) * table->num_entries;
+  table->size = num_entries * 2;
+  const size_t entry_size = sizeof(grpc_mdstr_hash_table_entry) * table->size;
   table->entries = gpr_malloc(entry_size);
   memset(table->entries, 0, entry_size);
   for (size_t i = 0; i < num_entries; ++i) {
@@ -98,7 +98,7 @@ grpc_mdstr_hash_table* grpc_mdstr_hash_table_ref(grpc_mdstr_hash_table* table) {
 
 int grpc_mdstr_hash_table_unref(grpc_mdstr_hash_table* table) {
   if (table != NULL && gpr_unref(&table->refs)) {
-    for (size_t i = 0; i < table->num_entries; ++i) {
+    for (size_t i = 0; i < table->size; ++i) {
       grpc_mdstr_hash_table_entry* entry = &table->entries[i];
       if (entry->key != NULL) {
         GRPC_MDSTR_UNREF(entry->key);
@@ -112,11 +112,15 @@ int grpc_mdstr_hash_table_unref(grpc_mdstr_hash_table* table) {
   return 0;
 }
 
+size_t grpc_mdstr_hash_table_num_entries(const grpc_mdstr_hash_table* table) {
+  return table->num_entries;
+}
+
 void* grpc_mdstr_hash_table_get(const grpc_mdstr_hash_table* table,
                                 const grpc_mdstr* key) {
   const size_t idx =
       grpc_mdstr_hash_table_find_index(table, key, false /* find_empty */);
-  if (idx == table->num_entries) return NULL;  // Not found.
+  if (idx == table->size) return NULL;  // Not found.
   return table->entries[idx].value;
 }
 
@@ -140,3 +144,14 @@ int grpc_mdstr_hash_table_cmp(const grpc_mdstr_hash_table* table1,
   }
   return 0;
 }
+
+void grpc_mdstr_hash_table_iterate(
+    const grpc_mdstr_hash_table* table,
+    void (*func)(const grpc_mdstr_hash_table_entry* entry, void* user_data),
+    void* user_data) {
+  for (size_t i = 0; i < table->size; ++i) {
+    if (table->entries[i].key != NULL) {
+      func(&table->entries[i], user_data);
+    }
+  }
+}
diff --git a/src/core/lib/transport/mdstr_hash_table.h b/src/core/lib/transport/mdstr_hash_table.h
index 52e5b023db..bceb4df93d 100644
--- a/src/core/lib/transport/mdstr_hash_table.h
+++ b/src/core/lib/transport/mdstr_hash_table.h
@@ -70,6 +70,9 @@ grpc_mdstr_hash_table* grpc_mdstr_hash_table_ref(grpc_mdstr_hash_table* table);
 /** Returns 1 when \a table is destroyed. */
 int grpc_mdstr_hash_table_unref(grpc_mdstr_hash_table* table);
 
+/** Returns the number of entries in \a table. */
+size_t grpc_mdstr_hash_table_num_entries(const grpc_mdstr_hash_table* table);
+
 /** Returns the value from \a table associated with \a key.
     Returns NULL if \a key is not found. */
 void* grpc_mdstr_hash_table_get(const grpc_mdstr_hash_table* table,
@@ -80,4 +83,10 @@ void* grpc_mdstr_hash_table_get(const grpc_mdstr_hash_table* table,
 int grpc_mdstr_hash_table_cmp(const grpc_mdstr_hash_table* table1,
                               const grpc_mdstr_hash_table* table2);
 
+/** Iterates over the entries in \a table, calling \a func for each entry. */
+void grpc_mdstr_hash_table_iterate(
+    const grpc_mdstr_hash_table* table,
+    void (*func)(const grpc_mdstr_hash_table_entry* entry, void* user_data),
+    void* user_data);
+
 #endif /* GRPC_CORE_LIB_TRANSPORT_MDSTR_HASH_TABLE_H */