Work towards removing grpc_server_shutdown

include/grpc/grpc.h

@@ -488,18 +488,17 @@ void grpc_server_start(grpc_server *server);
 /* Begin shutting down a server.
    After completion, no new calls or connections will be admitted.
    Existing calls will be allowed to complete.
-   Shutdown is idempotent. */
-void grpc_server_shutdown(grpc_server *server);
-
-/* As per grpc_server_shutdown, but send a GRPC_OP_COMPLETE event when
-   there are no more calls being serviced.
+   Send a GRPC_OP_COMPLETE event when there are no more calls being serviced.
    Shutdown is idempotent, and all tags will be notified at once if multiple
    grpc_server_shutdown_and_notify calls are made. */
 void grpc_server_shutdown_and_notify(grpc_server *server, void *tag);
 
+/* Cancel all in-progress calls. 
+   Only usable after shutdown. */
+void grpc_server_cancel_all_calls(grpc_server *server);
+
 /* Destroy a server.
-   Forcefully cancels all existing calls.
-   Implies grpc_server_shutdown() if one was not previously performed. */
+   Shutdown must have completed beforehand. */
 void grpc_server_destroy(grpc_server *server);
 
 #ifdef __cplusplus

src/core/iomgr/resolve_address_posix.c

@@ -166,13 +166,14 @@ void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) {
 void grpc_resolve_address(const char *name, const char *default_port,
                           grpc_resolve_cb cb, void *arg) {
   request *r = gpr_malloc(sizeof(request));
-  gpr_thd_id id;
+  /*gpr_thd_id id;*/
   grpc_iomgr_ref();
   r->name = gpr_strdup(name);
   r->default_port = gpr_strdup(default_port);
   r->cb = cb;
   r->arg = arg;
-  gpr_thd_new(&id, do_request, r, NULL);
+  /*gpr_thd_new(&id, do_request, r, NULL);*/
+  do_request(r);
 }
 
 #endif

src/core/surface/completion_queue.c

@@ -150,6 +150,7 @@ void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
                     int success) {
   event *ev;
   int shutdown = 0;
+  gpr_log(GPR_DEBUG, "end_op:%p", tag);
   gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
   ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call);
   ev->base.success = success;
@@ -167,26 +168,6 @@ void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
   }
 }
 
-void grpc_cq_begin_silent_op(grpc_completion_queue *cc) {
-  gpr_ref(&cc->refs);
-}
-
-void grpc_cq_end_silent_op(grpc_completion_queue *cc) {
-  int shutdown = 0;
-  gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
-  if (gpr_unref(&cc->refs)) {
-    GPR_ASSERT(!cc->shutdown);
-    GPR_ASSERT(cc->shutdown_called);
-    cc->shutdown = 1;
-    gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
-    shutdown = 1;
-  }
-  gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
-  if (shutdown) {
-    grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
-  }
-}
-
 /* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */
 static event *create_shutdown_event(void) {
   event *ev = gpr_malloc(sizeof(event));

src/core/surface/completion_queue.h

@@ -50,12 +50,6 @@ void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call);
 void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
                     int success);
 
-/* Begin a 'silent operation' - one that blocks completion queue shutdown
-   until it is ended */
-void grpc_cq_begin_silent_op(grpc_completion_queue *cc);
-/* End such an operation */
-void grpc_cq_end_silent_op(grpc_completion_queue *cc);
-
 /* disable polling for some tests */
 void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc);
 

src/core/surface/server.c

