Python3.6 multiprocessing.Pool响应KeyboardInterrupt间歇性退出失败求助

背景与问题 #

你用multiprocessing.Pool异步调度外部sleep子进程，期望收到SIGINT（也就是Ctrl+C）时能干净终止所有相关进程——外部子进程、池worker、主进程。当池大小设为4时一切正常，但增大池规模后，程序会间歇性出现退出失败的情况，部分worker和它们启动的sleep进程会残留。

你的测试代码如下：

#!/bin/env python3 
''' test.py ''' 
import multiprocessing.util 
from multiprocessing import Pool 
import shlex 
import subprocess 
from subprocess import PIPE 

multiprocessing.util.log_to_stderr(multiprocessing.util.DEBUG) 

def waiter(arg): 
    cmd = "sleep 360" 
    cmd_arg = shlex.split(cmd) 
    p = subprocess.Popen(cmd_arg, stdout=PIPE, stderr=PIPE) 
    so, se = p.communicate() 
    print(f"{so}\n{se}") 
    return arg 

def main1(): 
    proc_pool = Pool(4) 
    it = proc_pool.imap_unordered(waiter, range(0, 4)) 
    for r in it: 
        print(r) 

if __name__ == '__main__': 
    main1()

为什么池变大后会失败？ #

核心原因有两个：

1. 信号传递的竞争与阻塞：当按下Ctrl+C，SIGINT会发送给整个进程组。主进程收到信号后触发KeyboardInterrupt，但池中的worker如果正阻塞在subprocess.Popen.communicate()调用上（这是阻塞IO操作），内部的select()/poll()调用可能延迟信号处理，甚至部分worker根本没来得及接收信号就被主进程的默认逻辑遗漏。
2. 默认池清理逻辑的局限性：multiprocessing.Pool的默认退出逻辑在处理大量worker时，无法保证所有worker都能及时响应信号并清理自己启动的子进程。有些worker可能还在等待外部进程输出，导致整个清理流程卡住。

修复方案 #

要实现可靠的SIGINT终止，我们需要做两件事：

• 让每个worker进程在收到SIGINT时，先清理自己启动的外部子进程，再退出。
• 主进程捕获SIGINT后，主动强制终止所有worker进程，确保没有遗漏。

以下是修改后的完整代码：

#!/bin/env python3
''' test.py '''
import multiprocessing.util
from multiprocessing import Pool
import shlex
import subprocess
from subprocess import PIPE
import signal

multiprocessing.util.log_to_stderr(multiprocessing.util.DEBUG)

def cleanup_subprocess(signum, frame):
    """信号处理函数：终止当前worker启动的外部子进程"""
    global current_subprocess
    if current_subprocess and current_subprocess.poll() is None:
        current_subprocess.terminate()
    # 重新抛出KeyboardInterrupt，让worker进程正常退出
    raise KeyboardInterrupt

def waiter(arg):
    global current_subprocess
    cmd = "sleep 360"
    cmd_arg = shlex.split(cmd)
    # 给当前worker注册SIGINT信号处理函数
    signal.signal(signal.SIGINT, cleanup_subprocess)
    
    # 启动外部子进程，使用start_new_session确保可以独立终止
    current_subprocess = subprocess.Popen(
        cmd_arg, 
        stdout=PIPE, 
        stderr=PIPE,
        start_new_session=True
    )
    
    try:
        so, se = current_subprocess.communicate()
        print(f"{so}\n{se}")
    except KeyboardInterrupt:
        # 确保异常时子进程被彻底清理
        if current_subprocess.poll() is None:
            current_subprocess.wait()
        raise
    return arg

def main1():
    # 这里可以设置更大的池大小，比如8
    proc_pool = Pool(8)
    try:
        it = proc_pool.imap_unordered(waiter, range(0, 8))
        for r in it:
            print(r)
    except KeyboardInterrupt:
        print("收到SIGINT，正在终止所有进程...")
        # 强制终止所有worker进程，这会向每个worker发送SIGTERM
        proc_pool.terminate()
    finally:
        # 等待所有worker进程彻底退出
        proc_pool.join()

if __name__ == '__main__':
    # 确保主进程的SIGINT使用默认处理逻辑，触发KeyboardInterrupt
    signal.signal(signal.SIGINT, signal.default_int_handler)
    main1()

关键修改点说明 #

1. worker信号处理：给每个worker进程注册SIGINT处理函数cleanup_subprocess，确保收到信号时先终止自己启动的外部子进程，再抛出异常让worker退出。
2. 独立子进程会话：启动外部子进程时使用start_new_session=True，让子进程脱离当前进程组，避免信号传递的干扰，同时也方便后续批量终止操作。
3. 主动池终止：主进程捕获KeyboardInterrupt后，主动调用proc_pool.terminate()——这个方法会直接向所有worker进程发送SIGTERM信号，强制它们终止，比默认的清理逻辑更可靠。
4. 异常安全清理：在waiter函数中用try-except包裹communicate()，确保即使在信号中断时，外部子进程也能被彻底清理。

这样修改后，无论池大小设置为多少，按下Ctrl+C都能可靠终止所有相关进程，不会出现残留的情况。

内容的提问来源于stack exchange，提问作者dr-igor