python - Python中二维数组操作的有效并行化

Question

我正在尝试使用python中的joblib库并行处理二维数组上的操作。这是我的代码

from joblib import Parallel, delayed
import multiprocessing
import numpy as np

# The code below just aggregates the base_array to form a new two dimensional array
base_array = np.ones((2**12, 2**12), dtype=np.uint8)
def compute_average(i, j):
    return np.uint8(np.mean(base_array[i*4: (i+1)*4, j*4: (j+1)*4]))

num_cores = multiprocessing.cpu_count()
new_array = np.array(Parallel(n_jobs=num_cores)(delayed(compute_average)(i, j) 
                                        for i in xrange(0,1024) for j in xrange(0,1024)), dtype=np.uint8)

上面的代码比下面的基本嵌套for循环花费更多的时间。

new_array_nested = np.ones((2**10, 2**10), dtype=np.uint8)
for i in xrange(0,1024):
    for j in xrange(0,1024):
         new_array_nested[i,j] = compute_average(i,j)

为什么并行操作需要更多时间？如何提高上述代码的效率？

Answer 1

我们可以很容易地到达77 [ms]下的某个地方，但是需要掌握一些步骤才能到达那里，所以让我们开始：



问：为什么并行操作需要更多时间？

因为建议使用joblib的步骤创建了许多完整的过程副本-以便逃避GIL步进的pure-[SERIAL]跳舞（一个接一个），但是（！）包括所有附加成本。在开始对“有效载荷”计算进行“ usefull”工作的第一步之前，对所有变量以及整个python解释器及其内部状态进行内存传输（对于确实很大的numpy数组而言非常昂贵/敏感）战略，
所以
所有这些实例化开销的总和很容易变得大于与反比例1 / N因子的开销无关的期望，
设置N ~ num_cores的位置。

For details, read the mathematical formulation in the tail part of Amdahl's Law re-formulation here.



问：可以帮助提高上述代码的效率吗？

尽可能节省所有间接费用：
- 在可能的情况：
 -在进程生成侧，尝试使用n_jobs = ( num_cores - 1 )为正在进行的“主”进程留出更多空间，并在性能提高时进行基准测试
-在流程终止端，避免从返回值中收集和构造一个新的（可能是较大的）对象，而是预先分配一个足够大的本地流程数据结构，并返回一些有效的，易于序列化的数据以及各部分返回结果的对齐方式的无阻塞合并。

这两个“隐藏”成本都是您的主要设计敌人，因为它们会线性地添加到整个问题解决方案（ref.: the effects of both of these in the overhead-strict Amdahl's Law formula）的计算路径的pure- [SERIAL]部分中。



实验与结果：

>>> from zmq import Stopwatch; aClk = Stopwatch()
>>> base_array = np.ones( (2**12, 2**12), dtype = np.uint8 )
>>> base_array.flags
  C_CONTIGUOUS : True
  F_CONTIGUOUS : False
  OWNDATA      : True
  WRITEABLE    : True
  ALIGNED      : True
  UPDATEIFCOPY : False
>>> def compute_average_per_TILE(               TILE_i,   TILE_j ): // NAIVE MODE
...     return np.uint8( np.mean( base_array[ 4*TILE_i:4*(TILE_i+1),
...                                           4*TILE_j:4*(TILE_j+1)
...                                           ]
...                               )
...                      )
... 
>>> aClk.start(); _ = compute_average_per_TILE( 12,13 ); aClk.stop()
25110
  102
  109
   93

每一张照片大约需要 93 [us]。期望大约 1024*1024*93 ~ 97,517,568 [us]覆盖整个 base_array的均值处理。

通过实验，在这里可以很好地看到处理不当的开销所产生的影响，朴素的嵌套实验进行了以下操作：

>>> aClk.start(); _ = [ compute_average_per_TILE( i, j )
                                              for i    in xrange(1024)
                                              for    j in xrange(1024)
                        ]; aClk.stop()
26310594
^^...... 
26310594 / 1024. / 1024. == 25.09 [us/cell]

在终端分配时，开销减少了3.7倍（由于未发生“尾部”部分（分配单个返回值）），开销减少了2 ** 20倍，但只有一次。

然而，更多的惊喜来了。



这里有什么合适的工具？

从来没有一个普遍的规则，没有一个万能的。

给定
每次调用将不止一个4x4矩阵图块（实际上，每个建议的 25 [us]协调生成的 joblib调用产生的实际上少于 2**20，由原始实例分布在〜 .cpu_count()个完全实例化的过程中）提案

...( joblib.Parallel( n_jobs = num_cores )( 
     joblib.delayed( compute_average )( i, j ) 
                                    for i    in xrange( 1024 )
                                    for    j in xrange( 1024 )
     )

确实有提高性能的空间。

对于这些小型矩阵（并不是所有问题在这个意义上都这么高兴），人们可以从更智能的内存访问模式以及减少python GIL起源的弱点中获得最佳结果。

由于每次通话时间跨度仅为4x4微型计算，因此更好的方法是利用智能矢量化（所有数据都适合缓存，因此缓存内计算是寻求最大性能的度假之旅）

最好的（仍然是非常幼稚的矢量化代码）
能够从 ~ 25 [us/cell]减小到 ~ 74 [ns/cell]（还有空间可以更好地进行对齐处理，因为它花费了 ~ 4.6 [ns] / base_array单元格处理），因此如果在-cache优化的矢量化代码将正确制作。

在 77 [ms]中？值得做对，不是吗？

不是97秒
不是25秒
但只需几个按键，就少于 77 [ms]了，如果更好地优化呼叫签名，本来可以压缩更多的东西：

>>> import numba
>>> @numba.jit( nogil = True, nopython = True )
... def jit_avg2( base_IN, ret_OUT ):  // all pre-allocated memory for these data-structures
...     for i in np.arange( 1024 ):    // vectorised-code ready numpy iterator
...         for j in np.arange( 1024 ):// vectorised-code ready numpy iterator 
...             ret_OUT[i,j] = np.uint8( np.mean( base_IN[4*i:4*(i+1),
...                                                       4*j:4*(j+1)
...                                                       ]
...                                               )
...                                      )
...     return                         // avoid terminal assignment costs
... 

>>> aClk.start(); _ = jit_avg2( base_array, mean_array ); aClk.stop()
1586182 (even with all the jit-compilation circus, it was FASTER than GIL-stepped nested fors ...)
  76935
  77337