Split server lock into 2 pieces: one for call list and one for global
state (including channels)
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index c8ac559..dac374b 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -141,7 +141,8 @@
grpc_pollset **pollsets;
size_t cq_count;
- gpr_mu mu;
+ gpr_mu mu_global;
+ gpr_mu mu_call;
registered_method *registered_methods;
requested_call_array requested_calls;
@@ -200,6 +201,8 @@
static void fail_call(grpc_server *server, requested_call *rc);
static void shutdown_channel(channel_data *chand, int send_goaway,
int send_disconnect);
+/* Before calling maybe_finish_shutdown, we must hold mu_global and not
+ hold mu_call */
static void maybe_finish_shutdown(grpc_server *server);
static int call_list_join(call_data **root, call_data *call, call_list list) {
@@ -273,7 +276,8 @@
registered_method *rm;
size_t i;
grpc_channel_args_destroy(server->channel_args);
- gpr_mu_destroy(&server->mu);
+ gpr_mu_destroy(&server->mu_global);
+ gpr_mu_destroy(&server->mu_call);
gpr_free(server->channel_filters);
requested_call_array_destroy(&server->requested_calls);
while ((rm = server->registered_methods) != NULL) {
@@ -335,11 +339,11 @@
if (array->count == 0) {
calld->state = PENDING;
call_list_join(pending_root, calld, PENDING_START);
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_call);
} else {
rc = array->calls[--array->count];
calld->state = ACTIVATED;
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_call);
begin_call(server, calld, &rc);
}
}
@@ -352,7 +356,7 @@
gpr_uint32 hash;
channel_registered_method *rm;
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_call);
if (chand->registered_methods && calld->path && calld->host) {
/* TODO(ctiller): unify these two searches */
/* check for an exact match with host */
@@ -404,11 +408,16 @@
if (!server->shutdown || server->shutdown_published) {
return;
}
+
+ gpr_mu_lock(&server->mu_call);
if (server->lists[ALL_CALLS] != NULL) {
gpr_log(GPR_DEBUG,
"Waiting for all calls to finish before destroying server");
+ gpr_mu_unlock(&server->mu_call);
return;
}
+ gpr_mu_unlock(&server->mu_call);
+
if (server->root_channel_data.next != &server->root_channel_data) {
gpr_log(GPR_DEBUG,
"Waiting for all channels to close before destroying server");
@@ -452,6 +461,7 @@
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
+ int remove_res;
if (success && !calld->got_initial_metadata) {
size_t i;
@@ -476,16 +486,16 @@
case GRPC_STREAM_SEND_CLOSED:
break;
case GRPC_STREAM_RECV_CLOSED:
- gpr_mu_lock(&chand->server->mu);
+ gpr_mu_lock(&chand->server->mu_call);
if (calld->state == NOT_STARTED) {
calld->state = ZOMBIED;
grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
grpc_iomgr_add_callback(&calld->kill_zombie_closure);
}
- gpr_mu_unlock(&chand->server->mu);
+ gpr_mu_unlock(&chand->server->mu_call);
break;
case GRPC_STREAM_CLOSED:
- gpr_mu_lock(&chand->server->mu);
+ gpr_mu_lock(&chand->server->mu_call);
if (calld->state == NOT_STARTED) {
calld->state = ZOMBIED;
grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
@@ -496,10 +506,13 @@
grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
grpc_iomgr_add_callback(&calld->kill_zombie_closure);
}
- if (call_list_remove(calld, ALL_CALLS)) {
+ remove_res = call_list_remove(calld, ALL_CALLS);
+ gpr_mu_unlock(&chand->server->mu_call);
+ gpr_mu_lock(&chand->server->mu_global);
+ if (remove_res) {
decrement_call_count(chand);
}
- gpr_mu_unlock(&chand->server->mu);
+ gpr_mu_unlock(&chand->server->mu_global);
break;
}
@@ -542,10 +555,10 @@
case GRPC_TRANSPORT_CLOSED:
/* if the transport is closed for a server channel, we destroy the
channel */
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_global);
server_ref(server);
destroy_channel(chand);
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_global);
server_unref(server);
break;
case GRPC_TRANSPORT_GOAWAY:
@@ -612,10 +625,13 @@
calld->deadline = gpr_inf_future;
calld->call = grpc_call_from_top_element(elem);
- gpr_mu_lock(&chand->server->mu);
+ gpr_mu_lock(&chand->server->mu_call);
call_list_join(&chand->server->lists[ALL_CALLS], calld, ALL_CALLS);
+ gpr_mu_unlock(&chand->server->mu_call);
+
+ gpr_mu_lock(&chand->server->mu_global);
chand->num_calls++;
- gpr_mu_unlock(&chand->server->mu);
+ gpr_mu_unlock(&chand->server->mu_global);
server_ref(chand->server);
@@ -628,14 +644,16 @@
int removed[CALL_LIST_COUNT];
size_t i;
- gpr_mu_lock(&chand->server->mu);
+ gpr_mu_lock(&chand->server->mu_call);
for (i = 0; i < CALL_LIST_COUNT; i++) {
removed[i] = call_list_remove(elem->call_data, i);
}
+ gpr_mu_unlock(&chand->server->mu_call);
+ gpr_mu_lock(&chand->server->mu_global);
if (removed[ALL_CALLS]) {
decrement_call_count(chand);
}
- gpr_mu_unlock(&chand->server->mu);
+ gpr_mu_unlock(&chand->server->mu_global);
if (calld->host) {
grpc_mdstr_unref(calld->host);
@@ -678,12 +696,12 @@
gpr_free(chand->registered_methods);
}
if (chand->server) {
- gpr_mu_lock(&chand->server->mu);
+ gpr_mu_lock(&chand->server->mu_global);
chand->next->prev = chand->prev;
chand->prev->next = chand->next;
chand->next = chand->prev = chand;
maybe_finish_shutdown(chand->server);
- gpr_mu_unlock(&chand->server->mu);
+ gpr_mu_unlock(&chand->server->mu_global);
grpc_mdstr_unref(chand->path_key);
grpc_mdstr_unref(chand->authority_key);
server_unref(chand->server);
@@ -730,7 +748,8 @@
memset(server, 0, sizeof(grpc_server));
- gpr_mu_init(&server->mu);
+ gpr_mu_init(&server->mu_global);
+ gpr_mu_init(&server->mu_call);
/* decremented by grpc_server_destroy */
gpr_ref_init(&server->internal_refcount, 1);
@@ -880,11 +899,11 @@
result = grpc_connected_channel_bind_transport(
grpc_channel_get_channel_stack(channel), transport);
- gpr_mu_lock(&s->mu);
+ gpr_mu_lock(&s->mu_global);
chand->next = &s->root_channel_data;
chand->prev = chand->next->prev;
chand->next->prev = chand->prev->next = chand;
- gpr_mu_unlock(&s->mu);
+ gpr_mu_unlock(&s->mu_global);
gpr_free(filters);
@@ -901,7 +920,7 @@
shutdown_tag *sdt;
/* lock, and gather up some stuff to do */
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_global);
grpc_cq_begin_op(cq, NULL);
server->shutdown_tags =
gpr_realloc(server->shutdown_tags,
@@ -910,7 +929,7 @@
sdt->tag = tag;
sdt->cq = cq;
if (server->shutdown) {
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_global);
return;
}
@@ -920,6 +939,7 @@
}
/* collect all unregistered then registered calls */
+ gpr_mu_lock(&server->mu_call);
requested_calls = server->requested_calls;
memset(&server->requested_calls, 0, sizeof(server->requested_calls));
for (rm = server->registered_methods; rm; rm = rm->next) {
@@ -938,10 +958,11 @@
gpr_free(rm->requested.calls);
memset(&rm->requested, 0, sizeof(rm->requested));
}
+ gpr_mu_unlock(&server->mu_call);
server->shutdown = 1;
maybe_finish_shutdown(server);
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_global);
/* terminate all the requested calls */
for (i = 0; i < requested_calls.count; i++) {
@@ -957,10 +978,10 @@
void grpc_server_listener_destroy_done(void *s) {
grpc_server *server = s;
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_global);
server->listeners_destroyed++;
maybe_finish_shutdown(server);
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_global);
}
void grpc_server_cancel_all_calls(grpc_server *server) {
@@ -971,12 +992,12 @@
int is_first = 1;
size_t i;
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_call);
GPR_ASSERT(server->shutdown);
if (!server->lists[ALL_CALLS]) {
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_call);
return;
}
@@ -996,7 +1017,7 @@
is_first = 0;
}
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_call);
for (i = 0; i < call_count; i++) {
grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE,
@@ -1010,7 +1031,7 @@
void grpc_server_destroy(grpc_server *server) {
listener *l;
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_global);
GPR_ASSERT(server->shutdown || !server->listeners);
GPR_ASSERT(server->listeners_destroyed == num_listeners(server));
@@ -1020,7 +1041,7 @@
gpr_free(l);
}
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_global);
server_unref(server);
}
@@ -1042,9 +1063,9 @@
requested_call *rc) {
call_data *calld = NULL;
requested_call_array *requested_calls = NULL;
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_call);
if (server->shutdown) {
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_call);
fail_call(server, rc);
return GRPC_CALL_OK;
}
@@ -1063,12 +1084,12 @@
if (calld) {
GPR_ASSERT(calld->state == PENDING);
calld->state = ACTIVATED;
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_call);
begin_call(server, calld, rc);
return GRPC_CALL_OK;
} else {
*requested_call_array_add(requested_calls) = *rc;
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_call);
return GRPC_CALL_OK;
}
}
@@ -1212,8 +1233,8 @@
int grpc_server_has_open_connections(grpc_server *server) {
int r;
- gpr_mu_lock(&server->mu);
+ gpr_mu_lock(&server->mu_global);
r = server->root_channel_data.next != &server->root_channel_data;
- gpr_mu_unlock(&server->mu);
+ gpr_mu_unlock(&server->mu_global);
return r;
}