diff options
author | 2015-06-19 15:56:41 -0700 | |
---|---|---|
committer | 2015-06-19 15:56:41 -0700 | |
commit | 40ecfaf0610b0e3f8c79164526e016b2ad4af485 (patch) | |
tree | 0aaf34cf52d859f1182ca47a586c18650f744c65 /src/core/surface/server.c | |
parent | 87e430089caeb779ad9abce89cb1bc45c164d84b (diff) | |
parent | c7299987a0c7621d5d6bef7438e9baf04fc70de2 (diff) |
Merge github.com:grpc/grpc into oops-i-split-it-again
Diffstat (limited to 'src/core/surface/server.c')
-rw-r--r-- | src/core/surface/server.c | 102 |
1 files changed, 65 insertions, 37 deletions
diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 43e9afd56b..13ec5bee94 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -141,7 +141,15 @@ struct grpc_server { grpc_pollset **pollsets; size_t cq_count; - gpr_mu mu; + /* The two following mutexes control access to server-state + mu_global controls access to non-call-related state (e.g., channel state) + mu_call controls access to call-related state (e.g., the call lists) + + If they are ever required to be nested, you must lock mu_global + before mu_call. This is currently used in shutdown processing + (grpc_server_shutdown_and_notify and maybe_finish_shutdown) */ + gpr_mu mu_global; /* mutex for server and channel state */ + gpr_mu mu_call; /* mutex for call-specific state */ registered_method *registered_methods; requested_call_array requested_calls; @@ -200,6 +208,8 @@ static void begin_call(grpc_server *server, call_data *calld, 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 +283,8 @@ 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_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 +346,11 @@ static void finish_start_new_rpc_and_unlock(grpc_server *server, 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 +363,7 @@ static void start_new_rpc(grpc_call_element *elem) { 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 +415,16 @@ static void maybe_finish_shutdown(grpc_server *server) { 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 +468,7 @@ 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; + int remove_res; if (success && !calld->got_initial_metadata) { size_t i; @@ -476,16 +493,16 @@ static void server_on_recv(void *ptr, int success) { 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 +513,13 @@ static void server_on_recv(void *ptr, int success) { 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; } @@ -540,10 +560,10 @@ static void channel_op(grpc_channel_element *elem, 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 +632,13 @@ static void init_call_elem(grpc_call_element *elem, grpc_iomgr_closure_init(&calld->server_on_recv, server_on_recv, 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 +651,16 @@ static void destroy_call_elem(grpc_call_element *elem) { 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); if (removed[ALL_CALLS]) { + gpr_mu_lock(&chand->server->mu_global); decrement_call_count(chand); + gpr_mu_unlock(&chand->server->mu_global); } - gpr_mu_unlock(&chand->server->mu); if (calld->host) { grpc_mdstr_unref(calld->host); @@ -678,12 +703,12 @@ static void destroy_channel_elem(grpc_channel_element *elem) { 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 +755,8 @@ grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters, 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 +906,11 @@ grpc_transport_setup_result grpc_server_setup_transport( 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 +927,7 @@ void grpc_server_shutdown_and_notify(grpc_server *server, 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 +936,7 @@ void grpc_server_shutdown_and_notify(grpc_server *server, sdt->tag = tag; sdt->cq = cq; if (server->shutdown) { - gpr_mu_unlock(&server->mu); + gpr_mu_unlock(&server->mu_global); return; } @@ -920,6 +946,7 @@ void grpc_server_shutdown_and_notify(grpc_server *server, } /* 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 +965,11 @@ void grpc_server_shutdown_and_notify(grpc_server *server, 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 +985,10 @@ void grpc_server_shutdown_and_notify(grpc_server *server, 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 +999,12 @@ void grpc_server_cancel_all_calls(grpc_server *server) { 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 +1024,7 @@ void grpc_server_cancel_all_calls(grpc_server *server) { 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 +1038,7 @@ void grpc_server_cancel_all_calls(grpc_server *server) { 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 +1048,7 @@ void grpc_server_destroy(grpc_server *server) { gpr_free(l); } - gpr_mu_unlock(&server->mu); + gpr_mu_unlock(&server->mu_global); server_unref(server); } @@ -1042,9 +1070,9 @@ static grpc_call_error queue_call_request(grpc_server *server, 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 +1091,12 @@ static grpc_call_error queue_call_request(grpc_server *server, 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 +1240,8 @@ const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) { 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; } |