From cc357bf74e27888a354f361ad3b66e4e2b565dbc Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Mon, 22 Aug 2016 16:08:50 -0700 Subject: [PATCH 1/2] Remove completion queue parameter from grpc_server_add_insecure_channel_from_fd API --- include/grpc/grpc_posix.h | 11 +++++++---- .../server/insecure/server_chttp2_posix.c | 19 +++++++++++++------ src/core/lib/surface/server.c | 6 ++++++ src/core/lib/surface/server.h | 5 +++++ src/cpp/server/server_posix.cc | 3 +-- src/ruby/ext/grpc/rb_grpc_imports.generated.h | 2 +- test/core/end2end/fixtures/h2_fd.c | 2 +- 7 files changed, 34 insertions(+), 14 deletions(-) diff --git a/include/grpc/grpc_posix.h b/include/grpc/grpc_posix.h index 5e89ae3b1ee..b426c329ef3 100644 --- a/include/grpc/grpc_posix.h +++ b/include/grpc/grpc_posix.h @@ -57,11 +57,14 @@ GRPCAPI grpc_channel *grpc_insecure_channel_create_from_fd( /** Add the connected communication channel based on file descriptor 'fd' to the 'server'. The 'fd' must be an open file descriptor corresponding to a - connected socket. The 'cq' is a completion queue that will be getting events - from that descriptor. */ + connected socket. Events from the file descriptor may come on any of the + server completion queues (i.e completion queues registered via the + grpc_server_register_completion_queue API). + + The 'reserved' pointer MUST be NULL. + */ GRPCAPI void grpc_server_add_insecure_channel_from_fd(grpc_server *server, - grpc_completion_queue *cq, - int fd); + void *reserved, int fd); /** GRPC Core POSIX library may internally use signals to optimize some work. The library uses (SIGRTMIN + 2) signal by default. Use this API to instruct diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c index 4350543c27a..9af17fb5ae7 100644 --- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c +++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.c @@ -50,10 +50,10 @@ #include "src/core/lib/surface/server.h" void grpc_server_add_insecure_channel_from_fd(grpc_server *server, - grpc_completion_queue *cq, - int fd) { - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + void *reserved, int fd) { + GPR_ASSERT(reserved == NULL); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; char *name; gpr_asprintf(&name, "fd:%d", fd); @@ -65,7 +65,15 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server *server, const grpc_channel_args *server_args = grpc_server_get_channel_args(server); grpc_transport *transport = grpc_create_chttp2_transport( &exec_ctx, server_args, server_endpoint, 0 /* is_client */); - grpc_endpoint_add_to_pollset(&exec_ctx, server_endpoint, grpc_cq_pollset(cq)); + + grpc_pollset **pollsets; + size_t num_pollsets = 0; + grpc_server_get_pollsets(server, &pollsets, &num_pollsets); + + for (size_t i = 0; i < num_pollsets; i++) { + grpc_endpoint_add_to_pollset(&exec_ctx, server_endpoint, pollsets[i]); + } + grpc_server_setup_transport(&exec_ctx, server, transport, NULL, server_args); grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL); grpc_exec_ctx_finish(&exec_ctx); @@ -74,8 +82,7 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server *server, #else // !GPR_SUPPORT_CHANNELS_FROM_FD void grpc_server_add_insecure_channel_from_fd(grpc_server *server, - grpc_completion_queue *cq, - int fd) { + void *reserved, int fd) { GPR_ASSERT(0); } diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 64afcecc072..8811b6e8033 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -1098,6 +1098,12 @@ 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) { + *pollset_count = server->cq_count; + *pollsets = server->pollsets; +} + void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, grpc_transport *transport, grpc_pollset *accepting_pollset, diff --git a/src/core/lib/surface/server.h b/src/core/lib/surface/server.h index fb6e4d60c50..551a40a4ff7 100644 --- a/src/core/lib/surface/server.h +++ b/src/core/lib/surface/server.h @@ -60,4 +60,9 @@ const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server); int grpc_server_has_open_connections(grpc_server *server); +/* Do not call this before grpc_server_start. Returns the pollsets and the + * number of pollsets via 'pollsets' and 'pollset_count'. */ +void grpc_server_get_pollsets(grpc_server *server, grpc_pollset ***pollsets, + size_t *pollset_count); + #endif /* GRPC_CORE_LIB_SURFACE_SERVER_H */ diff --git a/src/cpp/server/server_posix.cc b/src/cpp/server/server_posix.cc index c3aa2adc60e..33d42a8dc70 100644 --- a/src/cpp/server/server_posix.cc +++ b/src/cpp/server/server_posix.cc @@ -40,8 +40,7 @@ namespace grpc { #ifdef GPR_SUPPORT_CHANNELS_FROM_FD void AddInsecureChannelFromFd(Server* server, int fd) { - grpc_server_add_insecure_channel_from_fd( - server->c_server(), server->completion_queue()->cq(), fd); + grpc_server_add_insecure_channel_from_fd(server->c_server(), NULL, fd); } #endif // GPR_SUPPORT_CHANNELS_FROM_FD diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h index 3bb76fbb97e..b3960b462dd 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h @@ -317,7 +317,7 @@ extern grpc_call_error_to_string_type grpc_call_error_to_string_import; typedef grpc_channel *(*grpc_insecure_channel_create_from_fd_type)(const char *target, int fd, const grpc_channel_args *args); extern grpc_insecure_channel_create_from_fd_type grpc_insecure_channel_create_from_fd_import; #define grpc_insecure_channel_create_from_fd grpc_insecure_channel_create_from_fd_import -typedef void(*grpc_server_add_insecure_channel_from_fd_type)(grpc_server *server, grpc_completion_queue *cq, int fd); +typedef void(*grpc_server_add_insecure_channel_from_fd_type)(grpc_server *server, void *reserved, int fd); extern grpc_server_add_insecure_channel_from_fd_type grpc_server_add_insecure_channel_from_fd_import; #define grpc_server_add_insecure_channel_from_fd grpc_server_add_insecure_channel_from_fd_import typedef void(*grpc_use_signal_type)(int signum); diff --git a/test/core/end2end/fixtures/h2_fd.c b/test/core/end2end/fixtures/h2_fd.c index 89fa02517df..8561feed703 100644 --- a/test/core/end2end/fixtures/h2_fd.c +++ b/test/core/end2end/fixtures/h2_fd.c @@ -95,7 +95,7 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f, grpc_server_register_completion_queue(f->server, f->cq, NULL); grpc_server_start(f->server); - grpc_server_add_insecure_channel_from_fd(f->server, f->cq, sfd->fd_pair[1]); + grpc_server_add_insecure_channel_from_fd(f->server, NULL, sfd->fd_pair[1]); grpc_exec_ctx_finish(&exec_ctx); } From 509ee4c11127700986ceb05c020d2ceef5667852 Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Mon, 22 Aug 2016 17:48:41 -0700 Subject: [PATCH 2/2] Remove the API to return the underlying completion queue. This API is no longer needed --- include/grpc++/server.h | 3 --- src/cpp/server/server.cc | 2 -- 2 files changed, 5 deletions(-) diff --git a/include/grpc++/server.h b/include/grpc++/server.h index 6876961e210..87536edbe8d 100644 --- a/include/grpc++/server.h +++ b/include/grpc++/server.h @@ -96,9 +96,6 @@ class Server GRPC_FINAL : public ServerInterface, private GrpcLibraryCodegen { // Returns a \em raw pointer to the underlying grpc_server instance. grpc_server* c_server(); - // Returns a \em raw pointer to the underlying CompletionQueue. - CompletionQueue* completion_queue(); - private: friend class AsyncGenericService; friend class ServerBuilder; diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc index af04fd4ca64..0effa9f125e 100644 --- a/src/cpp/server/server.cc +++ b/src/cpp/server/server.cc @@ -332,8 +332,6 @@ void Server::SetGlobalCallbacks(GlobalCallbacks* callbacks) { grpc_server* Server::c_server() { return server_; } -CompletionQueue* Server::completion_queue() { return &cq_; } - static grpc_server_register_method_payload_handling PayloadHandlingForMethod( RpcServiceMethod* method) { switch (method->method_type()) {