Merge github.com:grpc/grpc into footprints-on-the-sands-of-time

4 files changed, 86 insertions, 174 deletions

diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 48b4b5799c..8e6047d89c 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -298,8 +298,6 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
   if (call->is_client) {
     call->request_set[GRPC_IOREQ_SEND_TRAILING_METADATA] = REQSET_DONE;
     call->request_set[GRPC_IOREQ_SEND_STATUS] = REQSET_DONE;
-    call->context[GRPC_CONTEXT_TRACING].value = grpc_census_context_create();
-    call->context[GRPC_CONTEXT_TRACING].destroy = grpc_census_context_destroy;
   }
   GPR_ASSERT(add_initial_metadata_count < MAX_SEND_INITIAL_METADATA_COUNT);
   for (i = 0; i < add_initial_metadata_count; i++) {
@@ -369,18 +367,18 @@ static void destroy_call(void *call, int ignored_success) {
   gpr_mu_destroy(&c->mu);
   for (i = 0; i < STATUS_SOURCE_COUNT; i++) {
     if (c->status[i].details) {
-      grpc_mdstr_unref(c->status[i].details);
+      GRPC_MDSTR_UNREF(c->status[i].details);
     }
   }
   for (i = 0; i < c->owned_metadata_count; i++) {
-    grpc_mdelem_unref(c->owned_metadata[i]);
+    GRPC_MDELEM_UNREF(c->owned_metadata[i]);
   }
   gpr_free(c->owned_metadata);
   for (i = 0; i < GPR_ARRAY_SIZE(c->buffered_metadata); i++) {
     gpr_free(c->buffered_metadata[i].metadata);
   }
   for (i = 0; i < c->send_initial_metadata_count; i++) {
-    grpc_mdelem_unref(c->send_initial_metadata[i].md);
+    GRPC_MDELEM_UNREF(c->send_initial_metadata[i].md);
   }
   for (i = 0; i < GRPC_CONTEXT_COUNT; i++) {
     if (c->context[i].destroy) {
@@ -437,7 +435,7 @@ static void set_decode_compression_level(grpc_call *call,
 static void set_status_details(grpc_call *call, status_source source,
                                grpc_mdstr *status) {
   if (call->status[source].details != NULL) {
-    grpc_mdstr_unref(call->status[source].details);
+    GRPC_MDSTR_UNREF(call->status[source].details);
   }
   call->status[source].details = status;
 }
@@ -616,7 +614,7 @@ static void finish_live_ioreq_op(grpc_call *call, grpc_ioreq_op op,
         case GRPC_IOREQ_SEND_STATUS:
           if (call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details !=
               NULL) {
-            grpc_mdstr_unref(
+            GRPC_MDSTR_UNREF(
                 call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details);
             call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details =
                 NULL;
@@ -945,7 +943,7 @@ static int fill_send_ops(grpc_call *call, grpc_transport_stream_op *op) {
                 &mdb, &call->details_link,
                 grpc_mdelem_from_metadata_strings(
                     call->metadata_context,
-                    grpc_mdstr_ref(
+                    GRPC_MDSTR_REF(
                         grpc_channel_get_message_string(call->channel)),
                     data.send_status.details));
             call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details =
@@ -1053,7 +1051,7 @@ static grpc_call_error start_ioreq(grpc_call *call, const grpc_ioreq *reqs,
                       reqs[i].data.send_status.code);
       if (reqs[i].data.send_status.details) {
         set_status_details(call, STATUS_FROM_SERVER_STATUS,
-                           grpc_mdstr_ref(reqs[i].data.send_status.details));
+                           GRPC_MDSTR_REF(reqs[i].data.send_status.details));
       }
     }
     have_ops |= 1u << op;
@@ -1258,7 +1256,7 @@ static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) {
     if (key == grpc_channel_get_status_string(call->channel)) {
       set_status_code(call, STATUS_FROM_WIRE, decode_status(md));
     } else if (key == grpc_channel_get_message_string(call->channel)) {
-      set_status_details(call, STATUS_FROM_WIRE, grpc_mdstr_ref(md->value));
+      set_status_details(call, STATUS_FROM_WIRE, GRPC_MDSTR_REF(md->value));
     } else if (key ==
                grpc_channel_get_compresssion_level_string(call->channel)) {
       set_decode_compression_level(call, decode_compression(md));
@@ -1294,10 +1292,10 @@ static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) {
 
   grpc_mdctx_lock(mdctx);
   for (l = md->list.head; l; l = l->next) {
-    if (l->md) grpc_mdctx_locked_mdelem_unref(mdctx, l->md);
+    if (l->md) GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
   }
   for (l = md->garbage.head; l; l = l->next) {
-    grpc_mdctx_locked_mdelem_unref(mdctx, l->md);
+    GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
   }
   grpc_mdctx_unlock(mdctx);
 }
diff --git a/src/core/surface/call_log_batch.c b/src/core/surface/call_log_batch.c
index 55663298c9..997046d954 100644
--- a/src/core/surface/call_log_batch.c
+++ b/src/core/surface/call_log_batch.c
@@ -46,8 +46,8 @@ static void add_metadata(gpr_strvec *b, const grpc_metadata *md, size_t count) {
     gpr_strvec_add(b, gpr_strdup(md[i].key));
 
     gpr_strvec_add(b, gpr_strdup(" value="));
-    gpr_strvec_add(b, gpr_hexdump(md[i].value, md[i].value_length,
-                                  GPR_HEXDUMP_PLAINTEXT));
+    gpr_strvec_add(b, gpr_dump(md[i].value, md[i].value_length,
+                               GPR_DUMP_HEX | GPR_DUMP_ASCII));
   }
 }
 
diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c
index f8151c121c..eeae3b507c 100644
--- a/src/core/surface/channel.c
+++ b/src/core/surface/channel.c
@@ -104,7 +104,7 @@ grpc_channel *grpc_channel_create_from_filters(
     char buf[GPR_LTOA_MIN_BUFSIZE];
     gpr_ltoa(i, buf);
     channel->grpc_status_elem[i] = grpc_mdelem_from_metadata_strings(
-        mdctx, grpc_mdstr_ref(channel->grpc_status_string),
+        mdctx, GRPC_MDSTR_REF(channel->grpc_status_string),
         grpc_mdstr_from_string(mdctx, buf));
   }
   channel->path_string = grpc_mdstr_from_string(mdctx, ":path");
@@ -157,10 +157,10 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
   return grpc_channel_create_call_internal(
       channel, cq,
       grpc_mdelem_from_metadata_strings(
-          channel->metadata_context, grpc_mdstr_ref(channel->path_string),
+          channel->metadata_context, GRPC_MDSTR_REF(channel->path_string),
           grpc_mdstr_from_string(channel->metadata_context, method)),
       grpc_mdelem_from_metadata_strings(
-          channel->metadata_context, grpc_mdstr_ref(channel->authority_string),
+          channel->metadata_context, GRPC_MDSTR_REF(channel->authority_string),
           grpc_mdstr_from_string(channel->metadata_context, host)),
       deadline);
 }
@@ -169,10 +169,10 @@ void *grpc_channel_register_call(grpc_channel *channel, const char *method,
                                  const char *host) {
   registered_call *rc = gpr_malloc(sizeof(registered_call));
   rc->path = grpc_mdelem_from_metadata_strings(
-      channel->metadata_context, grpc_mdstr_ref(channel->path_string),
+      channel->metadata_context, GRPC_MDSTR_REF(channel->path_string),
       grpc_mdstr_from_string(channel->metadata_context, method));
   rc->authority = grpc_mdelem_from_metadata_strings(
-      channel->metadata_context, grpc_mdstr_ref(channel->authority_string),
+      channel->metadata_context, GRPC_MDSTR_REF(channel->authority_string),
       grpc_mdstr_from_string(channel->metadata_context, host));
   gpr_mu_lock(&channel->registered_call_mu);
   rc->next = channel->registered_calls;
@@ -186,8 +186,8 @@ grpc_call *grpc_channel_create_registered_call(
     void *registered_call_handle, gpr_timespec deadline) {
   registered_call *rc = registered_call_handle;
   return grpc_channel_create_call_internal(
-      channel, completion_queue, grpc_mdelem_ref(rc->path),
-      grpc_mdelem_ref(rc->authority), deadline);
+      channel, completion_queue, GRPC_MDELEM_REF(rc->path),
+      GRPC_MDELEM_REF(rc->authority), deadline);
 }
 
 #ifdef GRPC_CHANNEL_REF_COUNT_DEBUG
@@ -205,18 +205,18 @@ static void destroy_channel(void *p, int ok) {
   size_t i;
   grpc_channel_stack_destroy(CHANNEL_STACK_FROM_CHANNEL(channel));
   for (i = 0; i < NUM_CACHED_STATUS_ELEMS; i++) {
-    grpc_mdelem_unref(channel->grpc_status_elem[i]);
+    GRPC_MDELEM_UNREF(channel->grpc_status_elem[i]);
   }
-  grpc_mdstr_unref(channel->grpc_status_string);
-  grpc_mdstr_unref(channel->grpc_compression_level_string);
-  grpc_mdstr_unref(channel->grpc_message_string);
-  grpc_mdstr_unref(channel->path_string);
-  grpc_mdstr_unref(channel->authority_string);
+  GRPC_MDSTR_UNREF(channel->grpc_status_string);
+  GRPC_MDSTR_UNREF(channel->grpc_compression_level_string);
+  GRPC_MDSTR_UNREF(channel->grpc_message_string);
+  GRPC_MDSTR_UNREF(channel->path_string);
+  GRPC_MDSTR_UNREF(channel->authority_string);
   while (channel->registered_calls) {
     registered_call *rc = channel->registered_calls;
     channel->registered_calls = rc->next;
-    grpc_mdelem_unref(rc->path);
-    grpc_mdelem_unref(rc->authority);
+    GRPC_MDELEM_UNREF(rc->path);
+    GRPC_MDELEM_UNREF(rc->authority);
     gpr_free(rc);
   }
   grpc_mdctx_unref(channel->metadata_context);
@@ -267,12 +267,12 @@ grpc_mdstr *grpc_channel_get_compresssion_level_string(grpc_channel *channel) {
 
 grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) {
   if (i >= 0 && i < NUM_CACHED_STATUS_ELEMS) {
-    return grpc_mdelem_ref(channel->grpc_status_elem[i]);
+    return GRPC_MDELEM_REF(channel->grpc_status_elem[i]);
   } else {
     char tmp[GPR_LTOA_MIN_BUFSIZE];
     gpr_ltoa(i, tmp);
     return grpc_mdelem_from_metadata_strings(
-        channel->metadata_context, grpc_mdstr_ref(channel->grpc_status_string),
+        channel->metadata_context, GRPC_MDSTR_REF(channel->grpc_status_string),
         grpc_mdstr_from_string(channel->metadata_context, tmp));
   }
 }
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index f29d47c17c..a9d8940631 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -51,7 +51,7 @@
 #include <grpc/support/string_util.h>
 #include <grpc/support/useful.h>
 
-typedef enum { PENDING_START, ALL_CALLS, CALL_LIST_COUNT } call_list;
+typedef enum { PENDING_START, CALL_LIST_COUNT } call_list;
 
 typedef struct listener {
   void *arg;
@@ -114,7 +114,6 @@ typedef struct channel_registered_method {
 
 struct channel_data {
   grpc_server *server;
-  size_t num_calls;
   grpc_connectivity_state connectivity_state;
   grpc_channel *channel;
   grpc_mdstr *path_key;
@@ -183,7 +182,11 @@ typedef enum {
 struct call_data {
   grpc_call *call;
 
+  /** protects state */
+  gpr_mu mu_state;
+  /** the current state of a call - see call_state */
   call_state state;
+
   grpc_mdstr *path;
   grpc_mdstr *host;
   gpr_timespec deadline;
@@ -204,9 +207,7 @@ struct call_data {
 
 typedef struct {
   grpc_channel **channels;
-  grpc_channel **disconnects;
   size_t num_channels;
-  size_t num_disconnects;
 } channel_broadcaster;
 
 #define SERVER_FROM_CALL_ELEM(elem) \
@@ -225,26 +226,15 @@ static void maybe_finish_shutdown(grpc_server *server);
 static void channel_broadcaster_init(grpc_server *s, channel_broadcaster *cb) {
   channel_data *c;
   size_t count = 0;
-  size_t dc_count = 0;
   for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
     count++;
-    if (c->num_calls == 0) {
-      dc_count++;
-    }
   }
   cb->num_channels = count;
-  cb->num_disconnects = dc_count;
   cb->channels = gpr_malloc(sizeof(*cb->channels) * cb->num_channels);
-  cb->disconnects = gpr_malloc(sizeof(*cb->channels) * cb->num_disconnects);
   count = 0;
-  dc_count = 0;
   for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
     cb->channels[count++] = c->channel;
     GRPC_CHANNEL_INTERNAL_REF(c->channel, "broadcast");
-    if (c->num_calls == 0) {
-      cb->disconnects[dc_count++] = c->channel;
-      GRPC_CHANNEL_INTERNAL_REF(c->channel, "broadcast-disconnect");
-    }
   }
 }
 
@@ -280,19 +270,14 @@ static void send_shutdown(grpc_channel *channel, int send_goaway,
 }
 
 static void channel_broadcaster_shutdown(channel_broadcaster *cb,
-                                         int send_goaway, int send_disconnect) {
+                                         int send_goaway, int force_disconnect) {
   size_t i;
 
   for (i = 0; i < cb->num_channels; i++) {
-    send_shutdown(cb->channels[i], 1, 0);
+    send_shutdown(cb->channels[i], send_goaway, force_disconnect);
     GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast");
   }
-  for (i = 0; i < cb->num_disconnects; i++) {
-    send_shutdown(cb->disconnects[i], 0, 1);
-    GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast-disconnect");
-  }
   gpr_free(cb->channels);
-  gpr_free(cb->disconnects);
 }
 
 /* call list */
@@ -422,19 +407,23 @@ static void destroy_channel(channel_data *chand) {
   grpc_iomgr_add_callback(&chand->finish_destroy_channel_closure);
 }
 
-static void finish_start_new_rpc_and_unlock(grpc_server *server,
-                                            grpc_call_element *elem,
-                                            call_data **pending_root,
-                                            requested_call_array *array) {
+static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem,
+                                 call_data **pending_root,
+                                 requested_call_array *array) {
   requested_call rc;
   call_data *calld = elem->call_data;
+  gpr_mu_lock(&server->mu_call);
   if (array->count == 0) {
+    gpr_mu_lock(&calld->mu_state);
     calld->state = PENDING;
+    gpr_mu_unlock(&calld->mu_state);
     call_list_join(pending_root, calld, PENDING_START);
     gpr_mu_unlock(&server->mu_call);
   } else {
     rc = array->calls[--array->count];
+    gpr_mu_lock(&calld->mu_state);
     calld->state = ACTIVATED;
+    gpr_mu_unlock(&calld->mu_state);
     gpr_mu_unlock(&server->mu_call);
     begin_call(server, calld, &rc);
   }
@@ -448,7 +437,6 @@ static void start_new_rpc(grpc_call_element *elem) {
   gpr_uint32 hash;
   channel_registered_method *rm;
 
-  gpr_mu_lock(&server->mu_call);
   if (chand->registered_methods && calld->path && calld->host) {
     /* TODO(ctiller): unify these two searches */
     /* check for an exact match with host */
@@ -459,9 +447,8 @@ static void start_new_rpc(grpc_call_element *elem) {
       if (!rm) break;
       if (rm->host != calld->host) continue;
       if (rm->method != calld->path) continue;
-      finish_start_new_rpc_and_unlock(server, elem,
-                                      &rm->server_registered_method->pending,
-                                      &rm->server_registered_method->requested);
+      finish_start_new_rpc(server, elem, &rm->server_registered_method->pending,
+                           &rm->server_registered_method->requested);
       return;
     }
     /* check for a wildcard method definition (no host set) */
@@ -472,14 +459,13 @@ static void start_new_rpc(grpc_call_element *elem) {
       if (!rm) break;
       if (rm->host != NULL) continue;
       if (rm->method != calld->path) continue;
-      finish_start_new_rpc_and_unlock(server, elem,
-                                      &rm->server_registered_method->pending,
-                                      &rm->server_registered_method->requested);
+      finish_start_new_rpc(server, elem, &rm->server_registered_method->pending,
+                           &rm->server_registered_method->requested);
       return;
     }
   }
-  finish_start_new_rpc_and_unlock(server, elem, &server->lists[PENDING_START],
-                                  &server->requested_calls);
+  finish_start_new_rpc(server, elem, &server->lists[PENDING_START],
+                       &server->requested_calls);
 }
 
 static void kill_zombie(void *elem, int success) {
@@ -501,15 +487,6 @@ static void maybe_finish_shutdown(grpc_server *server) {
     return;
   }
 
-  gpr_mu_lock(&server->mu_call);
-  if (server->lists[ALL_CALLS] != NULL) {
-    gpr_log(GPR_DEBUG,
-            "Waiting for all calls to finish before destroying server");
-    gpr_mu_unlock(&server->mu_call);
-    return;
-  }
-  gpr_mu_unlock(&server->mu_call);
-
   if (server->root_channel_data.next != &server->root_channel_data) {
     gpr_log(GPR_DEBUG,
             "Waiting for all channels to close before destroying server");
@@ -532,31 +509,19 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
   channel_data *chand = elem->channel_data;
   call_data *calld = elem->call_data;
   if (md->key == chand->path_key) {
-    calld->path = grpc_mdstr_ref(md->value);
+    calld->path = GRPC_MDSTR_REF(md->value);
     return NULL;
   } else if (md->key == chand->authority_key) {
-    calld->host = grpc_mdstr_ref(md->value);
+    calld->host = GRPC_MDSTR_REF(md->value);
     return NULL;
   }
   return md;
 }
 
-static int decrement_call_count(channel_data *chand) {
-  int disconnect = 0;
-  chand->num_calls--;
-  if (0 == chand->num_calls && chand->server->shutdown) {
-    disconnect = 1;
-  }
-  maybe_finish_shutdown(chand->server);
-  return disconnect;
-}
-
 static void server_on_recv(void *ptr, int success) {
   grpc_call_element *elem = ptr;
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
-  int remove_res;
-  int disconnect = 0;
 
   if (success && !calld->got_initial_metadata) {
     size_t i;
@@ -581,39 +546,33 @@ static void server_on_recv(void *ptr, int success) {
     case GRPC_STREAM_SEND_CLOSED:
       break;
     case GRPC_STREAM_RECV_CLOSED:
-      gpr_mu_lock(&chand->server->mu_call);
+      gpr_mu_lock(&calld->mu_state);
       if (calld->state == NOT_STARTED) {
         calld->state = ZOMBIED;
+        gpr_mu_unlock(&calld->mu_state);
         grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
         grpc_iomgr_add_callback(&calld->kill_zombie_closure);
+      } else {
+        gpr_mu_unlock(&calld->mu_state);
       }
-      gpr_mu_unlock(&chand->server->mu_call);
       break;
     case GRPC_STREAM_CLOSED:
-      gpr_mu_lock(&chand->server->mu_call);
+      gpr_mu_lock(&calld->mu_state);
       if (calld->state == NOT_STARTED) {
         calld->state = ZOMBIED;
+        gpr_mu_unlock(&calld->mu_state);
         grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
         grpc_iomgr_add_callback(&calld->kill_zombie_closure);
       } else if (calld->state == PENDING) {
-        call_list_remove(calld, PENDING_START);
         calld->state = ZOMBIED;
+        gpr_mu_unlock(&calld->mu_state);
+        gpr_mu_lock(&chand->server->mu_call);
+        call_list_remove(calld, PENDING_START);
+        gpr_mu_unlock(&chand->server->mu_call);
         grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
         grpc_iomgr_add_callback(&calld->kill_zombie_closure);
-      }
-      remove_res = call_list_remove(calld, ALL_CALLS);
-      gpr_mu_unlock(&chand->server->mu_call);
-      gpr_mu_lock(&chand->server->mu_global);
-      if (remove_res) {
-        disconnect = decrement_call_count(chand);
-        if (disconnect) {
-          GRPC_CHANNEL_INTERNAL_REF(chand->channel, "send-disconnect");
-        }
-      }
-      gpr_mu_unlock(&chand->server->mu_global);
-      if (disconnect) {
-        send_shutdown(chand->channel, 0, 1);
-        GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "send-disconnect");
+      } else {
+        gpr_mu_unlock(&calld->mu_state);
       }
       break;
   }
@@ -676,17 +635,10 @@ static void init_call_elem(grpc_call_element *elem,
   memset(calld, 0, sizeof(call_data));
   calld->deadline = gpr_inf_future;
   calld->call = grpc_call_from_top_element(elem);
+  gpr_mu_init(&calld->mu_state);
 
   grpc_iomgr_closure_init(&calld->server_on_recv, server_on_recv, elem);
 
-  gpr_mu_lock(&chand->server->mu_call);
-  call_list_join(&chand->server->lists[ALL_CALLS], calld, ALL_CALLS);
-  gpr_mu_unlock(&chand->server->mu_call);
-
-  gpr_mu_lock(&chand->server->mu_global);
-  chand->num_calls++;
-  gpr_mu_unlock(&chand->server->mu_global);
-
   server_ref(chand->server);
 
   if (initial_op) server_mutate_op(elem, initial_op);
@@ -695,27 +647,22 @@ static void init_call_elem(grpc_call_element *elem,
 static void destroy_call_elem(grpc_call_element *elem) {
   channel_data *chand = elem->channel_data;
   call_data *calld = elem->call_data;
-  int removed[CALL_LIST_COUNT];
-  size_t i;
 
-  gpr_mu_lock(&chand->server->mu_call);
-  for (i = 0; i < CALL_LIST_COUNT; i++) {
-    removed[i] = call_list_remove(elem->call_data, i);
-  }
-  gpr_mu_unlock(&chand->server->mu_call);
-  if (removed[ALL_CALLS]) {
-    gpr_mu_lock(&chand->server->mu_global);
-    decrement_call_count(chand);
-    gpr_mu_unlock(&chand->server->mu_global);
+  if (calld->state == PENDING) {
+    gpr_mu_lock(&chand->server->mu_call);
+    call_list_remove(elem->call_data, PENDING_START);
+    gpr_mu_unlock(&chand->server->mu_call);
   }
 
   if (calld->host) {
-    grpc_mdstr_unref(calld->host);
+    GRPC_MDSTR_UNREF(calld->host);
   }
   if (calld->path) {
-    grpc_mdstr_unref(calld->path);
+    GRPC_MDSTR_UNREF(calld->path);
   }
 
+  gpr_mu_destroy(&calld->mu_state);
+
   server_unref(chand->server);
 }
 
@@ -727,7 +674,6 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
   GPR_ASSERT(is_first);
   GPR_ASSERT(!is_last);
   chand->server = NULL;
-  chand->num_calls = 0;
   chand->channel = NULL;
   chand->path_key = grpc_mdstr_from_string(metadata_context, ":path");
   chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority");
@@ -744,10 +690,10 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
   if (chand->registered_methods) {
     for (i = 0; i < chand->registered_method_slots; i++) {
       if (chand->registered_methods[i].method) {
-        grpc_mdstr_unref(chand->registered_methods[i].method);
+        GRPC_MDSTR_UNREF(chand->registered_methods[i].method);
       }
       if (chand->registered_methods[i].host) {
-        grpc_mdstr_unref(chand->registered_methods[i].host);
+        GRPC_MDSTR_UNREF(chand->registered_methods[i].host);
       }
     }
     gpr_free(chand->registered_methods);
@@ -759,8 +705,8 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
     chand->next = chand->prev = chand;
     maybe_finish_shutdown(chand->server);
     gpr_mu_unlock(&chand->server->mu_global);
-    grpc_mdstr_unref(chand->path_key);
-    grpc_mdstr_unref(chand->authority_key);
+    GRPC_MDSTR_UNREF(chand->path_key);
+    GRPC_MDSTR_UNREF(chand->authority_key);
     server_unref(chand->server);
   }
 }
@@ -1049,47 +995,13 @@ void grpc_server_listener_destroy_done(void *s) {
 }
 
 void grpc_server_cancel_all_calls(grpc_server *server) {
-  call_data *calld;
-  grpc_call **calls;
-  size_t call_count;
-  size_t call_capacity;
-  int is_first = 1;
-  size_t i;
-
-  gpr_mu_lock(&server->mu_call);
-
-  GPR_ASSERT(server->shutdown);
-
-  if (!server->lists[ALL_CALLS]) {
-    gpr_mu_unlock(&server->mu_call);
-    return;
-  }
-
-  call_capacity = 8;
-  call_count = 0;
-  calls = gpr_malloc(sizeof(grpc_call *) * call_capacity);
-
-  for (calld = server->lists[ALL_CALLS];
-       calld != server->lists[ALL_CALLS] || is_first;
-       calld = calld->links[ALL_CALLS].next) {
-    if (call_count == call_capacity) {
-      call_capacity *= 2;
-      calls = gpr_realloc(calls, sizeof(grpc_call *) * call_capacity);
-    }
-    calls[call_count++] = calld->call;
-    GRPC_CALL_INTERNAL_REF(calld->call, "cancel_all");
-    is_first = 0;
-  }
-
-  gpr_mu_unlock(&server->mu_call);
+  channel_broadcaster broadcaster;
 
-  for (i = 0; i < call_count; i++) {
-    grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE,
-                                 "Unavailable");
-    GRPC_CALL_INTERNAL_UNREF(calls[i], "cancel_all", 1);
-  }
+  gpr_mu_lock(&server->mu_global);
+  channel_broadcaster_init(server, &broadcaster);
+  gpr_mu_unlock(&server->mu_global);
 
-  gpr_free(calls);
+  channel_broadcaster_shutdown(&broadcaster, 0, 1);
 }
 
 void grpc_server_destroy(grpc_server *server) {
@@ -1145,10 +1057,12 @@ static grpc_call_error queue_call_request(grpc_server *server,
       requested_calls = &rc->data.registered.registered_method->requested;
       break;
   }
-  if (calld) {
+  if (calld != NULL) {
+    gpr_mu_unlock(&server->mu_call);
+    gpr_mu_lock(&calld->mu_state);
     GPR_ASSERT(calld->state == PENDING);
     calld->state = ACTIVATED;
-    gpr_mu_unlock(&server->mu_call);
+    gpr_mu_unlock(&calld->mu_state);
     begin_call(server, calld, rc);
     return GRPC_CALL_OK;
   } else {