用python多处理优化一个简单的CPU绑定函数

Question

我正在努力理解multiprocessing.Pool是如何工作的，我开发了一个很小的例子来说明我的问题。简单地说，我使用pool.map并行处理一个运行在数组上的CPU绑定函数，方法是遵循使用多处理队列、池和锁定的简单示例示例。当我遵循这种模式时，我只得到了一个只有4个核心的加速比，但是如果我手动地将数组分块到num_threads中，然后对块使用pool.map，我就会发现加速比因子远远超过4倍，这对我来说是毫无意义的。接下来的细节。

首先，函数定义。

def take_up_time():
    n = 1e3
    while n > 0:
        n -= 1

def count_even_numbers(x):
    take_up_time()
    return np.where(np.mod(x, 2) == 0, 1, 0)

现在定义我们将要测试的函数。

首先，以串行方式运行的函数：

def serial(arr):
    return np.sum(map(count_even_numbers,arr))

现在，以“标准”方式使用Pool.map的函数：

def parallelization_strategy1(arr):
    num_threads = multiprocessing_count()
    pool = multiprocessing.Pool(num_threads)
    result = pool.map(count_even_numbers,arr)
    pool.close()
    return np.sum(result)

最后，第二种策略是手动块数组，然后在块上运行Pool.map (由于python numpy拆分数组为不等子数组而导致的拆分解决方案)。

def split_padded(a,n):
    """ Simple helper function for strategy 2
    """
    padding = (-len(a))%n
    if padding == 0:
        return np.split(a, n)
    else:
        sub_arrays = np.split(np.concatenate((a,np.zeros(padding))),n)
        sub_arrays[-1] = sub_arrays[-1][:-padding]
    return sub_arrays

def parallelization_strategy2(arr):
    num_threads = multiprocessing_count()
    sub_arrays = split_padded(arr, num_threads)
    pool = multiprocessing.Pool(num_threads)
    result = pool.map(count_even_numbers,sub_arrays)
    pool.close()
    return np.sum(np.array(result))

这是我的数组输入：

npts = 1e3
arr = np.arange(npts)

现在，我使用IPython %timeit函数来运行我的计时，对于1e3点，我得到以下内容：

• 串行：10个循环，最佳每循环3: 98.7ms
• parallelization_strategy1：10循环，最佳每环3: 77.7ms
• parallelization_strategy2：10循环，最佳每环3: 22 ms

由于我有4个核心，策略1是一个令人失望的适度加速，而策略2比最大的4x加速要大得令人怀疑。

当我将npts增加到1e4时，结果更加令人费解：

• 串行：1循环，最佳每循环3: 967 ms
• parallelization_strategy1：1循环，最佳每环3: 596 ms
• parallelization_strategy2：10循环，最佳每环3: 22.9ms

因此，造成混乱的两个根源是：

1. 策略2比单纯的理论限制要快得多。
2. 由于某些原因，使用npts=1e4的%timeit只触发串行和策略1的1循环，而对于策略2则触发10循环。

Answer 1

你的策略不一样！

在第一种策略中，Pool.map迭代数组，因此对每个数组项都调用count_even_numbers (因为数组的形状是一维的)。

第二个策略映射一个数组列表，因此对列表中的每个数组都调用count_even_numbers。

Answer 2

结果证明，您的示例完全符合毕斯兰模型。编译以下源代码count_even.py

#pythran export count_even(int [:])
import numpy as np

def count_even_numbers(x):
    return np.where(np.mod(x, 2) == 0, 1, 0)

def count_even(arr):
    s = 0
    #omp parallel for reduction(+:s)
    for elem in arr:
        s += count_even_numbers(elem)
    return s

使用命令行(-fopenmp激活对OpenMP注释的处理)：

pythran count_even.py -fopenmp

在此基础上运行timeit已经产生了大量的加速，这要归功于对本机代码的转换：

无Pythran

$ python -m timeit -s 'import numpy as np; arr = np.arange(1e7, dtype=int); from count_even import count_even' 'count_even(arr)'
verryyy long, more than several minutes :-/

带Pythran的，一个核心

$ OMP_NUM_THREADS=1 python -m timeit -s 'import numpy as np; arr = np.arange(1e7, dtype=int); from count_even import count_even' 'count_even(arr)'
100 loops, best of 3: 10.3 msec per loop

与Pythran，双核

$ OMP_NUM_THREADS=2 python -m timeit -s 'import numpy as np; arr = np.arange(1e7, dtype=int); from count_even import count_even' 'count_even(arr)'
100 loops, best of 3: 5.5 msec per loop

并行化的速度是原来的两倍:-)

请注意，OpenMP支持多线程，而不是多处理.