杀死fork生成的PTY子进程时asyncio事件循环崩溃的原因排查

我正在开发一款通过Websocket控制bash shell的软件，服务端基于FastAPI和fastapi_socketio，客户端采用Socket.io+JS实现。对asyncio了解较浅，仅能在自主控制场景下使用，对其他模块的事件循环管理不熟悉。

使用pty模块的fork()方法启动PTY（执行命令为/bin/bash），目前可通过Web UI的xtermjs无缝控制多个终端，但存在以下问题：

• 在xtermjs中输入exit后，预期子进程退出并释放文件描述符，由pty_read_and_forward方法中的fstat检测到后退出并关闭WebSocket连接，但实际Web终端快速收到大量异常。
• 使用CTRL+C关闭uvicorn时也会报错。

怀疑fork出的子进程与asyncio事件循环存在交互，比如子进程是否继承了事件循环并在退出时将其终止，但不确定原因。目前想到的解决方案是检测Web UI发出的kill命令，但无法覆盖向PTY子进程发送kill信号的场景，且实现不够优雅。

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的标准输出/错误发送到远程Socket的方法 #

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/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/lib/python3.10/dist-packages/uvicorn/server.py", line 80, in serve
    await self.main_loop()
  File "/usr/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

根源分析 #

1. 子进程继承父进程事件循环资源：fork创建的子进程会复制父进程所有文件描述符，包括asyncio事件循环监听的套接字、管道等，子进程退出时可能意外关闭这些共享资源，导致父进程事件循环崩溃。
2. 子进程执行逻辑错误：在fork后的子进程中直接调用subprocess.run()会创建嵌套子进程，未正确重定向PTY文件描述符，导致PTY状态异常。
3. fd有效性检测时机滞后：读取数据后才检测fd状态，此时已多次尝试读取无效fd，引发大量异常。

修复方案 #

1. 正确处理fork后的子进程环境 #

在子进程分支中重置信号处理、重定向PTY标准输入输出，用execvp替代subprocess.run避免嵌套子进程：

import os
import pty
import signal

async def pty_handle_pty_config(self, sio: AsyncServer, client_sid: str, message: dict):
    # 原有参数校验逻辑...

    child_pid, fd = pty.fork()
    if child_pid == 0:
        # 重置信号处理，避免继承父进程的asyncio信号处理器
        signal.signal(signal.SIGINT, signal.SIG_DFL)
        signal.signal(signal.SIGTERM, signal.SIG_DFL)
        
        # 重定向标准输入输出到PTY
        os.dup2(fd, 0)
        os.dup2(fd, 1)
        os.dup2(fd, 2)
        os.close(fd)
        
        # 执行bash，用execvp替代subprocess.run，确保PTY与子进程生命周期一致
        os.execvp('/bin/bash', ['/bin/bash'])
    else:
        self.ptys[client_sid] = {
            'fd': fd,
            'child_pid': child_pid  # 保存子进程ID用于后续监控
        }
        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读取逻辑，监控子进程状态 #

用子进程退出状态替代fd检测，设置PTY为非阻塞模式，避免定时sleep浪费资源：

import asyncio
import os
import select

async def pty_read_and_forward(self, sio: AsyncServer, client_sid: str, client_data: dict):
    log = get_logger()
    max_read_bytes = 1024 * 20
    pty_data = self.ptys[client_sid]
    pty_fd = pty_data['fd']
    child_pid = pty_data['child_pid']
    
    # 设置PTY为非阻塞模式
    os.set_blocking(pty_fd, False)
    
    loop = asyncio.get_running_loop()
    try:
        while True:
            # 等待子进程退出或PTY有数据
            done, _ = await asyncio.wait([
                loop.run_in_executor(None, os.waitpid, child_pid, 0),
                loop.run_in_executor(None, select.select, [pty_fd], [], [], 0.1)
            ], return_when=asyncio.FIRST_COMPLETED)
            
            # 检查子进程是否退出
            try:
                pid, status = os.waitpid(child_pid, os.WNOHANG)
                if pid != 0:
                    log.info(f'PTY子进程已退出 [pid={child_pid}]')
                    break
            except OSError:
                break
            
            # 读取PTY数据
            try:
                output = os.read(pty_fd, max_read_bytes)
                if not output:
                    # 读取空字节表示PTY已关闭
                    break
                await sio.emit(
                    event='pty_out',
                    data=dict(output=output.decode('utf-8', errors='ignore')),
                    namespace='/pty',
                    room=client_sid
                )
            except BlockingIOError:
                # 无数据可读，继续循环
                continue
            except OSError as exc:
                log.error(f'读取PTY失败: {exc}')
                break
    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"]}] - {str(exc)}')
    finally:
        # 清理PTY资源
        try:
            os.close(pty_fd)
            del self.ptys[client_sid]
        except:
            pass
        await self.disconnect_client(sio=sio, client_sid=client_sid)

3. 客户端断开时清理子进程资源 #

确保断开连接时杀死PTY子进程并关闭文件描述符：

async def disconnect_client(self, sio: AsyncServer, client_sid: str):
    if client_sid in self.ptys:
        pty_data = self.ptys[client_sid]
        try:
            os.kill(pty_data['child_pid'], 9)
            os.close(pty_data['fd'])
        except OSError:
            pass
        del self.ptys[client_sid]
    if client_sid in self.clients:
        del self.clients[client_sid]
    await sio.disconnect(client_sid, namespace='/pty')

内容的提问来源于stack exchange，提问作者Thomas B.