Ensure channel isolation is maintained while graceful close

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

@@ -15,7 +15,7 @@
 
 cdef class _AioCall(GrpcCallWrapper):
     cdef:
-        AioChannel _channel
+        readonly AioChannel _channel
         list _references
         object _deadline
         list _done_callbacks

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

@@ -61,11 +61,6 @@ class _BaseMultiCallable:
     Handles the initialization logic and stores common attributes.
     """
     _loop: asyncio.AbstractEventLoop
-    _channel: cygrpc.AioChannel
-    _method: bytes
-    _request_serializer: SerializingFunction
-    _response_deserializer: DeserializingFunction
-
     _channel: cygrpc.AioChannel
     _method: bytes
     _request_serializer: SerializingFunction
@@ -368,8 +363,9 @@ class Channel:
             candidate = frame.f_locals.get('self')
             if candidate:
                 if isinstance(candidate, _base_call.Call):
-                    calls.append(candidate)
+                    if candidate._cython_call._channel is self._channel:
-                    call_tasks.append(task)
+                        calls.append(candidate)
+                        call_tasks.append(task)
 
         # If needed, try to wait for them to finish.
         # Call objects are not always awaitables.

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

@@ -120,6 +120,18 @@ class TestCloseChannel(AioTestBase):
         for call in calls:
             self.assertTrue(call.cancelled())
 
+    async def test_channel_isolation(self):
+        async with aio.insecure_channel(self._server_target) as channel1:
+            async with aio.insecure_channel(self._server_target) as channel2:
+                stub1 = test_pb2_grpc.TestServiceStub(channel1)
+                stub2 = test_pb2_grpc.TestServiceStub(channel2)
+
+                call1 = stub1.UnaryCall(messages_pb2.SimpleRequest())
+                call2 = stub2.UnaryCall(messages_pb2.SimpleRequest())
+
+            self.assertFalse(call1.cancelled())
+            self.assertTrue(call2.cancelled())
+
 
 if __name__ == '__main__':
     logging.basicConfig(level=logging.INFO)