Merge pull request #21696 from Skyscanner/move_status_initial_metadata_cython

Move status and initial metadata handling to Cython

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

@@ -17,6 +17,9 @@ cdef class _AioCall(GrpcCallWrapper):
     cdef:
         AioChannel _channel
         list _references
+        object _deadline
+        list _done_callbacks
+
         # Caches the picked event loop, so we can avoid the 30ns overhead each
         # time we need access to the event loop.
         object _loop
@@ -28,6 +31,15 @@ cdef class _AioCall(GrpcCallWrapper):
         # because Core is holding a pointer for the callback handler.
         bint _is_locally_cancelled
 
-        object _deadline
+        # Following attributes are used for storing the status of the call and
+        # the initial metadata. Waiters are used for pausing the execution of
+        # tasks that are asking for one of the field when they are not yet
+        # available.
+        object _status
+        object _initial_metadata
+        list _waiters_status
+        list _waiters_initial_metadata
 
     cdef void _create_grpc_call(self, object timeout, bytes method, CallCredentials credentials) except *
+    cdef void _set_status(self, AioRpcStatus status) except *
+    cdef void _set_initial_metadata(self, tuple initial_metadata) except *

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

@@ -15,34 +15,68 @@
 
 _EMPTY_FLAGS = 0
 _EMPTY_MASK = 0
-_EMPTY_METADATA = None
+_IMMUTABLE_EMPTY_METADATA = tuple()
 
 _UNKNOWN_CANCELLATION_DETAILS = 'RPC cancelled for unknown reason.'
+_OK_CALL_REPRESENTATION = ('<{} of RPC that terminated with:\n'
+                           '\tstatus = {}\n'
+                           '\tdetails = "{}"\n'
+                           '>')
+
+_NON_OK_CALL_REPRESENTATION = ('<{} of RPC that terminated with:\n'
+                               '\tstatus = {}\n'
+                               '\tdetails = "{}"\n'
+                               '\tdebug_error_string = "{}"\n'
+                               '>')
 
 
 cdef class _AioCall(GrpcCallWrapper):
 
-    def __cinit__(self,
-                  AioChannel channel,
-                  object deadline,
-                  bytes method,
-                  CallCredentials call_credentials):
+    def __cinit__(self, AioChannel channel, object deadline,
+                  bytes method, CallCredentials call_credentials):
         self.call = NULL
         self._channel = channel
+        self._loop = channel.loop
         self._references = []
-        self._loop = asyncio.get_event_loop()
-        self._create_grpc_call(deadline, method, call_credentials)
+        self._status = None
+        self._initial_metadata = None
+        self._waiters_status = []
+        self._waiters_initial_metadata = []
+        self._done_callbacks = []
         self._is_locally_cancelled = False
         self._deadline = deadline
+        self._create_grpc_call(deadline, method, call_credentials)
 
     def __dealloc__(self):
         if self.call:
             grpc_call_unref(self.call)
 
