Merge pull request #7833 from sreecha/fix_channel_from_fd_api

Remove completion queue parameter from `grpc_server_add_insecure_channel_from_fd` API

9 files changed, 34 insertions, 19 deletions

diff --git a/include/grpc++/server.h b/include/grpc++/server.h
index 6e15035125..f51a6c658f 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/include/grpc/grpc_posix.h b/include/grpc/grpc_posix.h
index 6d77f0fb82..a8c031548b 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 + 6) 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 4350543c27..9af17fb5ae 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 56fb80e92e..7300d79b9f 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -1106,6 +1106,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 fb6e4d60c5..551a40a4ff 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_cc.cc b/src/cpp/server/server_cc.cc
index a693ce9b8e..3f89275370 100644
--- a/src/cpp/server/server_cc.cc
+++ b/src/cpp/server/server_cc.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()) {
diff --git a/src/cpp/server/server_posix.cc b/src/cpp/server/server_posix.cc
index c3aa2adc60..33d42a8dc7 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 b87abfd8ef..00a67b0b2c 100644
--- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h
+++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
@@ -350,7 +350,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 89fa02517d..8561feed70 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);
 }