2

I'm trying to parallelize operations on two dimensional array using joblib library in python. Here is the code I have 

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)

The above code takes more time than the basic nested for loop below. 

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)

Why are parallel operations taking more time? How can the efficiency of the above code be improved?

4

1 回答 1

4

哇！绝对喜欢你的代码。它就像一种魅力，将总效率提高了 400 倍。我将尝试阅读有关 numba 和 jit 编译器的更多信息，但您能否简要说明一下为什么它如此高效。再次感谢所有的帮助！–拉姆 2018年1 月 3 日 20:30

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

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

因为提议的步骤joblib创建了那么多完整的流程副本——以便摆脱 GIL步骤的纯[SERIAL]跳舞（一个接一个）但是（！）这包括所有内存传输的附加成本（非常所有变量和整个python解释器及其内部状态的昂贵/敏感numpy），在它开始对你的“有效载荷”计算策略的“有用”工作进行第一步之前，
所以
所有的总和这些实例化开销很容易变得比与开销无关的反比例1 / N因子预期更大，
您在其中设置N ~ num_cores.

有关详细信息，请在此处阅读阿姆达尔定律重新公式化尾部的数学公式。

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

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

这两个“隐藏”成本都是您的主要设计敌人，因为它们会线性添加到[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-orchestrated2**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

在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
于 2018-01-02T22:49:07.960 回答