使用NumPy数组和共享内存并行化Python循环

Question

我知道这个话题已经有几个问题和答案，但是还没有找到一个令人满意的解决方案：

在使用numpy/scipy函数操纵numpy数组的python循环中，最简单的共享内存并行化方法是什么？

我不是在寻找最有效的方法，只是想要一些简单的实现方式，当循环不并行运行时，不需要进行重大改写。就像OpenMP在低级语言中实现一样。

在这方面，我看到的最好的答案是这个，但这是一个相当笨拙的方法，需要将循环表达为接受单个参数的函数，几行共享数组转换代码，似乎需要从__main__调用并且似乎不能从交互式提示符（我花了很多时间在那里）良好地工作。

在Python的所有简单性中，这真的是并行化循环的最佳方法吗？真的吗？这是一种在OpenMP风格下并行化的微不足道的事情。

我费尽心思地阅读了多进程模块的晦涩文档，只发现它太泛化了，似乎适用于除简单循环并行化以外的所有内容。我不感兴趣设置管理器、代理、管道等，我只有一个简单的循环，完全并行，任务之间没有任何通信。在这种简单情况下，使用MPI并行化似乎过度kill了，更不用说会浪费内存。

我还没有时间学习Python的许多不同共享内存并行包，但想知道是否有更简单的方法。请不要建议串行优化技术，如Cython（我已经使用它），或者使用并行numpy/scipy函数，如BLAS（我的情况更普遍，更并行）。

- tiago

---

相关：OpenMP和Python。请参考我的回答中的示例。 - jfs

在Linux上，你链接的答案中的代码可以从交互式提示符中正常工作。此外，Cython支持基于openmp的并行化，并且非常容易使用（在循环中用prange替换range）：http://docs.cython.org/src/userguide/parallelism.html - pv.

@ pv，谢谢你提供的链接。看起来很简单。但我认为prange只能用于C函数？这带来了其他问题，比如从Cython内部使用numpy/scipy数组函数。我不认为有一个简单的接口可以在Cython内部使用这些函数的C等效版本？ - tiago

OpenMP通常用于紧密循环的细粒度并行处理。你找不到Python中等效的东西是因为Python在紧密循环方面性能表现不佳。如果你不需要紧密循环，那么可以使用multiprocessing模块。如果需要，则建议使用Cython。 - DaveP

@tiago：你可以将prange循环内部包装在“with nogil:”中，以使用任何Python结构。一些Numpy函数在操作期间确实会释放GIL，因此您可能会获得一些并行性。但是，对Python对象的访问始终是串行化的，因此线程不可避免地部分同步。这是Python在单个进程中实现并行性的最佳方式---您需要使用多进程来获得更多。 - pv.

Answer 1

使用Cython并行支持：

# asd.pyx
from cython.parallel cimport prange

import numpy as np

def foo():
    cdef int i, j, n

    x = np.zeros((200, 2000), float)

    n = x.shape[0]
    for i in prange(n, nogil=True):
        with gil:
            for j in range(100):
                x[i,:] = np.cos(x[i,:])

    return x

在一台双核机器上：

$ cython asd.pyx
$ gcc -fPIC -fopenmp -shared -o asd.so asd.c -I/usr/include/python2.7
$ export OMP_NUM_THREADS=1
$ time python -c 'import asd; asd.foo()'
real    0m1.548s
user    0m1.442s
sys 0m0.061s

$ export OMP_NUM_THREADS=2
$ time python -c 'import asd; asd.foo()'
real    0m0.602s
user    0m0.826s
sys 0m0.075s

由于np.cos（像其他ufunc一样）释放了全局解释器锁，因此此代码可以并行运行。

如果您想要交互式使用：

# asd.pyxbdl
def make_ext(modname, pyxfilename):
    from distutils.extension import Extension
    return Extension(name=modname,
                     sources=[pyxfilename],
                     extra_link_args=['-fopenmp'],
                     extra_compile_args=['-fopenmp'])

然后（先删除asd.so和asd.c）：

