为什么我的 asyncio 事件循环在我终止分叉子进程 (PTY) 时终止？

Question

我尝试创建一个生成 bash shell 并通过 websockets 控制它们的软件。

它在服务器端基于 fastapi 和 fastapi_socketio，在客户端基于 socket.io + JS。

必须承认，当谈到 asyncio 时，我绝对是个菜鸟。当我自己控制它时我可以使用它但是我不熟悉管理来自其他模块的事件循环等。

要启动 PTY，我使用 pty 模块中的 fork() 方法，如图“1 - Forking a PTY”（提交的命令是“/bin/bash”）：

它实际上工作得很好。 client_sid 是客户端的 socket.io 会话 ID，我可以从我的 Web UI 通过 xtermjs 无缝控制多个终端。

不过我遇到了一个问题。当我向 xtermjs 发出“退出”时，我希望子进程退出并释放文件描述符。这应该由图“2 - 将 PTYs STDOUT/ERR 发送到远程套接字的方法”中显示的方法中的 fstat 方法检测到，然后该方法应该退出并关闭 websocket 连接。

相反，Web 终端以非常快的方式接收到多个异常（图“3 - 向客户端显示的错误”），当我尝试使用 CTRL+C 关闭 uvicorn 时，我从图“4 - The当我尝试使用 CTRL+C 关闭 uvicorn 时显示错误”。

我非常感谢有关此主题的任何帮助，因为我对异步 python（可能还有 OS/PTY）的了解还不够深入。

对我来说，感觉就像从我的主进程派生出来的子进程以某种方式与 asyncio 循环交互，但我真的不知道如何交互。子进程可能会继承 asyncio 循环并在它死亡时将其杀死，这有意义吗？

我想到的唯一解决方案是检测从 Web UI 发出的“kill”命令，但这会错过例如一个发送到 PTY 子进程的终止信号，它不是很干净。

谢谢。

1 - 分叉 PTY

async def pty_handle_pty_config(self, sio: AsyncServer, client_sid: str, message: dict):
    
    if not client_sid in self.clients or self.clients[client_sid] is None:
        await self.disconnect_client(sio=sio, client_sid=client_sid)
        return

    if not isinstance(message, dict) or not 'command' in message or not isinstance(message['command'], str):
        await self.disconnect_client(sio=sio, client_sid=client_sid)
        return

    child_pid, fd = fork() # pty.fork()
    if child_pid == 0:
        subproc_run(message['command']) # subprocess.run()
    else:
        self.ptys[client_sid] = {
            'fd': fd
        }
        self.set_winsize(client_sid, 50, 50)
        await sio.emit('pty_begin', data=dict(state='success'), namespace='/pty', room=client_sid)
        sio.start_background_task(
            target=self.pty_read_and_forward,
            sio=sio,
            client_sid=client_sid,
            client_data=self.clients[client_sid]
        )

2 - 将 PTY STDOUT/ERR 发送到远程套接字的方法

async def pty_read_and_forward(self, sio: AsyncServer, client_sid: str, client_data: dict):
    log = get_logger()
    max_read_bytes = 1024 * 20
    loop = get_event_loop()
    while True:
        try:
            await async_sleep(.05) # asyncio.sleep
            timeout_sec = 0
            (data_ready, _, _) = await loop.run_in_executor(None, select, [self.ptys[client_sid]['fd']], [], [], timeout_sec) 
            if data_ready:
                output = await loop.run_in_executor(None, os_read, self.ptys[client_sid]['fd'], max_read_bytes) # os.read
                try:
                    fstat(self.ptys[client_sid]['fd']) # os.fstat
                except OSError as exc:
                    log.error(exc)
                    break
                await sio.emit(
                    event='pty_out',
                    data=dict(
                        output=output.decode('utf-8', errors='ignore')
                    ),
                    namespace='/pty',
                    room=client_sid
                )
        except Exception as exc:
            if not client_sid in self.clients:
                log.info(f'PTY session closed [sid={client_sid};user={client_data["username"]}]')
            else:
                log.warn(f'PTY session closed unexpectedly [sid={client_sid};user={client_data["username"]}] - {excstr(exc)}')
            break

3 - 显示给客户端的错误

asyncio.exceptions.CancelledError

Process SpawnProcess-2:
Traceback (most recent call last):
  File "/usr/lib/python3.10/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/usr/lib/python3.10/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.10/dist-packages/uvicorn/_subprocess.py", line 76, in subprocess_started
    target(sockets=sockets)
  File "/usr/local/lib/python3.10/dist-packages/uvicorn/server.py", line 60, in run
    return asyncio.run(self.serve(sockets=sockets))
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.10/dist-packages/uvicorn/server.py", line 80, in serve
    await self.main_loop()
  File "/usr/local/lib/python3.10/dist-packages/uvicorn/server.py", line 221, in main_loop
    await asyncio.sleep(0.1)
  File "/usr/lib/python3.10/asyncio/tasks.py", line 599, in sleep
    loop = events.get_running_loop()
RuntimeError: no running event loop

4 - 当我尝试使用 CTRL+C 关闭 uvicorn 时显示的错误

Traceback (most recent call last):
  File "/usr/lib/python3.10/asyncio/unix_events.py", line 42, in _sighandler_noop
    def _sighandler_noop(signum, frame):
BlockingIOError: [Errno 11] Resource temporarily unavailable

Answer 1

幸运的是，我通过调用 os.execvpe() 而不是使用 subprocess.run() 解决了问题。这种方法用一个全新的进程替换子进程，同时保持与父进程能够读取的文件描述符的连接。

当在父进程中使用 asyncio 时，我真的不知道在子进程中使用 fork() 的后果。这是一种非常糟糕的方法，因为 asyncio 无法将子事件循环与父事件循环分开。如果你关闭孩子的，父母的也将停止工作，因为双方都“共享”他们的环境，这要归功于 fork()。

有关详细信息，请参阅此问题：mixing asyncio with fork in python: bad idea?

它是这样工作的：

child_pid, fd = fork()
if child_pid == 0:
    # This is the forked process. Replace it with the command
    # the user wants to run.
    env = os.environ.copy()
    execvpe(message['command'], [message['command'],], env)