Address leak when using request stream interceptors (#25449)

[bostikforever]

This addresses the leak demonstrated in #25449

The root cause seems to be the creation of lots of unresolved asyncio Tasks because of how the call status code is used to write to the interceptor request stream in an interruptible manner.

The first attempt that (sort of) worked is just to cancel whatever pending calls that are not required: ae3f88e

However, while this slows the rate of leakage, it does not quite solve the problem completely.
It turns out that making many calls to the status code itself is a problem, as this creates many Futures in the cython layer which persist even after the Tasks that create them are cancelled (see:

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

Lines 235 to 252 in fd3bd70

	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

)
While Future objects seem to be cheaper than the Task objects, creating a lot of them still adds up over time.
For the purposes of the interceptor request stream, it seems that it is sufficient to just have a single status code task that can be "listened" on for the lifetime of the interceptor request stream. This way we avoid polluting the callback queue of the corresponding cython call object: c8801bb

@lidizheng

[linux-foundation-easycla]

The committers are authorized under a signed CLA.

• ✅ Bolutife Ogunsola (c848b61)

[bostikforever]

I see some builds are failing, looking at other PRs though (including the merged ones) it seems that this is "normal". Is there any thing for me to do?

[bostikforever]

I have rebased on master and resolved the merge conflict that was preventing merge.
@lidizheng

[bostikforever]

@lidizheng Fixed typo causing build to fail.

[bostikforever]

@lidizheng I do not think the current failing tests have anything to do with my changes, can you please confirm.

[lidizheng]

Sorry for the late reply. This is a hard problem to investigate and solve. Thanks so much for fixing it. I think the fix and comment looks good. Just a minor comment, and please check our Sanity Tests' output, as following:

---

+ yapf_virtual_environment/bin/python -m yapf --diff --parallel --recursive --style=setup.cfg examples src test tools setup.py
--- src/python/grpcio/grpc/aio/_interceptor.py	(original)
+++ src/python/grpcio/grpc/aio/_interceptor.py	(reformatted)
@@ -505,8 +505,8 @@
                 break
             yield value

-    async def _write_to_iterator_queue_interruptible(
-            self, request: RequestType, call: InterceptedCall):
+    async def _write_to_iterator_queue_interruptible(self, request: RequestType,
+                                                     call: InterceptedCall):
         # Write the specified 'request' to the request iterator queue using the
         # specified 'call' to allow for interruption of the write in the case
         # of abrupt termination of the call.
@@ -516,8 +516,7 @@
         _, _ = await asyncio.wait(
             (self._loop.create_task(self._write_to_iterator_queue.put(request)),
              self._status_code_task),
-            return_when=asyncio.FIRST_COMPLETED
-        )
+            return_when=asyncio.FIRST_COMPLETED)

     async def write(self, request: RequestType) -> None:
         # If no queue was created it means that requests

[bostikforever]

Addressed the formatting fix and your nit.

[lidizheng]

Thanks for the update!

[bostikforever]

@lidizheng Thanks for accepting. What version to look out for this fix in?

[lidizheng]

We will have a branch cut tomorrow for v1.42.0. It will be available in release candidate soon, like within the week. The official release might take longer, as there might be release blockers.

[ctiller]

lgtm