-    def __repr__(self):
-        class_name = self.__class__.__name__
-        id_ = id(self)
-        return f"<{class_name} {id_}>"
+    def _repr(self) -> str:
+        """Assembles the RPC representation string."""
+        # This needs to be loaded at run time once everything
+        # has been loaded.
+        from grpc import _common
+
+        if not self.done():
+            return '<{} object>'.format(self.__class__.__name__)
+
+        if self._status.code() is StatusCode.ok:
+            return _OK_CALL_REPRESENTATION.format(
+                self.__class__.__name__,
+                _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE[self._status.code()],
+                self._status.details())
+        else:
+            return _NON_OK_CALL_REPRESENTATION.format(
+                self.__class__.__name__,
+                self._status.details(),
+                _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE[self._status.code()],
+                self._status.debug_error_string())
+
+    def __repr__(self) -> str:
+        return self._repr()
+
+    def __str__(self) -> str:
+        return self._repr()
 
     cdef void _create_grpc_call(self,
                                 object deadline,
@@ -82,13 +116,55 @@ cdef class _AioCall(GrpcCallWrapper):
 
         grpc_slice_unref(method_slice)
 
+    cdef void _set_status(self, AioRpcStatus status) except *:
+        cdef list waiters
+
+        if self._initial_metadata is None:
+            self._set_initial_metadata(_IMMUTABLE_EMPTY_METADATA)
+
+        self._status = status
+        waiters = self._waiters_status
+
+        # No more waiters should be expected since status
+        # has been set.
+        self._waiters_status = None
+
+        for waiter in waiters:
+            if not waiter.done():
+                waiter.set_result(None)
+
+        for callback in self._done_callbacks:
+            callback()
+
+    cdef void _set_initial_metadata(self, tuple initial_metadata) except *:
+        cdef list waiters
+
+        self._initial_metadata = initial_metadata
+
+        waiters = self._waiters_initial_metadata
+
+        # No more waiters should be expected since initial metadata
+        # has been set.
+        self._waiters_initial_metadata = None
+
+        for waiter in waiters:
+            if not waiter.done():
+                waiter.set_result(None)
+
+
+    def add_done_callback(self, callback):
+        if self.done():
+            callback()
+        else:
+            self._done_callbacks.append(callback)
+
     def time_remaining(self):
         if self._deadline is None:
             return None
         else:
             return max(0, self._deadline - time.time())
 
-    def cancel(self, AioRpcStatus status):
+    def cancel(self, str details):
         """Cancels the RPC in Core with given RPC status.
         
         Above abstractions must invoke this method to set Core objects into
@@ -96,44 +172,108 @@ cdef class _AioCall(GrpcCallWrapper):
         """
         self._is_locally_cancelled = True
 
-        cdef object details
+        cdef object details_bytes
         cdef char *c_details
         cdef grpc_call_error error
-        # Try to fetch application layer cancellation details in the future.
-        # * If cancellation details present, cancel with status;
-        # * If details not present, cancel with unknown reason.
-        if status is not None:
-            details = str_to_bytes(status.details())
-            self._references.append(details)
-            c_details = <char *>details
+
+        self._set_status(AioRpcStatus(
+            StatusCode.cancelled,
+            details,
+            None,
+            None,
+        ))
+
+        details_bytes = str_to_bytes(details)
+        self._references.append(details_bytes)
+        c_details = <char *>details_bytes
         # By implementation, grpc_call_cancel_with_status always return OK
         error = grpc_call_cancel_with_status(
             self.call,
-                status.c_code(),
+            StatusCode.cancelled,
             c_details,
             NULL,
         )
         assert error == GRPC_CALL_OK
-        else:
-            # By implementation, grpc_call_cancel always return OK
-            error = grpc_call_cancel(self.call, NULL)
-            assert error == GRPC_CALL_OK
+
+    def done(self):
+        """Returns if the RPC call has finished.
+        
+        Checks if the status has been provided, either
+        because the RPC finished or because was cancelled..
+
+        Returns:
+            True if the RPC can be considered finished.
+        """
+        return self._status is not None
+
+    def cancelled(self):
+        """Returns if the RPC was cancelled.
+        
+        Returns:
+            True if the RPC was cancelled.
+        """
+        if not self.done():
+            return False
+
+        return self._status.code() == StatusCode.cancelled
+
+    async def status(self):
+        """Returns the status of the RPC call.
+        
+        It returns the finshed status of the RPC. If the RPC
+        has not finished yet this function will wait until the RPC
+        gets finished.
+
+        Returns:
+            Finished status of the RPC as an AioRpcStatus object.
+        """
+        if self._status is not None:
+            return self._status
+
+        future = self._loop.create_future()
+        self._waiters_status.append(future)
+        await future
+
+        return self._status
+
+    async def initial_metadata(self):
+        """Returns the initial metadata of the RPC call.
+        
+        If the initial metadata has not been received yet this function will
+        wait until the RPC gets finished.
+
+        Returns:
+            The tuple object with the initial metadata.
+        """
+        if self._initial_metadata is not None:
+            return self._initial_metadata
+
+        future = self._loop.create_future()
+        self._waiters_initial_metadata.append(future)
+        await future
+
+        return self._initial_metadata
+
+    def is_locally_cancelled(self):
+        """Returns if the RPC was cancelled locally.
+
+        Returns:
+            True when was cancelled locally, False when was cancelled remotelly or
+            is still ongoing.
+        """
+        if self._is_locally_cancelled:
+            return True
+
+        return False
 
     async def unary_unary(self,
                           bytes request,
-                          tuple outbound_initial_metadata,
-                          object initial_metadata_observer,
-                          object status_observer):
+                          tuple outbound_initial_metadata):
         """Performs a unary unary RPC.
         
         Args:
-          method: name of the calling method in bytes.
           request: the serialized requests in bytes.
-          deadline: optional deadline of the RPC in float.
-          cancellation_future: the future that meant to transport the
-            cancellation reason from the application layer.
-          initial_metadata_observer: a callback for received initial metadata.
-          status_observer: a callback for received final status.
+          outbound_initial_metadata: optional outbound metadata.
         """
         cdef tuple ops
 
@@ -156,25 +296,24 @@ cdef class _AioCall(GrpcCallWrapper):
                             ops,
                             self._loop)
 
-        # Reports received initial metadata.
-        initial_metadata_observer(receive_initial_metadata_op.initial_metadata())
+        self._set_initial_metadata(receive_initial_metadata_op.initial_metadata())
 
-        status = AioRpcStatus(
-            receive_status_on_client_op.code(),
+        cdef grpc_status_code code
+        code = receive_status_on_client_op.code()
+
+        self._set_status(AioRpcStatus(
+            code,
             receive_status_on_client_op.details(),
             receive_status_on_client_op.trailing_metadata(),
             receive_status_on_client_op.error_string(),
-        )
-        # Reports the final status of the RPC to Python layer. The observer
-        # pattern is used here to unify unary and streaming code path.
-        status_observer(status)
+        ))
 
-        if status.code() == StatusCode.ok:
+        if code == StatusCode.ok:
             return receive_message_op.message()
         else:
             return None
 
-    async def _handle_status_once_received(self, object status_observer):
+    async def _handle_status_once_received(self):
         """Handles the status sent by peer once received."""
         cdef ReceiveStatusOnClientOperation op = ReceiveStatusOnClientOperation(_EMPTY_FLAGS)
         cdef tuple ops = (op,)
@@ -184,13 +323,12 @@ cdef class _AioCall(GrpcCallWrapper):
         if self._is_locally_cancelled:
             return
 
-        cdef AioRpcStatus status = AioRpcStatus(
+        self._set_status(AioRpcStatus(
             op.code(),
             op.details(),
             op.trailing_metadata(),
             op.error_string(),
-        )
-        status_observer(status)
+        ))
 
     async def receive_serialized_message(self):
         """Receives one single raw message in bytes."""
