为何子进程退出后asyncio.run仍无法返回？

通过pkill ssh杀死SSH子进程后，Python程序的asyncio.run无法正常退出。虽然所有异步协程（run_command、read_from_process、write_to_process）均已执行完毕，日志输出如下：

finally... worker 1 done
finally... worker 2 done
t1, t2 done
finally
finally .. DONE  returncode=255

但终端始终未打印ALL DONE，用ps命令查看能确认Python进程还在运行。相关代码如下：

import asyncio
import os
import pty
import sys
import termios
import traceback
import tty

async def run_command(command):
    # Create a pseudoterminal pair (master and slave)
    master_fd, slave_fd = pty.openpty()
    old_tty = termios.tcgetattr(sys.stdin)

    try:
        # Set the terminal in raw mode for capturing control characters
        tty.setraw(sys.stdin.fileno())
        # Start the process using the slave as stdin, stdout, stderr
        process = await asyncio.create_subprocess_exec(
            *command,
            stdin=slave_fd,
            stdout=slave_fd,
            stderr=slave_fd
        )

        # Close the slave in the parent process
        os.close(slave_fd)


        # Start tasks to handle reading from and writing to the terminal
        async def onCancel(coro, proc):
            try:
                await coro
            except asyncio.CancelledError as e:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
            except OSError as e:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
                # Cancel other end.
                if proc == 1:
                    t2.cancel()
                else:
                    t1.cancel()
                # process.kill()

                # print("".join(traceback.format_exception(type(e), e, e.__traceback__)), file=sys.stderr)
                # cancel_task(t1)
                # cancel_task(t2)
                # print("".join(traceback.format_exception(type(e), e, e.__traceback__)), file=sys.stderr)
            except Exception as e:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
                print("".join(traceback.format_exception(type(e), e, e.__traceback__)), file=sys.stderr)
            finally:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
                print(f"finally... worker {proc} done")
                # cancel_task(t1)
                # cancel_task(t2)

        t1 = asyncio.create_task(onCancel(read_from_process(master_fd), 1))
        t2 = asyncio.create_task(onCancel(write_to_process(master_fd), 2))

        await asyncio.gather(t1, t2)
        print("t1, t2 done")
        await process.wait()

        # asyncUtil.printWorkers() # Shows only run_command coro remains

    finally:
        print("finally")
        # Restore the terminal to its previous state
        os.close(master_fd)
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
        print(f"finally .. DONE  returncode={process.returncode}")


async def read_from_process(master_fd):
    """Asynchronously read from the master side of the PTY and print to stdout. Note that stdin/stdout are combined."""
    loop = asyncio.get_event_loop()
    while True:
        data = await loop.run_in_executor(None, os.read, master_fd, 1024)
        if not data:  # EOF
            break
        sys.stdout.write(data.decode())
        sys.stdout.flush()


async def write_to_process(master_fd):
    """Asynchronously read from stdin and send input to the process."""
    try:
        loop = asyncio.get_event_loop()
        while True:
            try:
                user_input = await loop.run_in_executor(None, sys.stdin.read, 1)  # Read one character
                if user_input == "\x04":  # CTRL+D to signal EOF
                    os.write(master_fd, user_input.encode())
                    # print("CTRL+D")
                    break
                if user_input == "":
                    break  # EOF, such as CTRL-D
                os.write(master_fd, user_input.encode())
            except EOFError:
                break
    except Exception as e:
        print("".join(traceback.format_exception(type(e), e, e.__traceback__)), file=sys.stderr)


# Run the main event loop with the command
command = ["ssh", "localhost"]  # replace with your SSH command
asyncio.run(run_command(command))
print("ALL DONE")
sys.stdout.flush()

核心问题出在阻塞的非守护线程上：

• write_to_process协程通过asyncio.run_in_executor启动了一个线程，专门执行sys.stdin.read(1)读取用户输入。这个线程属于非守护线程，只要它还在运行，Python进程就不会主动退出。
• 当SSH进程被杀死后，这个读取线程会一直阻塞等待用户输入，不会自动终止。哪怕所有协程的逻辑都执行完毕，后台残留的这个线程会卡住整个进程。

