gh-139373: Fix asyncio Process.communicate() losing output when cancelled (#154223)

Output already read when communicate() is cancelled (e.g. by a
wait_for() timeout) is now retained on the Process and returned by a
subsequent communicate() call, matching subprocess.Popen.communicate()
behaviour on timeout. Passing input to a communicate() call following a
cancelled one now raises ValueError.

Author: kumaraditya303

Doc/library/asyncio-subprocess.rst

@@ -240,10 +240,34 @@ their completion.
       Note, that the data read is buffered in memory, so do not use
       this method if the data size is large or unlimited.
 
+      If this coroutine is cancelled (for example, when a timeout is
+      set with :func:`~asyncio.wait_for`), the output that was already
+      read is not lost: call :meth:`!communicate` again to read the
+      remaining output and get the complete data::
+
+         try:
+             stdout, stderr = await asyncio.wait_for(
+                 proc.communicate(), timeout=5.0)
+         except TimeoutError:
+             proc.kill()
+             stdout, stderr = await proc.communicate()
+
+      Passing *input* after a previous :meth:`!communicate` call was
+      cancelled raises :exc:`ValueError`; pass ``input=None`` to
+      continue the communication, the original *input* is used.
+
       .. versionchanged:: 3.12
 
          *stdin* gets closed when ``input=None`` too.
 
+      .. versionchanged:: next
+
+         If :meth:`!communicate` is cancelled, the output that was
+         already read is now preserved and returned by a subsequent
+         :meth:`!communicate` call.  Passing *input* to a
+         :meth:`!communicate` call following a cancelled one now raises
+         :exc:`ValueError`.
+
    .. method:: send_signal(signal)
 
       Sends the signal *signal* to the child process.

Lib/asyncio/subprocess.py

@@ -124,6 +124,11 @@ def __init__(self, transport, protocol, loop):
         self.stdout = protocol.stdout
         self.stderr = protocol.stderr
         self.pid = transport.get_pid()
+        self._communication_started = False
+        self._input = None
+        self._input_written = False
+        self._stdout_buf = bytearray()
+        self._stderr_buf = bytearray()
 
     def __repr__(self):
         return f'<{self.__class__.__name__} {self.pid}>'
@@ -148,8 +153,9 @@ def kill(self):
     async def _feed_stdin(self, input):
         debug = self._loop.get_debug()
         try:
-            if input is not None:
+            if input is not None and not self._input_written:
                 self.stdin.write(input)
+                self._input_written = True
                 if debug:
                     logger.debug(
                         '%r communicate: feed stdin (%s bytes)', self, len(input))
@@ -172,22 +178,33 @@ async def _read_stream(self, fd):
         transport = self._transport.get_pipe_transport(fd)
         if fd == 2:
             stream = self.stderr
+            buf = self._stderr_buf
         else:
             assert fd == 1
             stream = self.stdout
+            buf = self._stdout_buf
         if self._loop.get_debug():
             name = 'stdout' if fd == 1 else 'stderr'
             logger.debug('%r communicate: read %s', self, name)
-        output = await stream.read()
+        # Append each block to the persistent buffer as soon as it is
+        # read so that no output is lost if this coroutine is cancelled.
+        while block := await stream.read(stream._limit):
+            buf += block
         if self._loop.get_debug():
             name = 'stdout' if fd == 1 else 'stderr'
             logger.debug('%r communicate: close %s', self, name)
         transport.close()
-        return output
 
     async def communicate(self, input=None):
+        if self._communication_started:
+            if input:
+                raise ValueError(
+                    "Cannot send input after starting communication")
+        else:
+            self._input = input
+            self._communication_started = True
         if self.stdin is not None:
-            stdin = self._feed_stdin(input)
+            stdin = self._feed_stdin(self._input)
         else:
             stdin = self._noop()
         if self.stdout is not None:
@@ -198,8 +215,16 @@ async def communicate(self, input=None):
             stderr = self._read_stream(2)
         else:
             stderr = self._noop()
-        stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr)
+        await tasks.gather(stdin, stdout, stderr)
         await self.wait()
+        if self.stdout is not None:
+            stdout = self._stdout_buf.take_bytes()
+        else:
+            stdout = None
+        if self.stderr is not None:
+            stderr = self._stderr_buf.take_bytes()
+        else:
+            stderr = None
         return (stdout, stderr)
 
 