>>> import pyximport
>>> pyximport.install(reload_support=True)
>>> import asd
>>> q1 = asd.foo()
# Go to an editor and change asd.pyx
>>> reload(asd)
>>> q2 = asd.foo()

所以，在某些情况下，您可以仅使用线程并行化。 OpenMP只是线程的高级封装，因此Cython仅在这里需要更简单的语法。如果没有Cython，您可以使用threading模块 --- 它的工作方式与多进程类似（可能更加稳健），但您无需特别声明数组为共享内存。

然而，并非所有操作都会释放GIL，因此性能因人而异。

***

这是从其他Stackoverflow答案中爬取的另一个可能有用的链接 --- 另一个与多进程交互的接口：http://packages.python.org/joblib/parallel.html

- pv.

---

谢谢，听起来很不错。我会尝试一些代码。刚刚发现在MacPorts中使用OpenMP并不直接，因为它默认使用clang。但是手动使用gcc，我可以让你的示例工作。 - tiago

嗨pv.，一个快速的问题 - 这个在Windows上也能用吗？因为我不知道在Windows上设置OMP_NUM_THREADS的位置...有什么链接可以让我开始吗？ - Yuxiang Wang

Answer 2

使用映射操作(在本例中为multiprocessing.Pool.map())是在单台计算机上并行化循环的规范方式。除非内置的map()被并行化。

不同可能性的概述可以在这里找到。

您可以使用Python和OpenMP(或者更确切地说是Cython)，但看起来并不容易。

如我所记，只从__main__运行多进程任务的原因是因为与Windows兼容性的需要。由于Windows缺少fork()，它会启动一个新的Python解释器，并在其中导入代码。

编辑

Numpy可以像dot()，vdot()和innerproduct()这样的一些操作进行并行化处理，当配备良好的多线程BLAS库（例如OpenBLAS）时。 （另请参见this question。）

由于numpy数组操作大多是按元素进行的，因此似乎可以将它们并行化处理。但是，这将涉及设置Python对象的共享内存段或将数组分成片段并将其馈送到不同的进程中，类似于multiprocessing.Pool所做的事情。无论采取何种方法，都需要处理所有这些的内存和处理开销。必须运行广泛的测试，以查看对于哪些数组大小实际上值得付出努力。这些测试的结果可能会因硬件架构，操作系统和RAM数量而有很大差异。

- Roland Smith

---

谢谢您提供Cython中OpenMP的链接，我之前并不知道。但遗憾的是似乎不是我正在寻找的答案。我已经看到了您在scipy.org上提到的页面，还有另一个页面：http://wiki.python.org/moin/ParallelProcessing。但是，这些选项大多需要对现有代码进行复杂的重写。我只是想寻找一种简单的方法来并行处理数组上的numpy/scipy操作。 - tiago

修复了scipy.org的链接。euroscipy的链接显示“暂时不可用”，所以它应该会恢复。 - Roland Smith

Answer 3

ParallelRegression 中 mathDict( ) 类的 .map( ) 方法恰好可以在两行代码中实现你所需的功能，非常适用于交互式提示。它使用真正的多进程技术，所以要求被并行运行的函数是可 pickle 的，在共享内存中从多个进程循环矩阵提供了一个简单的方式。

假设你有一个可 pickle 的函数：

def sum_row( matrix, row ):
    return( sum( matrix[row,:] ) )

然后，您只需要创建一个表示它的mathDict()对象，并使用mathDict().map()：

matrix = np.array( [i for i in range( 24 )] ).reshape( (6, 4) )

RA, MD = mathDictMaker.fromMatrix( matrix, integer=True )
res = MD.map( [(i,) for i in range( 6 )], sum_row, ordered=True )

print( res )
# [6, 22, 38, 54, 70, 86]

文档（上面的链接）解释了如何将位置参数和关键字参数的组合传递到您的函数中，包括在任何位置或作为关键字参数中的矩阵本身。这应该使您能够使用几乎任何已经编写的函数而无需修改它。

- RichardB