5 files changed, 80 insertions, 15 deletions

diff --git a/src/core/channel/channel_args.c b/src/core/channel/channel_args.c
index 0427ce0b8d..bae7a90a01 100644
--- a/src/core/channel/channel_args.c
+++ b/src/core/channel/channel_args.c
@@ -37,6 +37,7 @@
 
 #include <grpc/census.h>
 #include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include <grpc/support/useful.h>
 
@@ -55,9 +56,8 @@ static grpc_arg copy_arg(const grpc_arg *src) {
       break;
     case GRPC_ARG_POINTER:
       dst.value.pointer = src->value.pointer;
-      dst.value.pointer.p = src->value.pointer.copy
-                                ? src->value.pointer.copy(src->value.pointer.p)
-                                : src->value.pointer.p;
+      dst.value.pointer.p =
+          src->value.pointer.vtable->copy(src->value.pointer.p);
       break;
   }
   return dst;
@@ -94,6 +94,58 @@ grpc_channel_args *grpc_channel_args_merge(const grpc_channel_args *a,
   return grpc_channel_args_copy_and_add(a, b->args, b->num_args);
 }
 
+static int cmp_arg(const grpc_arg *a, const grpc_arg *b) {
+  int c = GPR_ICMP(a->type, b->type);
+  if (c != 0) return c;
+  c = strcmp(a->key, b->key);
+  if (c != 0) return c;
+  switch (a->type) {
+    case GRPC_ARG_STRING:
+      return strcmp(a->value.string, b->value.string);
+    case GRPC_ARG_INTEGER:
+      return GPR_ICMP(a->value.integer, b->value.integer);
+    case GRPC_ARG_POINTER:
+      c = GPR_ICMP(a->value.pointer.p, b->value.pointer.p);
+      if (c != 0) {
+        c = GPR_ICMP(a->value.pointer.vtable, b->value.pointer.vtable);
+        if (c == 0) {
+          c = a->value.pointer.vtable->cmp(a->value.pointer.p,
+                                           b->value.pointer.p);
+        }
+      }
+      return c;
+  }
+  GPR_UNREACHABLE_CODE(return 0);
+}
+
+/* stabilizing comparison function: since channel_args ordering matters for
+ * keys with the same name, we need to preserve that ordering */
+static int cmp_key_stable(const void *ap, const void *bp) {
+  const grpc_arg *const *a = ap;
+  const grpc_arg *const *b = bp;
+  int c = strcmp((*a)->key, (*b)->key);
+  if (c == 0) c = GPR_ICMP(*a, *b);
+  return c;
+}
+
+grpc_channel_args *grpc_channel_args_normalize(const grpc_channel_args *a) {
+  grpc_arg **args = gpr_malloc(sizeof(grpc_arg *) * a->num_args);
+  for (size_t i = 0; i < a->num_args; i++) {
+    args[i] = &a->args[i];
+  }
+  qsort(args, a->num_args, sizeof(grpc_arg *), cmp_key_stable);
+
+  grpc_channel_args *b = gpr_malloc(sizeof(grpc_channel_args));
+  b->num_args = a->num_args;
+  b->args = gpr_malloc(sizeof(grpc_arg) * b->num_args);
+  for (size_t i = 0; i < a->num_args; i++) {
+    b->args[i] = copy_arg(args[i]);
+  }
+
+  gpr_free(args);
+  return b;
+}
+
 void grpc_channel_args_destroy(grpc_channel_args *a) {
   size_t i;
   for (i = 0; i < a->num_args; i++) {
@@ -104,9 +156,7 @@ void grpc_channel_args_destroy(grpc_channel_args *a) {
       case GRPC_ARG_INTEGER:
         break;
       case GRPC_ARG_POINTER:
-        if (a->args[i].value.pointer.destroy) {
-          a->args[i].value.pointer.destroy(a->args[i].value.pointer.p);
-        }
+        a->args[i].value.pointer.vtable->destroy(a->args[i].value.pointer.p);
         break;
     }
     gpr_free(a->args[i].key);
@@ -208,3 +258,14 @@ int grpc_channel_args_compression_algorithm_get_states(
     return (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1; /* All algs. enabled */
   }
 }
+
+int grpc_channel_args_compare(const grpc_channel_args *a,
+                              const grpc_channel_args *b) {
+  int c = GPR_ICMP(a->num_args, b->num_args);
+  if (c != 0) return c;
+  for (size_t i = 0; i < a->num_args; i++) {
+    c = cmp_arg(&a->args[i], &b->args[i]);
+    if (c != 0) return c;
+  }
+  return 0;
+}
diff --git a/src/core/channel/channel_args.h b/src/core/channel/channel_args.h
index 480cc9aec2..b3a7c9f434 100644
--- a/src/core/channel/channel_args.h
+++ b/src/core/channel/channel_args.h
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -40,6 +40,9 @@
 /* Copy some arguments */
 grpc_channel_args *grpc_channel_args_copy(const grpc_channel_args *src);
 
+/* Copy some arguments, stably sorting keys */
+grpc_channel_args *grpc_channel_args_normalize(const grpc_channel_args *a);
+
 /** Copy some arguments and add the to_add parameter in the end.
    If to_add is NULL, it is equivalent to call grpc_channel_args_copy. */
 grpc_channel_args *grpc_channel_args_copy_and_add(const grpc_channel_args *src,
@@ -85,4 +88,7 @@ grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
 int grpc_channel_args_compression_algorithm_get_states(
     const grpc_channel_args *a);
 
+int grpc_channel_args_compare(const grpc_channel_args *a,
+                              const grpc_channel_args *b);
+
 #endif /* GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H */
diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c
index 43eee046b8..1aa27208c2 100644
--- a/src/core/channel/http_client_filter.c
+++ b/src/core/channel/http_client_filter.c
@@ -127,8 +127,8 @@ static void hc_mutate_op(grpc_call_element *elem,
   if (op->recv_initial_metadata != NULL) {
     /* substitute our callback for the higher callback */
     calld->recv_initial_metadata = op->recv_initial_metadata;
-    calld->on_done_recv = op->on_complete;
-    op->on_complete = &calld->hc_on_recv;
+    calld->on_done_recv = op->recv_initial_metadata_ready;
+    op->recv_initial_metadata_ready = &calld->hc_on_recv;
   }
 }
 
diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c
index bb75323933..370f8dbe42 100644
--- a/src/core/channel/http_server_filter.c
+++ b/src/core/channel/http_server_filter.c
@@ -186,8 +186,8 @@ static void hs_mutate_op(grpc_call_element *elem,
   if (op->recv_initial_metadata) {
     /* substitute our callback for the higher callback */
     calld->recv_initial_metadata = op->recv_initial_metadata;
-    calld->on_done_recv = op->on_complete;
-    op->on_complete = &calld->hs_on_recv;
+    calld->on_done_recv = op->recv_initial_metadata_ready;
+    op->recv_initial_metadata_ready = &calld->hs_on_recv;
   }
 }
 
diff --git a/src/core/channel/subchannel_call_holder.c b/src/core/channel/subchannel_call_holder.c
index 3ad9fd9efb..81297c8d44 100644
--- a/src/core/channel/subchannel_call_holder.c
+++ b/src/core/channel/subchannel_call_holder.c
@@ -241,10 +241,8 @@ static void fail_locked(grpc_exec_ctx *exec_ctx,
                         grpc_subchannel_call_holder *holder) {
   size_t i;
   for (i = 0; i < holder->waiting_ops_count; i++) {
-    grpc_exec_ctx_enqueue(exec_ctx, holder->waiting_ops[i].on_complete, false,
-                          NULL);
-    grpc_exec_ctx_enqueue(exec_ctx, holder->waiting_ops[i].recv_message_ready,
-                          false, NULL);
+    grpc_transport_stream_op_finish_with_failure(exec_ctx,
+                                                 &holder->waiting_ops[i]);
   }
   holder->waiting_ops_count = 0;
 }