@@ -134,7 +134,6 @@ struct grpc_server {
   size_t cq_count;
 
   gpr_mu mu;
-  gpr_cv cv;
 
   registered_method *registered_methods;
   requested_call_array requested_calls;
@@ -256,29 +255,32 @@ static void server_ref(grpc_server *server) {
   gpr_ref(&server->internal_refcount);
 }
 
-static void server_unref(grpc_server *server) {
+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_free(server->channel_filters);
+  requested_call_array_destroy(&server->requested_calls);
+  while ((rm = server->registered_methods) != NULL) {
+    server->registered_methods = rm->next;
+    gpr_free(rm->method);
+    gpr_free(rm->host);
+    requested_call_array_destroy(&rm->requested);
+    gpr_free(rm);
+  }
+  for (i = 0; i < server->cq_count; i++) {
+    grpc_cq_internal_unref(server->cqs[i]);
+  }
+  gpr_free(server->cqs);
+  gpr_free(server->pollsets);
+  gpr_free(server->shutdown_tags);
+  gpr_free(server);
+}
+
+static void server_unref(grpc_server *server) {
   if (gpr_unref(&server->internal_refcount)) {
-    grpc_channel_args_destroy(server->channel_args);
-    gpr_mu_destroy(&server->mu);
-    gpr_cv_destroy(&server->cv);
-    gpr_free(server->channel_filters);
-    requested_call_array_destroy(&server->requested_calls);
-    while ((rm = server->registered_methods) != NULL) {
-      server->registered_methods = rm->next;
-      gpr_free(rm->method);
-      gpr_free(rm->host);
-      requested_call_array_destroy(&rm->requested);
-      gpr_free(rm);
-    }
-    for (i = 0; i < server->cq_count; i++) {
-      grpc_cq_internal_unref(server->cqs[i]);
-    }
-    gpr_free(server->cqs);
-    gpr_free(server->pollsets);
-    gpr_free(server->shutdown_tags);
-    gpr_free(server);
+    server_delete(server);
   }
 }
 
@@ -371,12 +373,20 @@ static void kill_zombie(void *elem, int success) {
   grpc_call_destroy(grpc_call_from_top_element(elem));
 }
 
+static int num_listeners(grpc_server *server) {
+  listener *l;
+  int n = 0;
+  for (l = server->listeners; l; l = l->next) {
+    n++;
+  }
+  return n;
+}
+
 static void maybe_finish_shutdown(grpc_server *server) {
   size_t i, j;
-  if (server->shutdown && server->lists[ALL_CALLS] == NULL) {
-    for (j = 0; j < server->cq_count; j++) {
-      grpc_cq_end_silent_op(server->cqs[j]);
-      for (i = 0; i < server->num_shutdown_tags; i++) {
+  if (server->shutdown && server->lists[ALL_CALLS] == NULL && server->listeners_destroyed == num_listeners(server)) {
+    for (i = 0; i < server->num_shutdown_tags; i++) {
+      for (j = 0; j < server->cq_count; j++) {
         grpc_cq_end_op(server->cqs[j], server->shutdown_tags[i],
                        NULL, 1);
       }
@@ -541,13 +551,16 @@ 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);
   for (i = 0; i < CALL_LIST_COUNT; i++) {
-    call_list_remove(elem->call_data, i);
+    removed[i] = call_list_remove(elem->call_data, i);
+  }
+  if (removed[ALL_CALLS]) {
+    maybe_finish_shutdown(chand->server);
   }
-  maybe_finish_shutdown(chand->server);
   gpr_mu_unlock(&chand->server->mu);
 
   if (calld->host) {
@@ -633,7 +646,6 @@ grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters,
   memset(server, 0, sizeof(grpc_server));
 
   gpr_mu_init(&server->mu);
-  gpr_cv_init(&server->cv);
 
   /* decremented by grpc_server_destroy */
   gpr_ref_init(&server->internal_refcount, 1);
@@ -792,17 +804,7 @@ grpc_transport_setup_result grpc_server_setup_transport(
   return result;
 }
 
-static int num_listeners(grpc_server *server) {
-  listener *l;
-  int n = 0;
-  for (l = server->listeners; l; l = l->next) {
-    n++;
-  }
-  return n;
-}
-
-static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
-                              void *shutdown_tag) {
+void grpc_server_shutdown_and_notify(grpc_server *server, void *shutdown_tag) {
   listener *l;
   requested_call_array requested_calls;
   channel_data **channels;
@@ -815,24 +817,18 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
 
   /* lock, and gather up some stuff to do */
   gpr_mu_lock(&server->mu);
-  if (have_shutdown_tag) {
-    for (i = 0; i < server->cq_count; i++) {
-      grpc_cq_begin_op(server->cqs[i], NULL);
-    }
-    server->shutdown_tags =
-        gpr_realloc(server->shutdown_tags,
-                    sizeof(void *) * (server->num_shutdown_tags + 1));
-    server->shutdown_tags[server->num_shutdown_tags++] = shutdown_tag;
+  for (i = 0; i < server->cq_count; i++) {
+    grpc_cq_begin_op(server->cqs[i], NULL);
   }
+  server->shutdown_tags =
+      gpr_realloc(server->shutdown_tags,
+                  sizeof(void *) * (server->num_shutdown_tags + 1));
+  server->shutdown_tags[server->num_shutdown_tags++] = shutdown_tag;
   if (server->shutdown) {
     gpr_mu_unlock(&server->mu);
     return;
   }
 
-  for (i = 0; i < server->cq_count; i++) {
-    grpc_cq_begin_silent_op(server->cqs[i]);
-  }
-
   nchannels = 0;
   for (c = server->root_channel_data.next; c != &server->root_channel_data;
        c = c->next) {
@@ -898,49 +894,22 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
   }
 }
 
-void grpc_server_shutdown(grpc_server *server) {
-  shutdown_internal(server, 0, NULL);
-}
-
-void grpc_server_shutdown_and_notify(grpc_server *server, void *tag) {
-  shutdown_internal(server, 1, tag);
-}
-
 void grpc_server_listener_destroy_done(void *s) {
   grpc_server *server = s;
   gpr_mu_lock(&server->mu);
   server->listeners_destroyed++;
-  gpr_cv_signal(&server->cv);
+  maybe_finish_shutdown(server);
   gpr_mu_unlock(&server->mu);
 }
 
-static void continue_server_shutdown(void *server, int iomgr_success) {
-  grpc_server_destroy(server);
-}
-
 void grpc_server_destroy(grpc_server *server) {
   channel_data *c;
   listener *l;
-  size_t i;
   call_data *calld;
 
   gpr_mu_lock(&server->mu);
-  if (!server->shutdown) {
-    gpr_mu_unlock(&server->mu);
-    grpc_server_shutdown(server);
-    gpr_mu_lock(&server->mu);
-  }
-
-  if (server->listeners_destroyed != num_listeners(server)) {
-    gpr_mu_unlock(&server->mu);
-    for (i = 0; i < server->cq_count; i++) {
-      grpc_cq_hack_spin_pollset(server->cqs[i]);
-    }
-
-    /* delay execution some, and return early */
-    grpc_iomgr_add_callback(continue_server_shutdown, server);
-    return;
-  }
+  GPR_ASSERT(server->shutdown);
+  GPR_ASSERT(server->listeners_destroyed == num_listeners(server));
 
   while (server->listeners) {
     l = server->listeners;

test/core/end2end/dualstack_socket_test.c

@@ -206,7 +206,8 @@ void test_connect(const char *server_host, const char *client_host, int port,
   grpc_completion_queue_destroy(client_cq);
 
   /* Destroy server. */
-  grpc_server_shutdown(server);
+  grpc_server_shutdown_and_notify(server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(server);
   grpc_completion_queue_shutdown(server_cq);
   drain_cq(server_cq);

test/core/end2end/tests/bad_hostname.c

@@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/cancel_after_accept.c

@@ -75,7 +75,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/cancel_after_accept_and_writes_closed.c

@@ -75,7 +75,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/cancel_after_invoke.c

@@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/cancel_before_invoke.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/cancel_in_a_vacuum.c

@@ -46,6 +46,8 @@
 
 enum { TIMEOUT = 200000 };
 
+static void *tag(gpr_intptr t) { return (void *)t; }
+
 static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
                                             const char *test_name,
                                             grpc_channel_args *client_args,
@@ -73,7 +75,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/census_simple_request.c

@@ -60,9 +60,12 @@ static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
   return f;
 }
 
+static void *tag(gpr_intptr t) { return (void *)t; }
+
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }
@@ -92,8 +95,6 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_destroy(f->client_cq);
 }
 
-static void *tag(gpr_intptr t) { return (void *)t; }
-
 static void test_body(grpc_end2end_test_fixture f) {
   grpc_call *c;
   grpc_call *s;

test/core/end2end/tests/disappearing_server.c

@@ -62,7 +62,6 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }
@@ -137,7 +136,7 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f,
 
   /* should be able to shut down the server early
      - and still complete the request */
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
 
   op = ops;
   op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -154,6 +153,7 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f,
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
   cq_expect_completion(v_server, tag(102), 1);
+  cq_expect_completion(v_server, tag(1000), 1);
   cq_verify(v_server);
 
   cq_expect_completion(v_client, tag(1), 1);

test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c

@@ -72,13 +72,6 @@ static void drain_cq(grpc_completion_queue *cq) {
   } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
-static void shutdown_server(grpc_end2end_test_fixture *f) {
-  if (!f->server) return;
-  grpc_server_shutdown(f->server);
-  grpc_server_destroy(f->server);
-  f->server = NULL;
-}
-
 static void shutdown_client(grpc_end2end_test_fixture *f) {
   if (!f->client) return;
   grpc_channel_destroy(f->client);
@@ -86,7 +79,6 @@ static void shutdown_client(grpc_end2end_test_fixture *f) {
 }
 
 static void end_test(grpc_end2end_test_fixture *f) {
-  shutdown_server(f);
   shutdown_client(f);
 
   grpc_completion_queue_shutdown(f->server_cq);
@@ -157,11 +149,15 @@ static void test_early_server_shutdown_finishes_inflight_calls(
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
   /* shutdown and destroy the server */
-  shutdown_server(&f);
+  grpc_server_shutdown_and_notify(f.server, tag(1000));
+  grpc_server_cancel_all_calls(f.server);
 
   cq_expect_completion(v_server, tag(102), 1);
+  cq_expect_completion(v_server, tag(1000), 1);
   cq_verify(v_server);
 
+  grpc_server_destroy(f.server);
+
   cq_expect_completion(v_client, tag(1), 1);
   cq_verify(v_client);
 

test/core/end2end/tests/early_server_shutdown_finishes_tags.c

@@ -72,13 +72,6 @@ static void drain_cq(grpc_completion_queue *cq) {
   } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
-static void shutdown_server(grpc_end2end_test_fixture *f) {
-  if (!f->server) return;
-  /* don't shutdown, just destroy, to tickle this code edge */
-  grpc_server_destroy(f->server);
-  f->server = NULL;
-}
-
 static void shutdown_client(grpc_end2end_test_fixture *f) {
   if (!f->client) return;
   grpc_channel_destroy(f->client);
@@ -86,7 +79,6 @@ static void shutdown_client(grpc_end2end_test_fixture *f) {
 }
 
 static void end_test(grpc_end2end_test_fixture *f) {
-  shutdown_server(f);
   shutdown_client(f);
 
   grpc_completion_queue_shutdown(f->server_cq);
@@ -114,11 +106,14 @@ static void test_early_server_shutdown_finishes_tags(
              grpc_server_request_call(f.server, &s, &call_details,
                                       &request_metadata_recv, f.server_cq,
                                       f.server_cq, tag(101)));
-  grpc_server_shutdown(f.server);
+  grpc_server_shutdown_and_notify(f.server, tag(1000));
   cq_expect_completion(v_server, tag(101), 0);
+  cq_expect_completion(v_server, tag(1000), 1);
   cq_verify(v_server);
   GPR_ASSERT(s == NULL);
 
+  grpc_server_destroy(f.server);
+
   end_test(&f);
   config.tear_down_data(&f);
   cq_verifier_destroy(v_server);

test/core/end2end/tests/empty_batch.c

@@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/graceful_server_shutdown.c

@@ -168,11 +168,10 @@ static void test_early_server_shutdown_finishes_inflight_calls(
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
 
   cq_expect_completion(v_server, tag(102), 1);
+  cq_expect_completion(v_server, tag(0xdead), 1);
   cq_verify(v_server);
 
   grpc_call_destroy(s);
-  cq_expect_completion(v_server, tag(0xdead), 1);
-  cq_verify(v_server);
 
   cq_expect_completion(v_client, tag(1), 1);
   cq_verify(v_client);

test/core/end2end/tests/invoke_large_request.c

@@ -72,7 +72,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/max_concurrent_streams.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/max_message_length.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/no_op.c

@@ -45,6 +45,8 @@
 
 enum { TIMEOUT = 200000 };
 
+static void *tag(gpr_intptr t) { return (void *)t; }
+
 static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
                                             const char *test_name,
                                             grpc_channel_args *client_args,
@@ -72,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/ping_pong_streaming.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/registered_call.c

@@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/request_response_with_metadata_and_payload.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/request_response_with_payload.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/request_response_with_payload_and_call_creds.c

@@ -88,7 +88,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/request_with_large_metadata.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/request_with_payload.c

@@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/server_finishes_request.c

@@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/simple_delayed_request.c

@@ -62,7 +62,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/simple_request.c

@@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c

@@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) {
 
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
-  grpc_server_shutdown(f->server);
+  grpc_server_shutdown_and_notify(f->server, tag(1000));
+  GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/fling/server.c

@@ -233,7 +233,8 @@ int main(int argc, char **argv) {
   while (!shutdown_finished) {
     if (got_sigint && !shutdown_started) {
       gpr_log(GPR_INFO, "Shutting down due to SIGINT");
-      grpc_server_shutdown(server);
+      grpc_server_shutdown_and_notify(server, tag(1000));
+      GPR_ASSERT(grpc_completion_queue_pluck(cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE);
       grpc_completion_queue_shutdown(cq);
       shutdown_started = 1;
     }