diff options
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 b26637a6ac..8f49f1c351 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -182,7 +182,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; @@ -403,19 +407,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); } @@ -429,7 +437,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 */ @@ -440,9 +447,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) */ @@ -453,14 +459,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) { @@ -504,10 +509,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; @@ -541,27 +546,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; } @@ -623,6 +635,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); @@ -634,21 +647,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); } @@ -676,10 +690,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); @@ -691,8 +705,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); } } @@ -1043,10 +1057,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 { |