Add close() method to asyncio.StreamReader

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.

Author: rianhunter

Doc/library/asyncio-stream.rst

@@ -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).
+
 
 StreamWriter
 ============

Lib/asyncio/streams.py

@@ -465,6 +465,10 @@ def set_exception(self, exc):
             if not waiter.cancelled():
                 waiter.set_exception(exc)
 
+    def close(self):
+        if self._transport is not None:
+            self._transport.close()
+
     def _wakeup_waiter(self):
         """Wakeup read*() functions waiting for data or EOF."""
         waiter = self._waiter

Lib/test/test_asyncio/test_subprocess.py

@@ -35,6 +35,17 @@
               'data = sys.stdin.buffer.read()',
               'sys.stdout.buffer.write(data)'))]
 
+# Program generating infinite data
+PROGRAM_YES = [
+    sys.executable, '-c', """\
+import sys
+while True:
+    try:
+        sys.stdout.buffer.write(b"y\\n")
+    except BrokenPipeError:
+        break
+"""]
+
 
 def tearDownModule():
     asyncio._set_event_loop_policy(None)
@@ -879,6 +890,24 @@ async def main():
 
         self.loop.run_until_complete(main())
 
+    def test_subprocess_break_pipe(self):
+        # See https://github.com/python/cpython/issues/130925
+        async def main():
+            proc = await asyncio.create_subprocess_exec(*PROGRAM_YES,
+                                                        stdout=asyncio.subprocess.PIPE,
+                                                        stderr=asyncio.subprocess.DEVNULL)
+            try:
+                # just make sure the program has executed correctly
+                data = await proc.stdout.readline()
+                self.assertEqual(data, b"y\n")
+            finally:
+                # we are testing that the following method exists and
+                # has the intended effect of signaling the sub-process to terminate
+                proc.stdout.close()
+                await proc.wait()
+
+        self.loop.run_until_complete(main())
+
 
 if sys.platform != 'win32':
     # Unix

Misc/ACKS

@@ -836,6 +836,7 @@ Roberto Hueso Gomez
 Jim Hugunin
 Greg Humphreys
 Chris Hunt
+Rian Hunter
 Eric Huss
 Nehal Hussain
 Taihyun Hwang

Misc/NEWS.d/next/Library/2025-03-06-15-56-12.gh-issue-130925.USr9bm.rst

@@ -0,0 +1,3 @@
+: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.