使用ProcessPoolExecutor的map方法在大负载下出现挂起问题

问题现象 #

• 大负载场景下，使用ProcessPoolExecutor.map执行任务，运行约1分钟后陷入挂起状态，CPU使用率骤降并趋于空闲
• 不同时间点采集的栈跟踪内容基本一致，阻塞点集中在多进程队列通信与线程join环节

相关代码 #

import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor

def work_wrapper(args):
    return work(*args)

def work():
    # 具体任务逻辑
    ...

def start_working(num_threads, ...):
    with ProcessPoolExecutor(max_workers=num_threads, mp_context=mp.get_context('fork')) as executor:
        args = [arg_list1, arg_list2, ...]
        for res in executor.map(work_wrapper, args):
            pass

if __name__ == "__main__":
    mp.set_start_method('fork', force=True)
    start_working(...)

栈跟踪核心阻塞点 #

从栈信息可看出所有线程均卡在队列相关操作：

• 队列_feed线程阻塞于connection._send，进程间通信管道被填满
• 主线程与executor管理线程卡在threading.join，等待队列线程退出，但队列线程因管道阻塞无法推进
• 关键调用链：shutdown → join → _wait_for_tstate_lock；_feed → send_bytes → _send

环境信息 #

• Python版本：3.10.8
• Docker基础镜像：python:3.10-slim
• 已尝试操作：升级Python版本、切换fork/spawn多进程上下文，问题未解决

问题根源 #

这是典型的多进程队列生产者-消费者死锁：

1. 大负载下，子进程生成结果的速度远快于主进程通过map读取结果的速度
2. 队列管道缓冲区被填满后，子进程的结果写入线程直接阻塞在send操作
3. 当executor尝试关闭时，管理线程等待队列线程结束，但队列线程因阻塞无法退出，形成死锁

解决方法 #

1. 分批次提交任务，避免队列堆积 #

不要一次性提交所有任务给map，按worker数量分批次提交，处理完一批再提交下一批：

def start_working(num_threads, ...):
    batch_size = num_threads * 2  # 按worker数量的2倍设置批次大小
    args = [arg_list1, arg_list2, ...]
    with ProcessPoolExecutor(max_workers=num_threads, mp_context=mp.get_context('fork')) as executor:
        for i in range(0, len(args), batch_size):
            batch = args[i:i+batch_size]
            for res in executor.map(work_wrapper, batch):
                # 即时处理结果，避免队列堆积
                pass

2. 用submit+as_completed替代map，主动控制任务流 #

map会一次性提交所有任务，改用submit配合as_completed可灵活控制并发数，防止队列溢出：

from concurrent.futures import as_completed

def start_working(num_threads, ...):
    args = [arg_list1, arg_list2, ...]
    with ProcessPoolExecutor(max_workers=num_threads, mp_context=mp.get_context('fork')) as executor:
        futures = []
        for arg in args:
            futures.append(executor.submit(work_wrapper, arg))
            # 当待处理任务数达到worker数2倍时，先处理已完成的任务
            if len(futures) >= num_threads * 2:
                for future in as_completed(futures):
                    res = future.result()
                    # 处理结果
                futures = []
        # 处理剩余任务
        for future in as_completed(futures):
            res = future.result()

3. 检查work函数稳定性 #

• 确保work函数无未释放的文件句柄、网络连接等资源，这类泄漏会导致子进程异常退出，引发executor清理死锁
• 给work添加异常捕获，避免子进程崩溃：

def work(*args):
    try:
        # 原任务逻辑
        ...
    except Exception as e:
        print(f"任务失败: {e}")
        return None

4. 改用multiprocessing.Pool，调整队列缓冲区 #

若ProcessPoolExecutor修改成本高，可直接使用multiprocessing.Pool，设置队列容量与子进程重启策略：

from multiprocessing import Pool

def start_working(num_threads, ...):
    args = [arg_list1, arg_list2, ...]
    # maxtasksperchild=1让子进程处理完一个任务后重启，避免资源泄漏；queue_size限制队列容量
    with Pool(processes=num_threads, maxtasksperchild=1, queue_size=num_threads*4) as pool:
        for res in pool.imap(work_wrapper, args):
            pass

验证建议 #

优先尝试方案1和2，二者直接解决任务堆积导致的队列阻塞问题，是最常用的修复手段。若问题仍存在，重点排查work函数是否有隐性资源泄漏或未处理异常。

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