Adopt reviewer's advices

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

@@ -27,7 +27,7 @@ cdef grpc_custom_poller_vtable asyncio_pollset_vtable
 cdef grpc_error* asyncio_socket_init(
         grpc_custom_socket* grpc_socket,
         int domain) with gil:
-    socket = _AsyncioSocket.create(grpc_socket)
+    socket = _AsyncioSocket.create(grpc_socket, None, None)
     Py_INCREF(socket)
     grpc_socket.impl = <void*>socket
     return <grpc_error*>0
@@ -115,12 +115,11 @@ cdef grpc_error* asyncio_socket_listen(grpc_custom_socket* grpc_socket) with gil
     return grpc_error_none()
 
 
-# TODO(lidiz) connects the so_reuse_port option to channel arguments
 def _asyncio_apply_socket_options(object s, so_reuse_port=False):
-  s.setsockopt(native_socket.SOL_SOCKET, native_socket.SO_REUSEADDR, 1)
-  if so_reuse_port:
-    s.setsockopt(native_socket.SOL_SOCKET, native_socket.SO_REUSEPORT, 1)
-  s.setsockopt(native_socket.IPPROTO_TCP, native_socket.TCP_NODELAY, True)
+    # TODO(https://github.com/grpc/grpc/issues/20667)
+    # Connects the so_reuse_port option to channel arguments
+    s.setsockopt(native_socket.SOL_SOCKET, native_socket.SO_REUSEADDR, 1)
+    s.setsockopt(native_socket.IPPROTO_TCP, native_socket.TCP_NODELAY, True)
 
 
 cdef grpc_error* asyncio_socket_bind(

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/socket.pxd.pxi

@@ -35,7 +35,10 @@ cdef class _AsyncioSocket:
         object _peername
 
     @staticmethod
-    cdef _AsyncioSocket create(grpc_custom_socket * grpc_socket)
+    cdef _AsyncioSocket create(
+            grpc_custom_socket * grpc_socket,
+            object reader,
+            object writer)
     @staticmethod
     cdef _AsyncioSocket create_with_py_socket(grpc_custom_socket * grpc_socket, object py_socket)
 

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/socket.pyx.pxi

@@ -28,11 +28,18 @@ cdef class _AsyncioSocket:
         self._read_buffer = NULL
         self._server = None
         self._py_socket = None
+        self._peername = None
 
     @staticmethod
-    cdef _AsyncioSocket create(grpc_custom_socket * grpc_socket):
+    cdef _AsyncioSocket create(grpc_custom_socket * grpc_socket,
+                               object reader,
+                               object writer):
         socket = _AsyncioSocket()
         socket._grpc_socket = grpc_socket
+        socket._reader = reader
+        socket._writer = writer
+        if writer is not None:
+            socket._peername = writer.get_extra_info('peername')
         return socket
 
     @staticmethod
@@ -101,7 +108,13 @@ cdef class _AsyncioSocket:
                 grpc_socket_error("read {}".format(error_msg).encode())
             )
 
-    cdef void connect(self, object host, object port, grpc_custom_connect_callback grpc_connect_cb):
+    cdef void connect(self,
+                      object host,
+                      object port,
+                      grpc_custom_connect_callback grpc_connect_cb):
+        if self._reader:
+            return
+
         assert not self._task_connect
 
         self._task_connect = asyncio.ensure_future(
@@ -143,10 +156,11 @@ cdef class _AsyncioSocket:
             self._writer.close()
 
     def _new_connection_callback(self, object reader, object writer):
-        client_socket = _AsyncioSocket.create(self._grpc_client_socket)
-        client_socket._reader = reader
-        client_socket._writer = writer
-        client_socket._peername = addr = writer.get_extra_info('peername')
+        client_socket = _AsyncioSocket.create(
+            self._grpc_client_socket,
+            reader,
+            writer,
+        )
 
         self._grpc_client_socket.impl = <void*>client_socket
         cpython.Py_INCREF(client_socket)

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

@@ -25,10 +25,18 @@ cdef class RPCState:
     cdef bytes method(self)
 
 
+cdef enum AioServerStatus:
+    AIO_SERVER_STATUS_UNKNOWN
+    AIO_SERVER_STATUS_READY
+    AIO_SERVER_STATUS_RUNNING
+    AIO_SERVER_STATUS_STOPPED
+
+
 cdef class _AioServerState:
     cdef Server server
     cdef grpc_completion_queue *cq
     cdef list generic_handlers
+    cdef AioServerStatus status
 
 
 cdef class AioServer:

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

@@ -21,21 +21,26 @@ cdef class _HandlerCallDetails:
 class _ServicerContextPlaceHolder(object): pass
 
 
+# TODO(https://github.com/grpc/grpc/issues/20669)
+# Apply this to the client-side
 cdef class CallbackWrapper:
     cdef CallbackContext context
-    cdef object future
+    cdef object _keep_reference
 
     def __cinit__(self, object future):
         self.context.functor.functor_run = self.functor_run
         self.context.waiter = <cpython.PyObject*>(future)
-        self.future = future
+        self._keep_reference = future
 
     @staticmethod
     cdef void functor_run(
             grpc_experimental_completion_queue_functor* functor,
             int succeed):
         cdef CallbackContext *context = <CallbackContext *>functor
-        (<object>context.waiter).set_result(None)
+        if succeed == 0:
+            (<object>context.waiter).set_exception(RuntimeError())
+        else:
+            (<object>context.waiter).set_result(None)
 
     cdef grpc_experimental_completion_queue_functor *c_functor(self):
         return &self.context.functor
@@ -178,13 +183,11 @@ async def _server_call_request_call(_AioServerState server_state, object loop):
 async def _server_main_loop(_AioServerState server_state):
     cdef object loop = asyncio.get_event_loop()
     cdef RPCState rpc_state
-    cdef object waiter
 
     while True:
         rpc_state = await _server_call_request_call(
             server_state,
             loop)
-        # await waiter
 
         loop.create_task(_handle_rpc(server_state, rpc_state, loop))
         await asyncio.sleep(0)
@@ -212,6 +215,7 @@ cdef class AioServer:
             NULL,
             NULL
         )
