如何在Python中控制固定数量多进程执行基准测试

我需要分析不同基准测试的评估结果，在服务器运行时希望每次并行执行10个评估任务。现有Python评估函数：

def load_single_stat(benchmark,prefetcher, retry=False):
    # 评估逻辑实现
    pass

当前调用该函数时需等待返回才能执行后续逻辑，之前用Shell脚本控制同时运行10个Python进程，但想直接在Python脚本内实现。已知可以用多进程运行函数，但不知道如何控制同时运行的进程数量（服务器还有其他用户，不想占用全部资源），求高效实现方案。

原Shell脚本参考：

for((i=0;i<${#PREFETCH_METHODS[@]};i++))
do
    for ((j=1; j<=$BENCHMARK_NUM; j++))
    do
        sleep 2
        array=($(ps -aux | grep -o  ${PREFETCH_METHODS[i]}))
        echo ${#array[@]} 
        while [ ${#array[@]} -ge 10 ]
        do
            sleep 60
            array=($(ps -aux | grep -o  ${PREFETCH_METHODS[i]}))
        done

        cmd="python my_script.py ${PREFETCH_METHODS[i]} $BENCHMARK_NUM " 

        $cmd &

    done
done

方法1：使用concurrent.futures.ProcessPoolExecutor（推荐） #

这是Python标准库中最简洁的实现方式，直接通过max_workers参数限制并发进程数，无需手动管理进程生命周期。

示例代码 #

from concurrent.futures import ProcessPoolExecutor
import itertools

# 替换为你的实际基准测试列表和预取器方法列表
BENCHMARKS = [1, 2, 3, ...]
PREFETCH_METHODS = ["method_a", "method_b", "method_c", ...]

# 生成所有任务组合（对应原脚本的双层循环）
task_args = list(itertools.product(BENCHMARKS, PREFETCH_METHODS))

# 控制最大并发进程数为10
with ProcessPoolExecutor(max_workers=10) as executor:
    # 批量提交任务，自动分配进程执行
    results = executor.map(
        lambda args: load_single_stat(args[0], args[1], retry=False),
        task_args
    )

# 若需要处理任务返回结果，遍历results即可
for result in results:
    # 处理单个任务的返回值
    pass

关键说明 #

• ProcessPoolExecutor会自动维护进程池，最多同时运行指定数量的进程，无需手动监控进程数量
• itertools.product用于生成基准测试与预取器的所有组合，完美替代原Shell脚本的双层循环
• 若任务不需要返回值，可改用executor.submit()逐个提交，无需收集结果

方法2：使用multiprocessing.Pool #

这是更底层的多进程池实现，适合需要更精细控制的场景，原理与ProcessPoolExecutor一致。

示例代码 #

from multiprocessing import Pool
import itertools

def task_wrapper(args):
    # 封装参数，适配Pool.map的单参数要求
    benchmark, prefetcher = args
    return load_single_stat(benchmark, prefetcher, retry=False)

if __name__ == "__main__":
    BENCHMARKS = [1, 2, 3, ...]
    PREFETCH_METHODS = ["method_a", "method_b", "method_c", ...]
    task_args = list(itertools.product(BENCHMARKS, PREFETCH_METHODS))

    # 初始化进程池，设置最大进程数10
    with Pool(processes=10) as pool:
        results = pool.map(task_wrapper, task_args)

    # 处理返回结果
    for result in results:
        pass

关键说明 #

• 必须将主逻辑放在if __name__ == "__main__":代码块中，避免多进程启动时的递归导入问题
• Pool.map会按任务提交顺序返回结果，便于对应任务的结果处理

额外优化建议 #

• 如果评估任务以IO操作为主（如读写文件、磁盘IO），可改用ThreadPoolExecutor（线程池），减少进程创建开销；若为CPU密集型任务，必须用进程池（规避GIL限制）
• 若服务器资源紧张，可将max_workers设为8或9，预留少量资源给其他用户
• 若需要动态监控任务状态或调整并发数，可结合multiprocessing.Queue手动管理进程，但上述两种方法已足够满足大多数场景需求

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