1 files changed, 145 insertions, 81 deletions

diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 424734c54c..83caefcbc6 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -137,6 +137,7 @@ struct grpc_server {
   size_t cq_count;
 
   gpr_mu mu;
+  gpr_cv cv;
 
   registered_method *registered_methods;
   requested_call_array requested_calls;
@@ -149,6 +150,7 @@ struct grpc_server {
   channel_data root_channel_data;
 
   listener *listeners;
+  int listeners_destroyed;
   gpr_refcount internal_refcount;
 };
 
@@ -171,13 +173,19 @@ struct call_data {
   grpc_call *call;
 
   call_state state;
-  gpr_timespec deadline;
   grpc_mdstr *path;
   grpc_mdstr *host;
+  gpr_timespec deadline;
+  int got_initial_metadata;
 
   legacy_data *legacy;
   grpc_completion_queue *cq_new;
 
+  grpc_stream_op_buffer *recv_ops;
+  grpc_stream_state *recv_state;
+  void (*on_done_recv)(void *user_data, int success);
+  void *recv_user_data;
+
   call_data **root[CALL_LIST_COUNT];
   call_link links[CALL_LIST_COUNT];
 };
@@ -260,9 +268,11 @@ static void server_ref(grpc_server *server) {
 
 static void server_unref(grpc_server *server) {
   registered_method *rm;
+  size_t i;
   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) {
@@ -272,6 +282,9 @@ static void server_unref(grpc_server *server) {
       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);
@@ -368,88 +381,89 @@ static void kill_zombie(void *elem, int success) {
   grpc_call_destroy(grpc_call_from_top_element(elem));
 }
 
-static void stream_closed(grpc_call_element *elem) {
-  call_data *calld = elem->call_data;
+static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
+  grpc_call_element *elem = user_data;
   channel_data *chand = elem->channel_data;
-  gpr_mu_lock(&chand->server->mu);
-  switch (calld->state) {
-    case ACTIVATED:
-      break;
-    case PENDING:
-      call_list_remove(calld, PENDING_START);
-    /* fallthrough intended */
-    case NOT_STARTED:
-      calld->state = ZOMBIED;
-      grpc_iomgr_add_callback(kill_zombie, elem);
-      break;
-    case ZOMBIED:
-      break;
+  call_data *calld = elem->call_data;
+  if (md->key == chand->path_key) {
+    calld->path = grpc_mdstr_ref(md->value);
+    return NULL;
+  } else if (md->key == chand->authority_key) {
+    calld->host = grpc_mdstr_ref(md->value);
+    return NULL;
   }
-  gpr_mu_unlock(&chand->server->mu);
-  grpc_call_stream_closed(elem);
+  return md;
 }
 
-static void read_closed(grpc_call_element *elem) {
+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;
-  gpr_mu_lock(&chand->server->mu);
-  switch (calld->state) {
-    case ACTIVATED:
-    case PENDING:
-      grpc_call_read_closed(elem);
-      break;
-    case NOT_STARTED:
-      calld->state = ZOMBIED;
-      grpc_iomgr_add_callback(kill_zombie, elem);
-      break;
-    case ZOMBIED:
-      break;
-  }
-  gpr_mu_unlock(&chand->server->mu);
-}
 
-static void call_op(grpc_call_element *elem, grpc_call_element *from_elemn,
-                    grpc_call_op *op) {
-  channel_data *chand = elem->channel_data;
-  call_data *calld = elem->call_data;
-  grpc_mdelem *md;
-  GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
-  switch (op->type) {
-    case GRPC_RECV_METADATA:
-      md = op->data.metadata;
-      if (md->key == chand->path_key) {
-        calld->path = grpc_mdstr_ref(md->value);
-        grpc_mdelem_unref(md);
-      } else if (md->key == chand->authority_key) {
-        calld->host = grpc_mdstr_ref(md->value);
-        grpc_mdelem_unref(md);
-      } else {
-        grpc_call_recv_metadata(elem, md);
+  if (success && !calld->got_initial_metadata) {
+    size_t i;
+    size_t nops = calld->recv_ops->nops;
+    grpc_stream_op *ops = calld->recv_ops->ops;
+    for (i = 0; i < nops; i++) {
+      grpc_stream_op *op = &ops[i];
+      if (op->type != GRPC_OP_METADATA) continue;
+      grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem);
+      if (0 != gpr_time_cmp(op->data.metadata.deadline, gpr_inf_future)) {
+        calld->deadline = op->data.metadata.deadline;
       }
-      break;
-    case GRPC_RECV_END_OF_INITIAL_METADATA:
+      calld->got_initial_metadata = 1;
       start_new_rpc(elem);
-      grpc_call_initial_metadata_complete(elem);
-      break;
-    case GRPC_RECV_MESSAGE:
-      grpc_call_recv_message(elem, op->data.message);
-      op->done_cb(op->user_data, GRPC_OP_OK);
       break;
-    case GRPC_RECV_HALF_CLOSE:
-      read_closed(elem);
+    }
+  }
+
+  switch (*calld->recv_state) {
+    case GRPC_STREAM_OPEN:
       break;
-    case GRPC_RECV_FINISH:
-      stream_closed(elem);
+    case GRPC_STREAM_SEND_CLOSED:
       break;
-    case GRPC_RECV_DEADLINE:
-      grpc_call_set_deadline(elem, op->data.deadline);
-      ((call_data *)elem->call_data)->deadline = op->data.deadline;
+    case GRPC_STREAM_RECV_CLOSED:
+      gpr_mu_lock(&chand->server->mu);
+      if (calld->state == NOT_STARTED) {
+        calld->state = ZOMBIED;
+        grpc_iomgr_add_callback(kill_zombie, elem);
+      }
+      gpr_mu_unlock(&chand->server->mu);
       break;
-    default:
-      GPR_ASSERT(op->dir == GRPC_CALL_DOWN);
-      grpc_call_next_op(elem, op);
+    case GRPC_STREAM_CLOSED:
+      gpr_mu_lock(&chand->server->mu);
+      if (calld->state == NOT_STARTED) {
+        calld->state = ZOMBIED;
+        grpc_iomgr_add_callback(kill_zombie, elem);
+      } else if (calld->state == PENDING) {
+        call_list_remove(calld, PENDING_START);
+      }
+      gpr_mu_unlock(&chand->server->mu);
       break;
   }
+
+  calld->on_done_recv(calld->recv_user_data, success);
+}
+
+static void server_mutate_op(grpc_call_element *elem, grpc_transport_op *op) {
+  call_data *calld = elem->call_data;
+
+  if (op->recv_ops) {
+    /* substitute our callback for the higher callback */
+    calld->recv_ops = op->recv_ops;
+    calld->recv_state = op->recv_state;
+    calld->on_done_recv = op->on_done_recv;
+    calld->recv_user_data = op->recv_user_data;
+    op->on_done_recv = server_on_recv;
+    op->recv_user_data = elem;
+  }
+}
+
+static void server_start_transport_op(grpc_call_element *elem,
+                                      grpc_transport_op *op) {
+  GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
+  server_mutate_op(elem, op);
+  grpc_call_next_op(elem, op);
 }
 
 static void channel_op(grpc_channel_element *elem,
@@ -461,7 +475,8 @@ static void channel_op(grpc_channel_element *elem,
     case GRPC_ACCEPT_CALL:
       /* create a call */
       grpc_call_create(chand->channel, NULL,
-                       op->data.accept_call.transport_server_data);
+                       op->data.accept_call.transport_server_data, NULL, 0,
+                       gpr_inf_future);
       break;
     case GRPC_TRANSPORT_CLOSED:
       /* if the transport is closed for a server channel, we destroy the
@@ -499,7 +514,8 @@ static void shutdown_channel(channel_data *chand) {
 }
 
 static void init_call_elem(grpc_call_element *elem,
-                           const void *server_transport_data) {
+                           const void *server_transport_data,
+                           grpc_transport_op *initial_op) {
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
   memset(calld, 0, sizeof(call_data));
@@ -511,6 +527,8 @@ static void init_call_elem(grpc_call_element *elem,
   gpr_mu_unlock(&chand->server->mu);
 
   server_ref(chand->server);
+
+  if (initial_op) server_mutate_op(elem, initial_op);
 }
 
 static void destroy_call_elem(grpc_call_element *elem) {
@@ -589,9 +607,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
 }
 
 static const grpc_channel_filter server_surface_filter = {
-    call_op,           channel_op,           sizeof(call_data),
-    init_call_elem,    destroy_call_elem,    sizeof(channel_data),
-    init_channel_elem, destroy_channel_elem, "server",
+    server_start_transport_op, channel_op, sizeof(call_data), init_call_elem,
+    destroy_call_elem, sizeof(channel_data), init_channel_elem,
+    destroy_channel_elem, "server",
 };
 
 static void addcq(grpc_server *server, grpc_completion_queue *cq) {
@@ -599,6 +617,7 @@ static void addcq(grpc_server *server, grpc_completion_queue *cq) {
   for (i = 0; i < server->cq_count; i++) {
     if (server->cqs[i] == cq) return;
   }
+  grpc_cq_internal_ref(cq);
   n = server->cq_count++;
   server->cqs = gpr_realloc(server->cqs,
                             server->cq_count * sizeof(grpc_completion_queue *));
@@ -620,6 +639,7 @@ grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
   if (cq) addcq(server, cq);
 
   gpr_mu_init(&server->mu);
+  gpr_cv_init(&server->cv);
 
   server->unregistered_cq = cq;
   /* decremented by grpc_server_destroy */
@@ -733,7 +753,8 @@ grpc_transport_setup_result grpc_server_setup_transport(
   channel = grpc_channel_create_from_filters(filters, num_filters,
                                              s->channel_args, mdctx, 0);
   chand = (channel_data *)grpc_channel_stack_element(
-              grpc_channel_get_channel_stack(channel), 0)->channel_data;
+              grpc_channel_get_channel_stack(channel), 0)
+              ->channel_data;
   chand->server = s;
   server_ref(s);
   chand->channel = channel;
@@ -754,7 +775,7 @@ grpc_transport_setup_result grpc_server_setup_transport(
       method = grpc_mdstr_from_string(mdctx, rm->method);
       hash = GRPC_MDSTR_KV_HASH(host ? host->hash : 0, method->hash);
       for (probes = 0; chand->registered_methods[(hash + probes) % slots]
-                               .server_registered_method != NULL;
+                           .server_registered_method != NULL;
            probes++)
         ;
       if (probes > max_probes) max_probes = probes;
@@ -781,6 +802,15 @@ 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) {
   listener *l;
@@ -878,11 +908,6 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
   for (l = server->listeners; l; l = l->next) {
     l->destroy(server, l->arg);
   }
-  while (server->listeners) {
-    l = server->listeners;
-    server->listeners = l->next;
-    gpr_free(l);
-  }
 }
 
 void grpc_server_shutdown(grpc_server *server) {
@@ -893,8 +918,20 @@ 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);
+  gpr_mu_unlock(&server->mu);
+}
+
 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);
@@ -902,6 +939,32 @@ void grpc_server_destroy(grpc_server *server) {
     gpr_mu_lock(&server->mu);
   }
 
+  while (server->listeners_destroyed != num_listeners(server)) {
+    for (i = 0; i < server->cq_count; i++) {
+      gpr_mu_unlock(&server->mu);
+      grpc_cq_hack_spin_pollset(server->cqs[i]);
+      gpr_mu_lock(&server->mu);
+    }
+
+    gpr_cv_wait(&server->cv, &server->mu,
+                gpr_time_add(gpr_now(), gpr_time_from_millis(100)));
+  }
+
+  while (server->listeners) {
+    l = server->listeners;
+    server->listeners = l->next;
+    gpr_free(l);
+  }
+
+  while ((calld = call_list_remove_head(&server->lists[PENDING_START],
+                                        PENDING_START)) != NULL) {
+    gpr_log(GPR_DEBUG, "server destroys call %p", calld->call);
+    calld->state = ZOMBIED;
+    grpc_iomgr_add_callback(
+        kill_zombie,
+        grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
+  }
+
   for (c = server->root_channel_data.next; c != &server->root_channel_data;
        c = c->next) {
     shutdown_channel(c);
@@ -1048,6 +1111,7 @@ static void begin_call(grpc_server *server, call_data *calld,
             &rc->data.batch.details->host_capacity, calld->host);
       cpstr(&rc->data.batch.details->method,
             &rc->data.batch.details->method_capacity, calld->path);
+      rc->data.batch.details->deadline = calld->deadline;
       grpc_call_set_completion_queue(calld->call, rc->data.batch.cq_bind);
       *rc->data.batch.call = calld->call;
       r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
@@ -1073,7 +1137,7 @@ static void begin_call(grpc_server *server, call_data *calld,
       break;
   }
 
-  grpc_call_internal_ref(calld->call);
+  GRPC_CALL_INTERNAL_REF(calld->call, "server");
   grpc_call_start_ioreq_and_call_back(calld->call, req, r - req, publish,
                                       rc->tag);
 }