如何排查multiprocessing.Pool中阻塞进程及未就绪原因？

背景 #

我编写了如下Python代码，利用multiprocessing.Pool实现多进程任务：

import datetime
import multiprocessing

def my_function(a, b, c):
    c.append(a)
    # 一个耗时很长、返回大量数据的函数
    c.remove(a)
    return (random_tuple, very_large_list_of_data)

def main():
    start_time = datetime.datetime.now()
    prev_one_minute_passed = False
    large_list_a = list(range(100)) # 忽略写法不够Pythonic，只是示例
    large_list_b = [2] * 100
    l = [] # 实际用multiprocessing.manager管理，这里简化为普通列表

    with multiprocessing.Pool(processes=99) as pool:
        # 修正原代码笔误：zip对应large_list_a和large_list_b
        out_results = pool.starmap_async(my_function, [(a, b, l) for (a, b) in zip(large_list_a, large_list_b)])

        while not out_results.ready():
            # 执行一些日志记录操作
            if ((datetime.datetime.now() - start_time) > datetime.timedelta(minutes=1)):
                # 运行时间超过1分钟
                if not l:
                    # 列表为空，推测所有任务已完成
                    break

        out_tuple_list = out_results.get(timeout=300)

if __name__ == '__main__':
    main()

运行时偶尔会触发multiprocessing.Timeout异常：当共享列表l为空（推测所有任务已完成）5分钟后，结果仍未就绪。

我通过IPython进入调试环境后发现：所有进程均处于存活状态，池仍为RUN状态，out_results未就绪。

问题 #

如何确定multiprocessing.Pool中哪个进程阻止池进入完成状态？在Python3.10中是否有方法监控该进程未就绪的原因？

补充说明 #

我已知存在竞态条件，但任务通常不会耗时5分钟，且在IPython中查询列表l仍为空。

排查方案 #

一、定位阻塞的进程 #

• 获取进程PID并结合系统工具分析
  在调试会话中调用multiprocessing.active_children()，可以拿到池内所有存活进程的对象，从中提取每个进程的pid属性。接着用系统工具查看进程状态：
  • Linux环境：执行ps aux | grep <pid>查看进程基本信息，top -p <pid>观察CPU、内存占用，strace -p <pid>跟踪系统调用，判断是否卡在IO或系统调用上。
  • Windows环境：打开任务管理器，通过PID找到对应进程，查看资源占用，或用Process Explorer分析进程的线程栈。
• 给任务绑定进程标识
  在my_function开头记录当前进程ID和任务参数，方便后续对应任务与进程：

  import os
  
  def my_function(a, b, c):
      pid = os.getpid()
      print(f"进程{pid}开始处理任务a={a}")
      # 原函数逻辑
  

二、Python3.10中监控未就绪原因的方法 #

1. 增强任务的日志与异常捕获
   给my_function添加详细日志和全局异常捕获，避免子进程因未处理异常静默崩溃或卡住，同时记录关键节点的状态：

   import traceback
   import os
   
   def my_function(a, b, c):
       pid = os.getpid()
       try:
           print(f"进程{pid}: 开始执行任务，参数a={a}")
           c.append(a)
           print(f"进程{pid}: 已标记任务活跃")
           # 原耗时操作
           print(f"进程{pid}: 核心任务执行完成")
           c.remove(a)
           print(f"进程{pid}: 已清除任务标记")
           return (random_tuple, very_large_list_of_data)
       except Exception as e:
           tb_str = traceback.format_exc()
           print(f"进程{pid}处理任务a={a}失败: {e}\n{tb_str}")
           # 返回异常信息，便于后续结果分析
           return (None, None, str(e), tb_str)
   

2. 使用faulthandler生成栈追踪
   Python3.10中可以在子进程初始化时启用faulthandler，当怀疑进程卡住时，通过发送信号让进程输出当前调用栈，定位卡住的代码位置：

   import faulthandler
   import signal
   import os
   
   def init_child_process():
       # 启用faulthandler，绑定SIGUSR1信号触发栈追踪
       faulthandler.enable()
       # 将栈追踪写入单独文件，避免日志混乱
       trace_file = open(f"process_trace_{os.getpid()}.log", "w")
       faulthandler.register(signal.SIGUSR1, file=trace_file)
   
   # 创建Pool时指定初始化函数
   with multiprocessing.Pool(processes=99, initializer=init_child_process) as pool:
       # 后续代码不变
   

   当发现进程卡住时，在终端执行kill -SIGUSR1 <pid>（Linux），进程会将当前栈追踪写入对应日志文件，直接查看即可定位问题。
3. 修复共享列表的竞态隐患
   原代码中c.remove(a)存在竞态风险：如果多个进程处理相同的a值，可能出现某个进程执行remove时a已被其他进程移除，导致抛出ValueError。改用Manager.dict跟踪活跃任务计数，避免remove操作的竞态：

   from multiprocessing import Manager
   
   def main():
       # ... 其他代码
       manager = Manager()
       active_tasks = manager.dict()  # 键为a，值为活跃任务数
   
       with multiprocessing.Pool(processes=99) as pool:
           out_results = pool.starmap_async(my_function, [(a, b, active_tasks) for (a, b) in zip(large_list_a, large_list_b)])
           # ... 其他代码
   
   def my_function(a, b, active_tasks):
       # 原子更新活跃计数
       active_tasks[a] = active_tasks.get(a, 0) + 1
       try:
           # 原耗时操作
           pass
       finally:
           # 原子减少计数，计数为0时删除键
           active_tasks[a] -= 1
           if active_tasks[a] == 0:
               del active_tasks[a]
   

4. 查看Pool的剩余任务数
   在调试时，通过out_results._number_left可以直接获取剩余未完成的任务数量，结合之前的任务日志，快速定位未完成的任务参数，进而找到对应的进程。

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