Linux环境下如何同时等待进程与文件事件？

我正在Python中构建事件循环，主要用于学习目的。我对涉及asyncio的方案不感兴趣，因为我的目标是了解底层Linux API。

对于类文件对象（管道、套接字等），有select/poll/epoll系统调用，在Python中这些被封装在selectors模块中，示例代码如下：

file1 = open('/var/run/foo.sock')
file2 = open('/var/run/bar.sock')

with selectors.DefaultSelector() as selector:
    selector.register(file1, selectors.EVENT_READ)
    selector.register(file2, selectors.EVENT_READ)

    while True:
        for key, mask in selector.select():
            if key.fileobj == file1:
                handle_data(file1)
            else:
                handle_data(file2)

对于进程，有wait系统调用，在Python中被封装在subprocess模块中，等待单个进程终止的代码如下：

proc = subprocess.Popen(['/usr/bin/foo'])
proc.wait()
handle_data(proc)

我该如何将两者结合，甚至同时等待多个进程？

selector.select()和proc.wait()都支持设置超时，因此我可以实现一个半忙循环：

file1 = open('/var/run/foo.sock')
file2 = open('/var/run/bar.sock')
proc = subprocess.Popen(['/usr/bin/foo'])
timeout = 1

with selectors.DefaultSelector() as selector:
    selector.register(file1, selectors.EVENT_READ)
    selector.register(file2, selectors.EVENT_READ)

    while True:
        for key, mask in selector.select(timeout):
            if key.fileobj == file1:
                handle_data(file1)
            else:
                handle_data(file2)

        if proc:
            try:
                proc.wait(timeout)
                handle_data(proc)
                proc = None
            except TimeoutExpired:
                pass

我认为肯定有更好的方法，通常这种需求是如何实现的？

在Linux系统中，要同时监听类文件对象和多个进程状态，有两种更高效的实现方式，可避免半忙循环的低效问题：

1. 将进程的标准输出/错误注册到Selector #

如果需要监听进程的IO输出，可直接把进程的stdout/stderr管道文件描述符注册到selectors中，同时处理IO事件和进程退出：

import selectors
import subprocess
from subprocess import TimeoutExpired

def handle_file_data(fileobj):
    data = fileobj.read()
    if data:
        print(f"Received data from file: {data}")
    else:
        # 文件描述符关闭，取消注册并清理
        selector.unregister(fileobj)
        fileobj.close()

def handle_proc_exit(proc):
    print(f"Process {proc.pid} exited with code {proc.returncode}")
    # 清理进程关联的管道描述符
    for fd in [proc.stdout, proc.stderr]:
        if fd:
            selector.unregister(fd)
            fd.close()

selector = selectors.DefaultSelector()

# 注册类文件对象（套接字/管道）
file1 = open('/var/run/foo.sock', 'rb')
selector.register(file1, selectors.EVENT_READ, data={'type': 'file', 'handler': handle_file_data})

file2 = open('/var/run/bar.sock', 'rb')
selector.register(file2, selectors.EVENT_READ, data={'type': 'file', 'handler': handle_file_data})

# 启动并注册多个进程的stdout/stderr
active_procs = []
# 第一个进程
proc1 = subprocess.Popen(
    ['/usr/bin/foo'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)
selector.register(proc1.stdout, selectors.EVENT_READ, data={'type': 'proc', 'proc': proc1, 'handler': handle_file_data})
selector.register(proc1.stderr, selectors.EVENT_READ, data={'type': 'proc', 'proc': proc1, 'handler': handle_file_data})
active_procs.append(proc1)

# 第二个进程
proc2 = subprocess.Popen(
    ['/usr/bin/bar'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)
selector.register(proc2.stdout, selectors.EVENT_READ, data={'type': 'proc', 'proc': proc2, 'handler': handle_file_data})
selector.register(proc2.stderr, selectors.EVENT_READ, data={'type': 'proc', 'proc': proc2, 'handler': handle_file_data})
active_procs.append(proc2)

try:
    while True:
        events = selector.select()
        for key, mask in events:
            data = key.data
            if data['type'] == 'file':
                data['handler'](key.fileobj)
            elif data['type'] == 'proc':
                data['handler'](key.fileobj)
                # 检查进程是否已退出
                proc = data['proc']
                if proc.poll() is not None:
                    handle_proc_exit(proc)
                    active_procs.remove(proc)
        # 无活跃进程和监听描述符时退出循环
        if not active_procs and not selector.get_map():
            break
finally:
    selector.close()
    for proc in active_procs:
        proc.terminate()

核心逻辑是利用进程管道fd的关闭事件感知进程退出，同时统一处理IO和进程状态变化。

2. 监听SIGCHLD信号结合Selector #

如果不需要监听进程IO，仅需感知进程退出，可借助Linux的SIGCHLD信号机制，结合selectors避免轮询开销：

import selectors
import subprocess
import signal
import os

# 存储活跃进程的字典，key为进程PID
active_procs = {}

def handle_file_data(fileobj):
    data = fileobj.read()
    if data:
        print(f"Received data from file: {data}")
    else:
        selector.unregister(fileobj)
        fileobj.close()

def handle_proc_exit(pid, status):
    proc = active_procs.pop(pid)
    returncode = os.waitstatus_to_exitcode(status)
    print(f"Process {pid} exited with code {returncode}")
    handle_data(proc)

def sigchld_handler(signum, frame):
    # 批量回收已退出的子进程，避免僵尸进程
    while True:
        try:
            pid, status = os.waitpid(-1, os.WNOHANG)
            if pid == 0:
                break
            handle_proc_exit(pid, status)
        except OSError:
            break

# 注册SIGCHLD信号处理函数
signal.signal(signal.SIGCHLD, sigchld_handler)

selector = selectors.DefaultSelector()

# 注册类文件对象
file1 = open('/var/run/foo.sock', 'rb')
selector.register(file1, selectors.EVENT_READ, handle_file_data)

file2 = open('/var/run/bar.sock', 'rb')
selector.register(file2, selectors.EVENT_READ, handle_file_data)

# 启动多个进程
proc1 = subprocess.Popen(['/usr/bin/foo'])
active_procs[proc1.pid] = proc1

proc2 = subprocess.Popen(['/usr/bin/bar'])
active_procs[proc2.pid] = proc2

try:
    while True:
        events = selector.select()
        for key, mask in events:
            handler = key.data
            handler(key.fileobj)
        # 无活跃进程和监听描述符时退出循环
        if not active_procs and not selector.get_map():
            break
finally:
    selector.close()
    for proc in active_procs.values():
        proc.terminate()

这种方式完全基于Linux信号机制感知进程退出，无需轮询，是更贴近底层的实现方式。

半忙循环的问题 #

你之前的半忙循环存在两个明显缺陷：

• 低效：每次selector.select()超时后都要调用proc.wait()，即使进程未退出，会浪费CPU资源。
• 扩展性差：多进程场景下需逐个调用wait()，逻辑易混乱，还可能产生僵尸进程。

以上两种方案都是Linux底层API的标准用法，既能高效处理IO事件，又能正确管理进程生命周期。

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