在多处理过程之间共享大型只读Numpy数组

Question

我有一个60GB的SciPy数组(矩阵)我必须在5个以上的multiprocessing Process对象之间共享.我已经看过numpy-sharedmem并在SciPy列表上阅读了这个讨论.似乎有是两个approaches-- numpy-sharedmem和使用multiprocessing.RawArray(),并映射NumPy的dtypes到ctype秒.现在,numpy-sharedmem似乎是要走的路,但我还没有看到一个很好的参考例子.我不需要任何类型的锁,因为数组(实际上是矩阵)将是只读的.现在,由于它的大小,我想避免副本.这听起来像是正确的方法是创建唯一的数组作为副本sharedmem数组,然后将它传递给Process对象？几个具体问题:

1. 将sharedmem句柄实际传递给子的最佳方法是Process()什么？我是否需要一个队列来传递一个阵列？管道会更好吗？我可以将它作为参数传递给Process()子类的init(我假设它被腌制)吗？

2. 在上面我讨论过的讨论中,有人提到numpy-sharedmem不是64位安全吗？我肯定使用一些不是32位可寻址的结构.

3. 这种RawArray()方法是否存在权衡？更慢,更笨？

4. 我是否需要numpy-sharedmem方法的任何ctype-to-dtype映射？

5. 有没有人有一些OpenSource代码这样做的例子？我是一个非常亲力实践的人,如果没有任何好的例子,很难让它工作.

如果我可以提供任何其他信息以帮助其他人澄清这一点,请发表评论,我将添加.谢谢!

这需要在Ubuntu Linux和Maybe Mac OS上运行,但可移植性不是一个大问题.

Answer 1

如果您使用的是Linux(或任何符合POSIX的系统),则可以将此数组定义为全局变量.在Linux启动新的子进程时multiprocessing正在使用fork()它.新生成的子进程自动与其父进程共享内存,只要它不更改它(写时复制机制).

因为你说"我不需要任何类型的锁,因为数组(实际上是矩阵)将是只读的"利用这种行为将是一种非常简单但非常有效的方法:所有子进程都将访问读取这个大型numpy数组时,物理内存中的数据相同.

不要用手你的阵列的Process()构造,这将指示multiprocessing到pickle数据到孩子,这将是你的情况非常不充分或者不可能.在Linux上,在fork()子项是使用相同物理内存的父项的精确副本之后,所以您需要做的就是确保可以从target您交给的函数中访问"包含"矩阵的Python变量Process().这通常可以通过"全局"变量实现.

示例代码:

from multiprocessing import Process
from numpy import random


global_array = random.random(10**4)


def child():
    print sum(global_array)


def main():
    processes = [Process(target=child) for _ in xrange(10)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()


if __name__ == "__main__":
    main()

在Windows上 - 不支持fork()- multiprocessing正在使用win32 API调用CreateProcess.它从任何给定的可执行文件创建一个全新的进程.这就是为什么在Windows 上,如果需要在父运行时期间创建的数据,则需要向子进行数据处理.

Answer 2

@Velimir Mlaker给出了一个很好的答案.我想我可以添加一些评论和一个小例子.

(我找不到很多关于sharedmem的文档 - 这些是我自己实验的结果.)

1. 您是否需要在子进程启动时或启动后传递句柄？如果它只是前者,你可以使用target和args参数Process.这可能比使用全局变量更好.
2. 在您链接的讨论页面中,似乎有一段时间以来对sharedmem添加了对64位Linux的支持,因此它可能不是问题.
3. 我不知道这个.
4. 不.请参阅下面的示例.

例

#!/usr/bin/env python
from multiprocessing import Process
import sharedmem
import numpy

def do_work(data, start):
    data[start] = 0;

def split_work(num):
    n = 20
    width  = n/num
    shared = sharedmem.empty(n)
    shared[:] = numpy.random.rand(1, n)[0]
    print "values are %s" % shared

    processes = [Process(target=do_work, args=(shared, i*width)) for i in xrange(num)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

    print "values are %s" % shared
    print "type is %s" % type(shared[0])

if __name__ == '__main__':
    split_work(4)

产量

values are [ 0.81397784  0.59667692  0.10761908  0.6736734   0.46349645  0.98340718
  0.44056863  0.10701816  0.67167752  0.29158274  0.22242552  0.14273156
  0.34912309  0.43812636  0.58484507  0.81697513  0.57758441  0.4284959
  0.7292129   0.06063283]
values are [ 0.          0.59667692  0.10761908  0.6736734   0.46349645  0.
  0.44056863  0.10701816  0.67167752  0.29158274  0.          0.14273156
  0.34912309  0.43812636  0.58484507  0.          0.57758441  0.4284959
  0.7292129   0.06063283]
type is <type 'numpy.float64'>

这个相关问题可能有用.

Answer 3

您可能对我写的一小段代码感兴趣:github.com/vmlaker/benchmark-sharedmem

唯一感兴趣的文件是main.py.它是numpy-sharedmem的基准- 代码只是通过Pipe 将数组(numpy或者sharedmem)传递给生成的进程.工人只需要调用sum()数据.我只对比较两种实现之间的数据通信时间感兴趣.

我还写了另一个更复杂的代码:github.com/vmlaker/sherlock.

在这里,我使用numpy-sharedmem模块与OpenCV进行实时图像处理 - 图像是NumPy数组,根据OpenCV的新cv2API.图像(实际上是它们的引用)通过从multiprocessing.Manager(从使用Queue或Pipe)创建的字典对象在进程之间共享.与使用普通NumPy数组相比,我获得了很好的性能提升.

管道与队列:

根据我的经验,IPC with Pipe比Queue更快.这是有道理的,因为Queue增加了锁定,使其对多个生产者/消费者来说是安全的.管没有.但是如果你只有两个进程来回交谈,那么使用Pipe是安全的,或者,正如文档所读:

......同时使用管道不同端的进程不存在腐败风险.

sharedmem安全:

sharedmem模块的主要问题是在程序退出时出现内存泄漏的可能性.这在这里进行了长时间的讨论.虽然在2011年4月10日,Sturla提到了内存泄漏的修复,但自那时以来我仍然经历过泄漏,使用了Repos,Sturla Molden自己在GitHub(github.com/sturlamolden/sharedmem-numpy)和Chris Lee-Messer的Bitbucket(bitbucket.org/cleemesser/numpy-sharedmem).

Answer 4

如果您的阵列很大,您可以使用numpy.memmap.例如,如果你有一个存储在磁盘中的数组,比如说'test.array',你可以使用同步进程来访问它中的数据,即使是在"写入"模式,但你的情况更简单,因为你只需要"阅读"模式.

创建数组:

a = np.memmap('test.array', dtype='float32', mode='w+', shape=(100000,1000))

然后,您可以像使用普通数组一样填充此数组.例如:

a[:10,:100]=1.
a[10:,100:]=2.

删除变量时,数据将存储到磁盘中a.

稍后,您可以使用多个进程来访问以下数据test.array:

# read-only mode
b = np.memmap('test.array', dtype='float32', mode='r', shape=(100000,1000))

# read and writing mode
c = np.memmap('test.array', dtype='float32', mode='r+', shape=(100000,1000))

相关答案:

• 使用python和numpy中的大数据,没有足够的ram,如何在光盘上保存部分结果？

• 是否可以使用python将磁盘上的不连续数据映射到数组？