From: Kumar Aditya Date: Tue, 21 Jul 2026 08:53:58 +0000 (+0530) Subject: gh-139373: Fix asyncio Process.communicate() losing output when cancelled (#154223) X-Git-Url: http://git.ipfire.org/index.cgi?a=commitdiff_plain;h=027a8336891cf7dab6a437a5d4f9213a6a7828cd;p=thirdparty%2FPython%2Fcpython.git 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. --- diff --git a/Doc/library/asyncio-subprocess.rst b/Doc/library/asyncio-subprocess.rst index a6514649bf9a..70f711b779ed 100644 --- a/Doc/library/asyncio-subprocess.rst +++ b/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. diff --git a/Lib/asyncio/subprocess.py b/Lib/asyncio/subprocess.py index 043359bbd03f..26d7b755105c 100644 --- a/Lib/asyncio/subprocess.py +++ b/Lib/asyncio/subprocess.py @@ -124,6 +124,11 @@ class Process: 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 @@ class Process: 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 @@ class Process: 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 @@ class Process: 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) diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py index 848a682dd689..ce127e73f50d 100644 --- a/Lib/test/test_asyncio/test_subprocess.py +++ b/Lib/test/test_asyncio/test_subprocess.py @@ -177,6 +177,116 @@ class SubprocessMixin: 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') diff --git a/Misc/NEWS.d/next/Library/2026-07-20-12-45-00.gh-issue-139373.qL7xKm.rst b/Misc/NEWS.d/next/Library/2026-07-20-12-45-00.gh-issue-139373.qL7xKm.rst new file mode 100644 index 000000000000..be7d5b79f099 --- /dev/null +++ b/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.