Call list progress
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index e38c602..24545c6 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -57,9 +57,11 @@
typedef struct listener {
void *arg;
void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets,
- size_t pollset_count);
- void (*destroy)(grpc_server *server, void *arg);
+ size_t pollset_count, grpc_call_list *call_list);
+ void (*destroy)(grpc_server *server, void *arg, grpc_closure *closure,
+ grpc_call_list *call_list);
struct listener *next;
+ grpc_closure destroy_done;
} listener;
typedef struct call_data call_data;
@@ -219,19 +221,19 @@
/** when did we print the last shutdown progress message */
gpr_timespec last_shutdown_message_time;
-
- grpc_workqueue *workqueue;
};
#define SERVER_FROM_CALL_ELEM(elem) \
(((channel_data *)(elem)->channel_data)->server)
static void begin_call(grpc_server *server, call_data *calld,
- requested_call *rc);
-static void fail_call(grpc_server *server, requested_call *rc);
+ requested_call *rc, grpc_call_list *call_list);
+static void fail_call(grpc_server *server, requested_call *rc,
+ grpc_call_list *call_list);
/* Before calling maybe_finish_shutdown, we must hold mu_global and not
hold mu_call */
-static void maybe_finish_shutdown(grpc_server *server);
+static void maybe_finish_shutdown(grpc_server *server,
+ grpc_call_list *call_list);
/*
* channel broadcaster
@@ -258,14 +260,15 @@
gpr_slice slice;
};
-static void shutdown_cleanup(void *arg, int iomgr_status_ignored) {
+static void shutdown_cleanup(void *arg, int iomgr_status_ignored,
+ grpc_call_list *call_list) {
struct shutdown_cleanup_args *a = arg;
gpr_slice_unref(a->slice);
gpr_free(a);
}
static void send_shutdown(grpc_channel *channel, int send_goaway,
- int send_disconnect) {
+ int send_disconnect, grpc_call_list *call_list) {
grpc_transport_op op;
struct shutdown_cleanup_args *sc;
grpc_channel_element *elem;
@@ -281,17 +284,17 @@
op.on_consumed = &sc->closure;
elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
- elem->filter->start_transport_op(elem, &op);
+ elem->filter->start_transport_op(elem, &op, call_list);
}
static void channel_broadcaster_shutdown(channel_broadcaster *cb,
- int send_goaway,
- int force_disconnect) {
+ int send_goaway, int force_disconnect,
+ grpc_call_list *call_list) {
size_t i;
for (i = 0; i < cb->num_channels; i++) {
- send_shutdown(cb->channels[i], send_goaway, force_disconnect);
- GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast");
+ send_shutdown(cb->channels[i], send_goaway, force_disconnect, call_list);
+ GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast", call_list);
}
gpr_free(cb->channels);
}
@@ -311,12 +314,12 @@
gpr_stack_lockfree_destroy(request_matcher->requests);
}
-static void kill_zombie(void *elem, int success) {
+static void kill_zombie(void *elem, int success, grpc_call_list *call_list) {
grpc_call_destroy(grpc_call_from_top_element(elem));
}
static void request_matcher_zombify_all_pending_calls(
- request_matcher *request_matcher, grpc_workqueue *workqueue) {
+ request_matcher *request_matcher, grpc_call_list *call_list) {
while (request_matcher->pending_head) {
call_data *calld = request_matcher->pending_head;
request_matcher->pending_head = calld->pending_next;
@@ -326,15 +329,16 @@
grpc_closure_init(
&calld->kill_zombie_closure, kill_zombie,
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
- grpc_workqueue_push(workqueue, &calld->kill_zombie_closure, 1);
+ grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1);
}
}
static void request_matcher_kill_requests(grpc_server *server,
- request_matcher *rm) {
+ request_matcher *rm,
+ grpc_call_list *call_list) {
int request_id;
while ((request_id = gpr_stack_lockfree_pop(rm->requests)) != -1) {
- fail_call(server, &server->requested_calls[request_id]);
+ fail_call(server, &server->requested_calls[request_id], call_list);
}
}
@@ -346,7 +350,7 @@
gpr_ref(&server->internal_refcount);
}
-static void server_delete(grpc_server *server) {
+static void server_delete(grpc_server *server, grpc_call_list *call_list) {
registered_method *rm;
size_t i;
grpc_channel_args_destroy(server->channel_args);
@@ -365,7 +369,6 @@
}
request_matcher_destroy(&server->unregistered_request_matcher);
gpr_stack_lockfree_destroy(server->request_freelist);
- GRPC_WORKQUEUE_UNREF(server->workqueue, "destroy");
gpr_free(server->cqs);
gpr_free(server->pollsets);
gpr_free(server->shutdown_tags);
@@ -373,9 +376,9 @@
gpr_free(server);
}
-static void server_unref(grpc_server *server) {
+static void server_unref(grpc_server *server, grpc_call_list *call_list) {
if (gpr_unref(&server->internal_refcount)) {
- server_delete(server);
+ server_delete(server, call_list);
}
}
@@ -389,30 +392,29 @@
chand->next = chand->prev = chand;
}
-static void finish_destroy_channel(void *cd, int success) {
+static void finish_destroy_channel(void *cd, int success,
+ grpc_call_list *call_list) {
channel_data *chand = cd;
grpc_server *server = chand->server;
gpr_log(GPR_DEBUG, "finish_destroy_channel: %p", chand->channel);
- GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server");
- server_unref(server);
+ GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server", call_list);
+ server_unref(server, call_list);
}
-static void destroy_channel(channel_data *chand) {
+static void destroy_channel(channel_data *chand, grpc_call_list *call_list) {
if (is_channel_orphaned(chand)) return;
GPR_ASSERT(chand->server != NULL);
orphan_channel(chand);
server_ref(chand->server);
- maybe_finish_shutdown(chand->server);
+ maybe_finish_shutdown(chand->server, call_list);
chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
chand->finish_destroy_channel_closure.cb_arg = chand;
- gpr_log(GPR_DEBUG, "queue finish_destroy_channel: %p on %p", chand->channel,
- chand->server->workqueue);
- grpc_workqueue_push(chand->server->workqueue,
- &chand->finish_destroy_channel_closure, 1);
+ grpc_call_list_add(call_list, &chand->finish_destroy_channel_closure, 1);
}
static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem,
- request_matcher *request_matcher) {
+ request_matcher *request_matcher,
+ grpc_call_list *call_list) {
call_data *calld = elem->call_data;
int request_id;
@@ -421,7 +423,7 @@
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
- grpc_workqueue_push(server->workqueue, &calld->kill_zombie_closure, 1);
+ grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1);
return;
}
@@ -443,11 +445,11 @@
gpr_mu_lock(&calld->mu_state);
calld->state = ACTIVATED;
gpr_mu_unlock(&calld->mu_state);
- begin_call(server, calld, &server->requested_calls[request_id]);
+ begin_call(server, calld, &server->requested_calls[request_id], call_list);
}
}
-static void start_new_rpc(grpc_call_element *elem) {
+static void start_new_rpc(grpc_call_element *elem, grpc_call_list *call_list) {
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
grpc_server *server = chand->server;
@@ -466,7 +468,8 @@
if (rm->host != calld->host) continue;
if (rm->method != calld->path) continue;
finish_start_new_rpc(server, elem,
- &rm->server_registered_method->request_matcher);
+ &rm->server_registered_method->request_matcher,
+ call_list);
return;
}
/* check for a wildcard method definition (no host set) */
@@ -478,11 +481,13 @@
if (rm->host != NULL) continue;
if (rm->method != calld->path) continue;
finish_start_new_rpc(server, elem,
- &rm->server_registered_method->request_matcher);
+ &rm->server_registered_method->request_matcher,
+ call_list);
return;
}
}
- finish_start_new_rpc(server, elem, &server->unregistered_request_matcher);
+ finish_start_new_rpc(server, elem, &server->unregistered_request_matcher,
+ call_list);
}
static int num_listeners(grpc_server *server) {
@@ -494,8 +499,9 @@
return n;
}
-static void done_shutdown_event(void *server, grpc_cq_completion *completion) {
- server_unref(server);
+static void done_shutdown_event(void *server, grpc_cq_completion *completion,
+ grpc_call_list *call_list) {
+ server_unref(server, call_list);
}
static int num_channels(grpc_server *server) {
@@ -508,25 +514,27 @@
return n;
}
-static void kill_pending_work_locked(grpc_server *server) {
+static void kill_pending_work_locked(grpc_server *server,
+ grpc_call_list *call_list) {
registered_method *rm;
- request_matcher_kill_requests(server, &server->unregistered_request_matcher);
+ request_matcher_kill_requests(server, &server->unregistered_request_matcher,
+ call_list);
request_matcher_zombify_all_pending_calls(
- &server->unregistered_request_matcher, server->workqueue);
+ &server->unregistered_request_matcher, call_list);
for (rm = server->registered_methods; rm; rm = rm->next) {
- request_matcher_kill_requests(server, &rm->request_matcher);
- request_matcher_zombify_all_pending_calls(&rm->request_matcher,
- server->workqueue);
+ request_matcher_kill_requests(server, &rm->request_matcher, call_list);
+ request_matcher_zombify_all_pending_calls(&rm->request_matcher, call_list);
}
}
-static void maybe_finish_shutdown(grpc_server *server) {
+static void maybe_finish_shutdown(grpc_server *server,
+ grpc_call_list *call_list) {
size_t i;
if (!gpr_atm_acq_load(&server->shutdown_flag) || server->shutdown_published) {
return;
}
- kill_pending_work_locked(server);
+ kill_pending_work_locked(server, call_list);
if (server->root_channel_data.next != &server->root_channel_data ||
server->listeners_destroyed < num_listeners(server)) {
@@ -548,7 +556,7 @@
server_ref(server);
grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, 1,
done_shutdown_event, server,
- &server->shutdown_tags[i].completion);
+ &server->shutdown_tags[i].completion, call_list);
}
}
@@ -566,10 +574,9 @@
return md;
}
-static void server_on_recv(void *ptr, int success) {
+static void server_on_recv(void *ptr, int success, grpc_call_list *call_list) {
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
- channel_data *chand = elem->channel_data;
gpr_timespec op_deadline;
if (success && !calld->got_initial_metadata) {
@@ -587,7 +594,7 @@
}
if (calld->host && calld->path) {
calld->got_initial_metadata = 1;
- start_new_rpc(elem);
+ start_new_rpc(elem, call_list);
}
break;
}
@@ -604,8 +611,7 @@
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
- grpc_workqueue_push(chand->server->workqueue,
- &calld->kill_zombie_closure, 1);
+ grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1);
} else {
gpr_mu_unlock(&calld->mu_state);
}
@@ -616,8 +622,7 @@
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
- grpc_workqueue_push(chand->server->workqueue,
- &calld->kill_zombie_closure, 1);
+ grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1);
} else if (calld->state == PENDING) {
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
@@ -629,7 +634,7 @@
break;
}
- calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success);
+ calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success, call_list);
}
static void server_mutate_op(grpc_call_element *elem,
@@ -646,10 +651,11 @@
}
static void server_start_transport_stream_op(grpc_call_element *elem,
- grpc_transport_stream_op *op) {
+ grpc_transport_stream_op *op,
+ grpc_call_list *call_list) {
GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
server_mutate_op(elem, op);
- grpc_call_next_op(elem, op);
+ grpc_call_next_op(elem, op, call_list);
}
static void accept_stream(void *cd, grpc_transport *transport,
@@ -660,7 +666,8 @@
0, gpr_inf_future(GPR_CLOCK_MONOTONIC));
}
-static void channel_connectivity_changed(void *cd, int iomgr_status_ignored) {
+static void channel_connectivity_changed(void *cd, int iomgr_status_ignored,
+ grpc_call_list *call_list) {
channel_data *chand = cd;
grpc_server *server = chand->server;
if (chand->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) {
@@ -670,18 +677,19 @@
op.connectivity_state = &chand->connectivity_state;
grpc_channel_next_op(grpc_channel_stack_element(
grpc_channel_get_channel_stack(chand->channel), 0),
- &op);
+ &op, call_list);
} else {
gpr_mu_lock(&server->mu_global);
- destroy_channel(chand);
+ destroy_channel(chand, call_list);
gpr_mu_unlock(&server->mu_global);
- GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "connectivity");
+ GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "connectivity", call_list);
}
}
static void init_call_elem(grpc_call_element *elem,
const void *server_transport_data,
- grpc_transport_stream_op *initial_op) {
+ grpc_transport_stream_op *initial_op,
+ grpc_call_list *call_list) {
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
memset(calld, 0, sizeof(call_data));
@@ -696,7 +704,8 @@
if (initial_op) server_mutate_op(elem, initial_op);
}
-static void destroy_call_elem(grpc_call_element *elem) {
+static void destroy_call_elem(grpc_call_element *elem,
+ grpc_call_list *call_list) {
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
@@ -711,13 +720,13 @@
gpr_mu_destroy(&calld->mu_state);
- server_unref(chand->server);
+ server_unref(chand->server, call_list);
}
static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
const grpc_channel_args *args,
grpc_mdctx *metadata_context, int is_first,
- int is_last) {
+ int is_last, grpc_call_list *call_list) {
channel_data *chand = elem->channel_data;
GPR_ASSERT(is_first);
GPR_ASSERT(!is_last);
@@ -733,7 +742,8 @@
channel_connectivity_changed, chand);
}
-static void destroy_channel_elem(grpc_channel_element *elem) {
+static void destroy_channel_elem(grpc_channel_element *elem,
+ grpc_call_list *call_list) {
size_t i;
channel_data *chand = elem->channel_data;
if (chand->registered_methods) {
@@ -752,11 +762,11 @@
chand->next->prev = chand->prev;
chand->prev->next = chand->next;
chand->next = chand->prev = chand;
- maybe_finish_shutdown(chand->server);
+ maybe_finish_shutdown(chand->server, call_list);
gpr_mu_unlock(&chand->server->mu_global);
GRPC_MDSTR_UNREF(chand->path_key);
GRPC_MDSTR_UNREF(chand->authority_key);
- server_unref(chand->server);
+ server_unref(chand->server, call_list);
}
}
@@ -810,7 +820,6 @@
gpr_ref_init(&server->internal_refcount, 1);
server->root_channel_data.next = server->root_channel_data.prev =
&server->root_channel_data;
- server->workqueue = grpc_workqueue_create();
/* TODO(ctiller): expose a channel_arg for this */
server->max_requested_calls = 32768;
@@ -881,23 +890,26 @@
void grpc_server_start(grpc_server *server) {
listener *l;
size_t i;
+ grpc_call_list call_list = GRPC_CALL_LIST_INIT;
server->pollsets = gpr_malloc(sizeof(grpc_pollset *) * server->cq_count);
for (i = 0; i < server->cq_count; i++) {
server->pollsets[i] = grpc_cq_pollset(server->cqs[i]);
- grpc_workqueue_add_to_pollset(server->workqueue, server->pollsets[i]);
}
for (l = server->listeners; l; l = l->next) {
- l->start(server, l->arg, server->pollsets, server->cq_count);
+ l->start(server, l->arg, server->pollsets, server->cq_count, &call_list);
}
+
+ grpc_call_list_run(&call_list);
}
void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport,
grpc_channel_filter const **extra_filters,
size_t num_extra_filters, grpc_mdctx *mdctx,
grpc_workqueue *workqueue,
- const grpc_channel_args *args) {
+ const grpc_channel_args *args,
+ grpc_call_list *call_list) {
size_t num_filters = s->channel_filter_count + num_extra_filters + 1;
grpc_channel_filter const **filters =
gpr_malloc(sizeof(grpc_channel_filter *) * num_filters);
@@ -927,11 +939,11 @@
for (i = 0; i < s->cq_count; i++) {
memset(&op, 0, sizeof(op));
op.bind_pollset = grpc_cq_pollset(s->cqs[i]);
- grpc_transport_perform_op(transport, &op);
+ grpc_transport_perform_op(transport, &op, call_list);
}
channel = grpc_channel_create_from_filters(NULL, filters, num_filters, args,
- mdctx, workqueue, 0);
+ mdctx, workqueue, 0, call_list);
chand = (channel_data *)grpc_channel_stack_element(
grpc_channel_get_channel_stack(channel), 0)
->channel_data;
@@ -987,19 +999,30 @@
op.on_connectivity_state_change = &chand->channel_connectivity_changed;
op.connectivity_state = &chand->connectivity_state;
op.disconnect = gpr_atm_acq_load(&s->shutdown_flag) != 0;
- grpc_transport_perform_op(transport, &op);
+ grpc_transport_perform_op(transport, &op, call_list);
}
-void done_published_shutdown(void *done_arg, grpc_cq_completion *storage) {
+void done_published_shutdown(void *done_arg, grpc_cq_completion *storage,
+ grpc_call_list *call_list) {
(void) done_arg;
gpr_free(storage);
}
+static void listener_destroy_done(void *s, int success,
+ grpc_call_list *call_list) {
+ grpc_server *server = s;
+ gpr_mu_lock(&server->mu_global);
+ server->listeners_destroyed++;
+ maybe_finish_shutdown(server, call_list);
+ gpr_mu_unlock(&server->mu_global);
+}
+
void grpc_server_shutdown_and_notify(grpc_server *server,
grpc_completion_queue *cq, void *tag) {
listener *l;
shutdown_tag *sdt;
channel_broadcaster broadcaster;
+ grpc_call_list call_list = GRPC_CALL_LIST_INIT;
GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag);
@@ -1008,9 +1031,9 @@
grpc_cq_begin_op(cq);
if (server->shutdown_published) {
grpc_cq_end_op(cq, tag, 1, done_published_shutdown, NULL,
- gpr_malloc(sizeof(grpc_cq_completion)));
+ gpr_malloc(sizeof(grpc_cq_completion)), &call_list);
gpr_mu_unlock(&server->mu_global);
- return;
+ goto done;
}
server->shutdown_tags =
gpr_realloc(server->shutdown_tags,
@@ -1020,7 +1043,7 @@
sdt->cq = cq;
if (gpr_atm_acq_load(&server->shutdown_flag)) {
gpr_mu_unlock(&server->mu_global);
- return;
+ goto done;
}
server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME);
@@ -1029,41 +1052,40 @@
/* collect all unregistered then registered calls */
gpr_mu_lock(&server->mu_call);
- kill_pending_work_locked(server);
+ kill_pending_work_locked(server, &call_list);
gpr_mu_unlock(&server->mu_call);
gpr_atm_rel_store(&server->shutdown_flag, 1);
- maybe_finish_shutdown(server);
+ maybe_finish_shutdown(server, &call_list);
gpr_mu_unlock(&server->mu_global);
/* Shutdown listeners */
for (l = server->listeners; l; l = l->next) {
- l->destroy(server, l->arg);
+ grpc_closure_init(&l->destroy_done, listener_destroy_done, server);
+ l->destroy(server, l->arg, &l->destroy_done, &call_list);
}
- channel_broadcaster_shutdown(&broadcaster, 1, 0);
-}
+ channel_broadcaster_shutdown(&broadcaster, 1, 0, &call_list);
-void grpc_server_listener_destroy_done(void *s) {
- grpc_server *server = s;
- gpr_mu_lock(&server->mu_global);
- server->listeners_destroyed++;
- maybe_finish_shutdown(server);
- gpr_mu_unlock(&server->mu_global);
+done:
+ grpc_call_list_run(&call_list);
}
void grpc_server_cancel_all_calls(grpc_server *server) {
channel_broadcaster broadcaster;
+ grpc_call_list call_list = GRPC_CALL_LIST_INIT;
gpr_mu_lock(&server->mu_global);
channel_broadcaster_init(server, &broadcaster);
gpr_mu_unlock(&server->mu_global);
- channel_broadcaster_shutdown(&broadcaster, 0, 1);
+ channel_broadcaster_shutdown(&broadcaster, 0, 1, &call_list);
+ grpc_call_list_run(&call_list);
}
void grpc_server_destroy(grpc_server *server) {
listener *l;
+ grpc_call_list call_list = GRPC_CALL_LIST_INIT;
gpr_mu_lock(&server->mu_global);
GPR_ASSERT(gpr_atm_acq_load(&server->shutdown_flag) || !server->listeners);
@@ -1077,16 +1099,17 @@
gpr_mu_unlock(&server->mu_global);
- grpc_workqueue_flush(server->workqueue);
-
- server_unref(server);
+ server_unref(server, &call_list);
+ grpc_call_list_run(&call_list);
}
-void grpc_server_add_listener(grpc_server *server, void *arg,
- void (*start)(grpc_server *server, void *arg,
- grpc_pollset **pollsets,
- size_t pollset_count),
- void (*destroy)(grpc_server *server, void *arg)) {
+void grpc_server_add_listener(
+ grpc_server *server, void *arg,
+ void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets,
+ size_t pollset_count, grpc_call_list *call_list),
+ void (*destroy)(grpc_server *server, void *arg, grpc_closure *on_done,
+ grpc_call_list *call_list),
+ grpc_call_list *call_list) {
listener *l = gpr_malloc(sizeof(listener));
l->arg = arg;
l->start = start;
@@ -1096,18 +1119,19 @@
}
static grpc_call_error queue_call_request(grpc_server *server,
- requested_call *rc) {
+ requested_call *rc,
+ grpc_call_list *call_list) {
call_data *calld = NULL;
request_matcher *request_matcher = NULL;
int request_id;
if (gpr_atm_acq_load(&server->shutdown_flag)) {
- fail_call(server, rc);
+ fail_call(server, rc, call_list);
return GRPC_CALL_OK;
}
request_id = gpr_stack_lockfree_pop(server->request_freelist);
if (request_id == -1) {
/* out of request ids: just fail this one */
- fail_call(server, rc);
+ fail_call(server, rc, call_list);
return GRPC_CALL_OK;
}
switch (rc->type) {
@@ -1135,12 +1159,13 @@
grpc_closure_init(
&calld->kill_zombie_closure, kill_zombie,
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
- grpc_workqueue_push(server->workqueue, &calld->kill_zombie_closure, 1);
+ grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1);
} else {
GPR_ASSERT(calld->state == PENDING);
calld->state = ACTIVATED;
gpr_mu_unlock(&calld->mu_state);
- begin_call(server, calld, &server->requested_calls[request_id]);
+ begin_call(server, calld, &server->requested_calls[request_id],
+ call_list);
}
gpr_mu_lock(&server->mu_call);
}
@@ -1154,13 +1179,16 @@
grpc_metadata_array *initial_metadata,
grpc_completion_queue *cq_bound_to_call,
grpc_completion_queue *cq_for_notification, void *tag) {
+ grpc_call_error error;
+ grpc_call_list call_list = GRPC_CALL_LIST_INIT;
requested_call *rc = gpr_malloc(sizeof(*rc));
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)) {
gpr_free(rc);
- return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+ error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+ goto done;
}
grpc_cq_begin_op(cq_for_notification);
details->reserved = NULL;
@@ -1172,7 +1200,10 @@
rc->call = call;
rc->data.batch.details = details;
rc->data.batch.initial_metadata = initial_metadata;
- return queue_call_request(server, rc);
+ error = queue_call_request(server, rc, &call_list);
+done:
+ grpc_call_list_run(&call_list);
+ return error;
}
grpc_call_error grpc_server_request_registered_call(
@@ -1180,11 +1211,14 @@
grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload,
grpc_completion_queue *cq_bound_to_call,
grpc_completion_queue *cq_for_notification, void *tag) {
+ grpc_call_error error;
+ grpc_call_list call_list = GRPC_CALL_LIST_INIT;
requested_call *rc = gpr_malloc(sizeof(*rc));
registered_method *registered_method = rm;
if (!grpc_cq_is_server_cq(cq_for_notification)) {
gpr_free(rc);
- return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+ error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
+ goto done;
}
grpc_cq_begin_op(cq_for_notification);
rc->type = REGISTERED_CALL;
@@ -1197,12 +1231,16 @@
rc->data.registered.deadline = deadline;
rc->data.registered.initial_metadata = initial_metadata;
rc->data.registered.optional_payload = optional_payload;
- return queue_call_request(server, rc);
+ error = queue_call_request(server, rc, &call_list);
+done:
+ grpc_call_list_run(&call_list);
+ return error;
}
-static void publish_registered_or_batch(grpc_call *call, int success,
- void *tag);
-static void publish_was_not_set(grpc_call *call, int success, void *tag) {
+static void publish_registered_or_batch(grpc_call *call, int success, void *tag,
+ grpc_call_list *call_list);
+static void publish_was_not_set(grpc_call *call, int success, void *tag,
+ grpc_call_list *call_list) {
abort();
}
@@ -1218,7 +1256,7 @@
}
static void begin_call(grpc_server *server, call_data *calld,
- requested_call *rc) {
+ requested_call *rc, grpc_call_list *call_list) {
grpc_ioreq_completion_func publish = publish_was_not_set;
grpc_ioreq req[2];
grpc_ioreq *r = req;
@@ -1229,7 +1267,7 @@
fill in the metadata array passed by the client, we need to perform
an ioreq op, that should complete immediately. */
- grpc_call_set_completion_queue(calld->call, rc->cq_bound_to_call);
+ grpc_call_set_completion_queue(calld->call, rc->cq_bound_to_call, call_list);
*rc->call = calld->call;
calld->cq_new = rc->cq_for_notification;
switch (rc->type) {
@@ -1265,10 +1303,11 @@
GRPC_CALL_INTERNAL_REF(calld->call, "server");
grpc_call_start_ioreq_and_call_back(calld->call, req, (size_t)(r - req),
- publish, rc);
+ publish, rc, call_list);
}
-static void done_request_event(void *req, grpc_cq_completion *c) {
+static void done_request_event(void *req, grpc_cq_completion *c,
+ grpc_call_list *call_list) {
requested_call *rc = req;
grpc_server *server = rc->server;
@@ -1281,10 +1320,11 @@
gpr_free(req);
}
- server_unref(server);
+ server_unref(server, call_list);
}
-static void fail_call(grpc_server *server, requested_call *rc) {
+static void fail_call(grpc_server *server, requested_call *rc,
+ grpc_call_list *call_list) {
*rc->call = NULL;
switch (rc->type) {
case BATCH_CALL:
@@ -1296,11 +1336,11 @@
}
server_ref(server);
grpc_cq_end_op(rc->cq_for_notification, rc->tag, 0, done_request_event, rc,
- &rc->completion);
+ &rc->completion, call_list);
}
-static void publish_registered_or_batch(grpc_call *call, int success,
- void *prc) {
+static void publish_registered_or_batch(grpc_call *call, int success, void *prc,
+ grpc_call_list *call_list) {
grpc_call_element *elem =
grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
requested_call *rc = prc;
@@ -1308,7 +1348,7 @@
channel_data *chand = elem->channel_data;
server_ref(chand->server);
grpc_cq_end_op(calld->cq_new, rc->tag, success, done_request_event, rc,
- &rc->completion);
+ &rc->completion, call_list);
GRPC_CALL_INTERNAL_UNREF(call, "server", call_list);
}