ProcessPoolExecutor进程陷入Sl+状态问题排查与重启方案咨询

问题描述 #

代码结构如下：

with ProcessPoolExecutor(max_workers = n) as executor:
    for i in range(k):
        executor.submit(func, i)

执行时约半数进程会在6~8秒内陷入Sl+状态（单个函数正常单进程运行时长约20秒），但运行超过10秒后会稳定执行完成；单进程模式下函数运行完全正常，仅存在无法控制的日志输出部分。

可能的问题原因 #

• 日志输出的进程竞争：多进程并发写入同一日志文件时，若使用的日志库未做进程安全处理，会引发文件锁等待。6~8秒才出现卡住，可能是因为前序日志操作未触发竞争，直到多个进程同时进入日志密集输出阶段才触发阻塞，导致进程进入可中断睡眠状态。
• 隐性资源竞争：函数内部可能存在全局变量、共享文件句柄或第三方非进程安全组件，单进程下无冲突，但多进程并发时，在运行到6~8秒的逻辑节点触发了资源等待。
• 系统资源限制：进程运行到特定阶段（6~8秒）时需要申请新资源（如文件描述符、内存页），若此时系统资源紧张，会导致进程进入睡眠等待状态。

检测与重启卡住进程的方案 #

检测进程状态 #

• 使用psutil库监控子进程状态：遍历进程池内的进程，检查是否处于sleeping状态且运行时长超过设定阈值（如10秒），以此判断是否卡住：

import psutil
import time
from concurrent.futures import ProcessPoolExecutor

def monitor_pool(executor, check_interval=2, timeout=10):
    while True:
        for proc in executor._processes.values():
            p = psutil.Process(proc.pid)
            run_duration = time.time() - p.create_time()
            if p.status() == 'sleeping' and run_duration > timeout:
                print(f"进程 {p.pid} 疑似卡住，已运行 {run_duration:.1f} 秒")
                # 此处可调用重启逻辑
        time.sleep(check_interval)

• 任务超时检测：通过concurrent.futures.wait给任务设置超时，未在指定时间内完成的任务标记为异常：

from concurrent.futures import wait, FIRST_COMPLETED

# 提交任务后收集future对象
futures = [executor.submit(func, i) for i in range(k)]
# 每轮检查超时任务
while futures:
    done, pending = wait(futures, timeout=10, return_when=FIRST_COMPLETED)
    for future in pending:
        # 标记为卡住，处理重启
        print("任务超时，准备重启")
    futures = list(pending)

重启卡住进程 #

• 终止异常进程并重新提交任务：检测到卡住进程后，先终止进程，再重新提交对应任务（需维护任务参数与进程的映射，或通过参数追踪）：

def restart_stuck_proc(executor, func, task_args_map):
    for pid, proc in executor._processes.items():
        p = psutil.Process(pid)
        run_duration = time.time() - p.create_time()
        if p.status() == 'sleeping' and run_duration > 10:
            # 终止进程
            p.terminate()
            p.wait()
            # 重新提交对应任务（需提前维护pid与任务参数的映射）
            args = task_args_map.get(pid)
            if args:
                executor.submit(func, *args)
                print(f"重启进程 {pid} 对应的任务，参数：{args}")

• 自定义带超时重启的进程池：封装ProcessPoolExecutor，给每个任务绑定超时逻辑，超时自动终止进程并重启任务。

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