@@ -224,13 +362,11 @@ cdef class _AioCall(GrpcCallWrapper):
 
     async def initiate_unary_stream(self,
                            bytes request,
-                           tuple outbound_initial_metadata,
-                           object initial_metadata_observer,
-                           object status_observer):
+                           tuple outbound_initial_metadata):
         """Implementation of the start of a unary-stream call."""
         # Peer may prematurely end this RPC at any point. We need a corutine
         # that watches if the server sends the final status.
-        self._loop.create_task(self._handle_status_once_received(status_observer))
+        self._loop.create_task(self._handle_status_once_received())
 
         cdef tuple outbound_ops
         cdef Operation initial_metadata_op = SendInitialMetadataOperation(
@@ -254,16 +390,14 @@ cdef class _AioCall(GrpcCallWrapper):
                             self._loop)
 
         # Receives initial metadata.
-        initial_metadata_observer(
+        self._set_initial_metadata(
             await _receive_initial_metadata(self,
                                             self._loop),
         )
 
     async def stream_unary(self,
                            tuple outbound_initial_metadata,
-                           object metadata_sent_observer,
-                           object initial_metadata_observer,
-                           object status_observer):
+                           object metadata_sent_observer):
         """Actual implementation of the complete unary-stream call.
         
         Needs to pay extra attention to the raise mechanism. If we want to
@@ -278,9 +412,8 @@ cdef class _AioCall(GrpcCallWrapper):
         metadata_sent_observer()
 
         # Receives initial metadata.
-        initial_metadata_observer(
-            await _receive_initial_metadata(self,
-                                            self._loop),
+        self._set_initial_metadata(
+            await _receive_initial_metadata(self, self._loop)
         )
 
         cdef tuple inbound_ops
@@ -293,26 +426,24 @@ cdef class _AioCall(GrpcCallWrapper):
                             inbound_ops,
                             self._loop)
 
-        status = AioRpcStatus(
-            receive_status_on_client_op.code(),
+        cdef grpc_status_code code
+        code = receive_status_on_client_op.code()
+
+        self._set_status(AioRpcStatus(
+            code,
             receive_status_on_client_op.details(),
             receive_status_on_client_op.trailing_metadata(),
             receive_status_on_client_op.error_string(),
-        )
-        # Reports the final status of the RPC to Python layer. The observer
-        # pattern is used here to unify unary and streaming code path.
-        status_observer(status)
+        ))
 
-        if status.code() == StatusCode.ok:
+        if code == StatusCode.ok:
             return receive_message_op.message()
         else:
             return None
 
     async def initiate_stream_stream(self,
                            tuple outbound_initial_metadata,
-                           object metadata_sent_observer,
-                           object initial_metadata_observer,
-                           object status_observer):
+                           object metadata_sent_observer):
         """Actual implementation of the complete stream-stream call.
 
         Needs to pay extra attention to the raise mechanism. If we want to
