gh-130925: Add close() method to asyncio.StreamReader
#130929
Conversation
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: ```python 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. This change adds 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.
rianhunter
requested review from
1st1,
asvetlov,
kumaraditya303 and
willingc
as code owners
Doc/library/asyncio-stream.rst
Outdated
| @@ -290,6 +290,10 @@ StreamReader | |||
| Return ``True`` if the buffer is empty and :meth:`feed_eof` | |||
| was called. | |||
|
|
|||
| .. method:: close() | |||
|
|
|||
| Invoke ``close()`` on the underlying transport (if one exists). | |||
There was a problem hiding this comment.
it is not clear what "what" means here: "close() method" or "underlying transport"?
There was a problem hiding this comment.
A transport is a pervasive concept throughout the asyncio package. StreamReader objects usually have an internal transport member. I can add a link to the transport documentation page here to make that more clear. Or what do you suggest?
| @@ -35,6 +35,17 @@ | |||
| 'data = sys.stdin.buffer.read()', | |||
| 'sys.stdout.buffer.write(data)'))] | |||
|
|
|||
| # Program generating infinite data | |||
| PROGRAM_YES = [ | |||
There was a problem hiding this comment.
Is there a reason to make this a global var? Maybe we can just define it in test?
There was a problem hiding this comment.
No reason other than consistency with PROGRAM_CAT and PROGRAM_BLOCKED. I can make it a local variable.
| proc = await asyncio.create_subprocess_exec(*PROGRAM_YES, | ||
| stdout=asyncio.subprocess.PIPE) | ||
| try: | ||
| # just make sure the program has executed correctly |
There was a problem hiding this comment.
| # just make sure the program has executed correctly |
There was a problem hiding this comment.
I added that comment so it's clear why those otherwise unnecessary lines of code are there. They aren't conceptually part of the unit test though. I can be more clear.
| # we are testing that the following method exists and | ||
| # has the intended effect of signaling the sub-process to terminate |
There was a problem hiding this comment.
| # we are testing that the following method exists and | |
| # has the intended effect of signaling the sub-process to terminate |
There was a problem hiding this comment.
Same as before, the comment is there to make it explicit what the intent of the test is and the code included in the test.
| :func:`asyncio.StreamReader.close` now exists so that it's possible to | ||
| signal to sub-processes executed via :func:`asyncio.create_subprocess_exec` | ||
| that they may cease generating output and exit cleanly. |
There was a problem hiding this comment.
| :func:`asyncio.StreamReader.close` now exists so that it's possible to | |
| signal to sub-processes executed via :func:`asyncio.create_subprocess_exec` | |
| that they may cease generating output and exit cleanly. | |
| Add :func:`asyncio.StreamReader.close` method. It can signal to sub-processes executed via :func:`asyncio.create_subprocess_exec` | |
| that they may cease generating output and exit cleanly. |
There was a problem hiding this comment.
I think the "so" was a typo? Otherwise I can change this
When creating a sub-process using
asyncio.create_subprocess_exec(), it returns aProcessinstance that has astdoutproperty. This property is intended to be an asyncio version of thestdoutproperty of thePopeninstance from thesubprocessmodule.An important aspect of
Popen.stdoutproperty 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 atPopen.stdouthas aclose()method. This pattern is demonstrated below:Unfortunately this pattern cannot be reproduced easily with the
stdoutproperty of theProcessinstance returned fromasyncio.create_subprocess_exec()becausestdoutis an instance ofStreamReaderwhich does not have theclose()method.This change adds a
close()method to theStreamReaderclass so that asyncio version of thesubprocessmodule may support this pattern of managing sub-processes. This change is consistent with the asyncio ecosystem as the companionStreamWriterclass already has aclose()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'sclose()method.close()method toasyncio.StreamReader#130925📚 Documentation preview 📚: https://cpython-previews--130929.org.readthedocs.build/