Merge pull request #88 from ctiller/shutdown

Add an optional server shutdown event.

4 files changed, 36 insertions, 1 deletions

diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 0f09933fc0..652f23e888 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -155,6 +155,13 @@ static void end_op_locked(grpc_completion_queue *cc,
   }
 }
 
+void grpc_cq_end_server_shutdown(grpc_completion_queue *cc, void *tag) {
+  gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
+  add_locked(cc, GRPC_SERVER_SHUTDOWN, tag, NULL, NULL, NULL);
+  end_op_locked(cc, GRPC_SERVER_SHUTDOWN);
+  gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+}
+
 void grpc_cq_end_read(grpc_completion_queue *cc, void *tag, grpc_call *call,
                       grpc_event_finish_func on_finish, void *user_data,
                       grpc_byte_buffer *read) {
diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h
index 5e45749396..85984075f7 100644
--- a/src/core/surface/completion_queue.h
+++ b/src/core/surface/completion_queue.h
@@ -97,6 +97,8 @@ void grpc_cq_end_new_rpc(grpc_completion_queue *cc, void *tag, grpc_call *call,
                          gpr_timespec deadline, size_t metadata_count,
                          grpc_metadata *metadata_elements);
 
+void grpc_cq_end_server_shutdown(grpc_completion_queue *cc, void *tag);
+
 /* disable polling for some tests */
 void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc);
 
diff --git a/src/core/surface/event_string.c b/src/core/surface/event_string.c
index 8bd8049520..8ae2af7472 100644
--- a/src/core/surface/event_string.c
+++ b/src/core/surface/event_string.c
@@ -63,6 +63,9 @@ char *grpc_event_string(grpc_event *ev) {
   if (ev == NULL) return gpr_strdup("null");
 
   switch (ev->type) {
+    case GRPC_SERVER_SHUTDOWN:
+      p += sprintf(p, "SERVER_SHUTDOWN");
+      break;
     case GRPC_QUEUE_SHUTDOWN:
       p += sprintf(p, "QUEUE_SHUTDOWN");
       break;
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 167bfe97d1..846a00c94f 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -81,6 +81,8 @@ struct grpc_server {
   size_t tag_cap;
 
   gpr_uint8 shutdown;
+  gpr_uint8 have_shutdown_tag;
+  void *shutdown_tag;
 
   call_data *lists[CALL_LIST_COUNT];
   channel_data root_channel_data;
@@ -375,6 +377,10 @@ static void destroy_call_elem(grpc_call_element *elem) {
   for (i = 0; i < CALL_LIST_COUNT; i++) {
     call_list_remove(chand->server, elem->call_data, i);
   }
+  if (chand->server->shutdown && chand->server->have_shutdown_tag &&
+      chand->server->lists[ALL_CALLS] == NULL) {
+    grpc_cq_end_server_shutdown(chand->server->cq, chand->server->shutdown_tag);
+  }
   gpr_mu_unlock(&chand->server->mu);
 
   server_unref(chand->server);
@@ -513,7 +519,8 @@ grpc_transport_setup_result grpc_server_setup_transport(
       grpc_channel_get_channel_stack(channel), transport);
 }
 
-void grpc_server_shutdown(grpc_server *server) {
+void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
+                       void *shutdown_tag) {
   listener *l;
   void **tags;
   size_t ntags;
@@ -551,6 +558,14 @@ void grpc_server_shutdown(grpc_server *server) {
   server->ntags = 0;
 
   server->shutdown = 1;
+  server->have_shutdown_tag = have_shutdown_tag;
+  server->shutdown_tag = shutdown_tag;
+  if (have_shutdown_tag) {
+    grpc_cq_begin_op(server->cq, NULL, GRPC_SERVER_SHUTDOWN);
+    if (server->lists[ALL_CALLS] == NULL) {
+      grpc_cq_end_server_shutdown(server->cq, shutdown_tag);
+    }
+  }
   gpr_mu_unlock(&server->mu);
 
   for (i = 0; i < nchannels; i++) {
@@ -583,6 +598,14 @@ void grpc_server_shutdown(grpc_server *server) {
   }
 }
 
+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_destroy(grpc_server *server) {
   channel_data *c;
   gpr_mu_lock(&server->mu);