@@ -321,7 +452,7 @@ cdef class _AioCall(GrpcCallWrapper):
         """
         # Peer may prematurely end this RPC at any point. We need a corutine
         # that watches if the server sends the final status.
-        self._loop.create_task(self._handle_status_once_received(status_observer))
+        self._loop.create_task(self._handle_status_once_received())
 
         # Sends out initial_metadata ASAP.
         await _send_initial_metadata(self,
@@ -331,7 +462,6 @@ cdef class _AioCall(GrpcCallWrapper):
         metadata_sent_observer()
 
         # Receives initial metadata.
-        initial_metadata_observer(
-            await _receive_initial_metadata(self,
-                                            self._loop),
+        self._set_initial_metadata(
+            await _receive_initial_metadata(self, self._loop)
         )

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

@@ -21,6 +21,6 @@ cdef class AioChannel:
     cdef:
         grpc_channel * channel
         CallbackCompletionQueue cq
+        object loop
         bytes _target
-        object _loop
         AioChannelStatus _status

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

@@ -25,13 +25,13 @@ cdef CallbackFailureHandler _WATCH_CONNECTIVITY_FAILURE_HANDLER = CallbackFailur
 
 
 cdef class AioChannel:
-    def __cinit__(self, bytes target, tuple options, ChannelCredentials credentials):
+    def __cinit__(self, bytes target, tuple options, ChannelCredentials credentials, object loop):
         if options is None:
             options = ()
         cdef _ChannelArgs channel_args = _ChannelArgs(options)
         self._target = target
         self.cq = CallbackCompletionQueue()
-        self._loop = asyncio.get_event_loop()
+        self.loop = loop
         self._status = AIO_CHANNEL_STATUS_READY
 
         if credentials is None:
@@ -74,7 +74,7 @@ cdef class AioChannel:
             raise RuntimeError('Channel is closed.')
         cdef gpr_timespec c_deadline = _timespec_from_time(deadline)
 
-        cdef object future = self._loop.create_future()
+        cdef object future = self.loop.create_future()
         cdef CallbackWrapper wrapper = CallbackWrapper(
             future,
             _WATCH_CONNECTIVITY_FAILURE_HANDLER)
@@ -115,5 +115,4 @@ cdef class AioChannel:
         else:
             cython_call_credentials = None
 
-        cdef _AioCall call = _AioCall(self, deadline, method, cython_call_credentials)
-        return call
+        return _AioCall(self, deadline, method, cython_call_credentials)

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

@@ -38,7 +38,7 @@ cdef class RPCState:
         self.abort_exception = None
         self.metadata_sent = False
         self.status_sent = False
-        self.trailing_metadata = _EMPTY_METADATA
+        self.trailing_metadata = _IMMUTABLE_EMPTY_METADATA
 
     cdef bytes method(self):
         return _slice_bytes(self.details.method)
@@ -127,7 +127,7 @@ cdef class _ServicerContext:
     async def abort(self,
               object code,
               str details='',
-              tuple trailing_metadata=_EMPTY_METADATA):
+              tuple trailing_metadata=_IMMUTABLE_EMPTY_METADATA):
         if self._rpc_state.abort_exception is not None:
             raise RuntimeError('Abort already called!')
         else:
@@ -136,7 +136,7 @@ cdef class _ServicerContext:
             # could lead to undefined behavior.
             self._rpc_state.abort_exception = AbortError('Locally aborted.')
 
-            if trailing_metadata == _EMPTY_METADATA and self._rpc_state.trailing_metadata:
+            if trailing_metadata == _IMMUTABLE_EMPTY_METADATA and self._rpc_state.trailing_metadata:
                 trailing_metadata = self._rpc_state.trailing_metadata
 
             self._rpc_state.status_sent = True
@@ -469,7 +469,7 @@ async def _handle_rpc(list generic_handlers, RPCState rpc_state, object loop):
             rpc_state,
             StatusCode.unimplemented,
             'Method not found!',
-            _EMPTY_METADATA,
+            _IMMUTABLE_EMPTY_METADATA,
             rpc_state.metadata_sent,
             loop
         )

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

@@ -14,7 +14,8 @@
 """Invocation-side implementation of gRPC Asyncio Python."""
 
 import asyncio
-from typing import AsyncIterable, Awaitable, List, Dict, Optional
+from functools import partial
+from typing import AsyncIterable, Dict, Optional
 
 import grpc
 from grpc import _common
@@ -42,8 +43,6 @@ _NON_OK_CALL_REPRESENTATION = ('<{} of RPC that terminated with:\n'
                                '\tdebug_error_string = "{}"\n'
                                '>')
 
-_EMPTY_METADATA = tuple()
-
 
 class AioRpcError(grpc.RpcError):
     """An implementation of RpcError to be used by the asynchronous API.
