Python多进程简单场景性能不佳的问题排查与优化求助

问题场景 #

在8核Linux机器上，使用以下Python多进程代码生成排列大列表并计算指定数值：当设置num_threads=4时，可见4个进程，但每个进程CPU占用最高达200%，导致整机满载；同时性能表现极差，2线程加速比仅1.4，4线程仅2.3，远低于预期。

def multiprocessing_mci(arr, Smo, num_threads):
   import multiprocessing as mp

   permutations = unique_permutations(arr)

   
   chunk_size = int(np.ceil(len(permutations) / num_threads))
   permutations_chunks = [permutations[i:i+chunk_size] for i in range(0, len(permutations), chunk_size)]

   
   pool = mp.Pool(processes = num_threads)

   results = []
   for chunk in permutations_chunks:
      result = pool.starmap(compute_iring, [(order, Smo) for order in chunk])
      results.extend(result)
   
   trace = sum(results)

   return trace

一、CPU占用过高（单进程达200%）的解决 #

原因 #

compute_iring调用的numpy底层依赖OpenMP实现多线程并行，每个子进程默认会占用多个CPU核心，导致4个子进程实际占用的核心数远超4个，最终整机满载。

解决方法 #

在创建进程池前设置环境变量OMP_NUM_THREADS=1，强制每个子进程的numpy仅使用1个线程：

def multiprocessing_mci(arr, Smo, num_threads):
    import multiprocessing as mp
    import os
    # 限制OpenMP线程数，避免子进程内部多线程
    os.environ["OMP_NUM_THREADS"] = "1"
    
    permutations = unique_permutations(arr)
    
    chunk_size = int(np.ceil(len(permutations) / num_threads))
    permutations_chunks = [permutations[i:i+chunk_size] for i in range(0, len(permutations), chunk_size)]
    
    pool = mp.Pool(processes = num_threads)
    
    results = []
    for chunk in permutations_chunks:
        result = pool.starmap(compute_iring, [(order, Smo) for order in chunk])
        results.extend(result)
    
    trace = sum(results)
    
    return trace

设置后每个进程CPU占用会控制在100%左右，4个进程总占用约400%，不会占用全部8核资源。

二、加速比过低的优化方向 #

1. 减少进程间数据传输开销 #

当前代码提前生成全量排列并拆分chunk，大列表在进程间传递时会产生大量pickle序列化/反序列化开销，且占用大量内存。优化方案：

• 直接将排列迭代器传入starmap，无需手动拆分chunk，让进程池自动分配任务
• 改用imap_unordered流式处理结果，避免一次性加载所有结果到内存

修改示例：

def multiprocessing_mci(arr, Smo, num_threads):
    import multiprocessing as mp
    import os
    os.environ["OMP_NUM_THREADS"] = "1"
    
    permutations = unique_permutations(arr)
    
    pool = mp.Pool(processes = num_threads)
    # 直接传递排列迭代器，进程池自动分配任务
    results = pool.starmap(compute_iring, [(order, Smo) for order in permutations])
    
    trace = sum(results)
    
    return trace

2. 优化compute_iring函数本身 #

加速比低的核心可能是单线程性能不足，或者函数内有冗余操作：

• 检查是否存在Python循环，替换为numpy向量化操作
• 避免在函数内重复创建大数组，尽量复用内存
• 如果有重复计算逻辑，添加缓存（如functools.lru_cache，注意参数需可哈希）

3. 减少大参数传递开销 #

如果Smo是大数组，每次调用compute_iring都传递该数组会产生重复的内存拷贝开销。可以通过functools.partial绑定参数，或用进程池初始化函数将Smo设为子进程全局变量：

用partial绑定参数：

import functools

def compute_iring(order, Smo):
    # 原函数逻辑
    pass

def multiprocessing_mci(arr, Smo, num_threads):
    import multiprocessing as mp
    import os
    os.environ["OMP_NUM_THREADS"] = "1"
    
    permutations = unique_permutations(arr)
    # 绑定Smo参数，避免重复传递
    bound_compute = functools.partial(compute_iring, Smo=Smo)
    
    pool = mp.Pool(processes = num_threads)
    results = pool.map(bound_compute, permutations)
    
    trace = sum(results)
    
    return trace

用进程池初始化器：

def init_worker(smo):
    global shared_smo
    shared_smo = smo

def compute_iring_wrapper(order):
    return compute_iring(order, shared_smo)

def multiprocessing_mci(arr, Smo, num_threads):
    import multiprocessing as mp
    import os
    os.environ["OMP_NUM_THREADS"] = "1"
    
    permutations = unique_permutations(arr)
    
    pool = mp.Pool(processes = num_threads, initializer=init_worker, initargs=(Smo,))
    results = pool.map(compute_iring_wrapper, permutations)
    
    trace = sum(results)
    
    return trace

4. 优化排列生成逻辑 #

如果unique_permutations生成排列的耗时占比高，可替换为更高效的实现：

• 用itertools.permutations结合set去重（注意仅适用于小排列规模）
• 使用more_itertools.distinct_permutations等专门的排列生成函数提升效率

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