aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/core/ext/transport/chttp2/server/insecure/server_chttp2.c11
-rw-r--r--src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c74
-rw-r--r--src/core/lib/iomgr/tcp_server.h1
-rw-r--r--src/core/lib/iomgr/tcp_server_posix.c2
-rw-r--r--src/core/lib/iomgr/tcp_server_windows.c5
-rw-r--r--src/core/lib/surface/completion_queue.c5
-rw-r--r--src/core/lib/surface/completion_queue.h1
-rw-r--r--src/core/lib/surface/server.c176
-rw-r--r--src/core/lib/surface/server.h1
-rw-r--r--src/cpp/server/server.cc7
-rw-r--r--src/cpp/server/server_builder.cc17
11 files changed, 181 insertions, 119 deletions
diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
index e21fa2a072..0428bb1e3d 100644
--- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
+++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
@@ -43,14 +43,8 @@
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/server.h"
-static void setup_transport(grpc_exec_ctx *exec_ctx, void *server,
- grpc_transport *transport) {
- grpc_server_setup_transport(exec_ctx, server, transport,
- grpc_server_get_channel_args(server));
-}
-
static void new_transport(grpc_exec_ctx *exec_ctx, void *server,
- grpc_endpoint *tcp,
+ grpc_endpoint *tcp, grpc_pollset *accepting_pollset,
grpc_tcp_server_acceptor *acceptor) {
/*
* Beware that the call to grpc_create_chttp2_transport() has to happen before
@@ -61,7 +55,8 @@ static void new_transport(grpc_exec_ctx *exec_ctx, void *server,
*/
grpc_transport *transport = grpc_create_chttp2_transport(
exec_ctx, grpc_server_get_channel_args(server), tcp, 0);
- setup_transport(exec_ctx, server, transport);
+ grpc_server_setup_transport(exec_ctx, server, transport, accepting_pollset,
+ grpc_server_get_channel_args(server));
grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL, 0);
}
diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
index 698b2bef61..26b0f00e9e 100644
--- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
+++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
@@ -52,7 +52,7 @@
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/server.h"
-typedef struct grpc_server_secure_state {
+typedef struct server_secure_state {
grpc_server *server;
grpc_tcp_server *tcp;
grpc_server_security_connector *sc;
@@ -62,13 +62,16 @@ typedef struct grpc_server_secure_state {
gpr_refcount refcount;
grpc_closure destroy_closure;
grpc_closure *destroy_callback;
-} grpc_server_secure_state;
+} server_secure_state;
-static void state_ref(grpc_server_secure_state *state) {
- gpr_ref(&state->refcount);
-}
+typedef struct server_secure_connect {
+ server_secure_state *state;
+ grpc_pollset *accepting_pollset;
+} server_secure_connect;
+
+static void state_ref(server_secure_state *state) { gpr_ref(&state->refcount); }
-static void state_unref(grpc_server_secure_state *state) {
+static void state_unref(server_secure_state *state) {
if (gpr_unref(&state->refcount)) {
/* ensure all threads have unlocked */
gpr_mu_lock(&state->mu);
@@ -80,67 +83,66 @@ static void state_unref(grpc_server_secure_state *state) {
}
}
-static void setup_transport(grpc_exec_ctx *exec_ctx, void *statep,
- grpc_transport *transport,
- grpc_auth_context *auth_context) {
- grpc_server_secure_state *state = statep;
- grpc_channel_args *args_copy;
- grpc_arg args_to_add[2];
- args_to_add[0] = grpc_server_credentials_to_arg(state->creds);
- args_to_add[1] = grpc_auth_context_to_arg(auth_context);
- args_copy = grpc_channel_args_copy_and_add(
- grpc_server_get_channel_args(state->server), args_to_add,
- GPR_ARRAY_SIZE(args_to_add));
- grpc_server_setup_transport(exec_ctx, state->server, transport, args_copy);
- grpc_channel_args_destroy(args_copy);
-}
-
static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *statep,
grpc_security_status status,
grpc_endpoint *secure_endpoint,
grpc_auth_context *auth_context) {
- grpc_server_secure_state *state = statep;
+ server_secure_connect *state = statep;
grpc_transport *transport;
if (status == GRPC_SECURITY_OK) {
if (secure_endpoint) {
- gpr_mu_lock(&state->mu);
- if (!state->is_shutdown) {
+ gpr_mu_lock(&state->state->mu);
+ if (!state->state->is_shutdown) {
transport = grpc_create_chttp2_transport(
- exec_ctx, grpc_server_get_channel_args(state->server),
+ exec_ctx, grpc_server_get_channel_args(state->state->server),
secure_endpoint, 0);
- setup_transport(exec_ctx, state, transport, auth_context);
+ grpc_channel_args *args_copy;
+ grpc_arg args_to_add[2];
+ args_to_add[0] = grpc_server_credentials_to_arg(state->state->creds);
+ args_to_add[1] = grpc_auth_context_to_arg(auth_context);
+ args_copy = grpc_channel_args_copy_and_add(
+ grpc_server_get_channel_args(state->state->server), args_to_add,
+ GPR_ARRAY_SIZE(args_to_add));
+ grpc_server_setup_transport(exec_ctx, state->state->server, transport,
+ state->accepting_pollset, args_copy);
+ grpc_channel_args_destroy(args_copy);
grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL, 0);
} else {
/* We need to consume this here, because the server may already have
* gone away. */
grpc_endpoint_destroy(exec_ctx, secure_endpoint);
}
- gpr_mu_unlock(&state->mu);
+ gpr_mu_unlock(&state->state->mu);
}
} else {
gpr_log(GPR_ERROR, "Secure transport failed with error %d", status);
}
- state_unref(state);
+ state_unref(state->state);
+ gpr_free(state);
}
static void on_accept(grpc_exec_ctx *exec_ctx, void *statep, grpc_endpoint *tcp,
+ grpc_pollset *accepting_pollset,
grpc_tcp_server_acceptor *acceptor) {
- grpc_server_secure_state *state = statep;
- state_ref(state);
- grpc_server_security_connector_do_handshake(
- exec_ctx, state->sc, acceptor, tcp, on_secure_handshake_done, state);
+ server_secure_connect *state = gpr_malloc(sizeof(*state));
+ state->state = statep;
+ state_ref(state->state);
+ state->accepting_pollset = accepting_pollset;
+ grpc_server_security_connector_do_handshake(exec_ctx, state->state->sc,
+ acceptor, tcp,
+ on_secure_handshake_done, state);
}
/* Server callback: start listening on our ports */
static void start(grpc_exec_ctx *exec_ctx, grpc_server *server, void *statep,
grpc_pollset **pollsets, size_t pollset_count) {
- grpc_server_secure_state *state = statep;
+ server_secure_state *state = statep;
grpc_tcp_server_start(exec_ctx, state->tcp, pollsets, pollset_count,
on_accept, state);
}
static void destroy_done(grpc_exec_ctx *exec_ctx, void *statep, bool success) {
- grpc_server_secure_state *state = statep;
+ server_secure_state *state = statep;
if (state->destroy_callback != NULL) {
state->destroy_callback->cb(exec_ctx, state->destroy_callback->cb_arg,
success);
@@ -153,7 +155,7 @@ static void destroy_done(grpc_exec_ctx *exec_ctx, void *statep, bool success) {
callbacks) */
static void destroy(grpc_exec_ctx *exec_ctx, grpc_server *server, void *statep,
grpc_closure *callback) {
- grpc_server_secure_state *state = statep;
+ server_secure_state *state = statep;
grpc_tcp_server *tcp;
gpr_mu_lock(&state->mu);
state->is_shutdown = 1;
@@ -167,7 +169,7 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr,
grpc_server_credentials *creds) {
grpc_resolved_addresses *resolved = NULL;
grpc_tcp_server *tcp = NULL;
- grpc_server_secure_state *state = NULL;
+ server_secure_state *state = NULL;
size_t i;
unsigned count = 0;
int port_num = -1;
diff --git a/src/core/lib/iomgr/tcp_server.h b/src/core/lib/iomgr/tcp_server.h
index 99b9f29729..fee14ae661 100644
--- a/src/core/lib/iomgr/tcp_server.h
+++ b/src/core/lib/iomgr/tcp_server.h
@@ -52,6 +52,7 @@ typedef struct grpc_tcp_server_acceptor {
/* Called for newly connected TCP connections. */
typedef void (*grpc_tcp_server_cb)(grpc_exec_ctx *exec_ctx, void *arg,
grpc_endpoint *ep,
+ grpc_pollset *accepting_pollset,
grpc_tcp_server_acceptor *acceptor);
/* Create a server, initially not bound to any ports. The caller owns one ref.
diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c
index 97c945b834..c695621de8 100644
--- a/src/core/lib/iomgr/tcp_server_posix.c
+++ b/src/core/lib/iomgr/tcp_server_posix.c
@@ -362,7 +362,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
sp->server->on_accept_cb(
exec_ctx, sp->server->on_accept_cb_arg,
grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str),
- &acceptor);
+ read_notifier_pollset, &acceptor);
gpr_free(name);
gpr_free(addr_str);
diff --git a/src/core/lib/iomgr/tcp_server_windows.c b/src/core/lib/iomgr/tcp_server_windows.c
index 125f521d87..87cacfe979 100644
--- a/src/core/lib/iomgr/tcp_server_windows.c
+++ b/src/core/lib/iomgr/tcp_server_windows.c
@@ -379,9 +379,10 @@ static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, bool from_iocp) {
/* The only time we should call our callback, is where we successfully
managed to accept a connection, and created an endpoint. */
- if (ep)
+ if (ep) {
sp->server->on_accept_cb(exec_ctx, sp->server->on_accept_cb_arg, ep,
- &acceptor);
+ NULL, &acceptor);
+ }
/* As we were notified from the IOCP of one and exactly one accept,
the former socked we created has now either been destroy or assigned
to the new connection. We need to create a new one for the next
diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index ae78f8f616..5eb7cf1bf4 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -86,6 +86,7 @@ struct grpc_completion_queue {
};
#define POLLSET_FROM_CQ(cq) ((grpc_pollset *)(cq + 1))
+#define CQ_FROM_POLLSET(ps) (((grpc_completion_queue *)ps) - 1)
static gpr_mu g_freelist_mu;
static grpc_completion_queue *g_freelist;
@@ -514,6 +515,10 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
return POLLSET_FROM_CQ(cc);
}
+grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps) {
+ return CQ_FROM_POLLSET(ps);
+}
+
void grpc_cq_mark_non_listening_server_cq(grpc_completion_queue *cc) {
cc->is_non_listening_server_cq = 1;
}
diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h
index 1528ca4ad8..3d0dd13c53 100644
--- a/src/core/lib/surface/completion_queue.h
+++ b/src/core/lib/surface/completion_queue.h
@@ -81,6 +81,7 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
void *done_arg, grpc_cq_completion *storage);
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);
+grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps);
void grpc_cq_mark_non_listening_server_cq(grpc_completion_queue *cc);
bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc);
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index c9b458faf2..54b76d8aa5 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -69,11 +69,6 @@ typedef struct call_data call_data;
typedef struct channel_data channel_data;
typedef struct registered_method registered_method;
-typedef struct {
- call_data *next;
- call_data *prev;
-} call_link;
-
typedef enum { BATCH_CALL, REGISTERED_CALL } requested_call_type;
typedef struct requested_call {
@@ -81,7 +76,6 @@ typedef struct requested_call {
void *tag;
grpc_server *server;
grpc_completion_queue *cq_bound_to_call;
- grpc_completion_queue *cq_for_notification;
grpc_call **call;
grpc_cq_completion completion;
grpc_metadata_array *initial_metadata;
@@ -108,6 +102,7 @@ struct channel_data {
grpc_server *server;
grpc_connectivity_state connectivity_state;
grpc_channel *channel;
+ size_t cq_idx;
/* linked list of all channels on a server */
channel_data *next;
channel_data *prev;
@@ -172,7 +167,7 @@ struct request_matcher {
grpc_server *server;
call_data *pending_head;
call_data *pending_tail;
- gpr_stack_lockfree *requests;
+ gpr_stack_lockfree **requests_per_cq;
};
struct registered_method {
@@ -180,6 +175,7 @@ struct registered_method {
char *host;
grpc_server_register_method_payload_handling payload_handling;
uint32_t flags;
+ /* one request matcher per method per cq */
request_matcher request_matcher;
registered_method *next;
};
@@ -195,6 +191,7 @@ struct grpc_server {
grpc_completion_queue **cqs;
grpc_pollset **pollsets;
size_t cq_count;
+ bool started;
/* The two following mutexes control access to server-state
mu_global controls access to non-call-related state (e.g., channel state)
@@ -207,6 +204,7 @@ struct grpc_server {
gpr_mu mu_call; /* mutex for call-specific state */
registered_method *registered_methods;
+ /** one request matcher for unregistered methods per cq */
request_matcher unregistered_request_matcher;
/** free list of available requested_calls indices */
gpr_stack_lockfree *request_freelist;
@@ -234,7 +232,7 @@ struct grpc_server {
static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *calld, bool success);
static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
- requested_call *rc);
+ size_t cq_idx, requested_call *rc);
/* 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);
@@ -312,12 +310,19 @@ static void request_matcher_init(request_matcher *rm, size_t entries,
grpc_server *server) {
memset(rm, 0, sizeof(*rm));
rm->server = server;
- rm->requests = gpr_stack_lockfree_create(entries);
+ rm->requests_per_cq =
+ 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);
+ }
}
static void request_matcher_destroy(request_matcher *rm) {
- GPR_ASSERT(gpr_stack_lockfree_pop(rm->requests) == -1);
- gpr_stack_lockfree_destroy(rm->requests);
+ 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_free(rm->requests_per_cq);
}
static void kill_zombie(grpc_exec_ctx *exec_ctx, void *elem, bool success) {
@@ -343,8 +348,11 @@ static void request_matcher_kill_requests(grpc_exec_ctx *exec_ctx,
grpc_server *server,
request_matcher *rm) {
int request_id;
- while ((request_id = gpr_stack_lockfree_pop(rm->requests)) != -1) {
- fail_call(exec_ctx, server, &server->requested_calls[request_id]);
+ 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[request_id]);
+ }
}
}
@@ -364,15 +372,19 @@ static void server_delete(grpc_exec_ctx *exec_ctx, grpc_server *server) {
gpr_mu_destroy(&server->mu_call);
while ((rm = server->registered_methods) != NULL) {
server->registered_methods = rm->next;
- request_matcher_destroy(&rm->request_matcher);
+ if (server->started) {
+ request_matcher_destroy(&rm->request_matcher);
+ }
gpr_free(rm->method);
gpr_free(rm->host);
gpr_free(rm);
}
+ if (server->started) {
+ request_matcher_destroy(&server->unregistered_request_matcher);
+ }
for (i = 0; i < server->cq_count; i++) {
GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server");
}
- request_matcher_destroy(&server->unregistered_request_matcher);
gpr_stack_lockfree_destroy(server->request_freelist);
gpr_free(server->cqs);
gpr_free(server->pollsets);
@@ -453,11 +465,11 @@ static void done_request_event(grpc_exec_ctx *exec_ctx, void *req,
}
static void publish_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
- call_data *calld, requested_call *rc) {
+ 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;
*rc->call = call;
- calld->cq_new = rc->cq_for_notification;
+ calld->cq_new = server->cqs[cq_idx];
GPR_SWAP(grpc_metadata_array, *rc->initial_metadata, calld->initial_metadata);
switch (rc->type) {
case BATCH_CALL:
@@ -492,7 +504,9 @@ static void publish_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
}
static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
- call_data *calld = arg;
+ grpc_call_element *call_elem = arg;
+ call_data *calld = call_elem->call_data;
+ channel_data *chand = call_elem->channel_data;
request_matcher *rm = calld->request_matcher;
grpc_server *server = rm->server;
@@ -507,26 +521,34 @@ static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
return;
}
- int request_id = gpr_stack_lockfree_pop(rm->requests);
- if (request_id == -1) {
- gpr_mu_lock(&server->mu_call);
- gpr_mu_lock(&calld->mu_state);
- calld->state = PENDING;
- gpr_mu_unlock(&calld->mu_state);
- if (rm->pending_head == NULL) {
- rm->pending_tail = rm->pending_head = calld;
+ 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) {
+ continue;
} else {
- rm->pending_tail->pending_next = calld;
- rm->pending_tail = calld;
+ 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[request_id]);
+ return; /* early out */
}
- calld->pending_next = NULL;
- gpr_mu_unlock(&server->mu_call);
+ }
+
+ /* no cq to take the request found: queue it on the slow list */
+ gpr_mu_lock(&server->mu_call);
+ gpr_mu_lock(&calld->mu_state);
+ calld->state = PENDING;
+ gpr_mu_unlock(&calld->mu_state);
+ if (rm->pending_head == NULL) {
+ rm->pending_tail = rm->pending_head = calld;
} else {
- gpr_mu_lock(&calld->mu_state);
- calld->state = ACTIVATED;
- gpr_mu_unlock(&calld->mu_state);
- publish_call(exec_ctx, server, calld, &server->requested_calls[request_id]);
+ rm->pending_tail->pending_next = calld;
+ rm->pending_tail = calld;
}
+ calld->pending_next = NULL;
+ gpr_mu_unlock(&server->mu_call);
}
static void finish_start_new_rpc(
@@ -548,14 +570,14 @@ static void finish_start_new_rpc(
switch (payload_handling) {
case GRPC_SRM_PAYLOAD_NONE:
- publish_new_rpc(exec_ctx, calld, true);
+ publish_new_rpc(exec_ctx, elem, true);
break;
case GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER: {
grpc_op op;
memset(&op, 0, sizeof(op));
op.op = GRPC_OP_RECV_MESSAGE;
op.data.recv_message = &calld->payload;
- grpc_closure_init(&calld->publish, publish_new_rpc, calld);
+ grpc_closure_init(&calld->publish, publish_new_rpc, elem);
grpc_call_start_batch_and_execute(exec_ctx, calld->call, &op, 1,
&calld->publish);
break;
@@ -637,14 +659,16 @@ static int num_channels(grpc_server *server) {
static void kill_pending_work_locked(grpc_exec_ctx *exec_ctx,
grpc_server *server) {
- registered_method *rm;
- request_matcher_kill_requests(exec_ctx, server,
- &server->unregistered_request_matcher);
- request_matcher_zombify_all_pending_calls(
- exec_ctx, &server->unregistered_request_matcher);
- for (rm = server->registered_methods; rm; rm = rm->next) {
- request_matcher_kill_requests(exec_ctx, server, &rm->request_matcher);
- request_matcher_zombify_all_pending_calls(exec_ctx, &rm->request_matcher);
+ if (server->started) {
+ request_matcher_kill_requests(exec_ctx, server,
+ &server->unregistered_request_matcher);
+ request_matcher_zombify_all_pending_calls(
+ exec_ctx, &server->unregistered_request_matcher);
+ for (registered_method *rm = server->registered_methods; rm;
+ rm = rm->next) {
+ request_matcher_kill_requests(exec_ctx, server, &rm->request_matcher);
+ request_matcher_zombify_all_pending_calls(exec_ctx, &rm->request_matcher);
+ }
}
}
@@ -962,8 +986,6 @@ grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) {
for (i = 0; i < (size_t)server->max_requested_calls; i++) {
gpr_stack_lockfree_push(server->request_freelist, (int)i);
}
- request_matcher_init(&server->unregistered_request_matcher,
- server->max_requested_calls, server);
server->requested_calls = gpr_malloc(server->max_requested_calls *
sizeof(*server->requested_calls));
@@ -1007,8 +1029,6 @@ void *grpc_server_register_method(
}
m = gpr_malloc(sizeof(registered_method));
memset(m, 0, sizeof(*m));
- request_matcher_init(&m->request_matcher, server->max_requested_calls,
- server);
m->method = gpr_strdup(method);
m->host = gpr_strdup(host);
m->next = server->registered_methods;
@@ -1025,10 +1045,17 @@ void grpc_server_start(grpc_server *server) {
GRPC_API_TRACE("grpc_server_start(server=%p)", 1, (server));
+ server->started = true;
server->pollsets = gpr_malloc(sizeof(grpc_pollset *) * server->cq_count);
for (i = 0; i < server->cq_count; i++) {
server->pollsets[i] = grpc_cq_pollset(server->cqs[i]);
}
+ request_matcher_init(&server->unregistered_request_matcher,
+ server->max_requested_calls, server);
+ for (registered_method *rm = server->registered_methods; rm; rm = rm->next) {
+ request_matcher_init(&rm->request_matcher, server->max_requested_calls,
+ server);
+ }
for (l = server->listeners; l; l = l->next) {
l->start(&exec_ctx, server, l->arg, server->pollsets, server->cq_count);
@@ -1039,6 +1066,7 @@ void grpc_server_start(grpc_server *server) {
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;
@@ -1063,6 +1091,17 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
server_ref(s);
chand->channel = channel;
+ size_t cq_idx;
+ grpc_completion_queue *accepting_cq = grpc_cq_from_pollset(accepting_pollset);
+ for (cq_idx = 0; cq_idx < s->cq_count; cq_idx++) {
+ if (s->cqs[cq_idx] == accepting_cq) break;
+ }
+ if (cq_idx == s->cq_count) {
+ /* completion queue not found: pick a random one to publish new calls to */
+ cq_idx = (size_t)rand() % s->cq_count;
+ }
+ chand->cq_idx = cq_idx;
+
num_registered_methods = 0;
for (rm = s->registered_methods; rm; rm = rm->next) {
num_registered_methods++;
@@ -1233,19 +1272,19 @@ void grpc_server_add_listener(
}
static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
- grpc_server *server,
+ grpc_server *server, size_t cq_idx,
requested_call *rc) {
call_data *calld = NULL;
request_matcher *rm = NULL;
int request_id;
if (gpr_atm_acq_load(&server->shutdown_flag)) {
- fail_call(exec_ctx, server, rc);
+ fail_call(exec_ctx, server, cq_idx, rc);
return GRPC_CALL_OK;
}
request_id = gpr_stack_lockfree_pop(server->request_freelist);
if (request_id == -1) {
/* out of request ids: just fail this one */
- fail_call(exec_ctx, server, rc);
+ fail_call(exec_ctx, server, cq_idx, rc);
return GRPC_CALL_OK;
}
switch (rc->type) {
@@ -1258,12 +1297,12 @@ static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
}
server->requested_calls[request_id] = *rc;
gpr_free(rc);
- if (gpr_stack_lockfree_push(rm->requests, request_id)) {
+ if (gpr_stack_lockfree_push(rm->requests_per_cq[cq_idx], request_id)) {
/* 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);
+ request_id = gpr_stack_lockfree_pop(rm->requests_per_cq[cq_idx]);
if (request_id == -1) break;
rm->pending_head = calld->pending_next;
gpr_mu_unlock(&server->mu_call);
@@ -1279,7 +1318,7 @@ static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(calld->state == PENDING);
calld->state = ACTIVATED;
gpr_mu_unlock(&calld->mu_state);
- publish_call(exec_ctx, server, calld,
+ publish_call(exec_ctx, server, calld, cq_idx,
&server->requested_calls[request_id]);
}
gpr_mu_lock(&server->mu_call);
@@ -1303,7 +1342,13 @@ grpc_call_error grpc_server_request_call(
"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));
- if (!grpc_cq_is_server_cq(cq_for_notification)) {
+ size_t cq_idx;
+ for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) {
+ if (server->cqs[cq_idx] == cq_for_notification) {
+ break;
+ }
+ }
+ if (cq_idx == server->cq_count) {
gpr_free(rc);
error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
goto done;
@@ -1314,11 +1359,10 @@ grpc_call_error grpc_server_request_call(
rc->server = server;
rc->tag = tag;
rc->cq_bound_to_call = cq_bound_to_call;
- rc->cq_for_notification = cq_for_notification;
rc->call = call;
rc->data.batch.details = details;
rc->initial_metadata = initial_metadata;
- error = queue_call_request(&exec_ctx, server, rc);
+ error = queue_call_request(&exec_ctx, server, cq_idx, rc);
done:
grpc_exec_ctx_finish(&exec_ctx);
return error;
@@ -1340,7 +1384,14 @@ grpc_call_error grpc_server_request_registered_call(
"tag=%p)",
9, (server, rmp, call, deadline, initial_metadata, optional_payload,
cq_bound_to_call, cq_for_notification, tag));
- if (!grpc_cq_is_server_cq(cq_for_notification)) {
+
+ size_t cq_idx;
+ for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) {
+ if (server->cqs[cq_idx] == cq_for_notification) {
+ break;
+ }
+ }
+ if (cq_idx == server->cq_count) {
gpr_free(rc);
error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
goto done;
@@ -1356,26 +1407,25 @@ grpc_call_error grpc_server_request_registered_call(
rc->server = server;
rc->tag = tag;
rc->cq_bound_to_call = cq_bound_to_call;
- rc->cq_for_notification = cq_for_notification;
rc->call = call;
rc->data.registered.registered_method = rm;
rc->data.registered.deadline = deadline;
rc->initial_metadata = initial_metadata;
rc->data.registered.optional_payload = optional_payload;
- error = queue_call_request(&exec_ctx, server, rc);
+ error = queue_call_request(&exec_ctx, server, cq_idx, rc);
done:
grpc_exec_ctx_finish(&exec_ctx);
return error;
}
static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
- requested_call *rc) {
+ size_t cq_idx, requested_call *rc) {
*rc->call = NULL;
rc->initial_metadata->count = 0;
server_ref(server);
- grpc_cq_end_op(exec_ctx, rc->cq_for_notification, rc->tag, 0,
- done_request_event, rc, &rc->completion);
+ grpc_cq_end_op(exec_ctx, server->cqs[cq_idx], rc->tag, 0, done_request_event,
+ rc, &rc->completion);
}
const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {
diff --git a/src/core/lib/surface/server.h b/src/core/lib/surface/server.h
index 470ef23c69..fb6e4d60c5 100644
--- a/src/core/lib/surface/server.h
+++ b/src/core/lib/surface/server.h
@@ -53,6 +53,7 @@ void grpc_server_add_listener(
server */
void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *server,
grpc_transport *transport,
+ grpc_pollset *accepting_pollset,
const grpc_channel_args *args);
const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server);
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index f955a31494..854057efbc 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -295,7 +295,12 @@ Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
grpc_channel_args channel_args;
args->SetChannelArgs(&channel_args);
server_ = grpc_server_create(&channel_args, nullptr);
- grpc_server_register_completion_queue(server_, cq_.cq(), nullptr);
+ if (thread_pool_ == nullptr) {
+ grpc_server_register_non_listening_completion_queue(server_, cq_.cq(),
+ nullptr);
+ } else {
+ grpc_server_register_completion_queue(server_, cq_.cq(), nullptr);
+ }
}
Server::~Server() {
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index a5bcd3db31..391932c88e 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -128,10 +128,13 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
compression_options_.enabled_algorithms_bitset);
std::unique_ptr<Server> server(
new Server(thread_pool.release(), true, max_message_size_, &args));
-
ServerInitializer* initializer = server->initializer();
- int num_non_listening_cqs = 0;
+ // If the server has atleast one sync methods, we know that this is a Sync
+ // server or a Hybrid server and the completion queue (server->cq_) would be
+ // frequently polled.
+ int num_frequently_polled_cqs = has_sync_methods ? 1 : 0;
+
for (auto cq = cqs_.begin(); cq != cqs_.end(); ++cq) {
// A completion queue that is not polled frequently (by calling Next() or
// AsyncNext()) is not safe to use for listening to incoming channels.
@@ -147,12 +150,10 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
}
}
- // TODO: (sreek) - Find a good way to determine whether the server is a Sync
- // server or an Async server. In case of Async server, return an error if all
- // the completion queues are non-listening
- if (num_non_listening_cqs > 0) {
- gpr_log(GPR_INFO, "Number of non listening completion queues: %d out of %d",
- num_non_listening_cqs, cqs_.size());
+ if (num_frequently_polled_cqs == 0) {
+ gpr_log(GPR_ERROR,
+ "At least one of the completion queues must be frequently polled");
+ return nullptr;
}
for (auto service = services_.begin(); service != services_.end();
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-30 12:27:38 +0000