aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/surface
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/surface')
-rw-r--r--src/core/surface/server.c120
1 files changed, 9 insertions, 111 deletions
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index f29d47c17c..341ca2942c 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -51,7 +51,7 @@
#include <grpc/support/string_util.h>
#include <grpc/support/useful.h>
-typedef enum { PENDING_START, ALL_CALLS, CALL_LIST_COUNT } call_list;
+typedef enum { PENDING_START, CALL_LIST_COUNT } call_list;
typedef struct listener {
void *arg;
@@ -114,7 +114,6 @@ typedef struct channel_registered_method {
struct channel_data {
grpc_server *server;
- size_t num_calls;
grpc_connectivity_state connectivity_state;
grpc_channel *channel;
grpc_mdstr *path_key;
@@ -204,9 +203,7 @@ struct call_data {
typedef struct {
grpc_channel **channels;
- grpc_channel **disconnects;
size_t num_channels;
- size_t num_disconnects;
} channel_broadcaster;
#define SERVER_FROM_CALL_ELEM(elem) \
@@ -225,26 +222,15 @@ static void maybe_finish_shutdown(grpc_server *server);
static void channel_broadcaster_init(grpc_server *s, channel_broadcaster *cb) {
channel_data *c;
size_t count = 0;
- size_t dc_count = 0;
for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
count++;
- if (c->num_calls == 0) {
- dc_count++;
- }
}
cb->num_channels = count;
- cb->num_disconnects = dc_count;
cb->channels = gpr_malloc(sizeof(*cb->channels) * cb->num_channels);
- cb->disconnects = gpr_malloc(sizeof(*cb->channels) * cb->num_disconnects);
count = 0;
- dc_count = 0;
for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
cb->channels[count++] = c->channel;
GRPC_CHANNEL_INTERNAL_REF(c->channel, "broadcast");
- if (c->num_calls == 0) {
- cb->disconnects[dc_count++] = c->channel;
- GRPC_CHANNEL_INTERNAL_REF(c->channel, "broadcast-disconnect");
- }
}
}
@@ -280,19 +266,14 @@ static void send_shutdown(grpc_channel *channel, int send_goaway,
}
static void channel_broadcaster_shutdown(channel_broadcaster *cb,
- int send_goaway, int send_disconnect) {
+ int send_goaway, int force_disconnect) {
size_t i;
for (i = 0; i < cb->num_channels; i++) {
- send_shutdown(cb->channels[i], 1, 0);
+ send_shutdown(cb->channels[i], send_goaway, force_disconnect);
GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast");
}
- for (i = 0; i < cb->num_disconnects; i++) {
- send_shutdown(cb->disconnects[i], 0, 1);
- GRPC_CHANNEL_INTERNAL_UNREF(cb->channels[i], "broadcast-disconnect");
- }
gpr_free(cb->channels);
- gpr_free(cb->disconnects);
}
/* call list */
@@ -501,15 +482,6 @@ static void maybe_finish_shutdown(grpc_server *server) {
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");
@@ -541,22 +513,10 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
return md;
}
-static int decrement_call_count(channel_data *chand) {
- int disconnect = 0;
- chand->num_calls--;
- if (0 == chand->num_calls && chand->server->shutdown) {
- disconnect = 1;
- }
- maybe_finish_shutdown(chand->server);
- return disconnect;
-}
-
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;
- int disconnect = 0;
if (success && !calld->got_initial_metadata) {
size_t i;
@@ -601,20 +561,7 @@ 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);
}
- 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) {
- disconnect = decrement_call_count(chand);
- if (disconnect) {
- GRPC_CHANNEL_INTERNAL_REF(chand->channel, "send-disconnect");
- }
- }
- gpr_mu_unlock(&chand->server->mu_global);
- if (disconnect) {
- send_shutdown(chand->channel, 0, 1);
- GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "send-disconnect");
- }
break;
}
@@ -679,14 +626,6 @@ 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_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_global);
-
server_ref(chand->server);
if (initial_op) server_mutate_op(elem, initial_op);
@@ -695,19 +634,13 @@ 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;
- int removed[CALL_LIST_COUNT];
size_t i;
gpr_mu_lock(&chand->server->mu_call);
for (i = 0; i < CALL_LIST_COUNT; i++) {
- removed[i] = call_list_remove(elem->call_data, 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);
- }
if (calld->host) {
grpc_mdstr_unref(calld->host);
@@ -727,7 +660,6 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
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");
@@ -1049,47 +981,13 @@ void grpc_server_listener_destroy_done(void *s) {
}
void grpc_server_cancel_all_calls(grpc_server *server) {
- call_data *calld;
- grpc_call **calls;
- size_t call_count;
- size_t call_capacity;
- int is_first = 1;
- size_t i;
-
- gpr_mu_lock(&server->mu_call);
-
- GPR_ASSERT(server->shutdown);
-
- if (!server->lists[ALL_CALLS]) {
- gpr_mu_unlock(&server->mu_call);
- return;
- }
-
- call_capacity = 8;
- 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) {
- if (call_count == call_capacity) {
- call_capacity *= 2;
- calls = gpr_realloc(calls, sizeof(grpc_call *) * call_capacity);
- }
- calls[call_count++] = calld->call;
- GRPC_CALL_INTERNAL_REF(calld->call, "cancel_all");
- is_first = 0;
- }
-
- gpr_mu_unlock(&server->mu_call);
+ channel_broadcaster broadcaster;
- for (i = 0; i < call_count; i++) {
- grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE,
- "Unavailable");
- GRPC_CALL_INTERNAL_UNREF(calls[i], "cancel_all", 1);
- }
+ gpr_mu_lock(&server->mu_global);
+ channel_broadcaster_init(server, &broadcaster);
+ gpr_mu_unlock(&server->mu_global);
- gpr_free(calls);
+ channel_broadcaster_shutdown(&broadcaster, 0, 1);
}
void grpc_server_destroy(grpc_server *server) {