Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Multi-completion-queue-server

Browse Source 
Allow binding a different completion queue to each registered method.
This will allow multiplexing for the C++ server between sync & async
methods more easily.
pull/501/head
Craig Tiller 10 years ago
parent fada7d43b1
commit 20bc56d890
9 changed files with 77 additions and 40 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      include/grpc/grpc.h
  2. 5
      src/core/iomgr/tcp_server.h
  3. 11
      src/core/iomgr/tcp_server_posix.c
  4. 5
      src/core/security/server_secure_chttp2.c
  5. 79
      src/core/surface/server.c
  6. 2
      src/core/surface/server.h
  7. 4
      src/core/surface/server_chttp2.c
  8. 2
      src/cpp/server/server.cc
  9. 6
      test/core/iomgr/tcp_server_posix_test.c

3
include/grpc/grpc.h
Unescape View File

@ -565,7 +565,8 @@ grpc_call_error grpc_server_request_call(
Must be called before grpc_server_start.
Returns NULL on failure. */
void *grpc_server_register_method(grpc_server *server, const char *method,
const char *host);
const char *host,
grpc_completion_queue *new_call_cq);
/* Request notification of a new pre-registered call */
grpc_call_error grpc_server_request_registered_call(

5
src/core/iomgr/tcp_server.h
Unescape View File

@ -46,8 +46,9 @@ typedef void (*grpc_tcp_server_cb)(void *arg, grpc_endpoint *ep);
grpc_tcp_server *grpc_tcp_server_create(void);
/* Start listening to bound ports */
void grpc_tcp_server_start(grpc_tcp_server *server, grpc_pollset *pollset,
grpc_tcp_server_cb cb, void *cb_arg);
void grpc_tcp_server_start(grpc_tcp_server *server, grpc_pollset **pollsets,
size_t pollset_count, grpc_tcp_server_cb cb,
void *cb_arg);
/* Add a port to the server, returning port number on success, or negative
on failure.

11
src/core/iomgr/tcp_server_posix.c
Unescape View File

@ -353,9 +353,10 @@ int grpc_tcp_server_get_fd(grpc_tcp_server *s, unsigned index) {
return (index < s->nports) ? s->ports[index].fd : -1;
}
void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset *pollset,
grpc_tcp_server_cb cb, void *cb_arg) {
size_t i;
void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollsets,
size_t pollset_count, grpc_tcp_server_cb cb,
void *cb_arg) {
size_t i, j;
GPR_ASSERT(cb);
gpr_mu_lock(&s->mu);
GPR_ASSERT(!s->cb);
@ -363,8 +364,8 @@ void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset *pollset,
s->cb = cb;
s->cb_arg = cb_arg;
for (i = 0; i < s->nports; i++) {
if (pollset) {
grpc_pollset_add_fd(pollset, s->ports[i].emfd);
for (j = 0; j < pollset_count; j++) {
grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd);
}
grpc_fd_notify_on_read(s->ports[i].emfd, on_read, &s->ports[i]);
s->active_ports++;

5
src/core/security/server_secure_chttp2.c
Unescape View File

@ -76,9 +76,10 @@ static void on_accept(void *server, grpc_endpoint *tcp) {
/* Note: the following code is the same with server_chttp2.c */
/* Server callback: start listening on our ports */
static void start(grpc_server *server, void *tcpp, grpc_pollset *pollset) {
static void start(grpc_server *server, void *tcpp, grpc_pollset **pollsets,
size_t pollset_count) {
grpc_tcp_server *tcp = tcpp;
grpc_tcp_server_start(tcp, pollset, on_accept, server);
grpc_tcp_server_start(tcp, pollsets, pollset_count, on_accept, server);
}
/* Server callback: destroy the tcp listener (so we don't generate further

79
src/core/surface/server.c
Unescape View File

@ -53,7 +53,7 @@ typedef enum { PENDING_START, ALL_CALLS, CALL_LIST_COUNT } call_list;
typedef struct listener {
void *arg;
void (*start)(grpc_server *server, void *arg, grpc_pollset *pollset);
void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets, size_t pollset_count);
void (*destroy)(grpc_server *server, void *arg);
struct listener *next;
} listener;
@ -101,6 +101,7 @@ struct registered_method {
char *host;
call_data *pending;
requested_call_array requested;
grpc_completion_queue *cq;
registered_method *next;
};
@ -127,7 +128,11 @@ struct grpc_server {
size_t channel_filter_count;
const grpc_channel_filter **channel_filters;
grpc_channel_args *channel_args;
grpc_completion_queue *cq;
grpc_completion_queue *unregistered_cq;
grpc_completion_queue **cqs;
grpc_pollset **pollsets;
size_t cq_count;
gpr_mu mu;
@ -169,6 +174,7 @@ struct call_data {
grpc_mdstr *host;
legacy_data *legacy;
grpc_completion_queue *cq_new;
call_data **root[CALL_LIST_COUNT];
call_link links[CALL_LIST_COUNT];
@ -496,7 +502,7 @@ 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 i;
size_t i;
gpr_mu_lock(&chand->server->mu);
for (i = 0; i < CALL_LIST_COUNT; i++) {
@ -504,7 +510,9 @@ static void destroy_call_elem(grpc_call_element *elem) {
}
if (chand->server->shutdown && chand->server->have_shutdown_tag &&
chand->server->lists[ALL_CALLS] == NULL) {
grpc_cq_end_server_shutdown(chand->server->cq, chand->server->shutdown_tag);
for (i = 0; i < chand->server->cq_count; i++) {
grpc_cq_end_server_shutdown(chand->server->cqs[i], chand->server->shutdown_tag);
}
}
gpr_mu_unlock(&chand->server->mu);
@ -557,6 +565,16 @@ static const grpc_channel_filter server_surface_filter = {
sizeof(channel_data), init_channel_elem, destroy_channel_elem, "server",
};
static void addcq(grpc_server *server, grpc_completion_queue *cq) {
size_t i, n;
for (i = 0; i < server->cq_count; i++) {
if (server->cqs[i] == cq) return;
}
n = server->cq_count++;
server->cqs = gpr_realloc(server->cqs, server->cq_count * sizeof(grpc_completion_queue*));
server->cqs[n] = cq;
}
grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
grpc_channel_filter **filters,
size_t filter_count,
@ -566,10 +584,11 @@ grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
grpc_server *server = gpr_malloc(sizeof(grpc_server));
memset(server, 0, sizeof(grpc_server));
if (cq) addcq(server, cq);
gpr_mu_init(&server->mu);
server->cq = cq;
server->unregistered_cq = cq;
/* decremented by grpc_server_destroy */
gpr_ref_init(&server->internal_refcount, 1);
server->root_channel_data.next = server->root_channel_data.prev =
@ -605,7 +624,7 @@ static int streq(const char *a, const char *b) {
}
void *grpc_server_register_method(grpc_server *server, const char *method,
const char *host) {
const char *host, grpc_completion_queue *cq_new_rpc) {
registered_method *m;
if (!method) {
gpr_log(GPR_ERROR, "%s method string cannot be NULL", __FUNCTION__);
@ -618,20 +637,28 @@ void *grpc_server_register_method(grpc_server *server, const char *method,
return NULL;
}
}
addcq(server, cq_new_rpc);
m = gpr_malloc(sizeof(registered_method));
memset(m, 0, sizeof(*m));
m->method = gpr_strdup(method);
m->host = gpr_strdup(host);
m->next = server->registered_methods;
m->cq = cq_new_rpc;
server->registered_methods = m;
return m;
}
void grpc_server_start(grpc_server *server) {
listener *l;
size_t i;
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]);
}
for (l = server->listeners; l; l = l->next) {
l->start(server, l->arg, grpc_cq_pollset(server->cq));
l->start(server, l->arg, server->pollsets, server->cq_count);
}
}
@ -664,7 +691,9 @@ grpc_transport_setup_result grpc_server_setup_transport(
}
filters[i] = &grpc_connected_channel_filter;
grpc_transport_add_to_pollset(transport, grpc_cq_pollset(s->cq));
for (i = 0; i < s->cq_count; i++) {
grpc_transport_add_to_pollset(transport, grpc_cq_pollset(s->cqs[i]));
}
channel = grpc_channel_create_from_filters(filters, num_filters,
s->channel_args, mdctx, 0);
@ -765,9 +794,11 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
server->have_shutdown_tag = have_shutdown_tag;
server->shutdown_tag = shutdown_tag;
if (have_shutdown_tag) {
grpc_cq_begin_op(server->cq, NULL, GRPC_SERVER_SHUTDOWN);
if (server->lists[ALL_CALLS] == NULL) {
grpc_cq_end_server_shutdown(server->cq, shutdown_tag);
for (i = 0; i < server->cq_count; i++) {
grpc_cq_begin_op(server->cqs[i], NULL, GRPC_SERVER_SHUTDOWN);
if (server->lists[ALL_CALLS] == NULL) {
grpc_cq_end_server_shutdown(server->cqs[i], shutdown_tag);
}
}
}
gpr_mu_unlock(&server->mu);
@ -826,7 +857,7 @@ void grpc_server_destroy(grpc_server *server) {
void grpc_server_add_listener(grpc_server *server, void *arg,
void (*start)(grpc_server *server, void *arg,
grpc_pollset *pollset),
grpc_pollset **pollsets, size_t pollset_count),
void (*destroy)(grpc_server *server, void *arg)) {
listener *l = gpr_malloc(sizeof(listener));
l->arg = arg;
@ -878,7 +909,7 @@ grpc_call_error grpc_server_request_call(grpc_server *server, grpc_call **call,
grpc_completion_queue *cq_bind,
void *tag) {
requested_call rc;
grpc_cq_begin_op(server->cq, NULL, GRPC_OP_COMPLETE);
grpc_cq_begin_op(server->unregistered_cq, NULL, GRPC_OP_COMPLETE);
rc.type = BATCH_CALL;
rc.tag = tag;
rc.data.batch.cq_bind = cq_bind;
@ -889,12 +920,13 @@ grpc_call_error grpc_server_request_call(grpc_server *server, grpc_call **call,
}
grpc_call_error grpc_server_request_registered_call(
grpc_server *server, void *registered_method, grpc_call **call,
grpc_server *server, void *rm, grpc_call **call,
gpr_timespec *deadline, grpc_metadata_array *initial_metadata,
grpc_byte_buffer **optional_payload, grpc_completion_queue *cq_bind,
void *tag) {
requested_call rc;
grpc_cq_begin_op(server->cq, NULL, GRPC_OP_COMPLETE);
registered_method *registered_method = rm;
grpc_cq_begin_op(registered_method->cq, NULL, GRPC_OP_COMPLETE);
rc.type = REGISTERED_CALL;
rc.tag = tag;
rc.data.registered.cq_bind = cq_bind;
@ -909,7 +941,7 @@ grpc_call_error grpc_server_request_registered_call(
grpc_call_error grpc_server_request_call_old(grpc_server *server,
void *tag_new) {
requested_call rc;
grpc_cq_begin_op(server->cq, NULL, GRPC_SERVER_RPC_NEW);
grpc_cq_begin_op(server->unregistered_cq, NULL, GRPC_SERVER_RPC_NEW);
rc.type = LEGACY_CALL;
rc.tag = tag_new;
return queue_call_request(server, &rc);
@ -965,6 +997,7 @@ static void begin_call(grpc_server *server, call_data *calld,
r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
r->data.recv_metadata = rc->data.batch.initial_metadata;
r++;
calld->cq_new = server->unregistered_cq;
publish = publish_registered_or_batch;
break;
case REGISTERED_CALL:
@ -979,6 +1012,7 @@ static void begin_call(grpc_server *server, call_data *calld,
r->data.recv_message = rc->data.registered.optional_payload;
r++;
}
calld->cq_new = rc->data.registered.registered_method->cq;
publish = publish_registered_or_batch;
break;
}
@ -991,19 +1025,19 @@ static void begin_call(grpc_server *server, call_data *calld,
static void fail_call(grpc_server *server, requested_call *rc) {
switch (rc->type) {
case LEGACY_CALL:
grpc_cq_end_new_rpc(server->cq, rc->tag, NULL, do_nothing, NULL, NULL,
grpc_cq_end_new_rpc(server->unregistered_cq, rc->tag, NULL, do_nothing, NULL, NULL,
NULL, gpr_inf_past, 0, NULL);
break;
case BATCH_CALL:
*rc->data.batch.call = NULL;
rc->data.batch.initial_metadata->count = 0;
grpc_cq_end_op_complete(server->cq, rc->tag, NULL, do_nothing, NULL,
grpc_cq_end_op_complete(server->unregistered_cq, rc->tag, NULL, do_nothing, NULL,
GRPC_OP_ERROR);
break;
case REGISTERED_CALL:
*rc->data.registered.call = NULL;
rc->data.registered.initial_metadata->count = 0;
grpc_cq_end_op_complete(server->cq, rc->tag, NULL, do_nothing, NULL,
grpc_cq_end_op_complete(rc->data.registered.registered_method->cq, rc->tag, NULL, do_nothing, NULL,
GRPC_OP_ERROR);
break;
}
@ -1017,7 +1051,7 @@ static void publish_legacy(grpc_call *call, grpc_op_error status, void *tag) {
grpc_server *server = chand->server;
if (status == GRPC_OP_OK) {
grpc_cq_end_new_rpc(server->cq, tag, call, do_nothing, NULL,
grpc_cq_end_new_rpc(server->unregistered_cq, tag, call, do_nothing, NULL,
grpc_mdstr_as_c_string(calld->path),
grpc_mdstr_as_c_string(calld->host), calld->deadline,
calld->legacy->initial_metadata.count,
@ -1032,9 +1066,8 @@ static void publish_registered_or_batch(grpc_call *call, grpc_op_error status,
void *tag) {
grpc_call_element *elem =
grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
channel_data *chand = elem->channel_data;
grpc_server *server = chand->server;
grpc_cq_end_op_complete(server->cq, tag, call, do_nothing, NULL, status);
call_data *calld = elem->call_data;
grpc_cq_end_op_complete(calld->cq_new, tag, call, do_nothing, NULL, status);
}
const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {

2
src/core/surface/server.h
Unescape View File

@ -48,7 +48,7 @@ grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
and when it shuts down, it will call destroy */
void grpc_server_add_listener(grpc_server *server, void *listener,
void (*start)(grpc_server *server, void *arg,
grpc_pollset *pollset),
grpc_pollset **pollsets, size_t npollsets),
void (*destroy)(grpc_server *server, void *arg));
/* Setup a transport - creates a channel stack, binds the transport to the

4
src/core/surface/server_chttp2.c
Unescape View File

@ -59,9 +59,9 @@ static void new_transport(void *server, grpc_endpoint *tcp) {
}
/* Server callback: start listening on our ports */
static void start(grpc_server *server, void *tcpp, grpc_pollset *pollset) {
static void start(grpc_server *server, void *tcpp, grpc_pollset **pollsets, size_t pollset_count) {
grpc_tcp_server *tcp = tcpp;
grpc_tcp_server_start(tcp, pollset, new_transport, server);
grpc_tcp_server_start(tcp, pollsets, pollset_count, new_transport, server);
}
/* Server callback: destroy the tcp listener (so we don't generate further

2
src/cpp/server/server.cc
Unescape View File

@ -84,7 +84,7 @@ Server::~Server() {
bool Server::RegisterService(RpcService *service) {
for (int i = 0; i < service->GetMethodCount(); ++i) {
RpcServiceMethod *method = service->GetMethod(i);
void *tag = grpc_server_register_method(server_, method->name(), nullptr);
void *tag = grpc_server_register_method(server_, method->name(), nullptr, cq_.cq());
if (!tag) {
gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
method->name());

6
test/core/iomgr/tcp_server_posix_test.c
Unescape View File

@ -66,7 +66,7 @@ static void test_no_op(void) {
static void test_no_op_with_start(void) {
grpc_tcp_server *s = grpc_tcp_server_create();
LOG_TEST();
grpc_tcp_server_start(s, NULL, on_connect, NULL);
grpc_tcp_server_start(s, NULL, 0, on_connect, NULL);
grpc_tcp_server_destroy(s);
}
@ -93,7 +93,7 @@ static void test_no_op_with_port_and_start(void) {
GPR_ASSERT(
grpc_tcp_server_add_port(s, (struct sockaddr *)&addr, sizeof(addr)));
grpc_tcp_server_start(s, NULL, on_connect, NULL);
grpc_tcp_server_start(s, NULL, 0, on_connect, NULL);
grpc_tcp_server_destroy(s);
}
@ -120,7 +120,7 @@ static void test_connect(int n) {
GPR_ASSERT(getsockname(svrfd, (struct sockaddr *)&addr, &addr_len) == 0);
GPR_ASSERT(addr_len <= sizeof(addr));
grpc_tcp_server_start(s, NULL, on_connect, NULL);
grpc_tcp_server_start(s, NULL, 0, on_connect, NULL);
for (i = 0; i < n; i++) {
deadline = gpr_time_add(gpr_now(), gpr_time_from_micros(10000000));

Write Preview
Loading…
Cancel
Save