aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/surface/server.cc
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib/surface/server.cc')
-rw-r--r--src/core/lib/surface/server.cc675
1 files changed, 308 insertions, 367 deletions
diff --git a/src/core/lib/surface/server.cc b/src/core/lib/surface/server.cc
index dd09cb91de..da7ae17bef 100644
--- a/src/core/lib/surface/server.cc
+++ b/src/core/lib/surface/server.cc
@@ -33,7 +33,8 @@
#include "src/core/lib/iomgr/executor.h"
#include "src/core/lib/iomgr/iomgr.h"
#include "src/core/lib/slice/slice_internal.h"
-#include "src/core/lib/support/stack_lockfree.h"
+#include "src/core/lib/support/mpscq.h"
+#include "src/core/lib/support/spinlock.h"
#include "src/core/lib/support/string.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/call.h"
@@ -44,12 +45,12 @@
#include "src/core/lib/transport/static_metadata.h"
typedef struct listener {
- void *arg;
- void (*start)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
- grpc_pollset **pollsets, size_t pollset_count);
- void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
- grpc_closure *closure);
- struct listener *next;
+ void* arg;
+ void (*start)(grpc_exec_ctx* exec_ctx, grpc_server* server, void* arg,
+ grpc_pollset** pollsets, size_t pollset_count);
+ void (*destroy)(grpc_exec_ctx* exec_ctx, grpc_server* server, void* arg,
+ grpc_closure* closure);
+ struct listener* next;
grpc_closure destroy_done;
} listener;
@@ -63,28 +64,29 @@ grpc_tracer_flag grpc_server_channel_trace =
GRPC_TRACER_INITIALIZER(false, "server_channel");
typedef struct requested_call {
+ gpr_mpscq_node request_link; /* must be first */
requested_call_type type;
size_t cq_idx;
- void *tag;
- grpc_server *server;
- grpc_completion_queue *cq_bound_to_call;
- grpc_call **call;
+ void* tag;
+ grpc_server* server;
+ grpc_completion_queue* cq_bound_to_call;
+ grpc_call** call;
grpc_cq_completion completion;
- grpc_metadata_array *initial_metadata;
+ grpc_metadata_array* initial_metadata;
union {
struct {
- grpc_call_details *details;
+ grpc_call_details* details;
} batch;
struct {
- registered_method *method;
- gpr_timespec *deadline;
- grpc_byte_buffer **optional_payload;
+ registered_method* method;
+ gpr_timespec* deadline;
+ grpc_byte_buffer** optional_payload;
} registered;
} data;
} requested_call;
typedef struct channel_registered_method {
- registered_method *server_registered_method;
+ registered_method* server_registered_method;
uint32_t flags;
bool has_host;
grpc_slice method;
@@ -92,14 +94,14 @@ typedef struct channel_registered_method {
} channel_registered_method;
struct channel_data {
- grpc_server *server;
+ grpc_server* server;
grpc_connectivity_state connectivity_state;
- grpc_channel *channel;
+ grpc_channel* channel;
size_t cq_idx;
/* linked list of all channels on a server */
- channel_data *next;
- channel_data *prev;
- channel_registered_method *registered_methods;
+ channel_data* next;
+ channel_data* prev;
+ channel_registered_method* registered_methods;
uint32_t registered_method_slots;
uint32_t registered_method_max_probes;
grpc_closure finish_destroy_channel_closure;
@@ -107,8 +109,8 @@ struct channel_data {
};
typedef struct shutdown_tag {
- void *tag;
- grpc_completion_queue *cq;
+ void* tag;
+ grpc_completion_queue* cq;
grpc_cq_completion completion;
} shutdown_tag;
@@ -126,12 +128,9 @@ typedef enum {
typedef struct request_matcher request_matcher;
struct call_data {
- grpc_call *call;
+ grpc_call* call;
- /** protects state */
- gpr_mu mu_state;
- /** the current state of a call - see call_state */
- call_state state;
+ gpr_atm state;
bool path_set;
bool host_set;
@@ -139,52 +138,52 @@ struct call_data {
grpc_slice host;
grpc_millis deadline;
- grpc_completion_queue *cq_new;
+ grpc_completion_queue* cq_new;
- grpc_metadata_batch *recv_initial_metadata;
+ grpc_metadata_batch* recv_initial_metadata;
uint32_t recv_initial_metadata_flags;
grpc_metadata_array initial_metadata;
- request_matcher *matcher;
- grpc_byte_buffer *payload;
+ request_matcher* matcher;
+ grpc_byte_buffer* payload;
grpc_closure got_initial_metadata;
grpc_closure server_on_recv_initial_metadata;
grpc_closure kill_zombie_closure;
- grpc_closure *on_done_recv_initial_metadata;
+ grpc_closure* on_done_recv_initial_metadata;
grpc_closure publish;
- call_data *pending_next;
+ call_data* pending_next;
};
struct request_matcher {
- grpc_server *server;
- call_data *pending_head;
- call_data *pending_tail;
- gpr_stack_lockfree **requests_per_cq;
+ grpc_server* server;
+ call_data* pending_head;
+ call_data* pending_tail;
+ gpr_locked_mpscq* requests_per_cq;
};
struct registered_method {
- char *method;
- char *host;
+ char* method;
+ char* host;
grpc_server_register_method_payload_handling payload_handling;
uint32_t flags;
/* one request matcher per method */
request_matcher matcher;
- registered_method *next;
+ registered_method* next;
};
typedef struct {
- grpc_channel **channels;
+ grpc_channel** channels;
size_t num_channels;
} channel_broadcaster;
struct grpc_server {
- grpc_channel_args *channel_args;
+ grpc_channel_args* channel_args;
- grpc_completion_queue **cqs;
- grpc_pollset **pollsets;
+ grpc_completion_queue** cqs;
+ grpc_pollset** pollsets;
size_t cq_count;
size_t pollset_count;
bool started;
@@ -204,23 +203,18 @@ struct grpc_server {
bool starting;
gpr_cv starting_cv;
- registered_method *registered_methods;
+ registered_method* registered_methods;
/** one request matcher for unregistered methods */
request_matcher unregistered_request_matcher;
- /** free list of available requested_calls_per_cq indices */
- gpr_stack_lockfree **request_freelist_per_cq;
- /** requested call backing data */
- requested_call **requested_calls_per_cq;
- int max_requested_calls_per_cq;
gpr_atm shutdown_flag;
uint8_t shutdown_published;
size_t num_shutdown_tags;
- shutdown_tag *shutdown_tags;
+ shutdown_tag* shutdown_tags;
channel_data root_channel_data;
- listener *listeners;
+ listener* listeners;
int listeners_destroyed;
gpr_refcount internal_refcount;
@@ -229,30 +223,30 @@ struct grpc_server {
};
#define SERVER_FROM_CALL_ELEM(elem) \
- (((channel_data *)(elem)->channel_data)->server)
+ (((channel_data*)(elem)->channel_data)->server)
-static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *calld,
- grpc_error *error);
-static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
- size_t cq_idx, requested_call *rc, grpc_error *error);
+static void publish_new_rpc(grpc_exec_ctx* exec_ctx, void* calld,
+ grpc_error* error);
+static void fail_call(grpc_exec_ctx* exec_ctx, grpc_server* server,
+ size_t cq_idx, requested_call* rc, grpc_error* error);
/* Before calling maybe_finish_shutdown, we must hold mu_global and not
hold mu_call */
-static void maybe_finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_server *server);
+static void maybe_finish_shutdown(grpc_exec_ctx* exec_ctx, grpc_server* server);
/*
* channel broadcaster
*/
/* assumes server locked */
-static void channel_broadcaster_init(grpc_server *s, channel_broadcaster *cb) {
- channel_data *c;
+static void channel_broadcaster_init(grpc_server* s, channel_broadcaster* cb) {
+ channel_data* c;
size_t count = 0;
for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
count++;
}
cb->num_channels = count;
cb->channels =
- (grpc_channel **)gpr_malloc(sizeof(*cb->channels) * cb->num_channels);
+ (grpc_channel**)gpr_malloc(sizeof(*cb->channels) * cb->num_channels);
count = 0;
for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
cb->channels[count++] = c->channel;
@@ -265,21 +259,21 @@ struct shutdown_cleanup_args {
grpc_slice slice;
};
-static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- struct shutdown_cleanup_args *a = (struct shutdown_cleanup_args *)arg;
+static void shutdown_cleanup(grpc_exec_ctx* exec_ctx, void* arg,
+ grpc_error* error) {
+ struct shutdown_cleanup_args* a = (struct shutdown_cleanup_args*)arg;
grpc_slice_unref_internal(exec_ctx, a->slice);
gpr_free(a);
}
-static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
- bool send_goaway, grpc_error *send_disconnect) {
- struct shutdown_cleanup_args *sc =
- (struct shutdown_cleanup_args *)gpr_malloc(sizeof(*sc));
+static void send_shutdown(grpc_exec_ctx* exec_ctx, grpc_channel* channel,
+ bool send_goaway, grpc_error* send_disconnect) {
+ struct shutdown_cleanup_args* sc =
+ (struct shutdown_cleanup_args*)gpr_malloc(sizeof(*sc));
GRPC_CLOSURE_INIT(&sc->closure, shutdown_cleanup, sc,
grpc_schedule_on_exec_ctx);
- grpc_transport_op *op = grpc_make_transport_op(&sc->closure);
- grpc_channel_element *elem;
+ grpc_transport_op* op = grpc_make_transport_op(&sc->closure);
+ grpc_channel_element* elem;
op->goaway_error =
send_goaway ? grpc_error_set_int(
@@ -294,10 +288,10 @@ static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
elem->filter->start_transport_op(exec_ctx, elem, op);
}
-static void channel_broadcaster_shutdown(grpc_exec_ctx *exec_ctx,
- channel_broadcaster *cb,
+static void channel_broadcaster_shutdown(grpc_exec_ctx* exec_ctx,
+ channel_broadcaster* cb,
bool send_goaway,
- grpc_error *force_disconnect) {
+ grpc_error* force_disconnect) {
size_t i;
for (i = 0; i < cb->num_channels; i++) {
@@ -313,38 +307,35 @@ static void channel_broadcaster_shutdown(grpc_exec_ctx *exec_ctx,
* request_matcher
*/
-static void request_matcher_init(request_matcher *rm, size_t entries,
- grpc_server *server) {
+static void request_matcher_init(request_matcher* rm, grpc_server* server) {
memset(rm, 0, sizeof(*rm));
rm->server = server;
- rm->requests_per_cq = (gpr_stack_lockfree **)gpr_malloc(
+ rm->requests_per_cq = (gpr_locked_mpscq*)gpr_malloc(
sizeof(*rm->requests_per_cq) * server->cq_count);
for (size_t i = 0; i < server->cq_count; i++) {
- rm->requests_per_cq[i] = gpr_stack_lockfree_create(entries);
+ gpr_locked_mpscq_init(&rm->requests_per_cq[i]);
}
}
-static void request_matcher_destroy(request_matcher *rm) {
+static void request_matcher_destroy(request_matcher* rm) {
for (size_t i = 0; i < rm->server->cq_count; i++) {
- GPR_ASSERT(gpr_stack_lockfree_pop(rm->requests_per_cq[i]) == -1);
- gpr_stack_lockfree_destroy(rm->requests_per_cq[i]);
+ GPR_ASSERT(gpr_locked_mpscq_pop(&rm->requests_per_cq[i]) == NULL);
+ gpr_locked_mpscq_destroy(&rm->requests_per_cq[i]);
}
gpr_free(rm->requests_per_cq);
}
-static void kill_zombie(grpc_exec_ctx *exec_ctx, void *elem,
- grpc_error *error) {
- grpc_call_unref(grpc_call_from_top_element((grpc_call_element *)elem));
+static void kill_zombie(grpc_exec_ctx* exec_ctx, void* elem,
+ grpc_error* error) {
+ grpc_call_unref(grpc_call_from_top_element((grpc_call_element*)elem));
}
-static void request_matcher_zombify_all_pending_calls(grpc_exec_ctx *exec_ctx,
- request_matcher *rm) {
+static void request_matcher_zombify_all_pending_calls(grpc_exec_ctx* exec_ctx,
+ request_matcher* rm) {
while (rm->pending_head) {
- call_data *calld = rm->pending_head;
+ call_data* calld = rm->pending_head;
rm->pending_head = calld->pending_next;
- gpr_mu_lock(&calld->mu_state);
- calld->state = ZOMBIED;
- gpr_mu_unlock(&calld->mu_state);
+ gpr_atm_no_barrier_store(&calld->state, ZOMBIED);
GRPC_CLOSURE_INIT(
&calld->kill_zombie_closure, kill_zombie,
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0),
@@ -353,17 +344,21 @@ static void request_matcher_zombify_all_pending_calls(grpc_exec_ctx *exec_ctx,
}
}
-static void request_matcher_kill_requests(grpc_exec_ctx *exec_ctx,
- grpc_server *server,
- request_matcher *rm,
- grpc_error *error) {
- int request_id;
+static void request_matcher_kill_requests(grpc_exec_ctx* exec_ctx,
+ grpc_server* server,
+ request_matcher* rm,
+ grpc_error* error) {
+ requested_call* rc;
for (size_t i = 0; i < server->cq_count; i++) {
- while ((request_id = gpr_stack_lockfree_pop(rm->requests_per_cq[i])) !=
- -1) {
- fail_call(exec_ctx, server, i,
- &server->requested_calls_per_cq[i][request_id],
- GRPC_ERROR_REF(error));
+ /* Here we know:
+ 1. no requests are being added (since the server is shut down)
+ 2. no other threads are pulling (since the shut down process is single
+ threaded)
+ So, we can ignore the queue lock and just pop, with the guarantee that a
+ NULL returned here truly means that the queue is empty */
+ while ((rc = (requested_call*)gpr_mpscq_pop(
+ &rm->requests_per_cq[i].queue)) != NULL) {
+ fail_call(exec_ctx, server, i, rc, GRPC_ERROR_REF(error));
}
}
GRPC_ERROR_UNREF(error);
@@ -373,12 +368,12 @@ static void request_matcher_kill_requests(grpc_exec_ctx *exec_ctx,
* server proper
*/
-static void server_ref(grpc_server *server) {
+static void server_ref(grpc_server* server) {
gpr_ref(&server->internal_refcount);
}
-static void server_delete(grpc_exec_ctx *exec_ctx, grpc_server *server) {
- registered_method *rm;
+static void server_delete(grpc_exec_ctx* exec_ctx, grpc_server* server) {
+ registered_method* rm;
size_t i;
grpc_channel_args_destroy(exec_ctx, server->channel_args);
gpr_mu_destroy(&server->mu_global);
@@ -398,45 +393,39 @@ static void server_delete(grpc_exec_ctx *exec_ctx, grpc_server *server) {
}
for (i = 0; i < server->cq_count; i++) {
GRPC_CQ_INTERNAL_UNREF(exec_ctx, server->cqs[i], "server");
- if (server->started) {
- gpr_stack_lockfree_destroy(server->request_freelist_per_cq[i]);
- gpr_free(server->requested_calls_per_cq[i]);
- }
}
- gpr_free(server->request_freelist_per_cq);
- gpr_free(server->requested_calls_per_cq);
gpr_free(server->cqs);
gpr_free(server->pollsets);
gpr_free(server->shutdown_tags);
gpr_free(server);
}
-static void server_unref(grpc_exec_ctx *exec_ctx, grpc_server *server) {
+static void server_unref(grpc_exec_ctx* exec_ctx, grpc_server* server) {
if (gpr_unref(&server->internal_refcount)) {
server_delete(exec_ctx, server);
}
}
-static int is_channel_orphaned(channel_data *chand) {
+static int is_channel_orphaned(channel_data* chand) {
return chand->next == chand;
}
-static void orphan_channel(channel_data *chand) {
+static void orphan_channel(channel_data* chand) {
chand->next->prev = chand->prev;
chand->prev->next = chand->next;
chand->next = chand->prev = chand;
}
-static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd,
- grpc_error *error) {
- channel_data *chand = (channel_data *)cd;
- grpc_server *server = chand->server;
+static void finish_destroy_channel(grpc_exec_ctx* exec_ctx, void* cd,
+ grpc_error* error) {
+ channel_data* chand = (channel_data*)cd;
+ grpc_server* server = chand->server;
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "server");
server_unref(exec_ctx, server);
}
-static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand,
- grpc_error *error) {
+static void destroy_channel(grpc_exec_ctx* exec_ctx, channel_data* chand,
+ grpc_error* error) {
if (is_channel_orphaned(chand)) return;
GPR_ASSERT(chand->server != NULL);
orphan_channel(chand);
@@ -446,12 +435,12 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand,
finish_destroy_channel, chand, grpc_schedule_on_exec_ctx);
if (GRPC_TRACER_ON(grpc_server_channel_trace) && error != GRPC_ERROR_NONE) {
- const char *msg = grpc_error_string(error);
+ const char* msg = grpc_error_string(error);
gpr_log(GPR_INFO, "Disconnected client: %s", msg);
}
GRPC_ERROR_UNREF(error);
- grpc_transport_op *op =
+ grpc_transport_op* op =
grpc_make_transport_op(&chand->finish_destroy_channel_closure);
op->set_accept_stream = true;
grpc_channel_next_op(exec_ctx,
@@ -460,29 +449,15 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand,
op);
}
-static void done_request_event(grpc_exec_ctx *exec_ctx, void *req,
- grpc_cq_completion *c) {
- requested_call *rc = (requested_call *)req;
- grpc_server *server = rc->server;
-
- if (rc >= server->requested_calls_per_cq[rc->cq_idx] &&
- rc < server->requested_calls_per_cq[rc->cq_idx] +
- server->max_requested_calls_per_cq) {
- GPR_ASSERT(rc - server->requested_calls_per_cq[rc->cq_idx] <= INT_MAX);
- gpr_stack_lockfree_push(
- server->request_freelist_per_cq[rc->cq_idx],
- (int)(rc - server->requested_calls_per_cq[rc->cq_idx]));
- } else {
- gpr_free(req);
- }
-
- server_unref(exec_ctx, server);
+static void done_request_event(grpc_exec_ctx* exec_ctx, void* req,
+ grpc_cq_completion* c) {
+ gpr_free(req);
}
-static void publish_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
- call_data *calld, size_t cq_idx, requested_call *rc) {
+static void publish_call(grpc_exec_ctx* exec_ctx, grpc_server* server,
+ call_data* calld, size_t cq_idx, requested_call* rc) {
grpc_call_set_completion_queue(exec_ctx, calld->call, rc->cq_bound_to_call);
- grpc_call *call = calld->call;
+ grpc_call* call = calld->call;
*rc->call = call;
calld->cq_new = server->cqs[cq_idx];
GPR_SWAP(grpc_metadata_array, *rc->initial_metadata, calld->initial_metadata);
@@ -508,26 +483,20 @@ static void publish_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
GPR_UNREACHABLE_CODE(return );
}
- grpc_call_element *elem =
- grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
- channel_data *chand = (channel_data *)elem->channel_data;
- server_ref(chand->server);
grpc_cq_end_op(exec_ctx, calld->cq_new, rc->tag, GRPC_ERROR_NONE,
done_request_event, rc, &rc->completion);
}
-static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- grpc_call_element *call_elem = (grpc_call_element *)arg;
- call_data *calld = (call_data *)call_elem->call_data;
- channel_data *chand = (channel_data *)call_elem->channel_data;
- request_matcher *rm = calld->matcher;
- grpc_server *server = rm->server;
+static void publish_new_rpc(grpc_exec_ctx* exec_ctx, void* arg,
+ grpc_error* error) {
+ grpc_call_element* call_elem = (grpc_call_element*)arg;
+ call_data* calld = (call_data*)call_elem->call_data;
+ channel_data* chand = (channel_data*)call_elem->channel_data;
+ request_matcher* rm = calld->matcher;
+ grpc_server* server = rm->server;
if (error != GRPC_ERROR_NONE || gpr_atm_acq_load(&server->shutdown_flag)) {
- gpr_mu_lock(&calld->mu_state);
- calld->state = ZOMBIED;
- gpr_mu_unlock(&calld->mu_state);
+ gpr_atm_no_barrier_store(&calld->state, ZOMBIED);
GRPC_CLOSURE_INIT(
&calld->kill_zombie_closure, kill_zombie,
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0),
@@ -539,16 +508,14 @@ static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *arg,
for (size_t i = 0; i < server->cq_count; i++) {
size_t cq_idx = (chand->cq_idx + i) % server->cq_count;
- int request_id = gpr_stack_lockfree_pop(rm->requests_per_cq[cq_idx]);
- if (request_id == -1) {
+ requested_call* rc =
+ (requested_call*)gpr_locked_mpscq_try_pop(&rm->requests_per_cq[cq_idx]);
+ if (rc == NULL) {
continue;
} else {
GRPC_STATS_INC_SERVER_CQS_CHECKED(exec_ctx, i);
- gpr_mu_lock(&calld->mu_state);
- calld->state = ACTIVATED;
- gpr_mu_unlock(&calld->mu_state);
- publish_call(exec_ctx, server, calld, cq_idx,
- &server->requested_calls_per_cq[cq_idx][request_id]);
+ gpr_atm_no_barrier_store(&calld->state, ACTIVATED);
+ publish_call(exec_ctx, server, calld, cq_idx, rc);
return; /* early out */
}
}
@@ -556,9 +523,27 @@ static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *arg,
/* no cq to take the request found: queue it on the slow list */
GRPC_STATS_INC_SERVER_SLOWPATH_REQUESTS_QUEUED(exec_ctx);
gpr_mu_lock(&server->mu_call);
- gpr_mu_lock(&calld->mu_state);
- calld->state = PENDING;
- gpr_mu_unlock(&calld->mu_state);
+
+ // We need to ensure that all the queues are empty. We do this under
+ // the server mu_call lock to ensure that if something is added to
+ // an empty request queue, it will block until the call is actually
+ // added to the pending list.
+ for (size_t i = 0; i < server->cq_count; i++) {
+ size_t cq_idx = (chand->cq_idx + i) % server->cq_count;
+ requested_call* rc =
+ (requested_call*)gpr_locked_mpscq_pop(&rm->requests_per_cq[cq_idx]);
+ if (rc == NULL) {
+ continue;
+ } else {
+ gpr_mu_unlock(&server->mu_call);
+ GRPC_STATS_INC_SERVER_CQS_CHECKED(exec_ctx, i + server->cq_count);
+ gpr_atm_no_barrier_store(&calld->state, ACTIVATED);
+ publish_call(exec_ctx, server, calld, cq_idx, rc);
+ return; /* early out */
+ }
+ }
+
+ gpr_atm_no_barrier_store(&calld->state, PENDING);
if (rm->pending_head == NULL) {
rm->pending_tail = rm->pending_head = calld;
} else {
@@ -570,15 +555,13 @@ static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *arg,
}
static void finish_start_new_rpc(
- grpc_exec_ctx *exec_ctx, grpc_server *server, grpc_call_element *elem,
- request_matcher *rm,
+ grpc_exec_ctx* exec_ctx, grpc_server* server, grpc_call_element* elem,
+ request_matcher* rm,
grpc_server_register_method_payload_handling payload_handling) {
- call_data *calld = (call_data *)elem->call_data;
+ call_data* calld = (call_data*)elem->call_data;
if (gpr_atm_acq_load(&server->shutdown_flag)) {
- gpr_mu_lock(&calld->mu_state);
- calld->state = ZOMBIED;
- gpr_mu_unlock(&calld->mu_state);
+ gpr_atm_no_barrier_store(&calld->state, ZOMBIED);
GRPC_CLOSURE_INIT(&calld->kill_zombie_closure, kill_zombie, elem,
grpc_schedule_on_exec_ctx);
GRPC_CLOSURE_SCHED(exec_ctx, &calld->kill_zombie_closure, GRPC_ERROR_NONE);
@@ -605,13 +588,13 @@ static void finish_start_new_rpc(
}
}
-static void start_new_rpc(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
- channel_data *chand = (channel_data *)elem->channel_data;
- call_data *calld = (call_data *)elem->call_data;
- grpc_server *server = chand->server;
+static void start_new_rpc(grpc_exec_ctx* exec_ctx, grpc_call_element* elem) {
+ channel_data* chand = (channel_data*)elem->channel_data;
+ call_data* calld = (call_data*)elem->call_data;
+ grpc_server* server = chand->server;
uint32_t i;
uint32_t hash;
- channel_registered_method *rm;
+ channel_registered_method* rm;
if (chand->registered_methods && calld->path_set && calld->host_set) {
/* TODO(ctiller): unify these two searches */
@@ -659,8 +642,8 @@ static void start_new_rpc(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
GRPC_SRM_PAYLOAD_NONE);
}
-static int num_listeners(grpc_server *server) {
- listener *l;
+static int num_listeners(grpc_server* server) {
+ listener* l;
int n = 0;
for (l = server->listeners; l; l = l->next) {
n++;
@@ -668,13 +651,13 @@ static int num_listeners(grpc_server *server) {
return n;
}
-static void done_shutdown_event(grpc_exec_ctx *exec_ctx, void *server,
- grpc_cq_completion *completion) {
- server_unref(exec_ctx, (grpc_server *)server);
+static void done_shutdown_event(grpc_exec_ctx* exec_ctx, void* server,
+ grpc_cq_completion* completion) {
+ server_unref(exec_ctx, (grpc_server*)server);
}
-static int num_channels(grpc_server *server) {
- channel_data *chand;
+static int num_channels(grpc_server* server) {
+ channel_data* chand;
int n = 0;
for (chand = server->root_channel_data.next;
chand != &server->root_channel_data; chand = chand->next) {
@@ -683,15 +666,15 @@ static int num_channels(grpc_server *server) {
return n;
}
-static void kill_pending_work_locked(grpc_exec_ctx *exec_ctx,
- grpc_server *server, grpc_error *error) {
+static void kill_pending_work_locked(grpc_exec_ctx* exec_ctx,
+ grpc_server* server, grpc_error* error) {
if (server->started) {
request_matcher_kill_requests(exec_ctx, server,
&server->unregistered_request_matcher,
GRPC_ERROR_REF(error));
request_matcher_zombify_all_pending_calls(
exec_ctx, &server->unregistered_request_matcher);
- for (registered_method *rm = server->registered_methods; rm;
+ for (registered_method* rm = server->registered_methods; rm;
rm = rm->next) {
request_matcher_kill_requests(exec_ctx, server, &rm->matcher,
GRPC_ERROR_REF(error));
@@ -701,8 +684,8 @@ static void kill_pending_work_locked(grpc_exec_ctx *exec_ctx,
GRPC_ERROR_UNREF(error);
}
-static void maybe_finish_shutdown(grpc_exec_ctx *exec_ctx,
- grpc_server *server) {
+static void maybe_finish_shutdown(grpc_exec_ctx* exec_ctx,
+ grpc_server* server) {
size_t i;
if (!gpr_atm_acq_load(&server->shutdown_flag) || server->shutdown_published) {
return;
@@ -737,10 +720,10 @@ static void maybe_finish_shutdown(grpc_exec_ctx *exec_ctx,
}
}
-static void server_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
- grpc_error *error) {
- grpc_call_element *elem = (grpc_call_element *)ptr;
- call_data *calld = (call_data *)elem->call_data;
+static void server_on_recv_initial_metadata(grpc_exec_ctx* exec_ctx, void* ptr,
+ grpc_error* error) {
+ grpc_call_element* elem = (grpc_call_element*)ptr;
+ call_data* calld = (call_data*)elem->call_data;
grpc_millis op_deadline;
if (error == GRPC_ERROR_NONE) {
@@ -767,7 +750,7 @@ static void server_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
if (calld->host_set && calld->path_set) {
/* do nothing */
} else {
- grpc_error *src_error = error;
+ grpc_error* src_error = error;
error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Missing :authority or :path", &error, 1);
GRPC_ERROR_UNREF(src_error);
@@ -776,9 +759,9 @@ static void server_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
GRPC_CLOSURE_RUN(exec_ctx, calld->on_done_recv_initial_metadata, error);
}
-static void server_mutate_op(grpc_call_element *elem,
- grpc_transport_stream_op_batch *op) {
- call_data *calld = (call_data *)elem->call_data;
+static void server_mutate_op(grpc_call_element* elem,
+ grpc_transport_stream_op_batch* op) {
+ call_data* calld = (call_data*)elem->call_data;
if (op->recv_initial_metadata) {
GPR_ASSERT(op->payload->recv_initial_metadata.recv_flags == NULL);
@@ -794,58 +777,51 @@ static void server_mutate_op(grpc_call_element *elem,
}
static void server_start_transport_stream_op_batch(
- grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
- grpc_transport_stream_op_batch *op) {
+ grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
+ grpc_transport_stream_op_batch* op) {
server_mutate_op(elem, op);
grpc_call_next_op(exec_ctx, elem, op);
}
-static void got_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
- grpc_error *error) {
- grpc_call_element *elem = (grpc_call_element *)ptr;
- call_data *calld = (call_data *)elem->call_data;
+static void got_initial_metadata(grpc_exec_ctx* exec_ctx, void* ptr,
+ grpc_error* error) {
+ grpc_call_element* elem = (grpc_call_element*)ptr;
+ call_data* calld = (call_data*)elem->call_data;
if (error == GRPC_ERROR_NONE) {
start_new_rpc(exec_ctx, elem);
} else {
- gpr_mu_lock(&calld->mu_state);
- if (calld->state == NOT_STARTED) {
- calld->state = ZOMBIED;
- gpr_mu_unlock(&calld->mu_state);
+ if (gpr_atm_full_cas(&calld->state, NOT_STARTED, ZOMBIED)) {
GRPC_CLOSURE_INIT(&calld->kill_zombie_closure, kill_zombie, elem,
grpc_schedule_on_exec_ctx);
GRPC_CLOSURE_SCHED(exec_ctx, &calld->kill_zombie_closure,
GRPC_ERROR_NONE);
- } else if (calld->state == PENDING) {
- calld->state = ZOMBIED;
- gpr_mu_unlock(&calld->mu_state);
+ } else if (gpr_atm_full_cas(&calld->state, PENDING, ZOMBIED)) {
/* zombied call will be destroyed when it's removed from the pending
queue... later */
- } else {
- gpr_mu_unlock(&calld->mu_state);
}
}
}
-static void accept_stream(grpc_exec_ctx *exec_ctx, void *cd,
- grpc_transport *transport,
- const void *transport_server_data) {
- channel_data *chand = (channel_data *)cd;
+static void accept_stream(grpc_exec_ctx* exec_ctx, void* cd,
+ grpc_transport* transport,
+ const void* transport_server_data) {
+ channel_data* chand = (channel_data*)cd;
/* create a call */
grpc_call_create_args args;
memset(&args, 0, sizeof(args));
args.channel = chand->channel;
args.server_transport_data = transport_server_data;
args.send_deadline = GRPC_MILLIS_INF_FUTURE;
- grpc_call *call;
- grpc_error *error = grpc_call_create(exec_ctx, &args, &call);
- grpc_call_element *elem =
+ grpc_call* call;
+ grpc_error* error = grpc_call_create(exec_ctx, &args, &call);
+ grpc_call_element* elem =
grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
if (error != GRPC_ERROR_NONE) {
got_initial_metadata(exec_ctx, elem, error);
GRPC_ERROR_UNREF(error);
return;
}
- call_data *calld = (call_data *)elem->call_data;
+ call_data* calld = (call_data*)elem->call_data;
grpc_op op;
memset(&op, 0, sizeof(op));
op.op = GRPC_OP_RECV_INITIAL_METADATA;
@@ -857,12 +833,12 @@ static void accept_stream(grpc_exec_ctx *exec_ctx, void *cd,
&calld->got_initial_metadata);
}
-static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
- grpc_error *error) {
- channel_data *chand = (channel_data *)cd;
- grpc_server *server = chand->server;
+static void channel_connectivity_changed(grpc_exec_ctx* exec_ctx, void* cd,
+ grpc_error* error) {
+ channel_data* chand = (channel_data*)cd;
+ grpc_server* server = chand->server;
if (chand->connectivity_state != GRPC_CHANNEL_SHUTDOWN) {
- grpc_transport_op *op = grpc_make_transport_op(NULL);
+ grpc_transport_op* op = grpc_make_transport_op(NULL);
op->on_connectivity_state_change = &chand->channel_connectivity_changed,
op->connectivity_state = &chand->connectivity_state;
grpc_channel_next_op(exec_ctx,
@@ -877,15 +853,14 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
}
}
-static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- const grpc_call_element_args *args) {
- call_data *calld = (call_data *)elem->call_data;
- channel_data *chand = (channel_data *)elem->channel_data;
+static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
+ grpc_call_element* elem,
+ const grpc_call_element_args* args) {
+ call_data* calld = (call_data*)elem->call_data;
+ channel_data* chand = (channel_data*)elem->channel_data;
memset(calld, 0, sizeof(call_data));
calld->deadline = GRPC_MILLIS_INF_FUTURE;
calld->call = grpc_call_from_top_element(elem);
- gpr_mu_init(&calld->mu_state);
GRPC_CLOSURE_INIT(&calld->server_on_recv_initial_metadata,
server_on_recv_initial_metadata, elem,
@@ -895,11 +870,11 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
return GRPC_ERROR_NONE;
}
-static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
- const grpc_call_final_info *final_info,
- grpc_closure *ignored) {
- channel_data *chand = (channel_data *)elem->channel_data;
- call_data *calld = (call_data *)elem->call_data;
+static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
+ const grpc_call_final_info* final_info,
+ grpc_closure* ignored) {
+ channel_data* chand = (channel_data*)elem->channel_data;
+ call_data* calld = (call_data*)elem->call_data;
GPR_ASSERT(calld->state != PENDING);
@@ -912,15 +887,13 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
grpc_metadata_array_destroy(&calld->initial_metadata);
grpc_byte_buffer_destroy(calld->payload);
- gpr_mu_destroy(&calld->mu_state);
-
server_unref(exec_ctx, chand->server);
}
-static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
- grpc_channel_element *elem,
- grpc_channel_element_args *args) {
- channel_data *chand = (channel_data *)elem->channel_data;
+static grpc_error* init_channel_elem(grpc_exec_ctx* exec_ctx,
+ grpc_channel_element* elem,
+ grpc_channel_element_args* args) {
+ channel_data* chand = (channel_data*)elem->channel_data;
GPR_ASSERT(args->is_first);
GPR_ASSERT(!args->is_last);
chand->server = NULL;
@@ -934,10 +907,10 @@ static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
return GRPC_ERROR_NONE;
}
-static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
- grpc_channel_element *elem) {
+static void destroy_channel_elem(grpc_exec_ctx* exec_ctx,
+ grpc_channel_element* elem) {
size_t i;
- channel_data *chand = (channel_data *)elem->channel_data;
+ channel_data* chand = (channel_data*)elem->channel_data;
if (chand->registered_methods) {
for (i = 0; i < chand->registered_method_slots; i++) {
grpc_slice_unref_internal(exec_ctx, chand->registered_methods[i].method);
@@ -972,9 +945,9 @@ const grpc_channel_filter grpc_server_top_filter = {
"server",
};
-static void register_completion_queue(grpc_server *server,
- grpc_completion_queue *cq,
- void *reserved) {
+static void register_completion_queue(grpc_server* server,
+ grpc_completion_queue* cq,
+ void* reserved) {
size_t i, n;
GPR_ASSERT(!reserved);
for (i = 0; i < server->cq_count; i++) {
@@ -983,14 +956,14 @@ static void register_completion_queue(grpc_server *server,
GRPC_CQ_INTERNAL_REF(cq, "server");
n = server->cq_count++;
- server->cqs = (grpc_completion_queue **)gpr_realloc(
- server->cqs, server->cq_count * sizeof(grpc_completion_queue *));
+ server->cqs = (grpc_completion_queue**)gpr_realloc(
+ server->cqs, server->cq_count * sizeof(grpc_completion_queue*));
server->cqs[n] = cq;
}
-void grpc_server_register_completion_queue(grpc_server *server,
- grpc_completion_queue *cq,
- void *reserved) {
+void grpc_server_register_completion_queue(grpc_server* server,
+ grpc_completion_queue* cq,
+ void* reserved) {
GRPC_API_TRACE(
"grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3,
(server, cq, reserved));
@@ -1006,10 +979,10 @@ void grpc_server_register_completion_queue(grpc_server *server,
register_completion_queue(server, cq, reserved);
}
-grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) {
+grpc_server* grpc_server_create(const grpc_channel_args* args, void* reserved) {
GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved));
- grpc_server *server = (grpc_server *)gpr_zalloc(sizeof(grpc_server));
+ grpc_server* server = (grpc_server*)gpr_zalloc(sizeof(grpc_server));
gpr_mu_init(&server->mu_global);
gpr_mu_init(&server->mu_call);
@@ -1020,25 +993,23 @@ grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) {
server->root_channel_data.next = server->root_channel_data.prev =
&server->root_channel_data;
- /* TODO(ctiller): expose a channel_arg for this */
- server->max_requested_calls_per_cq = 32768;
server->channel_args = grpc_channel_args_copy(args);
return server;
}
-static int streq(const char *a, const char *b) {
+static int streq(const char* a, const char* b) {
if (a == NULL && b == NULL) return 1;
if (a == NULL) return 0;
if (b == NULL) return 0;
return 0 == strcmp(a, b);
}
-void *grpc_server_register_method(
- grpc_server *server, const char *method, const char *host,
+void* grpc_server_register_method(
+ grpc_server* server, const char* method, const char* host,
grpc_server_register_method_payload_handling payload_handling,
uint32_t flags) {
- registered_method *m;
+ registered_method* m;
GRPC_API_TRACE(
"grpc_server_register_method(server=%p, method=%s, host=%s, "
"flags=0x%08x)",
@@ -1060,7 +1031,7 @@ void *grpc_server_register_method(
flags);
return NULL;
}
- m = (registered_method *)gpr_zalloc(sizeof(registered_method));
+ m = (registered_method*)gpr_zalloc(sizeof(registered_method));
m->method = gpr_strdup(method);
m->host = gpr_strdup(host);
m->next = server->registered_methods;
@@ -1070,10 +1041,10 @@ void *grpc_server_register_method(
return m;
}
-static void start_listeners(grpc_exec_ctx *exec_ctx, void *s,
- grpc_error *error) {
- grpc_server *server = (grpc_server *)s;
- for (listener *l = server->listeners; l; l = l->next) {
+static void start_listeners(grpc_exec_ctx* exec_ctx, void* s,
+ grpc_error* error) {
+ grpc_server* server = (grpc_server*)s;
+ for (listener* l = server->listeners; l; l = l->next) {
l->start(exec_ctx, server, l->arg, server->pollsets, server->pollset_count);
}
@@ -1085,7 +1056,7 @@ static void start_listeners(grpc_exec_ctx *exec_ctx, void *s,
server_unref(exec_ctx, server);
}
-void grpc_server_start(grpc_server *server) {
+void grpc_server_start(grpc_server* server) {
size_t i;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
@@ -1094,30 +1065,16 @@ void grpc_server_start(grpc_server *server) {
server->started = true;
server->pollset_count = 0;
server->pollsets =
- (grpc_pollset **)gpr_malloc(sizeof(grpc_pollset *) * server->cq_count);
- server->request_freelist_per_cq = (gpr_stack_lockfree **)gpr_malloc(
- sizeof(*server->request_freelist_per_cq) * server->cq_count);
- server->requested_calls_per_cq = (requested_call **)gpr_malloc(
- sizeof(*server->requested_calls_per_cq) * server->cq_count);
+ (grpc_pollset**)gpr_malloc(sizeof(grpc_pollset*) * server->cq_count);
for (i = 0; i < server->cq_count; i++) {
if (grpc_cq_can_listen(server->cqs[i])) {
server->pollsets[server->pollset_count++] =
grpc_cq_pollset(server->cqs[i]);
}
- server->request_freelist_per_cq[i] =
- gpr_stack_lockfree_create((size_t)server->max_requested_calls_per_cq);
- for (int j = 0; j < server->max_requested_calls_per_cq; j++) {
- gpr_stack_lockfree_push(server->request_freelist_per_cq[i], j);
- }
- server->requested_calls_per_cq[i] = (requested_call *)gpr_malloc(
- (size_t)server->max_requested_calls_per_cq *
- sizeof(*server->requested_calls_per_cq[i]));
}
- request_matcher_init(&server->unregistered_request_matcher,
- (size_t)server->max_requested_calls_per_cq, server);
- for (registered_method *rm = server->registered_methods; rm; rm = rm->next) {
- request_matcher_init(&rm->matcher,
- (size_t)server->max_requested_calls_per_cq, server);
+ request_matcher_init(&server->unregistered_request_matcher, server);
+ for (registered_method* rm = server->registered_methods; rm; rm = rm->next) {
+ request_matcher_init(&rm->matcher, server);
}
server_ref(server);
@@ -1131,31 +1088,31 @@ void grpc_server_start(grpc_server *server) {
grpc_exec_ctx_finish(&exec_ctx);
}
-void grpc_server_get_pollsets(grpc_server *server, grpc_pollset ***pollsets,
- size_t *pollset_count) {
+void grpc_server_get_pollsets(grpc_server* server, grpc_pollset*** pollsets,
+ size_t* pollset_count) {
*pollset_count = server->pollset_count;
*pollsets = server->pollsets;
}
-void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
- grpc_transport *transport,
- grpc_pollset *accepting_pollset,
- const grpc_channel_args *args) {
+void grpc_server_setup_transport(grpc_exec_ctx* exec_ctx, grpc_server* s,
+ grpc_transport* transport,
+ grpc_pollset* accepting_pollset,
+ const grpc_channel_args* args) {
size_t num_registered_methods;
size_t alloc;
- registered_method *rm;
- channel_registered_method *crm;
- grpc_channel *channel;
- channel_data *chand;
+ registered_method* rm;
+ channel_registered_method* crm;
+ grpc_channel* channel;
+ channel_data* chand;
uint32_t hash;
size_t slots;
uint32_t probes;
uint32_t max_probes = 0;
- grpc_transport_op *op = NULL;
+ grpc_transport_op* op = NULL;
channel =
grpc_channel_create(exec_ctx, NULL, args, GRPC_SERVER_CHANNEL, transport);
- chand = (channel_data *)grpc_channel_stack_element(
+ chand = (channel_data*)grpc_channel_stack_element(
grpc_channel_get_channel_stack(channel), 0)
->channel_data;
chand->server = s;
@@ -1181,7 +1138,7 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
if (num_registered_methods > 0) {
slots = 2 * num_registered_methods;
alloc = sizeof(channel_registered_method) * slots;
- chand->registered_methods = (channel_registered_method *)gpr_zalloc(alloc);
+ chand->registered_methods = (channel_registered_method*)gpr_zalloc(alloc);
for (rm = s->registered_methods; rm; rm = rm->next) {
grpc_slice host;
bool has_host;
@@ -1234,25 +1191,25 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
grpc_transport_perform_op(exec_ctx, transport, op);
}
-void done_published_shutdown(grpc_exec_ctx *exec_ctx, void *done_arg,
- grpc_cq_completion *storage) {
+void done_published_shutdown(grpc_exec_ctx* exec_ctx, void* done_arg,
+ grpc_cq_completion* storage) {
(void)done_arg;
gpr_free(storage);
}
-static void listener_destroy_done(grpc_exec_ctx *exec_ctx, void *s,
- grpc_error *error) {
- grpc_server *server = (grpc_server *)s;
+static void listener_destroy_done(grpc_exec_ctx* exec_ctx, void* s,
+ grpc_error* error) {
+ grpc_server* server = (grpc_server*)s;
gpr_mu_lock(&server->mu_global);
server->listeners_destroyed++;
maybe_finish_shutdown(exec_ctx, server);
gpr_mu_unlock(&server->mu_global);
}
-void grpc_server_shutdown_and_notify(grpc_server *server,
- grpc_completion_queue *cq, void *tag) {
- listener *l;
- shutdown_tag *sdt;
+void grpc_server_shutdown_and_notify(grpc_server* server,
+ grpc_completion_queue* cq, void* tag) {
+ listener* l;
+ shutdown_tag* sdt;
channel_broadcaster broadcaster;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
@@ -1269,13 +1226,13 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
/* stay locked, and gather up some stuff to do */
GPR_ASSERT(grpc_cq_begin_op(cq, tag));
if (server->shutdown_published) {
- grpc_cq_end_op(
- &exec_ctx, cq, tag, GRPC_ERROR_NONE, done_published_shutdown, NULL,
- (grpc_cq_completion *)gpr_malloc(sizeof(grpc_cq_completion)));
+ grpc_cq_end_op(&exec_ctx, cq, tag, GRPC_ERROR_NONE, done_published_shutdown,
+ NULL,
+ (grpc_cq_completion*)gpr_malloc(sizeof(grpc_cq_completion)));
gpr_mu_unlock(&server->mu_global);
goto done;
}
- server->shutdown_tags = (shutdown_tag *)gpr_realloc(
+ server->shutdown_tags = (shutdown_tag*)gpr_realloc(
server->shutdown_tags,
sizeof(shutdown_tag) * (server->num_shutdown_tags + 1));
sdt = &server->shutdown_tags[server->num_shutdown_tags++];
@@ -1316,7 +1273,7 @@ done:
grpc_exec_ctx_finish(&exec_ctx);
}
-void grpc_server_cancel_all_calls(grpc_server *server) {
+void grpc_server_cancel_all_calls(grpc_server* server) {
channel_broadcaster broadcaster;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
@@ -1332,8 +1289,8 @@ void grpc_server_cancel_all_calls(grpc_server *server) {
grpc_exec_ctx_finish(&exec_ctx);
}
-void grpc_server_destroy(grpc_server *server) {
- listener *l;
+void grpc_server_destroy(grpc_server* server) {
+ listener* l;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
GRPC_API_TRACE("grpc_server_destroy(server=%p)", 1, (server));
@@ -1355,12 +1312,12 @@ void grpc_server_destroy(grpc_server *server) {
}
void grpc_server_add_listener(
- grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
- void (*start)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
- grpc_pollset **pollsets, size_t pollset_count),
- void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
- grpc_closure *on_done)) {
- listener *l = (listener *)gpr_malloc(sizeof(listener));
+ grpc_exec_ctx* exec_ctx, grpc_server* server, void* arg,
+ void (*start)(grpc_exec_ctx* exec_ctx, grpc_server* server, void* arg,
+ grpc_pollset** pollsets, size_t pollset_count),
+ void (*destroy)(grpc_exec_ctx* exec_ctx, grpc_server* server, void* arg,
+ grpc_closure* on_done)) {
+ listener* l = (listener*)gpr_malloc(sizeof(listener));
l->arg = arg;
l->start = start;
l->destroy = destroy;
@@ -1368,26 +1325,16 @@ void grpc_server_add_listener(
server->listeners = l;
}
-static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
- grpc_server *server, size_t cq_idx,
- requested_call *rc) {
- call_data *calld = NULL;
- request_matcher *rm = NULL;
- int request_id;
+static grpc_call_error queue_call_request(grpc_exec_ctx* exec_ctx,
+ grpc_server* server, size_t cq_idx,
+ requested_call* rc) {
+ call_data* calld = NULL;
+ request_matcher* rm = NULL;
if (gpr_atm_acq_load(&server->shutdown_flag)) {
fail_call(exec_ctx, server, cq_idx, rc,
GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
return GRPC_CALL_OK;
}
- request_id = gpr_stack_lockfree_pop(server->request_freelist_per_cq[cq_idx]);
- if (request_id == -1) {
- /* out of request ids: just fail this one */
- fail_call(exec_ctx, server, cq_idx, rc,
- grpc_error_set_int(
- GRPC_ERROR_CREATE_FROM_STATIC_STRING("Out of request ids"),
- GRPC_ERROR_INT_LIMIT, server->max_requested_calls_per_cq));
- return GRPC_CALL_OK;
- }
switch (rc->type) {
case BATCH_CALL:
rm = &server->unregistered_request_matcher;
@@ -1396,20 +1343,17 @@ static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
rm = &rc->data.registered.method->matcher;
break;
}
- server->requested_calls_per_cq[cq_idx][request_id] = *rc;
- gpr_free(rc);
- if (gpr_stack_lockfree_push(rm->requests_per_cq[cq_idx], request_id)) {
+ if (gpr_locked_mpscq_push(&rm->requests_per_cq[cq_idx], &rc->request_link)) {
/* this was the first queued request: we need to lock and start
matching calls */
gpr_mu_lock(&server->mu_call);
while ((calld = rm->pending_head) != NULL) {
- request_id = gpr_stack_lockfree_pop(rm->requests_per_cq[cq_idx]);
- if (request_id == -1) break;
+ rc = (requested_call*)gpr_locked_mpscq_pop(&rm->requests_per_cq[cq_idx]);
+ if (rc == NULL) break;
rm->pending_head = calld->pending_next;
gpr_mu_unlock(&server->mu_call);
- gpr_mu_lock(&calld->mu_state);
- if (calld->state == ZOMBIED) {
- gpr_mu_unlock(&calld->mu_state);
+ if (!gpr_atm_full_cas(&calld->state, PENDING, ACTIVATED)) {
+ // Zombied Call
GRPC_CLOSURE_INIT(
&calld->kill_zombie_closure, kill_zombie,
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0),
@@ -1417,11 +1361,7 @@ static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
GRPC_CLOSURE_SCHED(exec_ctx, &calld->kill_zombie_closure,
GRPC_ERROR_NONE);
} else {
- GPR_ASSERT(calld->state == PENDING);
- calld->state = ACTIVATED;
- gpr_mu_unlock(&calld->mu_state);
- publish_call(exec_ctx, server, calld, cq_idx,
- &server->requested_calls_per_cq[cq_idx][request_id]);
+ publish_call(exec_ctx, server, calld, cq_idx, rc);
}
gpr_mu_lock(&server->mu_call);
}
@@ -1431,20 +1371,21 @@ static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
}
grpc_call_error grpc_server_request_call(
- grpc_server *server, grpc_call **call, grpc_call_details *details,
- grpc_metadata_array *initial_metadata,
- grpc_completion_queue *cq_bound_to_call,
- grpc_completion_queue *cq_for_notification, void *tag) {
+ grpc_server* server, grpc_call** call, grpc_call_details* details,
+ grpc_metadata_array* initial_metadata,
+ grpc_completion_queue* cq_bound_to_call,
+ grpc_completion_queue* cq_for_notification, void* tag) {
grpc_call_error error;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- requested_call *rc = (requested_call *)gpr_malloc(sizeof(*rc));
+ requested_call* rc = (requested_call*)gpr_malloc(sizeof(*rc));
GRPC_STATS_INC_SERVER_REQUESTED_CALLS(&exec_ctx);
GRPC_API_TRACE(
"grpc_server_request_call("
"server=%p, call=%p, details=%p, initial_metadata=%p, "
"cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)",
- 7, (server, call, details, initial_metadata, cq_bound_to_call,
- cq_for_notification, tag));
+ 7,
+ (server, call, details, initial_metadata, cq_bound_to_call,
+ cq_for_notification, tag));
size_t cq_idx;
for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) {
if (server->cqs[cq_idx] == cq_for_notification) {
@@ -1477,22 +1418,23 @@ done:
}
grpc_call_error grpc_server_request_registered_call(
- grpc_server *server, void *rmp, grpc_call **call, gpr_timespec *deadline,
- grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload,
- grpc_completion_queue *cq_bound_to_call,
- grpc_completion_queue *cq_for_notification, void *tag) {
+ grpc_server* server, void* rmp, grpc_call** call, gpr_timespec* deadline,
+ grpc_metadata_array* initial_metadata, grpc_byte_buffer** optional_payload,
+ grpc_completion_queue* cq_bound_to_call,
+ grpc_completion_queue* cq_for_notification, void* tag) {
grpc_call_error error;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- requested_call *rc = (requested_call *)gpr_malloc(sizeof(*rc));
- registered_method *rm = (registered_method *)rmp;
+ requested_call* rc = (requested_call*)gpr_malloc(sizeof(*rc));
+ registered_method* rm = (registered_method*)rmp;
GRPC_STATS_INC_SERVER_REQUESTED_CALLS(&exec_ctx);
GRPC_API_TRACE(
"grpc_server_request_registered_call("
"server=%p, rmp=%p, call=%p, deadline=%p, initial_metadata=%p, "
"optional_payload=%p, cq_bound_to_call=%p, cq_for_notification=%p, "
"tag=%p)",
- 9, (server, rmp, call, deadline, initial_metadata, optional_payload,
- cq_bound_to_call, cq_for_notification, tag));
+ 9,
+ (server, rmp, call, deadline, initial_metadata, optional_payload,
+ cq_bound_to_call, cq_for_notification, tag));
size_t cq_idx;
for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) {
@@ -1532,22 +1474,21 @@ done:
return error;
}
-static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
- size_t cq_idx, requested_call *rc, grpc_error *error) {
+static void fail_call(grpc_exec_ctx* exec_ctx, grpc_server* server,
+ size_t cq_idx, requested_call* rc, grpc_error* error) {
*rc->call = NULL;
rc->initial_metadata->count = 0;
GPR_ASSERT(error != GRPC_ERROR_NONE);
- server_ref(server);
grpc_cq_end_op(exec_ctx, server->cqs[cq_idx], rc->tag, error,
done_request_event, rc, &rc->completion);
}
-const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {
+const grpc_channel_args* grpc_server_get_channel_args(grpc_server* server) {
return server->channel_args;
}
-int grpc_server_has_open_connections(grpc_server *server) {
+int grpc_server_has_open_connections(grpc_server* server) {
int r;
gpr_mu_lock(&server->mu_global);
r = server->root_channel_data.next != &server->root_channel_data;
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-22 03:11:27 +0000