+        self._state.status = AIO_SERVER_STATUS_READY
         grpc_server_register_completion_queue(
             self._state.server.c_server,
             self._state.cq,
@@ -240,9 +244,17 @@ cdef class AioServer:
                                           server_credentials._credentials)
 
     async def start(self):
+        if self._state.status == AIO_SERVER_STATUS_RUNNING:
+            return
+        elif self._state.status != AIO_SERVER_STATUS_READY:
+            raise RuntimeError('Server not in ready state')
+
+        self._state.status = AIO_SERVER_STATUS_RUNNING
         loop = asyncio.get_event_loop()
         loop.create_task(_server_start(self._state))
         await asyncio.sleep(0)
 
+    # TODO(https://github.com/grpc/grpc/issues/20668)
+    # Implement Destruction Methods for AsyncIO Server
     def stop(self, unused_grace):
         pass

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

@@ -112,7 +112,7 @@ class Server:
 
     async def wait_for_termination(self,
                                    timeout: Optional[float] = None) -> bool:
-        """Block current thread until the server stops.
+        """Block current coroutine until the server stops.
 
         This is an EXPERIMENTAL API.
 

src/python/grpcio_tests/tests_aio/unit/server_test.py

@@ -16,16 +16,25 @@ import asyncio
 import logging
 import unittest
 
+import grpc
 from grpc.experimental import aio
 from src.proto.grpc.testing import messages_pb2
 from src.proto.grpc.testing import benchmark_service_pb2_grpc
 
+_TEST_METHOD_PATH = ''
 
-class BenchmarkServer(benchmark_service_pb2_grpc.BenchmarkServiceServicer):
+_REQUEST = b'\x00\x00\x00'
+_RESPONSE = b'\x01\x01\x01'
 
-    async def UnaryCall(self, request, context):
-        payload = messages_pb2.Payload(body=b'\0' * request.response_size)
-        return messages_pb2.SimpleResponse(payload=payload)
+
+async def unary_unary(unused_request, unused_context):
+    return _RESPONSE
+
+
+class GenericHandler(grpc.GenericRpcHandler):
+
+    def service(self, unused_handler_details):
+        return grpc.unary_unary_rpc_method_handler(unary_unary)
 
 
 class TestServer(unittest.TestCase):
@@ -36,21 +45,13 @@ class TestServer(unittest.TestCase):
         async def test_unary_unary_body():
             server = aio.server()
             port = server.add_insecure_port(('[::]:0').encode('ASCII'))
-            benchmark_service_pb2_grpc.add_BenchmarkServiceServicer_to_server(
-                BenchmarkServer(), server)
+            server.add_generic_rpc_handlers((GenericHandler(),))
             await server.start()
 
             async with aio.insecure_channel(f'localhost:{port}') as channel:
-                unary_call = channel.unary_unary(
-                    '/grpc.testing.BenchmarkService/UnaryCall',
-                    request_serializer=messages_pb2.SimpleRequest.
-                    SerializeToString,
-                    response_deserializer=messages_pb2.SimpleResponse.FromString
-                )
-                response = await unary_call(
-                    messages_pb2.SimpleRequest(response_size=1))
-                self.assertIsInstance(response, messages_pb2.SimpleResponse)
-                self.assertEqual(1, len(response.payload.body))
+                unary_call = channel.unary_unary(_TEST_METHOD_PATH)
+                response = await unary_call(_REQUEST)
+                self.assertEqual(response, _RESPONSE)
 
         loop.run_until_complete(test_unary_unary_body())