@@ -153,116 +152,67 @@ class Call(_base_call.Call):
     """
     _loop: asyncio.AbstractEventLoop
     _code: grpc.StatusCode
-    _status: Awaitable[cygrpc.AioRpcStatus]
-    _initial_metadata: Awaitable[MetadataType]
-    _locally_cancelled: bool
     _cython_call: cygrpc._AioCall
-    _done_callbacks: List[DoneCallbackType]
-
-    def __init__(self, cython_call: cygrpc._AioCall) -> None:
-        self._loop = asyncio.get_event_loop()
-        self._code = None
-        self._status = self._loop.create_future()
-        self._initial_metadata = self._loop.create_future()
-        self._locally_cancelled = False
+
+    def __init__(self, cython_call: cygrpc._AioCall,
+                 loop: asyncio.AbstractEventLoop) -> None:
+        self._loop = loop
         self._cython_call = cython_call
-        self._done_callbacks = []
 
     def __del__(self) -> None:
-        if not self._status.done():
-            self._cancel(
-                cygrpc.AioRpcStatus(cygrpc.StatusCode.cancelled,
-                                    _GC_CANCELLATION_DETAILS, None, None))
+        if not self._cython_call.done():
+            self._cancel(_GC_CANCELLATION_DETAILS)
 
     def cancelled(self) -> bool:
-        return self._code == grpc.StatusCode.CANCELLED
+        return self._cython_call.cancelled()
 
-    def _cancel(self, status: cygrpc.AioRpcStatus) -> bool:
+    def _cancel(self, details: str) -> bool:
         """Forwards the application cancellation reasoning."""
-        if not self._status.done():
-            self._set_status(status)
-            self._cython_call.cancel(status)
+        if not self._cython_call.done():
+            self._cython_call.cancel(details)
             return True
         else:
             return False
 
     def cancel(self) -> bool:
-        return self._cancel(
-            cygrpc.AioRpcStatus(cygrpc.StatusCode.cancelled,
-                                _LOCAL_CANCELLATION_DETAILS, None, None))
+        return self._cancel(_LOCAL_CANCELLATION_DETAILS)
 
     def done(self) -> bool:
-        return self._status.done()
+        return self._cython_call.done()
 
     def add_done_callback(self, callback: DoneCallbackType) -> None:
-        if self.done():
-            callback(self)
-        else:
-            self._done_callbacks.append(callback)
+        cb = partial(callback, self)
+        self._cython_call.add_done_callback(cb)
 
     def time_remaining(self) -> Optional[float]:
         return self._cython_call.time_remaining()
 
     async def initial_metadata(self) -> MetadataType:
-        return await self._initial_metadata
+        return await self._cython_call.initial_metadata()
 
     async def trailing_metadata(self) -> MetadataType:
-        return (await self._status).trailing_metadata()
+        return (await self._cython_call.status()).trailing_metadata()
 
     async def code(self) -> grpc.StatusCode:
-        await self._status
-        return self._code
+        cygrpc_code = (await self._cython_call.status()).code()
+        return _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE[cygrpc_code]
 
     async def details(self) -> str:
-        return (await self._status).details()
+        return (await self._cython_call.status()).details()
 
     async def debug_error_string(self) -> str:
-        return (await self._status).debug_error_string()
-
-    def _set_initial_metadata(self, metadata: MetadataType) -> None:
-        self._initial_metadata.set_result(metadata)
-
-    def _set_status(self, status: cygrpc.AioRpcStatus) -> None:
-        """Private method to set final status of the RPC.
-
-        This method should only be invoked once.
-        """
-        # In case of local cancellation, flip the flag.
-        if status.details() is _LOCAL_CANCELLATION_DETAILS:
-            self._locally_cancelled = True
-
-        # In case of the RPC finished without receiving metadata.
-        if not self._initial_metadata.done():
-            self._initial_metadata.set_result(_EMPTY_METADATA)
-
-        # Sets final status
-        self._status.set_result(status)
-        self._code = _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE[status.code()]
-
-        for callback in self._done_callbacks:
-            callback(self)
+        return (await self._cython_call.status()).debug_error_string()
 
     async def _raise_for_status(self) -> None:
-        if self._locally_cancelled:
+        if self._cython_call.is_locally_cancelled():
             raise asyncio.CancelledError()
-        await self._status
-        if self._code != grpc.StatusCode.OK:
-            raise _create_rpc_error(await self.initial_metadata(),
-                                    self._status.result())
+        code = await self.code()
+        if code != grpc.StatusCode.OK:
+            raise _create_rpc_error(await self.initial_metadata(), await
+                                    self._cython_call.status())
 
     def _repr(self) -> str:
-        """Assembles the RPC representation string."""
-        if not self._status.done():
-            return '<{} object>'.format(self.__class__.__name__)
-        if self._code is grpc.StatusCode.OK:
-            return _OK_CALL_REPRESENTATION.format(
-                self.__class__.__name__, self._code,
-                self._status.result().details())
-        else:
-            return _NON_OK_CALL_REPRESENTATION.format(
-                self.__class__.__name__, self._code,
-                self._status.result().details(),
-                self._status.result().debug_error_string())
+        return repr(self._cython_call)
 
     def __repr__(self) -> str:
         return self._repr()
@@ -288,13 +238,14 @@ class UnaryUnaryCall(Call, _base_call.UnaryUnaryCall):
                  credentials: Optional[grpc.CallCredentials],
                  channel: cygrpc.AioChannel, method: bytes,
                  request_serializer: SerializingFunction,
-                 response_deserializer: DeserializingFunction) -> None:
-        super().__init__(channel.call(method, deadline, credentials))
+                 response_deserializer: DeserializingFunction,
+                 loop: asyncio.AbstractEventLoop) -> None:
+        super().__init__(channel.call(method, deadline, credentials), loop)
         self._request = request
         self._metadata = metadata
         self._request_serializer = request_serializer
         self._response_deserializer = response_deserializer
-        self._call = self._loop.create_task(self._invoke())
+        self._call = loop.create_task(self._invoke())
 
     def cancel(self) -> bool:
         if super().cancel():
@@ -312,11 +263,7 @@ class UnaryUnaryCall(Call, _base_call.UnaryUnaryCall):
         # https://github.com/python/cpython/blob/edad4d89e357c92f70c0324b937845d652b20afd/Lib/asyncio/tasks.py#L785
         try:
             serialized_response = await self._cython_call.unary_unary(
-                serialized_request,
-                self._metadata,
-                self._set_initial_metadata,
-                self._set_status,
-            )
+                serialized_request, self._metadata)
         except asyncio.CancelledError:
             if not self.cancelled():
                 self.cancel()
@@ -360,13 +307,14 @@ class UnaryStreamCall(Call, _base_call.UnaryStreamCall):
                  credentials: Optional[grpc.CallCredentials],
                  channel: cygrpc.AioChannel, method: bytes,
                  request_serializer: SerializingFunction,
-                 response_deserializer: DeserializingFunction) -> None:
-        super().__init__(channel.call(method, deadline, credentials))
+                 response_deserializer: DeserializingFunction,
+                 loop: asyncio.AbstractEventLoop) -> None:
+        super().__init__(channel.call(method, deadline, credentials), loop)
         self._request = request
         self._metadata = metadata
         self._request_serializer = request_serializer
         self._response_deserializer = response_deserializer
-        self._send_unary_request_task = self._loop.create_task(
+        self._send_unary_request_task = loop.create_task(
             self._send_unary_request())
         self._message_aiter = None
 
@@ -382,8 +330,7 @@ class UnaryStreamCall(Call, _base_call.UnaryStreamCall):
                                                self._request_serializer)
         try:
             await self._cython_call.initiate_unary_stream(
-                serialized_request, self._metadata, self._set_initial_metadata,
-                self._set_status)
+                serialized_request, self._metadata)
         except asyncio.CancelledError:
             if not self.cancelled():
                 self.cancel()
@@ -419,7 +366,7 @@ class UnaryStreamCall(Call, _base_call.UnaryStreamCall):
                                        self._response_deserializer)
 
     async def read(self) -> ResponseType:
-        if self._status.done():
+        if self._cython_call.done():
             await self._raise_for_status()
             return cygrpc.EOF
 
@@ -452,16 +399,17 @@ class StreamUnaryCall(Call, _base_call.StreamUnaryCall):
                  credentials: Optional[grpc.CallCredentials],
                  channel: cygrpc.AioChannel, method: bytes,
                  request_serializer: SerializingFunction,
-                 response_deserializer: DeserializingFunction) -> None:
-        super().__init__(channel.call(method, deadline, credentials))
+                 response_deserializer: DeserializingFunction,
+                 loop: asyncio.AbstractEventLoop) -> None:
+        super().__init__(channel.call(method, deadline, credentials), loop)
         self._metadata = metadata
         self._request_serializer = request_serializer
         self._response_deserializer = response_deserializer
 
-        self._metadata_sent = asyncio.Event(loop=self._loop)
+        self._metadata_sent = asyncio.Event(loop=loop)
         self._done_writing = False
 
-        self._call_finisher = self._loop.create_task(self._conduct_rpc())
+        self._call_finisher = loop.create_task(self._conduct_rpc())
 
         # If user passes in an async iterator, create a consumer Task.
         if request_async_iterator is not None:
@@ -485,11 +433,7 @@ class StreamUnaryCall(Call, _base_call.StreamUnaryCall):
     async def _conduct_rpc(self) -> ResponseType:
         try:
             serialized_response = await self._cython_call.stream_unary(
-                self._metadata,
-                self._metadata_sent_observer,
-                self._set_initial_metadata,
-                self._set_status,
-            )
+                self._metadata, self._metadata_sent_observer)
         except asyncio.CancelledError:
             if not self.cancelled():
                 self.cancel()
@@ -517,7 +461,7 @@ class StreamUnaryCall(Call, _base_call.StreamUnaryCall):
         return response
 
     async def write(self, request: RequestType) -> None:
-        if self._status.done():
+        if self._cython_call.done():
             raise asyncio.InvalidStateError(_RPC_ALREADY_FINISHED_DETAILS)
         if self._done_writing:
             raise asyncio.InvalidStateError(_RPC_HALF_CLOSED_DETAILS)
@@ -536,7 +480,7 @@ class StreamUnaryCall(Call, _base_call.StreamUnaryCall):
 
     async def done_writing(self) -> None:
         """Implementation of done_writing is idempotent."""
-        if self._status.done():
+        if self._cython_call.done():
             # If the RPC is finished, do nothing.
             return
         if not self._done_writing:
@@ -572,20 +516,21 @@ class StreamStreamCall(Call, _base_call.StreamStreamCall):
                  credentials: Optional[grpc.CallCredentials],
                  channel: cygrpc.AioChannel, method: bytes,
                  request_serializer: SerializingFunction,
-                 response_deserializer: DeserializingFunction) -> None:
-        super().__init__(channel.call(method, deadline, credentials))
+                 response_deserializer: DeserializingFunction,
+                 loop: asyncio.AbstractEventLoop) -> None:
+        super().__init__(channel.call(method, deadline, credentials), loop)
         self._metadata = metadata
         self._request_serializer = request_serializer
         self._response_deserializer = response_deserializer
 
-        self._metadata_sent = asyncio.Event(loop=self._loop)
+        self._metadata_sent = asyncio.Event(loop=loop)
         self._done_writing = False
 
         self._initializer = self._loop.create_task(self._prepare_rpc())
 
         # If user passes in an async iterator, create a consumer coroutine.
         if request_async_iterator is not None:
-            self._async_request_poller = self._loop.create_task(
+            self._async_request_poller = loop.create_task(
                 self._consume_request_iterator(request_async_iterator))
         else:
             self._async_request_poller = None
@@ -611,11 +556,7 @@ class StreamStreamCall(Call, _base_call.StreamStreamCall):
         """
         try:
             await self._cython_call.initiate_stream_stream(
-                self._metadata,
-                self._metadata_sent_observer,
-                self._set_initial_metadata,
-                self._set_status,
-            )
+                self._metadata, self._metadata_sent_observer)
         except asyncio.CancelledError:
             if not self.cancelled():
                 self.cancel()
