asyncio's Process.communicate() is unsafe to cancel

[uhx]

Bug report

Bug description:

Documentation on asyncio.subprocess.Process (link) says:

the communicate() and wait() methods don’t have a timeout parameter: use the wait_for() function;

but in the meantime, cancelling this Future may result in stdout/stderr loss, which might be unclear from the documentation and not very user-friendly if the user ever retries communicating with the process again to collect the whole output.

So in general, using the synchronous API I would do it like this:

import sys
import subprocess


process = subprocess.Popen(
    args=[
        sys.executable,
        '-c',
        'import time; print("first", flush=True); time.sleep(3); print("second")',
    ],
    stdout=subprocess.PIPE)
try:
    [stdout, _] = process.communicate(timeout=1)
except subprocess.TimeoutExpired:
    process.kill()
    [stdout, _] = process.communicate()
    print(f"Timed out, {stdout=}")

output:

Timed out, stdout=b'first\r\n'

which gives me the full process output. I think this is a common approach to using the subprocess API.

Doing the same in async will mostly lead to output loss:

import asyncio
import sys

async def main():
    process = await asyncio.create_subprocess_exec(
        sys.executable,
        '-c',
        'import time; print("first", flush=True); time.sleep(3); print("second")',
        stdout=asyncio.subprocess.PIPE,
        )
    try:
        [stdout, _] = await asyncio.wait_for(process.communicate(), timeout=1)
    except asyncio.TimeoutError:
        process.kill()
        [stdout, _] = await process.communicate()
        print(f"Timed out, {stdout=}")


if __name__ == '__main__':
    asyncio.run(main())

output:

Timed out, stdout=b''

I researched the source code a bit and don't see a clear solution for this other than implementing a timeout parameter for Process.communicate(). Cancellation may occur right between those two calls:

cpython/Lib/asyncio/subprocess.py

Lines 201 to 202 in 93ac352

	stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr)
	await self.wait()

so the transport buffer will already be drained, but the process is not yet .wait()ed for.

PS I don't know if I chose the issue type correctly, since it might be a bug, a documentation issue, or an enhancement proposal.

CPython versions tested on:

3.11, 3.13

Operating systems tested on:

No response

[Zheaoli]

confirmed on main branch

[Zheaoli]

The bug has been confirmed, I write a script about the cancel operation

import asyncio
import sys

# More precise monkey patch to track blocks list
original_read = None
blocks_data = []

async def patched_read(self, n=-1):
    """Track read method and blocks collection"""
    global blocks_data
    print(f"=== StreamReader.read({n}) called ===")

    if n < 0:
        print("=== Entering read(-1) logic, starting blocks collection ===")
        blocks_data = []  # Reset blocks record
        try:
            blocks = []
            while True:
                print(f"=== Calling read({self._limit}) in loop ===")
                block = await self.read(self._limit)
                print(f"=== Block read: {block!r} ===")
                if not block:
                    break
                blocks.append(block)
                blocks_data.append(block)  # Record to global variable
                print(f"=== Current blocks: {blocks} ===")

            result = b''.join(blocks)
            print(f"=== read(-1) successfully returned: {result!r} ===")
            return result

        except asyncio.CancelledError:
            print(f"=== read(-1) cancelled! Collected blocks: {blocks_data} ===")
            print(f"=== Total lost data: {b''.join(blocks_data)!r} ===")
            raise
    else:
        # Call original method for fixed-length reads
        return await original_read(self, n)

async def test_blocks_loss():
    """Verify blocks loss theory"""
    global original_read, blocks_data
    from asyncio import streams
    original_read = streams.StreamReader.read
    streams.StreamReader.read = patched_read

    try:
        process = await asyncio.create_subprocess_exec(
            sys.executable,
            '-c',
            'import time; print("first", flush=True); time.sleep(3); print("second")',
            stdout=asyncio.subprocess.PIPE,
        )

        # Wait for data to arrive
        await asyncio.sleep(0.5)

        try:
            print("=== Starting communicate test ===")
            result = await asyncio.wait_for(process.communicate(), timeout=1)
        except asyncio.TimeoutError:
            print("=== communicate timeout ===")
            print(f"=== Globally recorded lost data: {blocks_data} ===")

        process.kill()
        await process.wait()

    finally:
        streams.StreamReader.read = original_read

if __name__ == '__main__':
    asyncio.run(test_blocks_loss())

[Zheaoli]

I think maybe we need add a timeout argument just like what we do in subporcess lib.

cc @1st1 @asvetlov

[kumaraditya303]

This is more or less a duplicate of #103847

asyncio's subprocess implementation is not cancellation safe, I plan to rewrite it in 3.15 to fix that, closing in favor of other issue.