使用 numpy 数组和共享内存并行化 python 循环

Posted 2023-02-14

【中文标题】使用 numpy 数组和共享内存并行化 python 循环

【英文标题】：Parallelise python loop with numpy arrays and shared-memory

【发布时间】：2012-10-15 15:31:24

【问题描述】：

我知道关于这个主题的几个问题和答案，但还没有找到对这个特定问题的满意答案：

在通过 numpy/scipy 函数操作 numpy 数组的 python 循环中进行简单的共享内存并行化的最简单方法是什么？

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

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

Python 的所有简单性真的是并行化循环的最佳方式吗？真的吗？这对于以 OpenMP 方式进行并行化来说是微不足道的。

我煞费苦心地阅读了多处理模块的不透明文档，却发现它是如此通用，以至于它似乎适用于除了简单的循环并行化之外的所有东西。我对设置管理器、代理、管道等不感兴趣。我只有一个简单的循环，完全并行，任务之间没有任何通信。使用 MPI 并行化这样一个简单的情况似乎有点过头了，更不用说在这种情况下内存效率低了。

我还没有时间了解用于 Python 的大量不同的共享内存并行包，但想知道是否有人在这方面有更多经验并且可以告诉我一个更简单的方法。请不要建议使用 Cython 之类的串行优化技术（我已经使用它），或使用 BLAS 之类的并行 numpy/scipy 函数（我的情况更通用，更并行）。

【问题讨论】：

相关：OpenMP and Python。请参阅我的答案中的示例。

在 Linux 上，您链接到的答案中的代码可以在交互式提示中正常工作。此外，Cython 确实支持基于 openmp 的并行化，而且使用起来非常简单（在循环中将 range 替换为 prange）：docs.cython.org/src/userguide/parallelism.html

@pv，感谢您的链接。它看起来很简单。但我假设 prange 只能用于 C 函数？这带来了其他问题，例如在 Cython 内部使用 numpy/scipy 数组函数。我不认为有一个简单的接口可以在 Cython 中使用这些函数的 C 等价物？

OpenMP 通常用于紧密循环的细粒度并行性。您在 python 中找不到任何等价物的原因是因为 python 没有为紧密循环提供良好的性能。如果您不需要紧密循环，请使用多处理模块。如果你这样做，那么按照建议使用 cython。

@tiago：您可以将 prange 循环包装在 with nogil: 中以使用任何 Python 结构。一些 Numpy 函数在操作过程中会释放 GIL，因此您可能会获得一些并行性。但是，对 Python 对象的访问始终是序列化的，因此线程不可避免地会部分同步。这与 Python 在单个进程中的并行性一样好——您需要使用多处理来获得更多。

【参考方案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

在 2 核机器上：

$ 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 一样）发布了 GIL。

如果您想以交互方式使用它：

# 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，所以 YMMV 对于性能来说。

***

从其他 *** 答案中提取的另一个可能有用的链接 --- 多处理的另一个接口：http://packages.python.org/joblib/parallel.html

【讨论】：

谢谢，这看起来很棒。我将尝试一些代码。刚刚发现在 MacPorts 中将 OpenMP 与 Python 一起使用并不简单，因为它默认使用 clang。但是手动使用 gcc 我可以使您的示例正常工作。

您好 pv.，一个简单的问题 - 这也适用于 Windows 吗？因为我不知道在哪里为 Windows 设置 OMP_NUM_THREADS...有什么链接可以让我开始吗？

【参考方案2】：

使用映射操作（在本例中为 multiprocessing.Pool.map()）或多或少是在单台机器上并行化循环的规范方法。除非并且直到内置的 map() 被并行化。

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

你可以使用openmp with python（或者更确切地说是cython），但它看起来并不容易。

IIRC，如果只从__main__ 运行多处理的东西是必要的，因为它与 Windows 兼容。由于windows缺少fork()，它启动了一个新的python解释器，并且必须在其中导入代码。

编辑

当使用良好的多线程 BLAS 库（例如OpenBLAS。 （另见this question。）

由于 numpy 数组操作主要是按元素进行的，因此可能将它们并行化。但这将涉及为 python 对象设置共享内存段，或者将数组分成几部分并将它们提供给不同的进程，这与multiprocessing.Pool 所做的没有什么不同。无论采用何种方法，管理所有这些都会产生内存和处理开销。必须进行广泛的测试，以查看对于哪些大小的数组，这实际上值得付出努力。这些测试的结果可能会因硬件架构、操作系统和 RAM 量而有很大差异。

【讨论】：

感谢您提供带有 Cython 的 OpenMP 链接，我对此一无所知。可悲的是，这似乎不是我想要的答案。我看过你在 scipy.org 上提到的页面，还有this one。但似乎列出的大多数选项都需要对现有代码进行复杂的重写。我只是在寻找一种简单的方法来并行化数组上的 numpy/scipy 操作。

修复了 scipy.org 链接。 euroscipy 链接说“暂时不可用”，所以它应该会回来。

【参考方案3】：

ParallelRegression 中的 mathDict( ) 类的 .map( ) 方法完全符合您在两行代码中寻找的功能，在交互式提示下应该很容易。它使用真正的多处理，因此并行运行的函数是可腌制的要求是不可避免的，但这确实提供了一种简单的方法来循环来自多个进程的共享内存中的矩阵。

假设你有一个可腌制的功能：

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]

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