@@ -629,7 +570,7 @@ class StreamStreamCall(Call, _base_call.StreamStreamCall):
         await self.done_writing()
 
     async def write(self, request: RequestType) -> None:
-        if self._status.done():
+        if self._cython_call.done():
             raise asyncio.InvalidStateError(_RPC_ALREADY_FINISHED_DETAILS)
         if self._done_writing:
             raise asyncio.InvalidStateError(_RPC_HALF_CLOSED_DETAILS)
@@ -648,7 +589,7 @@ class StreamStreamCall(Call, _base_call.StreamStreamCall):
 
     async def done_writing(self) -> None:
         """Implementation of done_writing is idempotent."""
-        if self._status.done():
+        if self._cython_call.done():
             # If the RPC is finished, do nothing.
             return
         if not self._done_writing:
@@ -692,7 +633,7 @@ class StreamStreamCall(Call, _base_call.StreamStreamCall):
                                        self._response_deserializer)
 
     async def read(self) -> ResponseType:
-        if self._status.done():
+        if self._cython_call.done():
             await self._raise_for_status()
             return cygrpc.EOF
 

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

@@ -28,6 +28,8 @@ from ._typing import (ChannelArgumentType, DeserializingFunction, MetadataType,
                       SerializingFunction)
 from ._utils import _timeout_to_deadline
 