Lib/test/test_asyncio/test_subprocess.py

@@ -177,6 +177,116 @@ async def run():
         self.assertEqual(exitcode, 0)
         self.assertEqual(stdout, b'')
 
+    def test_communicate_cancelled_mid_read_retry(self):
+        # gh-139373: output read before communicate() was cancelled must
+        # be returned by a subsequent communicate() call.
+        code = ('import sys, time;'
+                'sys.stdout.buffer.write(b"first\\n");'
+                'sys.stdout.buffer.flush();'
+                'time.sleep(3600)')
+
+        async def run():
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable, '-c', code,
+                stdout=subprocess.PIPE,
+            )
+            task = asyncio.create_task(proc.communicate())
+            # wait until communicate() has read the first line
+            while not proc._stdout_buf:
+                await asyncio.sleep(0.01)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+            proc.kill()
+            stdout, stderr = await proc.communicate()
+            return stdout
+
+        task = asyncio.wait_for(run(), support.LONG_TIMEOUT)
+        stdout = self.loop.run_until_complete(task)
+        self.assertEqual(stdout, b'first\n')
+
+    def test_communicate_cancelled_during_wait_retry(self):
+        # gh-139373: cancellation landing after the output was fully read
+        # but before wait() completed must not lose the output.
+        code = ('import os, time;'
+                'os.write(1, b"all output\\n");'
+                'os.close(1);'
+                'time.sleep(3600)')
+
+        async def run():
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable, '-c', code,
+                stdout=subprocess.PIPE,
+            )
+            task = asyncio.create_task(proc.communicate())
+            # wait until the stdout reader has consumed everything up to
+            # EOF; communicate() is then blocked on wait()
+            while not proc.stdout.at_eof():
+                await asyncio.sleep(0.01)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+            proc.kill()
+            stdout, stderr = await proc.communicate()
+            return stdout
+
+        task = asyncio.wait_for(run(), support.LONG_TIMEOUT)
+        stdout = self.loop.run_until_complete(task)
+        self.assertEqual(stdout, b'all output\n')
+
+    def test_communicate_cancelled_input_not_resendable(self):
+        # gh-139373: like subprocess.Popen.communicate(), sending new
+        # input after a cancelled communicate() call is an error.
+        async def run():
+            proc = await asyncio.create_subprocess_exec(
+                *PROGRAM_CAT,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+            )
+            task = asyncio.create_task(proc.communicate(b'data'))
+            await asyncio.sleep(0)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+            with self.assertRaises(ValueError):
+                await proc.communicate(b'more data')
+            proc.kill()
+            await proc.communicate()
+
+        self.loop.run_until_complete(
+            asyncio.wait_for(run(), support.LONG_TIMEOUT))
+
+    def test_communicate_cancelled_stdin_retry(self):
+        # gh-139373: input already fed before cancellation is not re-sent
+        # by the retried communicate() call, and the output is preserved.
+        code = ('import sys, time;'
+                'sys.stdout.buffer.write(sys.stdin.buffer.read());'
+                'sys.stdout.buffer.flush();'
+                'time.sleep(3600)')
+
+        async def run():
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable, '-c', code,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+            )
+            task = asyncio.create_task(proc.communicate(b'hello'))
+            # the child echoes stdin only after it is closed, so once
+            # output arrives the input was fully written and
+            # communicate() is blocked on wait()
+            while not proc._stdout_buf:
+                await asyncio.sleep(0.01)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+            proc.kill()
+            stdout, stderr = await proc.communicate()
+            return stdout
+
+        task = asyncio.wait_for(run(), support.LONG_TIMEOUT)
+        stdout = self.loop.run_until_complete(task)
+        self.assertEqual(stdout, b'hello')
+
     def test_shell(self):
         proc = self.loop.run_until_complete(
             asyncio.create_subprocess_shell('exit 7')

Misc/NEWS.d/next/Library/2026-07-20-12-45-00.gh-issue-139373.qL7xKm.rst

@@ -0,0 +1,3 @@
+Fix :meth:`asyncio.subprocess.Process.communicate` losing already-read
+output when it is cancelled; the output is now retained and returned by a
+subsequent ``communicate()`` call. Patch by Kumar Aditya.