为何Python多进程脚本触发错误后未终止？

代码场景 #

我在Python脚本中通过multiprocessing.Pool实现并行计算，希望当特定条件满足时抛出错误并终止整个脚本，但实际触发错误后脚本仍持续运行，多次输出错误回溯。相关核心代码如下：

import sys

def my_func(some_arguments):
    # 部分业务代码
    if X.abs().max().max() > 1e50:
        # 输出错误信息到标准错误流
        print('------------ ERROR! line 287, extremely large X={} ----------------'.format(X.abs().max().max()), file=sys.stderr)
        print('selectedFeatures: '+('-'.join(sorted(list(selectedFeatures)))), file=sys.stderr)
        print('featurePool: '+('-'.join(sorted(list(featurePool)))), file=sys.stderr)
        print('p1: '+p1+',  p2: '+p2, file=sys.stderr)
        print('X.shape={}X{}'.format(X.shape[0],X.shape[1]), file=sys.stderr)
        tmp = 1/0 # 通过除零触发错误
    # 剩余业务代码

def caller_func():
    for ...:
        # 部分业务代码
        my_func(arguments)
        # 剩余业务代码

错误现象 #

触发条件后，脚本会输出错误信息和回溯，但不会终止，而是继续运行其他并行任务，重复输出类似错误：

# 部分标准输出内容

------------ ERROR! line 287, extremely large X=3.130749212136059e+58 ----------------
selectedFeatures: <some_string>
featurePool: <some_string>
p1: <some_string>, p2: <some_string>
X.shape=133615X41
/usr/lib/python3.10/multiprocessing/process.py:-1: ResourceWarning: unclosed file <_io.TextIOWrapper name='my_file_1.csv' mode='w' encoding='UTF-8'>
ResourceWarning: Enable tracemalloc to get the object allocation traceback
Traceback (most recent call last):
  File "//script_heavy_ga.py", line 332, in caller_func
    X, selectedFeatures, featurePool = my_func(p1, p2, X, X_original, selectedFeatures, featurePool)
  File "//script_heavy_ga.py", line 293, in my_func
    tmp = 1/0 # to create an error
ZeroDivisionError: division by zero

# 部分标准输出内容
# 重复出现类似错误输出...

原因分析 #

这是multiprocessing.Pool的核心特性导致的：

• Pool创建的每个子进程都是独立的操作系统进程，拥有独立的内存空间和执行上下文，子进程中抛出的异常只会终止当前子进程，不会直接影响主进程或其他子进程。
• 当子进程因异常退出后，Pool会自动重新启动新的子进程处理剩余任务（若存在未完成任务），或继续执行其他已分配的子进程任务，因此整个脚本不会终止。
• 你用1/0触发的ZeroDivisionError仅在当前子进程中生效，主进程无法主动感知该异常（除非显式处理），所以会继续调度其他并行任务。

解决方案 #

方案1：在子进程中主动终止整个进程组 #

如果需要触发条件时立即终止整个脚本（包括主进程和所有子进程），可在错误处理逻辑中直接杀死主进程：

import os
import signal
import sys

def my_func(some_arguments):
    # 部分业务代码
    if X.abs().max().max() > 1e50:
        # 输出错误信息（保留原有逻辑）
        print('------------ ERROR! line 287, extremely large X={} ----------------'.format(X.abs().max().max()), file=sys.stderr)
        # ...其他错误信息输出
        # 杀死主进程，终止整个脚本
        os.kill(os.getppid(), signal.SIGTERM)
        # 确保子进程快速退出
        os._exit(1)

方案2：主进程捕获异常并终止Pool #

在主进程中通过Pool的方法捕获子进程抛出的异常，一旦捕获到异常就立即终止整个Pool并退出脚本：

from multiprocessing import Pool

if __name__ == '__main__':
    tasks = [...] # 你的并行任务列表
    with Pool(processes=4) as pool:
        try:
            # 使用map/imap等方法执行任务，子进程异常会传递到主进程
            results = pool.map(caller_func, tasks)
        except Exception as e:
            # 终止所有子进程
            pool.terminate()
            # 重新抛出异常，终止主进程
            raise e

注意：这种方式要求子进程的异常没有被内部捕获，需向上传递到caller_func，最终传递到主进程。

方案3：自定义异常并强制终止 #

定义自定义异常，子进程抛出后，主进程捕获时立即终止所有任务：

from multiprocessing import Pool

class CriticalError(Exception):
    pass

def my_func(some_arguments):
    # 部分业务代码
    if X.abs().max().max() > 1e50:
        # 输出错误信息
        # ...
        raise CriticalError("Extremely large X detected")

# 主进程调用逻辑
if __name__ == '__main__':
    tasks = [...]
    with Pool(processes=4) as pool:
        try:
            results = pool.map(caller_func, tasks)
        except CriticalError as e:
            print(f"Critical error: {e}", file=sys.stderr)
            pool.terminate()
            exit(1)

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