+_IMMUTABLE_EMPTY_TUPLE = tuple()
+
 
 class _BaseMultiCallable:
     """Base class of all multi callable objects.
@@ -47,12 +49,16 @@ class _BaseMultiCallable:
     _interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]]
     _loop: asyncio.AbstractEventLoop
 
-    def __init__(self, channel: cygrpc.AioChannel, method: bytes,
+    def __init__(
+            self,
+            channel: cygrpc.AioChannel,
+            method: bytes,
             request_serializer: SerializingFunction,
             response_deserializer: DeserializingFunction,
-                 interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]]
+            interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]],
+            loop: asyncio.AbstractEventLoop,
     ) -> None:
-        self._loop = asyncio.get_event_loop()
+        self._loop = loop
         self._channel = channel
         self._method = method
         self._request_serializer = request_serializer
@@ -102,31 +108,20 @@ class UnaryUnaryMultiCallable(_BaseMultiCallable):
             raise NotImplementedError("TODO: compression not implemented yet")
 
         if metadata is None:
-            metadata = tuple()
+            metadata = _IMMUTABLE_EMPTY_TUPLE
 
         if not self._interceptors:
-            return UnaryUnaryCall(
-                request,
-                _timeout_to_deadline(timeout),
-                metadata,
-                credentials,
-                self._channel,
-                self._method,
-                self._request_serializer,
-                self._response_deserializer,
-            )
+            return UnaryUnaryCall(request, _timeout_to_deadline(timeout),
+                                  metadata, credentials, self._channel,
+                                  self._method, self._request_serializer,
+                                  self._response_deserializer, self._loop)
         else:
-            return InterceptedUnaryUnaryCall(
-                self._interceptors,
-                request,
-                timeout,
-                metadata,
-                credentials,
-                self._channel,
-                self._method,
+            return InterceptedUnaryUnaryCall(self._interceptors, request,
+                                             timeout, metadata, credentials,
+                                             self._channel, self._method,
                                              self._request_serializer,
                                              self._response_deserializer,
-            )
+                                             self._loop)
 
 
 class UnaryStreamMultiCallable(_BaseMultiCallable):
@@ -168,18 +163,12 @@ class UnaryStreamMultiCallable(_BaseMultiCallable):
 
         deadline = _timeout_to_deadline(timeout)
         if metadata is None:
-            metadata = tuple()
-
-        return UnaryStreamCall(
-            request,
-            deadline,
-            metadata,
-            credentials,
-            self._channel,
-            self._method,
+            metadata = _IMMUTABLE_EMPTY_TUPLE
+
+        return UnaryStreamCall(request, deadline, metadata, credentials,
+                               self._channel, self._method,
                                self._request_serializer,
-            self._response_deserializer,
-        )
+                               self._response_deserializer, self._loop)
 
 
 class StreamUnaryMultiCallable(_BaseMultiCallable):
@@ -225,18 +214,12 @@ class StreamUnaryMultiCallable(_BaseMultiCallable):
 
         deadline = _timeout_to_deadline(timeout)
         if metadata is None:
-            metadata = tuple()
-
-        return StreamUnaryCall(
-            request_async_iterator,
-            deadline,
-            metadata,
-            credentials,
-            self._channel,
-            self._method,
+            metadata = _IMMUTABLE_EMPTY_TUPLE
+
+        return StreamUnaryCall(request_async_iterator, deadline, metadata,
+                               credentials, self._channel, self._method,
                                self._request_serializer,
-            self._response_deserializer,
-        )
+                               self._response_deserializer, self._loop)
 
 
 class StreamStreamMultiCallable(_BaseMultiCallable):
@@ -282,18 +265,12 @@ class StreamStreamMultiCallable(_BaseMultiCallable):
 
         deadline = _timeout_to_deadline(timeout)
         if metadata is None:
-            metadata = tuple()
-
-        return StreamStreamCall(
-            request_async_iterator,
-            deadline,
-            metadata,
-            credentials,
-            self._channel,
-            self._method,
+            metadata = _IMMUTABLE_EMPTY_TUPLE
+
+        return StreamStreamCall(request_async_iterator, deadline, metadata,
+                                credentials, self._channel, self._method,
                                 self._request_serializer,
-            self._response_deserializer,
-        )
+                                self._response_deserializer, self._loop)
 
 
 class Channel:
@@ -301,6 +278,7 @@ class Channel:
 
     A cygrpc.AioChannel-backed implementation.
     """
+    _loop: asyncio.AbstractEventLoop
     _channel: cygrpc.AioChannel
     _unary_unary_interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]]
 
@@ -341,8 +319,9 @@ class Channel:
                     "UnaryUnaryClientInterceptors, the following are invalid: {}"\
                     .format(invalid_interceptors))
 
+        self._loop = asyncio.get_event_loop()
         self._channel = cygrpc.AioChannel(_common.encode(target), options,
-                                          credentials)
+                                          credentials, self._loop)
 
     def get_state(self,
                   try_to_connect: bool = False) -> grpc.ChannelConnectivity:
