Merge branch 'master' of github.com:grpc/grpc into decompression

1 files changed, 163 insertions, 111 deletions

diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 80bbe42af4..076301777b 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -114,6 +114,7 @@ typedef struct channel_registered_method {
 
 struct channel_data {
   grpc_server *server;
+  size_t num_calls;
   grpc_channel *channel;
   grpc_mdstr *path_key;
   grpc_mdstr *authority_key;
@@ -123,7 +124,6 @@ struct channel_data {
   channel_registered_method *registered_methods;
   gpr_uint32 registered_method_slots;
   gpr_uint32 registered_method_max_probes;
-  grpc_iomgr_closure finish_shutdown_channel_closure;
   grpc_iomgr_closure finish_destroy_channel_closure;
 };
 
@@ -141,7 +141,15 @@ struct grpc_server {
   grpc_pollset **pollsets;
   size_t cq_count;
 
-  gpr_mu mu;
+  /* The two following mutexes control access to server-state
+     mu_global controls access to non-call-related state (e.g., channel state)
+     mu_call controls access to call-related state (e.g., the call lists)
+
+     If they are ever required to be nested, you must lock mu_global
+     before mu_call. This is currently used in shutdown processing
+     (grpc_server_shutdown_and_notify and maybe_finish_shutdown) */
+  gpr_mu mu_global; /* mutex for server and channel state */
+  gpr_mu mu_call; /* mutex for call-specific state */
 
   registered_method *registered_methods;
   requested_call_array requested_calls;
@@ -198,6 +206,11 @@ struct call_data {
 static void begin_call(grpc_server *server, call_data *calld,
                        requested_call *rc);
 static void fail_call(grpc_server *server, requested_call *rc);
+static void shutdown_channel(channel_data *chand, int send_goaway,
+                             int send_disconnect);
+/* Before calling maybe_finish_shutdown, we must hold mu_global and not
+   hold mu_call */
+static void maybe_finish_shutdown(grpc_server *server);
 
 static int call_list_join(call_data **root, call_data *call, call_list list) {
   GPR_ASSERT(!call->root[list]);
@@ -270,7 +283,8 @@ static void server_delete(grpc_server *server) {
   registered_method *rm;
   size_t i;
   grpc_channel_args_destroy(server->channel_args);
-  gpr_mu_destroy(&server->mu);
+  gpr_mu_destroy(&server->mu_global);
+  gpr_mu_destroy(&server->mu_call);
   gpr_free(server->channel_filters);
   requested_call_array_destroy(&server->requested_calls);
   while ((rm = server->registered_methods) != NULL) {
@@ -281,7 +295,7 @@ static void server_delete(grpc_server *server) {
     gpr_free(rm);
   }
   for (i = 0; i < server->cq_count; i++) {
-    grpc_cq_internal_unref(server->cqs[i]);
+    GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server");
   }
   gpr_free(server->cqs);
   gpr_free(server->pollsets);
@@ -308,7 +322,7 @@ static void orphan_channel(channel_data *chand) {
 static void finish_destroy_channel(void *cd, int success) {
   channel_data *chand = cd;
   grpc_server *server = chand->server;
-  grpc_channel_internal_unref(chand->channel);
+  GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server");
   server_unref(server);
 }
 
@@ -317,6 +331,7 @@ static void destroy_channel(channel_data *chand) {
   GPR_ASSERT(chand->server != NULL);
   orphan_channel(chand);
   server_ref(chand->server);
+  maybe_finish_shutdown(chand->server);
   chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
   chand->finish_destroy_channel_closure.cb_arg = chand;
   grpc_iomgr_add_callback(&chand->finish_destroy_channel_closure);
@@ -331,11 +346,11 @@ static void finish_start_new_rpc_and_unlock(grpc_server *server,
   if (array->count == 0) {
     calld->state = PENDING;
     call_list_join(pending_root, calld, PENDING_START);
-    gpr_mu_unlock(&server->mu);
+    gpr_mu_unlock(&server->mu_call);
   } else {
     rc = array->calls[--array->count];
     calld->state = ACTIVATED;
-    gpr_mu_unlock(&server->mu);
+    gpr_mu_unlock(&server->mu_call);
     begin_call(server, calld, &rc);
   }
 }
@@ -348,7 +363,7 @@ static void start_new_rpc(grpc_call_element *elem) {
   gpr_uint32 hash;
   channel_registered_method *rm;
 
-  gpr_mu_lock(&server->mu);
+  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 */
@@ -397,12 +412,33 @@ static int num_listeners(grpc_server *server) {
 
 static void maybe_finish_shutdown(grpc_server *server) {
   size_t i;
-  if (server->shutdown && !server->shutdown_published && server->lists[ALL_CALLS] == NULL && server->listeners_destroyed == num_listeners(server)) {
-    server->shutdown_published = 1;
-    for (i = 0; i < server->num_shutdown_tags; i++) {
-      grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag,
-                     NULL, 1);
-    }
+  if (!server->shutdown || server->shutdown_published) {
+    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");
+    return;
+  }
+  if (server->listeners_destroyed < num_listeners(server)) {
+    gpr_log(GPR_DEBUG, "Waiting for all listeners to be destroyed (@ %d/%d)",
+            server->listeners_destroyed, num_listeners(server));
+    return;
+  }
+  server->shutdown_published = 1;
+  for (i = 0; i < server->num_shutdown_tags; i++) {
+    grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag,
+                   NULL, 1);
   }
 }
 
@@ -420,10 +456,19 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
   return md;
 }
 
+static void decrement_call_count(channel_data *chand) {
+  chand->num_calls--;
+  if (0 == chand->num_calls && chand->server->shutdown) {
+    shutdown_channel(chand, 0, 1);
+  }
+  maybe_finish_shutdown(chand->server);
+}
+
 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;
 
   if (success && !calld->got_initial_metadata) {
     size_t i;
@@ -448,16 +493,16 @@ 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);
+      gpr_mu_lock(&chand->server->mu_call);
       if (calld->state == NOT_STARTED) {
         calld->state = ZOMBIED;
         grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
         grpc_iomgr_add_callback(&calld->kill_zombie_closure);
       }
-      gpr_mu_unlock(&chand->server->mu);
+      gpr_mu_unlock(&chand->server->mu_call);
       break;
     case GRPC_STREAM_CLOSED:
-      gpr_mu_lock(&chand->server->mu);
+      gpr_mu_lock(&chand->server->mu_call);
       if (calld->state == NOT_STARTED) {
         calld->state = ZOMBIED;
         grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
@@ -467,12 +512,14 @@ static void server_on_recv(void *ptr, int success) {
         calld->state = ZOMBIED;
         grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
         grpc_iomgr_add_callback(&calld->kill_zombie_closure);
-
       }
-      if (call_list_remove(calld, ALL_CALLS)) {
-        maybe_finish_shutdown(chand->server);
+      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) {
+        decrement_call_count(chand);
       }
-      gpr_mu_unlock(&chand->server->mu);
+      gpr_mu_unlock(&chand->server->mu_global);
       break;
   }
 
@@ -515,10 +562,10 @@ static void channel_op(grpc_channel_element *elem,
     case GRPC_TRANSPORT_CLOSED:
       /* if the transport is closed for a server channel, we destroy the
          channel */
-      gpr_mu_lock(&server->mu);
+      gpr_mu_lock(&server->mu_global);
       server_ref(server);
       destroy_channel(chand);
-      gpr_mu_unlock(&server->mu);
+      gpr_mu_unlock(&server->mu_global);
       server_unref(server);
       break;
     case GRPC_TRANSPORT_GOAWAY:
@@ -531,22 +578,49 @@ static void channel_op(grpc_channel_element *elem,
   }
 }
 
-static void finish_shutdown_channel(void *cd, int success) {
-  channel_data *chand = cd;
+typedef struct {
+  channel_data *chand;
+  int send_goaway;
+  int send_disconnect;
+  grpc_iomgr_closure finish_shutdown_channel_closure;
+} shutdown_channel_args;
+
+static void finish_shutdown_channel(void *p, int success) {
+  shutdown_channel_args *sca = p;
   grpc_channel_op op;
-  op.type = GRPC_CHANNEL_DISCONNECT;
-  op.dir = GRPC_CALL_DOWN;
-  channel_op(grpc_channel_stack_element(
-                 grpc_channel_get_channel_stack(chand->channel), 0),
-             NULL, &op);
-  grpc_channel_internal_unref(chand->channel);
+
+  if (sca->send_goaway) {
+    op.type = GRPC_CHANNEL_GOAWAY;
+    op.dir = GRPC_CALL_DOWN;
+    op.data.goaway.status = GRPC_STATUS_OK;
+    op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown");
+    channel_op(grpc_channel_stack_element(
+                   grpc_channel_get_channel_stack(sca->chand->channel), 0),
+               NULL, &op);
+  }
+  if (sca->send_disconnect) {
+    op.type = GRPC_CHANNEL_DISCONNECT;
+    op.dir = GRPC_CALL_DOWN;
+    channel_op(grpc_channel_stack_element(
+                   grpc_channel_get_channel_stack(sca->chand->channel), 0),
+               NULL, &op);
+  }
+  GRPC_CHANNEL_INTERNAL_UNREF(sca->chand->channel, "shutdown");
+
+  gpr_free(sca);
 }
 
-static void shutdown_channel(channel_data *chand) {
-  grpc_channel_internal_ref(chand->channel);
-  chand->finish_shutdown_channel_closure.cb = finish_shutdown_channel;
-  chand->finish_shutdown_channel_closure.cb_arg = chand;
-  grpc_iomgr_add_callback(&chand->finish_shutdown_channel_closure);
+static void shutdown_channel(channel_data *chand, int send_goaway,
+                             int send_disconnect) {
+  shutdown_channel_args *sca;
+  GRPC_CHANNEL_INTERNAL_REF(chand->channel, "shutdown");
+  sca = gpr_malloc(sizeof(shutdown_channel_args));
+  sca->chand = chand;
+  sca->send_goaway = send_goaway;
+  sca->send_disconnect = send_disconnect;
+  sca->finish_shutdown_channel_closure.cb = finish_shutdown_channel;
+  sca->finish_shutdown_channel_closure.cb_arg = sca;
+  grpc_iomgr_add_callback(&sca->finish_shutdown_channel_closure);
 }
 
 static void init_call_elem(grpc_call_element *elem,
@@ -558,9 +632,13 @@ static void init_call_elem(grpc_call_element *elem,
   calld->deadline = gpr_inf_future;
   calld->call = grpc_call_from_top_element(elem);
 
-  gpr_mu_lock(&chand->server->mu);
+  gpr_mu_lock(&chand->server->mu_call);
   call_list_join(&chand->server->lists[ALL_CALLS], calld, ALL_CALLS);
-  gpr_mu_unlock(&chand->server->mu);
+  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);
 
@@ -573,14 +651,16 @@ static void destroy_call_elem(grpc_call_element *elem) {
   int removed[CALL_LIST_COUNT];
   size_t i;
 
-  gpr_mu_lock(&chand->server->mu);
+  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]) {
-    maybe_finish_shutdown(chand->server);
+    gpr_mu_lock(&chand->server->mu_global);
+    decrement_call_count(chand);
+    gpr_mu_unlock(&chand->server->mu_global);
   }
-  gpr_mu_unlock(&chand->server->mu);
 
   if (calld->host) {
     grpc_mdstr_unref(calld->host);
@@ -600,6 +680,7 @@ static void init_channel_elem(grpc_channel_element *elem,
   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");
@@ -622,11 +703,12 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
     gpr_free(chand->registered_methods);
   }
   if (chand->server) {
-    gpr_mu_lock(&chand->server->mu);
+    gpr_mu_lock(&chand->server->mu_global);
     chand->next->prev = chand->prev;
     chand->prev->next = chand->next;
     chand->next = chand->prev = chand;
-    gpr_mu_unlock(&chand->server->mu);
+    maybe_finish_shutdown(chand->server);
+    gpr_mu_unlock(&chand->server->mu_global);
     grpc_mdstr_unref(chand->path_key);
     grpc_mdstr_unref(chand->authority_key);
     server_unref(chand->server);
@@ -651,7 +733,8 @@ void grpc_server_register_completion_queue(grpc_server *server,
   for (i = 0; i < server->cq_count; i++) {
     if (server->cqs[i] == cq) return;
   }
-  grpc_cq_internal_ref(cq);
+  GRPC_CQ_INTERNAL_REF(cq, "server");
+  grpc_cq_mark_server_cq(cq);
   n = server->cq_count++;
   server->cqs = gpr_realloc(server->cqs,
                             server->cq_count * sizeof(grpc_completion_queue *));
@@ -672,7 +755,8 @@ grpc_server *grpc_server_create_from_filters(
 
   memset(server, 0, sizeof(grpc_server));
 
-  gpr_mu_init(&server->mu);
+  gpr_mu_init(&server->mu_global);
+  gpr_mu_init(&server->mu_call);
 
   /* decremented by grpc_server_destroy */
   gpr_ref_init(&server->internal_refcount, 1);
@@ -713,7 +797,8 @@ void *grpc_server_register_method(grpc_server *server, const char *method,
                                   const char *host) {
   registered_method *m;
   if (!method) {
-    gpr_log(GPR_ERROR, "grpc_server_register_method method string cannot be NULL");
+    gpr_log(GPR_ERROR,
+            "grpc_server_register_method method string cannot be NULL");
     return NULL;
   }
   for (m = server->registered_methods; m; m = m->next) {
@@ -821,11 +906,11 @@ grpc_transport_setup_result grpc_server_setup_transport(
   result = grpc_connected_channel_bind_transport(
       grpc_channel_get_channel_stack(channel), transport);
 
-  gpr_mu_lock(&s->mu);
+  gpr_mu_lock(&s->mu_global);
   chand->next = &s->root_channel_data;
   chand->prev = chand->next->prev;
   chand->next->prev = chand->prev->next = chand;
-  gpr_mu_unlock(&s->mu);
+  gpr_mu_unlock(&s->mu_global);
 
   gpr_free(filters);
 
@@ -836,17 +921,13 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
                                      grpc_completion_queue *cq, void *tag) {
   listener *l;
   requested_call_array requested_calls;
-  channel_data **channels;
   channel_data *c;
-  size_t nchannels;
   size_t i;
-  grpc_channel_op op;
-  grpc_channel_element *elem;
   registered_method *rm;
   shutdown_tag *sdt;
 
   /* lock, and gather up some stuff to do */
-  gpr_mu_lock(&server->mu);
+  gpr_mu_lock(&server->mu_global);
   grpc_cq_begin_op(cq, NULL);
   server->shutdown_tags =
       gpr_realloc(server->shutdown_tags,
@@ -855,25 +936,17 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
   sdt->tag = tag;
   sdt->cq = cq;
   if (server->shutdown) {
-    gpr_mu_unlock(&server->mu);
+    gpr_mu_unlock(&server->mu_global);
     return;
   }
 
-  nchannels = 0;
   for (c = server->root_channel_data.next; c != &server->root_channel_data;
        c = c->next) {
-    nchannels++;
-  }
-  channels = gpr_malloc(sizeof(channel_data *) * nchannels);
-  i = 0;
-  for (c = server->root_channel_data.next; c != &server->root_channel_data;
-       c = c->next) {
-    grpc_channel_internal_ref(c->channel);
-    channels[i] = c;
-    i++;
+    shutdown_channel(c, 1, c->num_calls == 0);
   }
 
   /* collect all unregistered then registered calls */
+  gpr_mu_lock(&server->mu_call);
   requested_calls = server->requested_calls;
   memset(&server->requested_calls, 0, sizeof(server->requested_calls));
   for (rm = server->registered_methods; rm; rm = rm->next) {
@@ -892,25 +965,11 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
     gpr_free(rm->requested.calls);
     memset(&rm->requested, 0, sizeof(rm->requested));
   }
+  gpr_mu_unlock(&server->mu_call);
 
   server->shutdown = 1;
   maybe_finish_shutdown(server);
-  gpr_mu_unlock(&server->mu);
-
-  for (i = 0; i < nchannels; i++) {
-    c = channels[i];
-    elem = grpc_channel_stack_element(
-        grpc_channel_get_channel_stack(c->channel), 0);
-
-    op.type = GRPC_CHANNEL_GOAWAY;
-    op.dir = GRPC_CALL_DOWN;
-    op.data.goaway.status = GRPC_STATUS_OK;
-    op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown");
-    elem->filter->channel_op(elem, NULL, &op);
-
-    grpc_channel_internal_unref(c->channel);
-  }
-  gpr_free(channels);
+  gpr_mu_unlock(&server->mu_global);
 
   /* terminate all the requested calls */
   for (i = 0; i < requested_calls.count; i++) {
@@ -926,10 +985,10 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
 
 void grpc_server_listener_destroy_done(void *s) {
   grpc_server *server = s;
-  gpr_mu_lock(&server->mu);
+  gpr_mu_lock(&server->mu_global);
   server->listeners_destroyed++;
   maybe_finish_shutdown(server);
-  gpr_mu_unlock(&server->mu);
+  gpr_mu_unlock(&server->mu_global);
 }
 
 void grpc_server_cancel_all_calls(grpc_server *server) {
@@ -940,12 +999,12 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
   int is_first = 1;
   size_t i;
 
-  gpr_mu_lock(&server->mu);
+  gpr_mu_lock(&server->mu_call);
 
   GPR_ASSERT(server->shutdown);
 
   if (!server->lists[ALL_CALLS]) {
-    gpr_mu_unlock(&server->mu);
+    gpr_mu_unlock(&server->mu_call);
     return;
   }
 
@@ -953,7 +1012,9 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
   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) {
+  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);
@@ -963,10 +1024,11 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
     is_first = 0;
   }
 
-  gpr_mu_unlock(&server->mu);
+  gpr_mu_unlock(&server->mu_call);
 
   for (i = 0; i < call_count; i++) {
-    grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE, "Unavailable");
+    grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE,
+                                 "Unavailable");
     GRPC_CALL_INTERNAL_UNREF(calls[i], "cancel_all", 1);
   }
 
@@ -974,11 +1036,9 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
 }
 
 void grpc_server_destroy(grpc_server *server) {
-  channel_data *c;
   listener *l;
-  call_data *calld;
 
-  gpr_mu_lock(&server->mu);
+  gpr_mu_lock(&server->mu_global);
   GPR_ASSERT(server->shutdown || !server->listeners);
   GPR_ASSERT(server->listeners_destroyed == num_listeners(server));
 
@@ -988,20 +1048,7 @@ void grpc_server_destroy(grpc_server *server) {
     gpr_free(l);
   }
 
-  while ((calld = call_list_remove_head(&server->lists[PENDING_START],
-                                        PENDING_START)) != NULL) {
-    calld->state = ZOMBIED;
-    grpc_iomgr_closure_init(
-        &calld->kill_zombie_closure, kill_zombie,
-        grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
-    grpc_iomgr_add_callback(&calld->kill_zombie_closure);
-  }
-
-  for (c = server->root_channel_data.next; c != &server->root_channel_data;
-       c = c->next) {
-    shutdown_channel(c);
-  }
-  gpr_mu_unlock(&server->mu);
+  gpr_mu_unlock(&server->mu_global);
 
   server_unref(server);
 }
@@ -1023,9 +1070,9 @@ static grpc_call_error queue_call_request(grpc_server *server,
                                           requested_call *rc) {
   call_data *calld = NULL;
   requested_call_array *requested_calls = NULL;
-  gpr_mu_lock(&server->mu);
+  gpr_mu_lock(&server->mu_call);
   if (server->shutdown) {
-    gpr_mu_unlock(&server->mu);
+    gpr_mu_unlock(&server->mu_call);
     fail_call(server, rc);
     return GRPC_CALL_OK;
   }
@@ -1044,12 +1091,12 @@ static grpc_call_error queue_call_request(grpc_server *server,
   if (calld) {
     GPR_ASSERT(calld->state == PENDING);
     calld->state = ACTIVATED;
-    gpr_mu_unlock(&server->mu);
+    gpr_mu_unlock(&server->mu_call);
     begin_call(server, calld, rc);
     return GRPC_CALL_OK;
   } else {
     *requested_call_array_add(requested_calls) = *rc;
-    gpr_mu_unlock(&server->mu);
+    gpr_mu_unlock(&server->mu_call);
     return GRPC_CALL_OK;
   }
 }
@@ -1063,6 +1110,9 @@ grpc_call_error grpc_server_request_call(
   GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details,
                                initial_metadata, cq_bound_to_call,
                                cq_for_notification, tag);
+  if (!grpc_cq_is_server_cq(cq_for_notification)) {
+    return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+  }
   grpc_cq_begin_op(cq_for_notification, NULL);
   rc.type = BATCH_CALL;
   rc.tag = tag;
@@ -1081,6 +1131,9 @@ grpc_call_error grpc_server_request_registered_call(
     grpc_completion_queue *cq_for_notification, void *tag) {
   requested_call rc;
   registered_method *registered_method = rm;
+  if (!grpc_cq_is_server_cq(cq_for_notification)) {
+    return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+  }
   grpc_cq_begin_op(cq_for_notification, NULL);
   rc.type = REGISTERED_CALL;
   rc.tag = tag;
@@ -1187,9 +1240,8 @@ const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {
 
 int grpc_server_has_open_connections(grpc_server *server) {
   int r;
-  gpr_mu_lock(&server->mu);
+  gpr_mu_lock(&server->mu_global);
   r = server->root_channel_data.next != &server->root_channel_data;
-  gpr_mu_unlock(&server->mu);
+  gpr_mu_unlock(&server->mu_global);
   return r;
 }
-