diff options
author | 2015-06-23 10:23:31 -0700 | |
---|---|---|
committer | 2015-06-23 10:37:58 -0700 | |
commit | e75d4c85315c4260f8eea5516dfba986982a6771 (patch) | |
tree | 329011db5c526acefcae917f54104998e9cb86fc /src/core/surface/server.c | |
parent | e25e92873d1a63d2a377d4b4803337a8b50baa11 (diff) | |
parent | 253aaa6b106673553c5cc94c87f7d8932729e54f (diff) |
Merge branch 'master' of github.com:grpc/grpc into decompression
Diffstat (limited to 'src/core/surface/server.c')
-rw-r--r-- | src/core/surface/server.c | 274 |
1 files changed, 163 insertions, 111 deletions
diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 80bbe42af4..076301777b 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -114,6 +114,7 @@ typedef struct channel_registered_method { struct channel_data { grpc_server *server; + size_t num_calls; grpc_channel *channel; grpc_mdstr *path_key; grpc_mdstr *authority_key; @@ -123,7 +124,6 @@ struct channel_data { channel_registered_method *registered_methods; gpr_uint32 registered_method_slots; gpr_uint32 registered_method_max_probes; - grpc_iomgr_closure finish_shutdown_channel_closure; grpc_iomgr_closure finish_destroy_channel_closure; }; @@ -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; @@ -198,6 +206,11 @@ struct call_data { static void begin_call(grpc_server *server, call_data *calld, requested_call *rc); 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) { GPR_ASSERT(!call->root[list]); @@ -270,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) { @@ -281,7 +295,7 @@ static void server_delete(grpc_server *server) { gpr_free(rm); } for (i = 0; i < server->cq_count; i++) { - grpc_cq_internal_unref(server->cqs[i]); + GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server"); } gpr_free(server->cqs); gpr_free(server->pollsets); @@ -308,7 +322,7 @@ static void orphan_channel(channel_data *chand) { static void finish_destroy_channel(void *cd, int success) { channel_data *chand = cd; grpc_server *server = chand->server; - grpc_channel_internal_unref(chand->channel); + GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server"); server_unref(server); } @@ -317,6 +331,7 @@ static void destroy_channel(channel_data *chand) { GPR_ASSERT(chand->server != NULL); orphan_channel(chand); server_ref(chand->server); + maybe_finish_shutdown(chand->server); chand->finish_destroy_channel_closure.cb = finish_destroy_channel; chand->finish_destroy_channel_closure.cb_arg = chand; grpc_iomgr_add_callback(&chand->finish_destroy_channel_closure); @@ -331,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); } } @@ -348,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 */ @@ -397,12 +412,33 @@ static int num_listeners(grpc_server *server) { static void maybe_finish_shutdown(grpc_server *server) { size_t i; - if (server->shutdown && !server->shutdown_published && server->lists[ALL_CALLS] == NULL && server->listeners_destroyed == num_listeners(server)) { - server->shutdown_published = 1; - for (i = 0; i < server->num_shutdown_tags; i++) { - grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, - NULL, 1); - } + 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"); + return; + } + if (server->listeners_destroyed < num_listeners(server)) { + gpr_log(GPR_DEBUG, "Waiting for all listeners to be destroyed (@ %d/%d)", + server->listeners_destroyed, num_listeners(server)); + return; + } + server->shutdown_published = 1; + for (i = 0; i < server->num_shutdown_tags; i++) { + grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, + NULL, 1); } } @@ -420,10 +456,19 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { return md; } +static void decrement_call_count(channel_data *chand) { + chand->num_calls--; + if (0 == chand->num_calls && chand->server->shutdown) { + shutdown_channel(chand, 0, 1); + } + maybe_finish_shutdown(chand->server); +} + 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; @@ -448,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); @@ -467,12 +512,14 @@ static void server_on_recv(void *ptr, int success) { calld->state = ZOMBIED; 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)) { - maybe_finish_shutdown(chand->server); + 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; } @@ -515,10 +562,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: @@ -531,22 +578,49 @@ static void channel_op(grpc_channel_element *elem, } } -static void finish_shutdown_channel(void *cd, int success) { - channel_data *chand = cd; +typedef struct { + channel_data *chand; + int send_goaway; + int send_disconnect; + grpc_iomgr_closure finish_shutdown_channel_closure; +} shutdown_channel_args; + +static void finish_shutdown_channel(void *p, int success) { + shutdown_channel_args *sca = p; grpc_channel_op op; - op.type = GRPC_CHANNEL_DISCONNECT; - op.dir = GRPC_CALL_DOWN; - channel_op(grpc_channel_stack_element( - grpc_channel_get_channel_stack(chand->channel), 0), - NULL, &op); - grpc_channel_internal_unref(chand->channel); + + if (sca->send_goaway) { + op.type = GRPC_CHANNEL_GOAWAY; + op.dir = GRPC_CALL_DOWN; + op.data.goaway.status = GRPC_STATUS_OK; + op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown"); + channel_op(grpc_channel_stack_element( + grpc_channel_get_channel_stack(sca->chand->channel), 0), + NULL, &op); + } + if (sca->send_disconnect) { + op.type = GRPC_CHANNEL_DISCONNECT; + op.dir = GRPC_CALL_DOWN; + channel_op(grpc_channel_stack_element( + grpc_channel_get_channel_stack(sca->chand->channel), 0), + NULL, &op); + } + GRPC_CHANNEL_INTERNAL_UNREF(sca->chand->channel, "shutdown"); + + gpr_free(sca); } -static void shutdown_channel(channel_data *chand) { - grpc_channel_internal_ref(chand->channel); - chand->finish_shutdown_channel_closure.cb = finish_shutdown_channel; - chand->finish_shutdown_channel_closure.cb_arg = chand; - grpc_iomgr_add_callback(&chand->finish_shutdown_channel_closure); +static void shutdown_channel(channel_data *chand, int send_goaway, + int send_disconnect) { + shutdown_channel_args *sca; + GRPC_CHANNEL_INTERNAL_REF(chand->channel, "shutdown"); + sca = gpr_malloc(sizeof(shutdown_channel_args)); + sca->chand = chand; + sca->send_goaway = send_goaway; + sca->send_disconnect = send_disconnect; + sca->finish_shutdown_channel_closure.cb = finish_shutdown_channel; + sca->finish_shutdown_channel_closure.cb_arg = sca; + grpc_iomgr_add_callback(&sca->finish_shutdown_channel_closure); } static void init_call_elem(grpc_call_element *elem, @@ -558,9 +632,13 @@ static void init_call_elem(grpc_call_element *elem, 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); + gpr_mu_unlock(&chand->server->mu_call); + + gpr_mu_lock(&chand->server->mu_global); + chand->num_calls++; + gpr_mu_unlock(&chand->server->mu_global); server_ref(chand->server); @@ -573,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]) { - maybe_finish_shutdown(chand->server); + 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); @@ -600,6 +680,7 @@ static void init_channel_elem(grpc_channel_element *elem, GPR_ASSERT(is_first); GPR_ASSERT(!is_last); chand->server = NULL; + chand->num_calls = 0; chand->channel = NULL; chand->path_key = grpc_mdstr_from_string(metadata_context, ":path"); chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority"); @@ -622,11 +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; - gpr_mu_unlock(&chand->server->mu); + maybe_finish_shutdown(chand->server); + gpr_mu_unlock(&chand->server->mu_global); grpc_mdstr_unref(chand->path_key); grpc_mdstr_unref(chand->authority_key); server_unref(chand->server); @@ -651,7 +733,8 @@ void grpc_server_register_completion_queue(grpc_server *server, for (i = 0; i < server->cq_count; i++) { if (server->cqs[i] == cq) return; } - grpc_cq_internal_ref(cq); + GRPC_CQ_INTERNAL_REF(cq, "server"); + grpc_cq_mark_server_cq(cq); n = server->cq_count++; server->cqs = gpr_realloc(server->cqs, server->cq_count * sizeof(grpc_completion_queue *)); @@ -672,7 +755,8 @@ grpc_server *grpc_server_create_from_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); @@ -713,7 +797,8 @@ void *grpc_server_register_method(grpc_server *server, const char *method, const char *host) { registered_method *m; if (!method) { - gpr_log(GPR_ERROR, "grpc_server_register_method method string cannot be NULL"); + gpr_log(GPR_ERROR, + "grpc_server_register_method method string cannot be NULL"); return NULL; } for (m = server->registered_methods; m; m = m->next) { @@ -821,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); @@ -836,17 +921,13 @@ void grpc_server_shutdown_and_notify(grpc_server *server, grpc_completion_queue *cq, void *tag) { listener *l; requested_call_array requested_calls; - channel_data **channels; channel_data *c; - size_t nchannels; size_t i; - grpc_channel_op op; - grpc_channel_element *elem; registered_method *rm; 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, @@ -855,25 +936,17 @@ 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; } - nchannels = 0; for (c = server->root_channel_data.next; c != &server->root_channel_data; c = c->next) { - nchannels++; - } - channels = gpr_malloc(sizeof(channel_data *) * nchannels); - i = 0; - for (c = server->root_channel_data.next; c != &server->root_channel_data; - c = c->next) { - grpc_channel_internal_ref(c->channel); - channels[i] = c; - i++; + shutdown_channel(c, 1, c->num_calls == 0); } /* 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) { @@ -892,25 +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); - - for (i = 0; i < nchannels; i++) { - c = channels[i]; - elem = grpc_channel_stack_element( - grpc_channel_get_channel_stack(c->channel), 0); - - op.type = GRPC_CHANNEL_GOAWAY; - op.dir = GRPC_CALL_DOWN; - op.data.goaway.status = GRPC_STATUS_OK; - op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown"); - elem->filter->channel_op(elem, NULL, &op); - - grpc_channel_internal_unref(c->channel); - } - gpr_free(channels); + gpr_mu_unlock(&server->mu_global); /* terminate all the requested calls */ for (i = 0; i < requested_calls.count; i++) { @@ -926,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) { @@ -940,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; } @@ -953,7 +1012,9 @@ void grpc_server_cancel_all_calls(grpc_server *server) { call_count = 0; calls = gpr_malloc(sizeof(grpc_call *) * call_capacity); - for (calld = server->lists[ALL_CALLS]; calld != server->lists[ALL_CALLS] || is_first; calld = calld->links[ALL_CALLS].next) { + for (calld = server->lists[ALL_CALLS]; + calld != server->lists[ALL_CALLS] || is_first; + calld = calld->links[ALL_CALLS].next) { if (call_count == call_capacity) { call_capacity *= 2; calls = gpr_realloc(calls, sizeof(grpc_call *) * call_capacity); @@ -963,10 +1024,11 @@ 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, "Unavailable"); + grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE, + "Unavailable"); GRPC_CALL_INTERNAL_UNREF(calls[i], "cancel_all", 1); } @@ -974,11 +1036,9 @@ void grpc_server_cancel_all_calls(grpc_server *server) { } void grpc_server_destroy(grpc_server *server) { - channel_data *c; listener *l; - call_data *calld; - 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)); @@ -988,20 +1048,7 @@ void grpc_server_destroy(grpc_server *server) { gpr_free(l); } - while ((calld = call_list_remove_head(&server->lists[PENDING_START], - PENDING_START)) != NULL) { - calld->state = ZOMBIED; - grpc_iomgr_closure_init( - &calld->kill_zombie_closure, kill_zombie, - grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0)); - grpc_iomgr_add_callback(&calld->kill_zombie_closure); - } - - for (c = server->root_channel_data.next; c != &server->root_channel_data; - c = c->next) { - shutdown_channel(c); - } - gpr_mu_unlock(&server->mu); + gpr_mu_unlock(&server->mu_global); server_unref(server); } @@ -1023,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; } @@ -1044,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; } } @@ -1063,6 +1110,9 @@ grpc_call_error grpc_server_request_call( 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)) { + return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; + } grpc_cq_begin_op(cq_for_notification, NULL); rc.type = BATCH_CALL; rc.tag = tag; @@ -1081,6 +1131,9 @@ grpc_call_error grpc_server_request_registered_call( grpc_completion_queue *cq_for_notification, void *tag) { requested_call rc; registered_method *registered_method = rm; + if (!grpc_cq_is_server_cq(cq_for_notification)) { + return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; + } grpc_cq_begin_op(cq_for_notification, NULL); rc.type = REGISTERED_CALL; rc.tag = tag; @@ -1187,9 +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; } - |