diff options
author | 2015-07-09 16:45:56 -0700 | |
---|---|---|
committer | 2015-07-09 16:45:56 -0700 | |
commit | 5aa65e820b98ac6c7ca0a0f82ed37369ee343d1b (patch) | |
tree | 083b9e9228b570a5088ba833e737abd1d3f77b76 /src/core/surface/server.c | |
parent | ab54f793ffbf8f1e805f85d6b0a95c93e619baec (diff) | |
parent | b97b988052f628a7831042acd4cfe1ab854e97cf (diff) |
Merge github.com:grpc/grpc into be-less-verbose
Diffstat (limited to 'src/core/surface/server.c')
-rw-r--r-- | src/core/surface/server.c | 82 |
1 files changed, 49 insertions, 33 deletions
diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 4053b22063..09c42c4b7e 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -185,7 +185,11 @@ typedef enum { struct call_data { grpc_call *call; + /** protects state */ + gpr_mu mu_state; + /** the current state of a call - see call_state */ call_state state; + grpc_mdstr *path; grpc_mdstr *host; gpr_timespec deadline; @@ -406,19 +410,23 @@ static void destroy_channel(channel_data *chand) { grpc_iomgr_add_callback(&chand->finish_destroy_channel_closure); } -static void finish_start_new_rpc_and_unlock(grpc_server *server, - grpc_call_element *elem, - call_data **pending_root, - requested_call_array *array) { +static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem, + call_data **pending_root, + requested_call_array *array) { requested_call rc; call_data *calld = elem->call_data; + gpr_mu_lock(&server->mu_call); if (array->count == 0) { + gpr_mu_lock(&calld->mu_state); calld->state = PENDING; + gpr_mu_unlock(&calld->mu_state); call_list_join(pending_root, calld, PENDING_START); gpr_mu_unlock(&server->mu_call); } else { rc = array->calls[--array->count]; + gpr_mu_lock(&calld->mu_state); calld->state = ACTIVATED; + gpr_mu_unlock(&calld->mu_state); gpr_mu_unlock(&server->mu_call); begin_call(server, calld, &rc); } @@ -432,7 +440,6 @@ static void start_new_rpc(grpc_call_element *elem) { gpr_uint32 hash; channel_registered_method *rm; - 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 */ @@ -443,9 +450,8 @@ static void start_new_rpc(grpc_call_element *elem) { if (!rm) break; if (rm->host != calld->host) continue; if (rm->method != calld->path) continue; - finish_start_new_rpc_and_unlock(server, elem, - &rm->server_registered_method->pending, - &rm->server_registered_method->requested); + finish_start_new_rpc(server, elem, &rm->server_registered_method->pending, + &rm->server_registered_method->requested); return; } /* check for a wildcard method definition (no host set) */ @@ -456,14 +462,13 @@ static void start_new_rpc(grpc_call_element *elem) { if (!rm) break; if (rm->host != NULL) continue; if (rm->method != calld->path) continue; - finish_start_new_rpc_and_unlock(server, elem, - &rm->server_registered_method->pending, - &rm->server_registered_method->requested); + finish_start_new_rpc(server, elem, &rm->server_registered_method->pending, + &rm->server_registered_method->requested); return; } } - finish_start_new_rpc_and_unlock(server, elem, &server->lists[PENDING_START], - &server->requested_calls); + finish_start_new_rpc(server, elem, &server->lists[PENDING_START], + &server->requested_calls); } static void kill_zombie(void *elem, int success) { @@ -522,10 +527,10 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; if (md->key == chand->path_key) { - calld->path = grpc_mdstr_ref(md->value); + calld->path = GRPC_MDSTR_REF(md->value); return NULL; } else if (md->key == chand->authority_key) { - calld->host = grpc_mdstr_ref(md->value); + calld->host = GRPC_MDSTR_REF(md->value); return NULL; } return md; @@ -559,27 +564,34 @@ 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_call); + gpr_mu_lock(&calld->mu_state); if (calld->state == NOT_STARTED) { calld->state = ZOMBIED; + gpr_mu_unlock(&calld->mu_state); grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); grpc_iomgr_add_callback(&calld->kill_zombie_closure); + } else { + gpr_mu_unlock(&calld->mu_state); } - gpr_mu_unlock(&chand->server->mu_call); break; case GRPC_STREAM_CLOSED: - gpr_mu_lock(&chand->server->mu_call); + gpr_mu_lock(&calld->mu_state); if (calld->state == NOT_STARTED) { calld->state = ZOMBIED; + gpr_mu_unlock(&calld->mu_state); grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); grpc_iomgr_add_callback(&calld->kill_zombie_closure); } else if (calld->state == PENDING) { - call_list_remove(calld, PENDING_START); calld->state = ZOMBIED; + gpr_mu_unlock(&calld->mu_state); + gpr_mu_lock(&chand->server->mu_call); + call_list_remove(calld, PENDING_START); + gpr_mu_unlock(&chand->server->mu_call); grpc_iomgr_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); grpc_iomgr_add_callback(&calld->kill_zombie_closure); + } else { + gpr_mu_unlock(&calld->mu_state); } - gpr_mu_unlock(&chand->server->mu_call); break; } @@ -641,6 +653,7 @@ static void init_call_elem(grpc_call_element *elem, memset(calld, 0, sizeof(call_data)); calld->deadline = gpr_inf_future; calld->call = grpc_call_from_top_element(elem); + gpr_mu_init(&calld->mu_state); grpc_iomgr_closure_init(&calld->server_on_recv, server_on_recv, elem); @@ -652,21 +665,22 @@ static void init_call_elem(grpc_call_element *elem, static void destroy_call_elem(grpc_call_element *elem) { channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; - size_t i; - gpr_mu_lock(&chand->server->mu_call); - for (i = 0; i < CALL_LIST_COUNT; i++) { - call_list_remove(elem->call_data, i); + if (calld->state == PENDING) { + gpr_mu_lock(&chand->server->mu_call); + call_list_remove(elem->call_data, PENDING_START); + gpr_mu_unlock(&chand->server->mu_call); } - gpr_mu_unlock(&chand->server->mu_call); if (calld->host) { - grpc_mdstr_unref(calld->host); + GRPC_MDSTR_UNREF(calld->host); } if (calld->path) { - grpc_mdstr_unref(calld->path); + GRPC_MDSTR_UNREF(calld->path); } + gpr_mu_destroy(&calld->mu_state); + server_unref(chand->server); } @@ -694,10 +708,10 @@ static void destroy_channel_elem(grpc_channel_element *elem) { if (chand->registered_methods) { for (i = 0; i < chand->registered_method_slots; i++) { if (chand->registered_methods[i].method) { - grpc_mdstr_unref(chand->registered_methods[i].method); + GRPC_MDSTR_UNREF(chand->registered_methods[i].method); } if (chand->registered_methods[i].host) { - grpc_mdstr_unref(chand->registered_methods[i].host); + GRPC_MDSTR_UNREF(chand->registered_methods[i].host); } } gpr_free(chand->registered_methods); @@ -709,8 +723,8 @@ static void destroy_channel_elem(grpc_channel_element *elem) { chand->next = chand->prev = chand; maybe_finish_shutdown(chand->server); gpr_mu_unlock(&chand->server->mu_global); - grpc_mdstr_unref(chand->path_key); - grpc_mdstr_unref(chand->authority_key); + GRPC_MDSTR_UNREF(chand->path_key); + GRPC_MDSTR_UNREF(chand->authority_key); server_unref(chand->server); } } @@ -1063,10 +1077,12 @@ static grpc_call_error queue_call_request(grpc_server *server, requested_calls = &rc->data.registered.registered_method->requested; break; } - if (calld) { + if (calld != NULL) { + gpr_mu_unlock(&server->mu_call); + gpr_mu_lock(&calld->mu_state); GPR_ASSERT(calld->state == PENDING); calld->state = ACTIVATED; - gpr_mu_unlock(&server->mu_call); + gpr_mu_unlock(&calld->mu_state); begin_call(server, calld, rc); return GRPC_CALL_OK; } else { |