Simplify the Cython code

src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pxd.pxi

@@ -13,20 +13,18 @@
 # limitations under the License.
 
 cdef class BaseCompletionQueue:
+    cdef grpc_completion_queue *_cq
 
     cdef grpc_completion_queue* c_ptr(self)
 
 cdef class PollerCompletionQueue(BaseCompletionQueue):
-    cdef grpc_completion_queue *_cq
     cdef bint _shutdown
     cdef object _shutdown_completed
-    cdef object _poller
-    cdef object _poller_running
+    cdef object _poller_thread
 
     cdef void _poll(self) except *
 
 
 cdef class CallbackCompletionQueue(BaseCompletionQueue):
-    cdef grpc_completion_queue *_cq
     cdef object _shutdown_completed  # asyncio.Future
     cdef CallbackWrapper _wrapper

src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pyx.pxi

@@ -25,7 +25,7 @@ cdef class BaseCompletionQueue:
         raise NotImplementedError()
 
     cdef grpc_completion_queue* c_ptr(self):
-        raise NotImplementedError()
+        return self._cq
 
 
 cdef class PollerCompletionQueue(BaseCompletionQueue):
@@ -34,17 +34,13 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
         self._cq = grpc_completion_queue_create_for_next(NULL)
         self._shutdown = False
         self._shutdown_completed = asyncio.get_event_loop().create_future()
-        self._poller = None
-        self._poller_running = asyncio.get_event_loop().create_future()
-        self._poller = threading.Thread(target=self._poll_wrapper)
-        self._poller.daemon = True
-        self._poller.start()
+        self._poller_thread = threading.Thread(target=self._poll_wrapper, daemon=True)
+        self._poller_thread.start()
 
     cdef void _poll(self) except *:
         cdef grpc_event event
         cdef CallbackContext *context
         cdef object waiter
-        grpc_call_soon_threadsafe(self._poller_running.set_result, None)
 
         while not self._shutdown:
             with nogil:
@@ -56,10 +52,10 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
                 raise AssertionError("Core should not return timeout error!")
             elif event.type == GRPC_QUEUE_SHUTDOWN:
                 self._shutdown = True
-                grpc_call_soon_threadsafe(self._shutdown_completed.set_result, None)
+                aio_loop_call_soon_threadsafe(self._shutdown_completed.set_result, None)
             else:
                 context = <CallbackContext *>event.tag
-                grpc_call_soon_threadsafe(
+                aio_loop_call_soon_threadsafe(
                     _handle_callback_wrapper,
                     <CallbackWrapper>context.callback_wrapper,
                     event.success)
@@ -71,9 +67,7 @@ cdef class PollerCompletionQueue(BaseCompletionQueue):
         grpc_completion_queue_shutdown(self._cq)
         await self._shutdown_completed
         grpc_completion_queue_destroy(self._cq)
-
-    cdef grpc_completion_queue* c_ptr(self):
-        return self._cq
+        self._poller_thread.join()
 
 
 cdef class CallbackCompletionQueue(BaseCompletionQueue):
@@ -88,9 +82,6 @@ cdef class CallbackCompletionQueue(BaseCompletionQueue):
             NULL
         )
 
-    cdef grpc_completion_queue* c_ptr(self):
-        return self._cq
-
     async def shutdown(self):
         grpc_completion_queue_shutdown(self._cq)
         await self._shutdown_completed

src/python/grpcio/grpc/_cython/_cygrpc/aio/grpc_aio.pyx.pxi

@@ -88,7 +88,7 @@ def grpc_aio_loop():
     return _grpc_aio_loop
 
 
-def grpc_schedule_coroutine(object coro):
+def aio_loop_schedule_coroutine(object coro):
     """Thread-safely schedules coroutine to gRPC Aio event loop.
 
     If invoked within the same thread as the event loop, return an
@@ -102,7 +102,7 @@ def grpc_schedule_coroutine(object coro):
         return _grpc_aio_loop.create_task(coro)
 
 
-def grpc_call_soon_threadsafe(object func, *args):
+def aio_loop_call_soon_threadsafe(object func, *args):
     # TODO(lidiz) After we are confident, we can drop this assert. Otherwsie,
     # we should limit this function to non-grpc-event-loop thread.
     assert _event_loop_thread_ident != threading.current_thread().ident

src/python/grpcio/grpc/experimental/aio/_server.py

@@ -162,7 +162,7 @@ class Server(_base_server.Server):
         be safe to slightly extend the underlying Cython object's life span.
         """
         if hasattr(self, '_server'):
-            cygrpc.grpc_schedule_coroutine(self._server.shutdown(None))
+            cygrpc.aio_loop_schedule_coroutine(self._server.shutdown(None))
 
 
 def server(migration_thread_pool: Optional[Executor] = None,