Add `close()` method to `asyncio.StreamReader`

[rianhunter]

Feature or enhancement

Proposal:

When creating a sub-process using asyncio.create_subprocess_exec(), it returns a Process instance that has a stdout property. This property is intended to be an asyncio version of the stdout property of the Popen instance from the subprocess module.

An important aspect of Popen.stdout property is that you can close it. This is a signal to the sub-process that is generating output that it should cleanly terminate. This is a common pattern in processes used in shell pipelines. Indeed, the object located at Popen.stdout has a close() method. This pattern is demonstrated below:

import subprocess
proc = subprocess.Popen(["yes"], stdout=subprocess.PIPE) # start subprocess
data = proc.stdout.read(4096) # get data
proc.stdout.close() # signal to process to cleanly shutdown
proc.wait() # wait for shutdown

Unfortunately this pattern cannot be reproduced easily with the stdout property of the Process instance returned from asyncio.create_subprocess_exec() because stdout is an instance of StreamReader which does not have the close() method.

I propose adding a close() method to the StreamReader class so that asyncio version of the subprocess module may support this pattern of managing sub-processes. This change is consistent with the asyncio ecosystem as the companion StreamWriter class already has a close() method, along with other methods that expose its inner "transport" object. It's also trivial to implement, since it's essentially a wrapper method around the inner transport object's close() method.

Has this already been discussed elsewhere?

This is a minor feature, which does not need previous discussion elsewhere

Links to previous discussion of this feature:

No response

Linked PRs

• gh-130925: Add close() method to asyncio.StreamReader #130929

[cmaloney]

A number of classes with close (ex. multiprocessing.pool.Pool, FileIO, tempfile.NamedTemporaryFile) emit a ResourceWarning if they are deleted (__del__) without being closed to help prevent resource leakage and accidental dropped data. Would that make sense to add for this case?

Sample:

cpython/Lib/tempfile.py

Lines 478 to 483 in 886a4d7

	def __del__(self):
	close_called = self.close_called
	self.cleanup()
	if not close_called:
	_warnings.warn(self.warn_message, ResourceWarning)

[rianhunter]

I did think about this ahead of time and my conclusion was that a __del__ method that auto-closes is not right in this instance because the StreamReader is not the owner of its internal "transport" object and thus not responsible for closing it. In all current uses of StreamReader, other code is responsible for closing the transport object and indeed does so. I.e. it would not be an error to not to call close() on StreamReader references you may have.

This is just to allow code that only has a reference to the StreamReader and not the transport to eagerly close the transport for the sake of signaling. This is the norm when dealing with sub-processes in asyncio using the convenience interface. Eventually once the actual resource cleanup happens, the transport will be closed again but that is fine since close() is designed to be idempotent.