这里提供两种简单有效的修复方式：

方案一：SSH退出后主动取消输入任务 #

在确认SSH进程退出后，主动取消write_to_process对应的任务，强制中断stdin读取线程的阻塞：
修改run_command函数中await process.wait()之后的代码，添加任务取消逻辑：

await process.wait()
# 主动取消输入任务，终止stdin读取线程
if t2:
    t2.cancel()
    try:
        await t2
    except asyncio.CancelledError:
        pass

方案二：改用异步IO读取stdin #

放弃用线程池阻塞读取stdin，改用asyncio.StreamReader实现异步读取，这样协程可以被正常取消，不会残留阻塞线程。修改write_to_process函数如下：

async def write_to_process(master_fd):
    """Asynchronously read from stdin and send input to the process."""
    loop = asyncio.get_event_loop()
    # 创建StreamReader异步读取stdin
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    
    try:
        while True:
            # 异步读取一个字节
            user_input = await reader.read(1)
            if not user_input:  # 读到EOF
                break
            user_input_str = user_input.decode()
            if user_input_str == "\x04":  # 捕获CTRL+D
                os.write(master_fd, user_input)
                break
            os.write(master_fd, user_input)
    except Exception as e:
        print("".join(traceback.format_exception(type(e), e, e.__traceback__)), file=sys.stderr)

import asyncio
import os
import pty
import sys
import termios
import traceback
import tty

async def run_command(command):
    master_fd, slave_fd = pty.openpty()
    old_tty = termios.tcgetattr(sys.stdin)
    process = None
    t1 = None
    t2 = None

    try:
        tty.setraw(sys.stdin.fileno())
        process = await asyncio.create_subprocess_exec(
            *command,
            stdin=slave_fd,
            stdout=slave_fd,
            stderr=slave_fd
        )
        os.close(slave_fd)

        async def onCancel(coro, proc):
            nonlocal t1, t2
            try:
                await coro
            except asyncio.CancelledError:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
            except OSError:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
                if proc == 1 and t2:
                    t2.cancel()
                elif proc == 2 and t1:
                    t1.cancel()
            except Exception as e:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
                print("".join(traceback.format_exception(type(e), e, e.__traceback__)), file=sys.stderr)
            finally:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
                print(f"finally... worker {proc} done")

        t1 = asyncio.create_task(onCancel(read_from_process(master_fd), 1))
        t2 = asyncio.create_task(onCancel(write_to_process(master_fd), 2))

        await asyncio.gather(t1, t2)
        print("t1, t2 done")
        await process.wait()

        # 主动取消输入任务，终止stdin读取线程
        if t2:
            t2.cancel()
            try:
                await t2
            except asyncio.CancelledError:
                pass

    finally:
        print("finally")
        if master_fd:
            os.close(master_fd)
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
        if process:
            print(f"finally .. DONE  returncode={process.returncode}")


async def read_from_process(master_fd):
    loop = asyncio.get_event_loop()
    while True:
        try:
            data = await loop.run_in_executor(None, os.read, master_fd, 1024)
            if not data:
                break
            sys.stdout.write(data.decode())
            sys.stdout.flush()
        except OSError:
            break


async def write_to_process(master_fd):
    try:
        loop = asyncio.get_event_loop()
        while True:
            try:
                user_input = await loop.run_in_executor(None, sys.stdin.read, 1)
                if user_input == "\x04":
                    os.write(master_fd, user_input.encode())
                    break
                if user_input == "":
                    break
                os.write(master_fd, user_input.encode())
            except EOFError:
                break
    except Exception as e:
        print("".join(traceback.format_exception(type(e), e, e.__traceback__)), file=sys.stderr)


command = ["ssh", "localhost"]
asyncio.run(run_command(command))
print("ALL DONE")
sys.stdout.flush()

内容的提问来源于stack exchange，提问作者user48956