Verify that completion queues are server registered

include/grpc/grpc.h

@@ -144,7 +144,10 @@ typedef enum grpc_call_error {
   /* the flags value was illegal for this call */
   GRPC_CALL_ERROR_INVALID_FLAGS,
   /* invalid metadata was passed to this call */
-  GRPC_CALL_ERROR_INVALID_METADATA
+  GRPC_CALL_ERROR_INVALID_METADATA,
+  /* completion queue for notification has not been registered with the server
+     */
+  GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE
 } grpc_call_error;
 
 /* Write Flags: */

src/core/surface/completion_queue.c

@@ -73,6 +73,7 @@ struct grpc_completion_queue {
   event *queue;
   /* Fixed size chained hash table of events for pluck() */
   event *buckets[NUM_TAG_BUCKETS];
+  int is_server_cq;
 };
 
 grpc_completion_queue *grpc_completion_queue_create(void) {
@@ -323,3 +324,7 @@ void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc) {
                     gpr_time_add(gpr_now(), gpr_time_from_millis(100)));
   gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
 }
+
+void grpc_cq_mark_server_cq(grpc_completion_queue *cc) { cc->is_server_cq = 1; }
+
+int grpc_cq_is_server_cq(grpc_completion_queue *cc) { return cc->is_server_cq; }

src/core/surface/completion_queue.h

@@ -63,4 +63,7 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);
 
 void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc);
 
+void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
+int grpc_cq_is_server_cq(grpc_completion_queue *cc);
+
 #endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */

src/core/surface/server.c

@@ -709,6 +709,7 @@ void grpc_server_register_completion_queue(grpc_server *server,
     if (server->cqs[i] == cq) return;
   }
   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 *));
@@ -1081,6 +1082,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;
@@ -1099,6 +1103,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;