使用Python多进程时线程越多运算越慢的原因与解决咨询

问题背景 #

编写经济模型时采用嵌套循环结构，对独立外循环做并行化后，发现内循环单次迭代耗时随进程数增加而变长。已将MKL BLAS线程数设为1，排除后台自动并行化干扰；手动打开多个Python终端运行也出现同等性能损耗，排除multiprocessing库本身的进程创建开销问题。

核心原因 #

• CPU共享资源竞争：多个进程同时运行时，即使每个进程绑定单核心，CPU的L3缓存、内存带宽等共享资源会被争抢。代码中大量的数组运算与矩阵乘法，会导致多进程同时读写内存时带宽饱和、延迟上升；同时L3缓存被多个进程占满，缓存命中率下降，直接拖慢运算速度。
• NUMA架构影响：若机器采用NUMA（非统一内存访问）架构，进程分散到不同NUMA节点时，跨节点访问内存的延迟远高于本地节点，多进程并行会加剧这种延迟。
• 操作系统调度间接开销：即使绑定核心，操作系统调度其他进程时的上下文切换，也会对当前进程的运行产生微小但累积的影响。

解决办法 #

1. 绑定进程到NUMA节点与核心

   • Linux系统可使用numactl工具将进程绑定到同一NUMA节点的核心，减少跨节点内存访问开销：

     numactl --cpunodebind=0 --membind=0 python your_script.py
     

   • 或在Python中通过psutil库手动设置进程CPU亲和性，配合spawn启动方式：

     import psutil
     import multiprocessing
     
     def worker_core_binding(i, x):
         p = psutil.Process()
         p.cpu_affinity([i])  # 绑定到第i个核心
         return second_loop(i, x)
     
     if __name__ == '__main__':
         multiprocessing.set_start_method('spawn')
         # 后续进程池调用改用worker_core_binding
     

2. 优化任务粒度与内存访问

   • 合并小粒度的外循环任务，减少同时运行的进程数量，降低资源争用占比；
   • 调整数组存储顺序为连续内存，提升缓存命中率：

     x = np.ascontiguousarray(np.random.rand(1000,5000))
     y = np.ascontiguousarray(np.random.rand(1,1000))
     

   • 确认Numba函数禁用并行：

     @njit(parallel=False)
     def task1(x):
         return x**(1-0.4)/(1-0.4)
     

3. 监控系统资源定位瓶颈

   • 用perf工具查看缓存命中情况：

     perf stat -e cache-misses,cache-references python your_script.py
     

   • 用vmstat查看内存带宽使用率，确认是否存在内存饱和。

Julia中的类似问题 #

Julia的多线程/多进程同样会面临硬件层面的资源争用问题，这不是语言特性导致的。不过Julia的多线程基于共享内存，比Python多进程的内存复制开销更小；且Julia的数组内存布局更缓存友好，能一定程度缓解性能下降。若使用Julia的多进程模式，仍需通过--bind-to core参数绑定核心，或调整任务粒度来减少资源竞争。

复现代码 #

import numpy as np
from numba import njit
import time
import mkl
import multiprocessing

mkl.set_num_threads(1)
np.random.seed(0)

@njit()
def task1(x):
    return x**(1-0.4)/(1-0.4)

def second_loop(i,x):
    y = np.random.rand(1,1000)
    time_passed = []
    for j in range(15):
        tic = time.time()
        x = task1(x)
        x = x+i
        np.matmul(y,x)
        toc = time.time()
        time_passed.append(toc-tic)
        print(toc-tic)
    if i == 0:
        return time_passed

import matplotlib.pyplot as plt 
from statistics import mean 
    
if __name__ == '__main__':
    average_time = []
    for cpu_numbers in range(1,9):
        x = np.random.rand(1000,5000)
        args = [(i,x) for i in range(cpu_numbers)]
        pool_obj = multiprocessing.Pool(cpu_numbers)
        time_passed = pool_obj.starmap(second_loop, args)
        pool_obj.close()
        average_time.append(mean(time_passed[0][2:]))

    plt.plot(np.arange(1,9),average_time,marker="X",ls="--",mfc = "r",mec="r",ms=8)
    plt.xlabel("进程数")
    plt.ylabel("耗时（秒）")
    plt.savefig("Figure.png")

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