@@ -411,7 +390,8 @@ class Channel:
         return UnaryUnaryMultiCallable(self._channel, _common.encode(method),
                                        request_serializer,
                                        response_deserializer,
-                                       self._unary_unary_interceptors)
+                                       self._unary_unary_interceptors,
+                                       self._loop)
 
     def unary_stream(
             self,
@@ -421,7 +401,7 @@ class Channel:
     ) -> UnaryStreamMultiCallable:
         return UnaryStreamMultiCallable(self._channel, _common.encode(method),
                                         request_serializer,
-                                        response_deserializer, None)
+                                        response_deserializer, None, self._loop)
 
     def stream_unary(
             self,
@@ -431,7 +411,7 @@ class Channel:
     ) -> StreamUnaryMultiCallable:
         return StreamUnaryMultiCallable(self._channel, _common.encode(method),
                                         request_serializer,
-                                        response_deserializer, None)
+                                        response_deserializer, None, self._loop)
 
     def stream_stream(
             self,
@@ -441,7 +421,8 @@ class Channel:
     ) -> StreamStreamMultiCallable:
         return StreamStreamMultiCallable(self._channel, _common.encode(method),
                                          request_serializer,
-                                         response_deserializer, None)
+                                         response_deserializer, None,
+                                         self._loop)
 
     async def _close(self):
         # TODO: Send cancellation status

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

@@ -110,12 +110,14 @@ class InterceptedUnaryUnaryCall(_base_call.UnaryUnaryCall):
                  credentials: Optional[grpc.CallCredentials],
                  channel: cygrpc.AioChannel, method: bytes,
                  request_serializer: SerializingFunction,
-                 response_deserializer: DeserializingFunction) -> None:
+                 response_deserializer: DeserializingFunction,
+                 loop: asyncio.AbstractEventLoop) -> None:
         self._channel = channel
-        self._loop = asyncio.get_event_loop()
-        self._interceptors_task = asyncio.ensure_future(
-            self._invoke(interceptors, method, timeout, metadata, credentials,
-                         request, request_serializer, response_deserializer))
+        self._loop = loop
+        self._interceptors_task = asyncio.ensure_future(self._invoke(
+            interceptors, method, timeout, metadata, credentials, request,
+            request_serializer, response_deserializer),
+                                                        loop=loop)
 
     def __del__(self):
         self.cancel()
@@ -154,7 +156,7 @@ class InterceptedUnaryUnaryCall(_base_call.UnaryUnaryCall):
                     client_call_details.metadata,
                     client_call_details.credentials, self._channel,
                     client_call_details.method, request_serializer,
-                    response_deserializer)
+                    response_deserializer, self._loop)
 
         client_call_details = ClientCallDetails(method, timeout, metadata,
                                                 credentials)

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

@@ -49,6 +49,17 @@ class _MulticallableTestMixin():
 
 class TestUnaryUnaryCall(_MulticallableTestMixin, AioTestBase):
 
+    async def test_call_to_string(self):
+        call = self._stub.UnaryCall(messages_pb2.SimpleRequest())
+
+        self.assertTrue(str(call) is not None)
+        self.assertTrue(repr(call) is not None)
+
+        response = await call
+
+        self.assertTrue(str(call) is not None)
+        self.assertTrue(repr(call) is not None)
+
     async def test_call_ok(self):
         call = self._stub.UnaryCall(messages_pb2.SimpleRequest())
 
@@ -105,6 +116,65 @@ class TestUnaryUnaryCall(_MulticallableTestMixin, AioTestBase):
         call = self._stub.UnaryCall(messages_pb2.SimpleRequest())
         self.assertEqual((), await call.trailing_metadata())
 
+    async def test_call_initial_metadata_cancelable(self):
+        coro_started = asyncio.Event()
+        call = self._stub.UnaryCall(messages_pb2.SimpleRequest())
+
+        async def coro():
+            coro_started.set()
+            await call.initial_metadata()
+
+        task = self.loop.create_task(coro())
+        await coro_started.wait()
+        task.cancel()
+
+        # Test that initial metadata can still be asked thought
+        # a cancellation happened with the previous task
+        self.assertEqual((), await call.initial_metadata())
+
+    async def test_call_initial_metadata_multiple_waiters(self):
+        call = self._stub.UnaryCall(messages_pb2.SimpleRequest())
+
+        async def coro():
+            return await call.initial_metadata()
+
+        task1 = self.loop.create_task(coro())
+        task2 = self.loop.create_task(coro())
+
+        await call
+
+        self.assertEqual([(), ()], await asyncio.gather(*[task1, task2]))
+
+    async def test_call_code_cancelable(self):
+        coro_started = asyncio.Event()
+        call = self._stub.UnaryCall(messages_pb2.SimpleRequest())
+
+        async def coro():
+            coro_started.set()
+            await call.code()
+
+        task = self.loop.create_task(coro())
+        await coro_started.wait()
+        task.cancel()
+
+        # Test that code can still be asked thought
+        # a cancellation happened with the previous task
+        self.assertEqual(grpc.StatusCode.OK, await call.code())
+
+    async def test_call_code_multiple_waiters(self):
+        call = self._stub.UnaryCall(messages_pb2.SimpleRequest())
+
+        async def coro():
+            return await call.code()
+
+        task1 = self.loop.create_task(coro())
+        task2 = self.loop.create_task(coro())
+
+        await call
+
+        self.assertEqual([grpc.StatusCode.OK, grpc.StatusCode.OK], await
+                         asyncio.gather(task1, task2))
+
     async def test_cancel_unary_unary(self):
         call = self._stub.UnaryCall(messages_pb2.SimpleRequest())