|
|
@ -57,9 +57,11 @@ |
|
|
|
typedef struct listener { |
|
|
|
typedef struct listener { |
|
|
|
void *arg; |
|
|
|
void *arg; |
|
|
|
void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets, |
|
|
|
void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets, |
|
|
|
size_t pollset_count); |
|
|
|
size_t pollset_count, grpc_call_list *call_list); |
|
|
|
void (*destroy)(grpc_server *server, void *arg); |
|
|
|
void (*destroy)(grpc_server *server, void *arg, grpc_closure *closure, |
|
|
|
|
|
|
|
grpc_call_list *call_list); |
|
|
|
struct listener *next; |
|
|
|
struct listener *next; |
|
|
|
|
|
|
|
grpc_closure destroy_done; |
|
|
|
} listener; |
|
|
|
} listener; |
|
|
|
|
|
|
|
|
|
|
|
typedef struct call_data call_data; |
|
|
|
typedef struct call_data call_data; |
|
|
@ -219,19 +221,19 @@ struct grpc_server { |
|
|
|
|
|
|
|
|
|
|
|
/** when did we print the last shutdown progress message */ |
|
|
|
/** when did we print the last shutdown progress message */ |
|
|
|
gpr_timespec last_shutdown_message_time; |
|
|
|
gpr_timespec last_shutdown_message_time; |
|
|
|
|
|
|
|
|
|
|
|
grpc_workqueue *workqueue; |
|
|
|
|
|
|
|
}; |
|
|
|
}; |
|
|
|
|
|
|
|
|
|
|
|
#define SERVER_FROM_CALL_ELEM(elem) \ |
|
|
|
#define SERVER_FROM_CALL_ELEM(elem) \ |
|
|
|
(((channel_data *)(elem)->channel_data)->server) |
|
|
|
(((channel_data *)(elem)->channel_data)->server) |
|
|
|
|
|
|
|
|
|
|
|
static void begin_call(grpc_server *server, call_data *calld, |
|
|
|
static void begin_call(grpc_server *server, call_data *calld, |
|
|
|
requested_call *rc); |
|
|
|
requested_call *rc, grpc_call_list *call_list); |
|
|
|
static void fail_call(grpc_server *server, requested_call *rc); |
|
|
|
static void fail_call(grpc_server *server, requested_call *rc, |
|
|
|
|
|
|
|
grpc_call_list *call_list); |
|
|
|
/* Before calling maybe_finish_shutdown, we must hold mu_global and not
|
|
|
|
/* Before calling maybe_finish_shutdown, we must hold mu_global and not
|
|
|
|
hold mu_call */ |
|
|
|
hold mu_call */ |
|
|
|
static void maybe_finish_shutdown(grpc_server *server); |
|
|
|
static void maybe_finish_shutdown(grpc_server *server, |
|
|
|
|
|
|
|
grpc_call_list *call_list); |
|
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
/*
|
|
|
|
* channel broadcaster |
|
|
|
* channel broadcaster |
|
|
@ -258,14 +260,15 @@ struct shutdown_cleanup_args { |
|
|
|
gpr_slice slice; |
|
|
|
gpr_slice slice; |
|
|
|
}; |
|
|
|
}; |
|
|
|
|
|
|
|
|
|
|
|
static void shutdown_cleanup(void *arg, int iomgr_status_ignored) { |
|
|
|
static void shutdown_cleanup(void *arg, int iomgr_status_ignored, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
struct shutdown_cleanup_args *a = arg; |
|
|
|
struct shutdown_cleanup_args *a = arg; |
|
|
|
gpr_slice_unref(a->slice); |
|
|
|
gpr_slice_unref(a->slice); |
|
|
|
gpr_free(a); |
|
|
|
gpr_free(a); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void send_shutdown(grpc_channel *channel, int send_goaway, |
|
|
|
static void send_shutdown(grpc_channel *channel, int send_goaway, |
|
|
|
int send_disconnect) { |
|
|
|
int send_disconnect, grpc_call_list *call_list) { |
|
|
|
grpc_transport_op op; |
|
|
|
grpc_transport_op op; |
|
|
|
struct shutdown_cleanup_args *sc; |
|
|
|
struct shutdown_cleanup_args *sc; |
|
|
|
grpc_channel_element *elem; |
|
|
|
grpc_channel_element *elem; |
|
|
@ -281,17 +284,17 @@ static void send_shutdown(grpc_channel *channel, int send_goaway, |
|
|
|
op.on_consumed = &sc->closure; |
|
|
|
op.on_consumed = &sc->closure; |
|
|
|
|
|
|
|
|
|
|
|
elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0); |
|
|
|
elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0); |
|
|
|
elem->filter->start_transport_op(elem, &op); |
|
|
|
elem->filter->start_transport_op(elem, &op, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void channel_broadcaster_shutdown(channel_broadcaster *cb, |
|
|
|
static void channel_broadcaster_shutdown(channel_broadcaster *cb, |
|
|
|
int send_goaway, |
|
|
|
int send_goaway, int force_disconnect, |
|
|
|
int force_disconnect) { |
|
|
|
grpc_call_list *call_list) { |
|
|
|
size_t i; |
|
|
|
size_t i; |
|
|
|
|
|
|
|
|
|
|
|
for (i = 0; i < cb->num_channels; i++) { |
|
|
|
for (i = 0; i < cb->num_channels; i++) { |
|
|
|
send_shutdown(cb->channels[i], send_goaway, force_disconnect); |
|
|
|
send_shutdown(cb->channels[i], send_goaway, force_disconnect, call_list); |
|
|
|
GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast"); |
|
|
|
GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast", call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
gpr_free(cb->channels); |
|
|
|
gpr_free(cb->channels); |
|
|
|
} |
|
|
|
} |
|
|
@ -311,12 +314,12 @@ static void request_matcher_destroy(request_matcher *request_matcher) { |
|
|
|
gpr_stack_lockfree_destroy(request_matcher->requests); |
|
|
|
gpr_stack_lockfree_destroy(request_matcher->requests); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void kill_zombie(void *elem, int success) { |
|
|
|
static void kill_zombie(void *elem, int success, grpc_call_list *call_list) { |
|
|
|
grpc_call_destroy(grpc_call_from_top_element(elem)); |
|
|
|
grpc_call_destroy(grpc_call_from_top_element(elem)); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void request_matcher_zombify_all_pending_calls( |
|
|
|
static void request_matcher_zombify_all_pending_calls( |
|
|
|
request_matcher *request_matcher, grpc_workqueue *workqueue) { |
|
|
|
request_matcher *request_matcher, grpc_call_list *call_list) { |
|
|
|
while (request_matcher->pending_head) { |
|
|
|
while (request_matcher->pending_head) { |
|
|
|
call_data *calld = request_matcher->pending_head; |
|
|
|
call_data *calld = request_matcher->pending_head; |
|
|
|
request_matcher->pending_head = calld->pending_next; |
|
|
|
request_matcher->pending_head = calld->pending_next; |
|
|
@ -326,15 +329,16 @@ static void request_matcher_zombify_all_pending_calls( |
|
|
|
grpc_closure_init( |
|
|
|
grpc_closure_init( |
|
|
|
&calld->kill_zombie_closure, kill_zombie, |
|
|
|
&calld->kill_zombie_closure, kill_zombie, |
|
|
|
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0)); |
|
|
|
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0)); |
|
|
|
grpc_workqueue_push(workqueue, &calld->kill_zombie_closure, 1); |
|
|
|
grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void request_matcher_kill_requests(grpc_server *server, |
|
|
|
static void request_matcher_kill_requests(grpc_server *server, |
|
|
|
request_matcher *rm) { |
|
|
|
request_matcher *rm, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
int request_id; |
|
|
|
int request_id; |
|
|
|
while ((request_id = gpr_stack_lockfree_pop(rm->requests)) != -1) { |
|
|
|
while ((request_id = gpr_stack_lockfree_pop(rm->requests)) != -1) { |
|
|
|
fail_call(server, &server->requested_calls[request_id]); |
|
|
|
fail_call(server, &server->requested_calls[request_id], call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -346,7 +350,7 @@ static void server_ref(grpc_server *server) { |
|
|
|
gpr_ref(&server->internal_refcount); |
|
|
|
gpr_ref(&server->internal_refcount); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void server_delete(grpc_server *server) { |
|
|
|
static void server_delete(grpc_server *server, grpc_call_list *call_list) { |
|
|
|
registered_method *rm; |
|
|
|
registered_method *rm; |
|
|
|
size_t i; |
|
|
|
size_t i; |
|
|
|
grpc_channel_args_destroy(server->channel_args); |
|
|
|
grpc_channel_args_destroy(server->channel_args); |
|
|
@ -365,7 +369,6 @@ static void server_delete(grpc_server *server) { |
|
|
|
} |
|
|
|
} |
|
|
|
request_matcher_destroy(&server->unregistered_request_matcher); |
|
|
|
request_matcher_destroy(&server->unregistered_request_matcher); |
|
|
|
gpr_stack_lockfree_destroy(server->request_freelist); |
|
|
|
gpr_stack_lockfree_destroy(server->request_freelist); |
|
|
|
GRPC_WORKQUEUE_UNREF(server->workqueue, "destroy"); |
|
|
|
|
|
|
|
gpr_free(server->cqs); |
|
|
|
gpr_free(server->cqs); |
|
|
|
gpr_free(server->pollsets); |
|
|
|
gpr_free(server->pollsets); |
|
|
|
gpr_free(server->shutdown_tags); |
|
|
|
gpr_free(server->shutdown_tags); |
|
|
@ -373,9 +376,9 @@ static void server_delete(grpc_server *server) { |
|
|
|
gpr_free(server); |
|
|
|
gpr_free(server); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void server_unref(grpc_server *server) { |
|
|
|
static void server_unref(grpc_server *server, grpc_call_list *call_list) { |
|
|
|
if (gpr_unref(&server->internal_refcount)) { |
|
|
|
if (gpr_unref(&server->internal_refcount)) { |
|
|
|
server_delete(server); |
|
|
|
server_delete(server, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -389,30 +392,29 @@ static void orphan_channel(channel_data *chand) { |
|
|
|
chand->next = chand->prev = chand; |
|
|
|
chand->next = chand->prev = chand; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void finish_destroy_channel(void *cd, int success) { |
|
|
|
static void finish_destroy_channel(void *cd, int success, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
channel_data *chand = cd; |
|
|
|
channel_data *chand = cd; |
|
|
|
grpc_server *server = chand->server; |
|
|
|
grpc_server *server = chand->server; |
|
|
|
gpr_log(GPR_DEBUG, "finish_destroy_channel: %p", chand->channel); |
|
|
|
gpr_log(GPR_DEBUG, "finish_destroy_channel: %p", chand->channel); |
|
|
|
GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server"); |
|
|
|
GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server", call_list); |
|
|
|
server_unref(server); |
|
|
|
server_unref(server, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void destroy_channel(channel_data *chand) { |
|
|
|
static void destroy_channel(channel_data *chand, grpc_call_list *call_list) { |
|
|
|
if (is_channel_orphaned(chand)) return; |
|
|
|
if (is_channel_orphaned(chand)) return; |
|
|
|
GPR_ASSERT(chand->server != NULL); |
|
|
|
GPR_ASSERT(chand->server != NULL); |
|
|
|
orphan_channel(chand); |
|
|
|
orphan_channel(chand); |
|
|
|
server_ref(chand->server); |
|
|
|
server_ref(chand->server); |
|
|
|
maybe_finish_shutdown(chand->server); |
|
|
|
maybe_finish_shutdown(chand->server, call_list); |
|
|
|
chand->finish_destroy_channel_closure.cb = finish_destroy_channel; |
|
|
|
chand->finish_destroy_channel_closure.cb = finish_destroy_channel; |
|
|
|
chand->finish_destroy_channel_closure.cb_arg = chand; |
|
|
|
chand->finish_destroy_channel_closure.cb_arg = chand; |
|
|
|
gpr_log(GPR_DEBUG, "queue finish_destroy_channel: %p on %p", chand->channel, |
|
|
|
grpc_call_list_add(call_list, &chand->finish_destroy_channel_closure, 1); |
|
|
|
chand->server->workqueue); |
|
|
|
|
|
|
|
grpc_workqueue_push(chand->server->workqueue, |
|
|
|
|
|
|
|
&chand->finish_destroy_channel_closure, 1); |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem, |
|
|
|
static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem, |
|
|
|
request_matcher *request_matcher) { |
|
|
|
request_matcher *request_matcher, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
int request_id; |
|
|
|
int request_id; |
|
|
|
|
|
|
|
|
|
|
@ -421,7 +423,7 @@ static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem, |
|
|
|
calld->state = ZOMBIED; |
|
|
|
calld->state = ZOMBIED; |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); |
|
|
|
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); |
|
|
|
grpc_workqueue_push(server->workqueue, &calld->kill_zombie_closure, 1); |
|
|
|
grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1); |
|
|
|
return; |
|
|
|
return; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -443,11 +445,11 @@ static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem, |
|
|
|
gpr_mu_lock(&calld->mu_state); |
|
|
|
gpr_mu_lock(&calld->mu_state); |
|
|
|
calld->state = ACTIVATED; |
|
|
|
calld->state = ACTIVATED; |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
begin_call(server, calld, &server->requested_calls[request_id]); |
|
|
|
begin_call(server, calld, &server->requested_calls[request_id], call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void start_new_rpc(grpc_call_element *elem) { |
|
|
|
static void start_new_rpc(grpc_call_element *elem, grpc_call_list *call_list) { |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
grpc_server *server = chand->server; |
|
|
|
grpc_server *server = chand->server; |
|
|
@ -466,7 +468,8 @@ static void start_new_rpc(grpc_call_element *elem) { |
|
|
|
if (rm->host != calld->host) continue; |
|
|
|
if (rm->host != calld->host) continue; |
|
|
|
if (rm->method != calld->path) continue; |
|
|
|
if (rm->method != calld->path) continue; |
|
|
|
finish_start_new_rpc(server, elem, |
|
|
|
finish_start_new_rpc(server, elem, |
|
|
|
&rm->server_registered_method->request_matcher); |
|
|
|
&rm->server_registered_method->request_matcher, |
|
|
|
|
|
|
|
call_list); |
|
|
|
return; |
|
|
|
return; |
|
|
|
} |
|
|
|
} |
|
|
|
/* check for a wildcard method definition (no host set) */ |
|
|
|
/* check for a wildcard method definition (no host set) */ |
|
|
@ -478,11 +481,13 @@ static void start_new_rpc(grpc_call_element *elem) { |
|
|
|
if (rm->host != NULL) continue; |
|
|
|
if (rm->host != NULL) continue; |
|
|
|
if (rm->method != calld->path) continue; |
|
|
|
if (rm->method != calld->path) continue; |
|
|
|
finish_start_new_rpc(server, elem, |
|
|
|
finish_start_new_rpc(server, elem, |
|
|
|
&rm->server_registered_method->request_matcher); |
|
|
|
&rm->server_registered_method->request_matcher, |
|
|
|
|
|
|
|
call_list); |
|
|
|
return; |
|
|
|
return; |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
finish_start_new_rpc(server, elem, &server->unregistered_request_matcher); |
|
|
|
finish_start_new_rpc(server, elem, &server->unregistered_request_matcher, |
|
|
|
|
|
|
|
call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static int num_listeners(grpc_server *server) { |
|
|
|
static int num_listeners(grpc_server *server) { |
|
|
@ -494,8 +499,9 @@ static int num_listeners(grpc_server *server) { |
|
|
|
return n; |
|
|
|
return n; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void done_shutdown_event(void *server, grpc_cq_completion *completion) { |
|
|
|
static void done_shutdown_event(void *server, grpc_cq_completion *completion, |
|
|
|
server_unref(server); |
|
|
|
grpc_call_list *call_list) { |
|
|
|
|
|
|
|
server_unref(server, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static int num_channels(grpc_server *server) { |
|
|
|
static int num_channels(grpc_server *server) { |
|
|
@ -508,25 +514,27 @@ static int num_channels(grpc_server *server) { |
|
|
|
return n; |
|
|
|
return n; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void kill_pending_work_locked(grpc_server *server) { |
|
|
|
static void kill_pending_work_locked(grpc_server *server, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
registered_method *rm; |
|
|
|
registered_method *rm; |
|
|
|
request_matcher_kill_requests(server, &server->unregistered_request_matcher); |
|
|
|
request_matcher_kill_requests(server, &server->unregistered_request_matcher, |
|
|
|
|
|
|
|
call_list); |
|
|
|
request_matcher_zombify_all_pending_calls( |
|
|
|
request_matcher_zombify_all_pending_calls( |
|
|
|
&server->unregistered_request_matcher, server->workqueue); |
|
|
|
&server->unregistered_request_matcher, call_list); |
|
|
|
for (rm = server->registered_methods; rm; rm = rm->next) { |
|
|
|
for (rm = server->registered_methods; rm; rm = rm->next) { |
|
|
|
request_matcher_kill_requests(server, &rm->request_matcher); |
|
|
|
request_matcher_kill_requests(server, &rm->request_matcher, call_list); |
|
|
|
request_matcher_zombify_all_pending_calls(&rm->request_matcher, |
|
|
|
request_matcher_zombify_all_pending_calls(&rm->request_matcher, call_list); |
|
|
|
server->workqueue); |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void maybe_finish_shutdown(grpc_server *server) { |
|
|
|
static void maybe_finish_shutdown(grpc_server *server, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
size_t i; |
|
|
|
size_t i; |
|
|
|
if (!gpr_atm_acq_load(&server->shutdown_flag) || server->shutdown_published) { |
|
|
|
if (!gpr_atm_acq_load(&server->shutdown_flag) || server->shutdown_published) { |
|
|
|
return; |
|
|
|
return; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
kill_pending_work_locked(server); |
|
|
|
kill_pending_work_locked(server, call_list); |
|
|
|
|
|
|
|
|
|
|
|
if (server->root_channel_data.next != &server->root_channel_data || |
|
|
|
if (server->root_channel_data.next != &server->root_channel_data || |
|
|
|
server->listeners_destroyed < num_listeners(server)) { |
|
|
|
server->listeners_destroyed < num_listeners(server)) { |
|
|
@ -548,7 +556,7 @@ static void maybe_finish_shutdown(grpc_server *server) { |
|
|
|
server_ref(server); |
|
|
|
server_ref(server); |
|
|
|
grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, 1, |
|
|
|
grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, 1, |
|
|
|
done_shutdown_event, server, |
|
|
|
done_shutdown_event, server, |
|
|
|
&server->shutdown_tags[i].completion); |
|
|
|
&server->shutdown_tags[i].completion, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -566,10 +574,9 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { |
|
|
|
return md; |
|
|
|
return md; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void server_on_recv(void *ptr, int success) { |
|
|
|
static void server_on_recv(void *ptr, int success, grpc_call_list *call_list) { |
|
|
|
grpc_call_element *elem = ptr; |
|
|
|
grpc_call_element *elem = ptr; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
|
|
|
gpr_timespec op_deadline; |
|
|
|
gpr_timespec op_deadline; |
|
|
|
|
|
|
|
|
|
|
|
if (success && !calld->got_initial_metadata) { |
|
|
|
if (success && !calld->got_initial_metadata) { |
|
|
@ -587,7 +594,7 @@ static void server_on_recv(void *ptr, int success) { |
|
|
|
} |
|
|
|
} |
|
|
|
if (calld->host && calld->path) { |
|
|
|
if (calld->host && calld->path) { |
|
|
|
calld->got_initial_metadata = 1; |
|
|
|
calld->got_initial_metadata = 1; |
|
|
|
start_new_rpc(elem); |
|
|
|
start_new_rpc(elem, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
break; |
|
|
|
break; |
|
|
|
} |
|
|
|
} |
|
|
@ -604,8 +611,7 @@ static void server_on_recv(void *ptr, int success) { |
|
|
|
calld->state = ZOMBIED; |
|
|
|
calld->state = ZOMBIED; |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); |
|
|
|
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); |
|
|
|
grpc_workqueue_push(chand->server->workqueue, |
|
|
|
grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1); |
|
|
|
&calld->kill_zombie_closure, 1); |
|
|
|
|
|
|
|
} else { |
|
|
|
} else { |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
} |
|
|
|
} |
|
|
@ -616,8 +622,7 @@ static void server_on_recv(void *ptr, int success) { |
|
|
|
calld->state = ZOMBIED; |
|
|
|
calld->state = ZOMBIED; |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); |
|
|
|
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); |
|
|
|
grpc_workqueue_push(chand->server->workqueue, |
|
|
|
grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1); |
|
|
|
&calld->kill_zombie_closure, 1); |
|
|
|
|
|
|
|
} else if (calld->state == PENDING) { |
|
|
|
} else if (calld->state == PENDING) { |
|
|
|
calld->state = ZOMBIED; |
|
|
|
calld->state = ZOMBIED; |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
@ -629,7 +634,7 @@ static void server_on_recv(void *ptr, int success) { |
|
|
|
break; |
|
|
|
break; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); |
|
|
|
calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void server_mutate_op(grpc_call_element *elem, |
|
|
|
static void server_mutate_op(grpc_call_element *elem, |
|
|
@ -646,10 +651,11 @@ static void server_mutate_op(grpc_call_element *elem, |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void server_start_transport_stream_op(grpc_call_element *elem, |
|
|
|
static void server_start_transport_stream_op(grpc_call_element *elem, |
|
|
|
grpc_transport_stream_op *op) { |
|
|
|
grpc_transport_stream_op *op, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
GRPC_CALL_LOG_OP(GPR_INFO, elem, op); |
|
|
|
GRPC_CALL_LOG_OP(GPR_INFO, elem, op); |
|
|
|
server_mutate_op(elem, op); |
|
|
|
server_mutate_op(elem, op); |
|
|
|
grpc_call_next_op(elem, op); |
|
|
|
grpc_call_next_op(elem, op, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void accept_stream(void *cd, grpc_transport *transport, |
|
|
|
static void accept_stream(void *cd, grpc_transport *transport, |
|
|
@ -660,7 +666,8 @@ static void accept_stream(void *cd, grpc_transport *transport, |
|
|
|
0, gpr_inf_future(GPR_CLOCK_MONOTONIC)); |
|
|
|
0, gpr_inf_future(GPR_CLOCK_MONOTONIC)); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void channel_connectivity_changed(void *cd, int iomgr_status_ignored) { |
|
|
|
static void channel_connectivity_changed(void *cd, int iomgr_status_ignored, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
channel_data *chand = cd; |
|
|
|
channel_data *chand = cd; |
|
|
|
grpc_server *server = chand->server; |
|
|
|
grpc_server *server = chand->server; |
|
|
|
if (chand->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) { |
|
|
|
if (chand->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) { |
|
|
@ -670,18 +677,19 @@ static void channel_connectivity_changed(void *cd, int iomgr_status_ignored) { |
|
|
|
op.connectivity_state = &chand->connectivity_state; |
|
|
|
op.connectivity_state = &chand->connectivity_state; |
|
|
|
grpc_channel_next_op(grpc_channel_stack_element( |
|
|
|
grpc_channel_next_op(grpc_channel_stack_element( |
|
|
|
grpc_channel_get_channel_stack(chand->channel), 0), |
|
|
|
grpc_channel_get_channel_stack(chand->channel), 0), |
|
|
|
&op); |
|
|
|
&op, call_list); |
|
|
|
} else { |
|
|
|
} else { |
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
destroy_channel(chand); |
|
|
|
destroy_channel(chand, call_list); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "connectivity"); |
|
|
|
GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "connectivity", call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void init_call_elem(grpc_call_element *elem, |
|
|
|
static void init_call_elem(grpc_call_element *elem, |
|
|
|
const void *server_transport_data, |
|
|
|
const void *server_transport_data, |
|
|
|
grpc_transport_stream_op *initial_op) { |
|
|
|
grpc_transport_stream_op *initial_op, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
memset(calld, 0, sizeof(call_data)); |
|
|
|
memset(calld, 0, sizeof(call_data)); |
|
|
@ -696,7 +704,8 @@ static void init_call_elem(grpc_call_element *elem, |
|
|
|
if (initial_op) server_mutate_op(elem, initial_op); |
|
|
|
if (initial_op) server_mutate_op(elem, initial_op); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void destroy_call_elem(grpc_call_element *elem) { |
|
|
|
static void destroy_call_elem(grpc_call_element *elem, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
|
|
|
|
|
|
@ -711,13 +720,13 @@ static void destroy_call_elem(grpc_call_element *elem) { |
|
|
|
|
|
|
|
|
|
|
|
gpr_mu_destroy(&calld->mu_state); |
|
|
|
gpr_mu_destroy(&calld->mu_state); |
|
|
|
|
|
|
|
|
|
|
|
server_unref(chand->server); |
|
|
|
server_unref(chand->server, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, |
|
|
|
static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, |
|
|
|
const grpc_channel_args *args, |
|
|
|
const grpc_channel_args *args, |
|
|
|
grpc_mdctx *metadata_context, int is_first, |
|
|
|
grpc_mdctx *metadata_context, int is_first, |
|
|
|
int is_last) { |
|
|
|
int is_last, grpc_call_list *call_list) { |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
GPR_ASSERT(is_first); |
|
|
|
GPR_ASSERT(is_first); |
|
|
|
GPR_ASSERT(!is_last); |
|
|
|
GPR_ASSERT(!is_last); |
|
|
@ -733,7 +742,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, |
|
|
|
channel_connectivity_changed, chand); |
|
|
|
channel_connectivity_changed, chand); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void destroy_channel_elem(grpc_channel_element *elem) { |
|
|
|
static void destroy_channel_elem(grpc_channel_element *elem, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
size_t i; |
|
|
|
size_t i; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
if (chand->registered_methods) { |
|
|
|
if (chand->registered_methods) { |
|
|
@ -752,11 +762,11 @@ static void destroy_channel_elem(grpc_channel_element *elem) { |
|
|
|
chand->next->prev = chand->prev; |
|
|
|
chand->next->prev = chand->prev; |
|
|
|
chand->prev->next = chand->next; |
|
|
|
chand->prev->next = chand->next; |
|
|
|
chand->next = chand->prev = chand; |
|
|
|
chand->next = chand->prev = chand; |
|
|
|
maybe_finish_shutdown(chand->server); |
|
|
|
maybe_finish_shutdown(chand->server, call_list); |
|
|
|
gpr_mu_unlock(&chand->server->mu_global); |
|
|
|
gpr_mu_unlock(&chand->server->mu_global); |
|
|
|
GRPC_MDSTR_UNREF(chand->path_key); |
|
|
|
GRPC_MDSTR_UNREF(chand->path_key); |
|
|
|
GRPC_MDSTR_UNREF(chand->authority_key); |
|
|
|
GRPC_MDSTR_UNREF(chand->authority_key); |
|
|
|
server_unref(chand->server); |
|
|
|
server_unref(chand->server, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -810,7 +820,6 @@ grpc_server *grpc_server_create_from_filters( |
|
|
|
gpr_ref_init(&server->internal_refcount, 1); |
|
|
|
gpr_ref_init(&server->internal_refcount, 1); |
|
|
|
server->root_channel_data.next = server->root_channel_data.prev = |
|
|
|
server->root_channel_data.next = server->root_channel_data.prev = |
|
|
|
&server->root_channel_data; |
|
|
|
&server->root_channel_data; |
|
|
|
server->workqueue = grpc_workqueue_create(); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/* TODO(ctiller): expose a channel_arg for this */ |
|
|
|
/* TODO(ctiller): expose a channel_arg for this */ |
|
|
|
server->max_requested_calls = 32768; |
|
|
|
server->max_requested_calls = 32768; |
|
|
@ -881,23 +890,26 @@ void *grpc_server_register_method(grpc_server *server, const char *method, |
|
|
|
void grpc_server_start(grpc_server *server) { |
|
|
|
void grpc_server_start(grpc_server *server) { |
|
|
|
listener *l; |
|
|
|
listener *l; |
|
|
|
size_t i; |
|
|
|
size_t i; |
|
|
|
|
|
|
|
grpc_call_list call_list = GRPC_CALL_LIST_INIT; |
|
|
|
|
|
|
|
|
|
|
|
server->pollsets = gpr_malloc(sizeof(grpc_pollset *) * server->cq_count); |
|
|
|
server->pollsets = gpr_malloc(sizeof(grpc_pollset *) * server->cq_count); |
|
|
|
for (i = 0; i < server->cq_count; i++) { |
|
|
|
for (i = 0; i < server->cq_count; i++) { |
|
|
|
server->pollsets[i] = grpc_cq_pollset(server->cqs[i]); |
|
|
|
server->pollsets[i] = grpc_cq_pollset(server->cqs[i]); |
|
|
|
grpc_workqueue_add_to_pollset(server->workqueue, server->pollsets[i]); |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
for (l = server->listeners; l; l = l->next) { |
|
|
|
for (l = server->listeners; l; l = l->next) { |
|
|
|
l->start(server, l->arg, server->pollsets, server->cq_count); |
|
|
|
l->start(server, l->arg, server->pollsets, server->cq_count, &call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
grpc_call_list_run(&call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport, |
|
|
|
void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport, |
|
|
|
grpc_channel_filter const **extra_filters, |
|
|
|
grpc_channel_filter const **extra_filters, |
|
|
|
size_t num_extra_filters, grpc_mdctx *mdctx, |
|
|
|
size_t num_extra_filters, grpc_mdctx *mdctx, |
|
|
|
grpc_workqueue *workqueue, |
|
|
|
grpc_workqueue *workqueue, |
|
|
|
const grpc_channel_args *args) { |
|
|
|
const grpc_channel_args *args, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
size_t num_filters = s->channel_filter_count + num_extra_filters + 1; |
|
|
|
size_t num_filters = s->channel_filter_count + num_extra_filters + 1; |
|
|
|
grpc_channel_filter const **filters = |
|
|
|
grpc_channel_filter const **filters = |
|
|
|
gpr_malloc(sizeof(grpc_channel_filter *) * num_filters); |
|
|
|
gpr_malloc(sizeof(grpc_channel_filter *) * num_filters); |
|
|
@ -927,11 +939,11 @@ void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport, |
|
|
|
for (i = 0; i < s->cq_count; i++) { |
|
|
|
for (i = 0; i < s->cq_count; i++) { |
|
|
|
memset(&op, 0, sizeof(op)); |
|
|
|
memset(&op, 0, sizeof(op)); |
|
|
|
op.bind_pollset = grpc_cq_pollset(s->cqs[i]); |
|
|
|
op.bind_pollset = grpc_cq_pollset(s->cqs[i]); |
|
|
|
grpc_transport_perform_op(transport, &op); |
|
|
|
grpc_transport_perform_op(transport, &op, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
channel = grpc_channel_create_from_filters(NULL, filters, num_filters, args, |
|
|
|
channel = grpc_channel_create_from_filters(NULL, filters, num_filters, args, |
|
|
|
mdctx, workqueue, 0); |
|
|
|
mdctx, workqueue, 0, call_list); |
|
|
|
chand = (channel_data *)grpc_channel_stack_element( |
|
|
|
chand = (channel_data *)grpc_channel_stack_element( |
|
|
|
grpc_channel_get_channel_stack(channel), 0) |
|
|
|
grpc_channel_get_channel_stack(channel), 0) |
|
|
|
->channel_data; |
|
|
|
->channel_data; |
|
|
@ -987,19 +999,30 @@ void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport, |
|
|
|
op.on_connectivity_state_change = &chand->channel_connectivity_changed; |
|
|
|
op.on_connectivity_state_change = &chand->channel_connectivity_changed; |
|
|
|
op.connectivity_state = &chand->connectivity_state; |
|
|
|
op.connectivity_state = &chand->connectivity_state; |
|
|
|
op.disconnect = gpr_atm_acq_load(&s->shutdown_flag) != 0; |
|
|
|
op.disconnect = gpr_atm_acq_load(&s->shutdown_flag) != 0; |
|
|
|
grpc_transport_perform_op(transport, &op); |
|
|
|
grpc_transport_perform_op(transport, &op, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
void done_published_shutdown(void *done_arg, grpc_cq_completion *storage) { |
|
|
|
void done_published_shutdown(void *done_arg, grpc_cq_completion *storage, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
(void) done_arg; |
|
|
|
(void) done_arg; |
|
|
|
gpr_free(storage); |
|
|
|
gpr_free(storage); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
static void listener_destroy_done(void *s, int success, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
|
|
|
|
grpc_server *server = s; |
|
|
|
|
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
|
|
|
|
server->listeners_destroyed++; |
|
|
|
|
|
|
|
maybe_finish_shutdown(server, call_list); |
|
|
|
|
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
void grpc_server_shutdown_and_notify(grpc_server *server, |
|
|
|
void grpc_server_shutdown_and_notify(grpc_server *server, |
|
|
|
grpc_completion_queue *cq, void *tag) { |
|
|
|
grpc_completion_queue *cq, void *tag) { |
|
|
|
listener *l; |
|
|
|
listener *l; |
|
|
|
shutdown_tag *sdt; |
|
|
|
shutdown_tag *sdt; |
|
|
|
channel_broadcaster broadcaster; |
|
|
|
channel_broadcaster broadcaster; |
|
|
|
|
|
|
|
grpc_call_list call_list = GRPC_CALL_LIST_INIT; |
|
|
|
|
|
|
|
|
|
|
|
GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag); |
|
|
|
GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag); |
|
|
|
|
|
|
|
|
|
|
@ -1008,9 +1031,9 @@ void grpc_server_shutdown_and_notify(grpc_server *server, |
|
|
|
grpc_cq_begin_op(cq); |
|
|
|
grpc_cq_begin_op(cq); |
|
|
|
if (server->shutdown_published) { |
|
|
|
if (server->shutdown_published) { |
|
|
|
grpc_cq_end_op(cq, tag, 1, done_published_shutdown, NULL, |
|
|
|
grpc_cq_end_op(cq, tag, 1, done_published_shutdown, NULL, |
|
|
|
gpr_malloc(sizeof(grpc_cq_completion))); |
|
|
|
gpr_malloc(sizeof(grpc_cq_completion)), &call_list); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
return; |
|
|
|
goto done; |
|
|
|
} |
|
|
|
} |
|
|
|
server->shutdown_tags = |
|
|
|
server->shutdown_tags = |
|
|
|
gpr_realloc(server->shutdown_tags, |
|
|
|
gpr_realloc(server->shutdown_tags, |
|
|
@ -1020,7 +1043,7 @@ void grpc_server_shutdown_and_notify(grpc_server *server, |
|
|
|
sdt->cq = cq; |
|
|
|
sdt->cq = cq; |
|
|
|
if (gpr_atm_acq_load(&server->shutdown_flag)) { |
|
|
|
if (gpr_atm_acq_load(&server->shutdown_flag)) { |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
return; |
|
|
|
goto done; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME); |
|
|
|
server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME); |
|
|
@ -1029,41 +1052,40 @@ void grpc_server_shutdown_and_notify(grpc_server *server, |
|
|
|
|
|
|
|
|
|
|
|
/* collect all unregistered then registered calls */ |
|
|
|
/* collect all unregistered then registered calls */ |
|
|
|
gpr_mu_lock(&server->mu_call); |
|
|
|
gpr_mu_lock(&server->mu_call); |
|
|
|
kill_pending_work_locked(server); |
|
|
|
kill_pending_work_locked(server, &call_list); |
|
|
|
gpr_mu_unlock(&server->mu_call); |
|
|
|
gpr_mu_unlock(&server->mu_call); |
|
|
|
|
|
|
|
|
|
|
|
gpr_atm_rel_store(&server->shutdown_flag, 1); |
|
|
|
gpr_atm_rel_store(&server->shutdown_flag, 1); |
|
|
|
maybe_finish_shutdown(server); |
|
|
|
maybe_finish_shutdown(server, &call_list); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
|
|
|
|
|
|
|
|
/* Shutdown listeners */ |
|
|
|
/* Shutdown listeners */ |
|
|
|
for (l = server->listeners; l; l = l->next) { |
|
|
|
for (l = server->listeners; l; l = l->next) { |
|
|
|
l->destroy(server, l->arg); |
|
|
|
grpc_closure_init(&l->destroy_done, listener_destroy_done, server); |
|
|
|
|
|
|
|
l->destroy(server, l->arg, &l->destroy_done, &call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
channel_broadcaster_shutdown(&broadcaster, 1, 0); |
|
|
|
channel_broadcaster_shutdown(&broadcaster, 1, 0, &call_list); |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
void grpc_server_listener_destroy_done(void *s) { |
|
|
|
done: |
|
|
|
grpc_server *server = s; |
|
|
|
grpc_call_list_run(&call_list); |
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
|
|
|
|
server->listeners_destroyed++; |
|
|
|
|
|
|
|
maybe_finish_shutdown(server); |
|
|
|
|
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
void grpc_server_cancel_all_calls(grpc_server *server) { |
|
|
|
void grpc_server_cancel_all_calls(grpc_server *server) { |
|
|
|
channel_broadcaster broadcaster; |
|
|
|
channel_broadcaster broadcaster; |
|
|
|
|
|
|
|
grpc_call_list call_list = GRPC_CALL_LIST_INIT; |
|
|
|
|
|
|
|
|
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
channel_broadcaster_init(server, &broadcaster); |
|
|
|
channel_broadcaster_init(server, &broadcaster); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
|
|
|
|
|
|
|
|
channel_broadcaster_shutdown(&broadcaster, 0, 1); |
|
|
|
channel_broadcaster_shutdown(&broadcaster, 0, 1, &call_list); |
|
|
|
|
|
|
|
grpc_call_list_run(&call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
void grpc_server_destroy(grpc_server *server) { |
|
|
|
void grpc_server_destroy(grpc_server *server) { |
|
|
|
listener *l; |
|
|
|
listener *l; |
|
|
|
|
|
|
|
grpc_call_list call_list = GRPC_CALL_LIST_INIT; |
|
|
|
|
|
|
|
|
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
gpr_mu_lock(&server->mu_global); |
|
|
|
GPR_ASSERT(gpr_atm_acq_load(&server->shutdown_flag) || !server->listeners); |
|
|
|
GPR_ASSERT(gpr_atm_acq_load(&server->shutdown_flag) || !server->listeners); |
|
|
@ -1077,16 +1099,17 @@ void grpc_server_destroy(grpc_server *server) { |
|
|
|
|
|
|
|
|
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
gpr_mu_unlock(&server->mu_global); |
|
|
|
|
|
|
|
|
|
|
|
grpc_workqueue_flush(server->workqueue); |
|
|
|
server_unref(server, &call_list); |
|
|
|
|
|
|
|
grpc_call_list_run(&call_list); |
|
|
|
server_unref(server); |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
void grpc_server_add_listener(grpc_server *server, void *arg, |
|
|
|
void grpc_server_add_listener( |
|
|
|
void (*start)(grpc_server *server, void *arg, |
|
|
|
grpc_server *server, void *arg, |
|
|
|
grpc_pollset **pollsets, |
|
|
|
void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets, |
|
|
|
size_t pollset_count), |
|
|
|
size_t pollset_count, grpc_call_list *call_list), |
|
|
|
void (*destroy)(grpc_server *server, void *arg)) { |
|
|
|
void (*destroy)(grpc_server *server, void *arg, grpc_closure *on_done, |
|
|
|
|
|
|
|
grpc_call_list *call_list), |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
listener *l = gpr_malloc(sizeof(listener)); |
|
|
|
listener *l = gpr_malloc(sizeof(listener)); |
|
|
|
l->arg = arg; |
|
|
|
l->arg = arg; |
|
|
|
l->start = start; |
|
|
|
l->start = start; |
|
|
@ -1096,18 +1119,19 @@ void grpc_server_add_listener(grpc_server *server, void *arg, |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static grpc_call_error queue_call_request(grpc_server *server, |
|
|
|
static grpc_call_error queue_call_request(grpc_server *server, |
|
|
|
requested_call *rc) { |
|
|
|
requested_call *rc, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
call_data *calld = NULL; |
|
|
|
call_data *calld = NULL; |
|
|
|
request_matcher *request_matcher = NULL; |
|
|
|
request_matcher *request_matcher = NULL; |
|
|
|
int request_id; |
|
|
|
int request_id; |
|
|
|
if (gpr_atm_acq_load(&server->shutdown_flag)) { |
|
|
|
if (gpr_atm_acq_load(&server->shutdown_flag)) { |
|
|
|
fail_call(server, rc); |
|
|
|
fail_call(server, rc, call_list); |
|
|
|
return GRPC_CALL_OK; |
|
|
|
return GRPC_CALL_OK; |
|
|
|
} |
|
|
|
} |
|
|
|
request_id = gpr_stack_lockfree_pop(server->request_freelist); |
|
|
|
request_id = gpr_stack_lockfree_pop(server->request_freelist); |
|
|
|
if (request_id == -1) { |
|
|
|
if (request_id == -1) { |
|
|
|
/* out of request ids: just fail this one */ |
|
|
|
/* out of request ids: just fail this one */ |
|
|
|
fail_call(server, rc); |
|
|
|
fail_call(server, rc, call_list); |
|
|
|
return GRPC_CALL_OK; |
|
|
|
return GRPC_CALL_OK; |
|
|
|
} |
|
|
|
} |
|
|
|
switch (rc->type) { |
|
|
|
switch (rc->type) { |
|
|
@ -1135,12 +1159,13 @@ static grpc_call_error queue_call_request(grpc_server *server, |
|
|
|
grpc_closure_init( |
|
|
|
grpc_closure_init( |
|
|
|
&calld->kill_zombie_closure, kill_zombie, |
|
|
|
&calld->kill_zombie_closure, kill_zombie, |
|
|
|
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0)); |
|
|
|
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0)); |
|
|
|
grpc_workqueue_push(server->workqueue, &calld->kill_zombie_closure, 1); |
|
|
|
grpc_call_list_add(call_list, &calld->kill_zombie_closure, 1); |
|
|
|
} else { |
|
|
|
} else { |
|
|
|
GPR_ASSERT(calld->state == PENDING); |
|
|
|
GPR_ASSERT(calld->state == PENDING); |
|
|
|
calld->state = ACTIVATED; |
|
|
|
calld->state = ACTIVATED; |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
gpr_mu_unlock(&calld->mu_state); |
|
|
|
begin_call(server, calld, &server->requested_calls[request_id]); |
|
|
|
begin_call(server, calld, &server->requested_calls[request_id], |
|
|
|
|
|
|
|
call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
gpr_mu_lock(&server->mu_call); |
|
|
|
gpr_mu_lock(&server->mu_call); |
|
|
|
} |
|
|
|
} |
|
|
@ -1154,13 +1179,16 @@ grpc_call_error grpc_server_request_call( |
|
|
|
grpc_metadata_array *initial_metadata, |
|
|
|
grpc_metadata_array *initial_metadata, |
|
|
|
grpc_completion_queue *cq_bound_to_call, |
|
|
|
grpc_completion_queue *cq_bound_to_call, |
|
|
|
grpc_completion_queue *cq_for_notification, void *tag) { |
|
|
|
grpc_completion_queue *cq_for_notification, void *tag) { |
|
|
|
|
|
|
|
grpc_call_error error; |
|
|
|
|
|
|
|
grpc_call_list call_list = GRPC_CALL_LIST_INIT; |
|
|
|
requested_call *rc = gpr_malloc(sizeof(*rc)); |
|
|
|
requested_call *rc = gpr_malloc(sizeof(*rc)); |
|
|
|
GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details, |
|
|
|
GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details, |
|
|
|
initial_metadata, cq_bound_to_call, |
|
|
|
initial_metadata, cq_bound_to_call, |
|
|
|
cq_for_notification, tag); |
|
|
|
cq_for_notification, tag); |
|
|
|
if (!grpc_cq_is_server_cq(cq_for_notification)) { |
|
|
|
if (!grpc_cq_is_server_cq(cq_for_notification)) { |
|
|
|
gpr_free(rc); |
|
|
|
gpr_free(rc); |
|
|
|
return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; |
|
|
|
error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; |
|
|
|
|
|
|
|
goto done; |
|
|
|
} |
|
|
|
} |
|
|
|
grpc_cq_begin_op(cq_for_notification); |
|
|
|
grpc_cq_begin_op(cq_for_notification); |
|
|
|
details->reserved = NULL; |
|
|
|
details->reserved = NULL; |
|
|
@ -1172,7 +1200,10 @@ grpc_call_error grpc_server_request_call( |
|
|
|
rc->call = call; |
|
|
|
rc->call = call; |
|
|
|
rc->data.batch.details = details; |
|
|
|
rc->data.batch.details = details; |
|
|
|
rc->data.batch.initial_metadata = initial_metadata; |
|
|
|
rc->data.batch.initial_metadata = initial_metadata; |
|
|
|
return queue_call_request(server, rc); |
|
|
|
error = queue_call_request(server, rc, &call_list); |
|
|
|
|
|
|
|
done: |
|
|
|
|
|
|
|
grpc_call_list_run(&call_list); |
|
|
|
|
|
|
|
return error; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
grpc_call_error grpc_server_request_registered_call( |
|
|
|
grpc_call_error grpc_server_request_registered_call( |
|
|
@ -1180,11 +1211,14 @@ grpc_call_error grpc_server_request_registered_call( |
|
|
|
grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload, |
|
|
|
grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload, |
|
|
|
grpc_completion_queue *cq_bound_to_call, |
|
|
|
grpc_completion_queue *cq_bound_to_call, |
|
|
|
grpc_completion_queue *cq_for_notification, void *tag) { |
|
|
|
grpc_completion_queue *cq_for_notification, void *tag) { |
|
|
|
|
|
|
|
grpc_call_error error; |
|
|
|
|
|
|
|
grpc_call_list call_list = GRPC_CALL_LIST_INIT; |
|
|
|
requested_call *rc = gpr_malloc(sizeof(*rc)); |
|
|
|
requested_call *rc = gpr_malloc(sizeof(*rc)); |
|
|
|
registered_method *registered_method = rm; |
|
|
|
registered_method *registered_method = rm; |
|
|
|
if (!grpc_cq_is_server_cq(cq_for_notification)) { |
|
|
|
if (!grpc_cq_is_server_cq(cq_for_notification)) { |
|
|
|
gpr_free(rc); |
|
|
|
gpr_free(rc); |
|
|
|
return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; |
|
|
|
error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; |
|
|
|
|
|
|
|
goto done; |
|
|
|
} |
|
|
|
} |
|
|
|
grpc_cq_begin_op(cq_for_notification); |
|
|
|
grpc_cq_begin_op(cq_for_notification); |
|
|
|
rc->type = REGISTERED_CALL; |
|
|
|
rc->type = REGISTERED_CALL; |
|
|
@ -1197,12 +1231,16 @@ grpc_call_error grpc_server_request_registered_call( |
|
|
|
rc->data.registered.deadline = deadline; |
|
|
|
rc->data.registered.deadline = deadline; |
|
|
|
rc->data.registered.initial_metadata = initial_metadata; |
|
|
|
rc->data.registered.initial_metadata = initial_metadata; |
|
|
|
rc->data.registered.optional_payload = optional_payload; |
|
|
|
rc->data.registered.optional_payload = optional_payload; |
|
|
|
return queue_call_request(server, rc); |
|
|
|
error = queue_call_request(server, rc, &call_list); |
|
|
|
|
|
|
|
done: |
|
|
|
|
|
|
|
grpc_call_list_run(&call_list); |
|
|
|
|
|
|
|
return error; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void publish_registered_or_batch(grpc_call *call, int success, |
|
|
|
static void publish_registered_or_batch(grpc_call *call, int success, void *tag, |
|
|
|
void *tag); |
|
|
|
grpc_call_list *call_list); |
|
|
|
static void publish_was_not_set(grpc_call *call, int success, void *tag) { |
|
|
|
static void publish_was_not_set(grpc_call *call, int success, void *tag, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
abort(); |
|
|
|
abort(); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -1218,7 +1256,7 @@ static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void begin_call(grpc_server *server, call_data *calld, |
|
|
|
static void begin_call(grpc_server *server, call_data *calld, |
|
|
|
requested_call *rc) { |
|
|
|
requested_call *rc, grpc_call_list *call_list) { |
|
|
|
grpc_ioreq_completion_func publish = publish_was_not_set; |
|
|
|
grpc_ioreq_completion_func publish = publish_was_not_set; |
|
|
|
grpc_ioreq req[2]; |
|
|
|
grpc_ioreq req[2]; |
|
|
|
grpc_ioreq *r = req; |
|
|
|
grpc_ioreq *r = req; |
|
|
@ -1229,7 +1267,7 @@ static void begin_call(grpc_server *server, call_data *calld, |
|
|
|
fill in the metadata array passed by the client, we need to perform |
|
|
|
fill in the metadata array passed by the client, we need to perform |
|
|
|
an ioreq op, that should complete immediately. */ |
|
|
|
an ioreq op, that should complete immediately. */ |
|
|
|
|
|
|
|
|
|
|
|
grpc_call_set_completion_queue(calld->call, rc->cq_bound_to_call); |
|
|
|
grpc_call_set_completion_queue(calld->call, rc->cq_bound_to_call, call_list); |
|
|
|
*rc->call = calld->call; |
|
|
|
*rc->call = calld->call; |
|
|
|
calld->cq_new = rc->cq_for_notification; |
|
|
|
calld->cq_new = rc->cq_for_notification; |
|
|
|
switch (rc->type) { |
|
|
|
switch (rc->type) { |
|
|
@ -1265,10 +1303,11 @@ static void begin_call(grpc_server *server, call_data *calld, |
|
|
|
|
|
|
|
|
|
|
|
GRPC_CALL_INTERNAL_REF(calld->call, "server"); |
|
|
|
GRPC_CALL_INTERNAL_REF(calld->call, "server"); |
|
|
|
grpc_call_start_ioreq_and_call_back(calld->call, req, (size_t)(r - req), |
|
|
|
grpc_call_start_ioreq_and_call_back(calld->call, req, (size_t)(r - req), |
|
|
|
publish, rc); |
|
|
|
publish, rc, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void done_request_event(void *req, grpc_cq_completion *c) { |
|
|
|
static void done_request_event(void *req, grpc_cq_completion *c, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
requested_call *rc = req; |
|
|
|
requested_call *rc = req; |
|
|
|
grpc_server *server = rc->server; |
|
|
|
grpc_server *server = rc->server; |
|
|
|
|
|
|
|
|
|
|
@ -1281,10 +1320,11 @@ static void done_request_event(void *req, grpc_cq_completion *c) { |
|
|
|
gpr_free(req); |
|
|
|
gpr_free(req); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
server_unref(server); |
|
|
|
server_unref(server, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void fail_call(grpc_server *server, requested_call *rc) { |
|
|
|
static void fail_call(grpc_server *server, requested_call *rc, |
|
|
|
|
|
|
|
grpc_call_list *call_list) { |
|
|
|
*rc->call = NULL; |
|
|
|
*rc->call = NULL; |
|
|
|
switch (rc->type) { |
|
|
|
switch (rc->type) { |
|
|
|
case BATCH_CALL: |
|
|
|
case BATCH_CALL: |
|
|
@ -1296,11 +1336,11 @@ static void fail_call(grpc_server *server, requested_call *rc) { |
|
|
|
} |
|
|
|
} |
|
|
|
server_ref(server); |
|
|
|
server_ref(server); |
|
|
|
grpc_cq_end_op(rc->cq_for_notification, rc->tag, 0, done_request_event, rc, |
|
|
|
grpc_cq_end_op(rc->cq_for_notification, rc->tag, 0, done_request_event, rc, |
|
|
|
&rc->completion); |
|
|
|
&rc->completion, call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
static void publish_registered_or_batch(grpc_call *call, int success, |
|
|
|
static void publish_registered_or_batch(grpc_call *call, int success, void *prc, |
|
|
|
void *prc) { |
|
|
|
grpc_call_list *call_list) { |
|
|
|
grpc_call_element *elem = |
|
|
|
grpc_call_element *elem = |
|
|
|
grpc_call_stack_element(grpc_call_get_call_stack(call), 0); |
|
|
|
grpc_call_stack_element(grpc_call_get_call_stack(call), 0); |
|
|
|
requested_call *rc = prc; |
|
|
|
requested_call *rc = prc; |
|
|
@ -1308,7 +1348,7 @@ static void publish_registered_or_batch(grpc_call *call, int success, |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
server_ref(chand->server); |
|
|
|
server_ref(chand->server); |
|
|
|
grpc_cq_end_op(calld->cq_new, rc->tag, success, done_request_event, rc, |
|
|
|
grpc_cq_end_op(calld->cq_new, rc->tag, success, done_request_event, rc, |
|
|
|
&rc->completion); |
|
|
|
&rc->completion, call_list); |
|
|
|
GRPC_CALL_INTERNAL_UNREF(call, "server", call_list); |
|
|
|
GRPC_CALL_INTERNAL_UNREF(call, "server", call_list); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|