python - 以大型 numpy 数组作为参数的多处理

Question

我希望使用多处理，其中一个参数是一个非常大的 numpy 数组。我研究了其他一些似乎有类似问题的帖子

Large numpy arrays in shared memory for multiprocessing: Is sth wrong with this approach?

Share Large, Read-Only Numpy Array Between Multiprocessing Processes

但作为 python 的新手，我一直无法将解决方案调整到此模板和此表单中。我想知道我是否可以寻求您的帮助以了解我的选择是什么，以便以只读方式将 X 传递给函数。我的简化代码片段在这里:

import multiprocessing as mp
import numpy as np

def funcA(X):
    # do something with X
    print 'funcA OK'

def funcB(X):
    # do something else with X
    print 'funcB OK'

if __name__=='__main__':
    X=np.random.rand(int(5.00e8),)
    funcA(X) # OK
    funcB(X) # OK
    X=np.random.rand(int(2.65e8),)
    P=[]
    P.append(mp.Process(target=funcA,args=(X,))) # OK
    P.append(mp.Process(target=funcB,args=(X,))) # OK
    for p in P:
        p.start()

    for p in P:
        p.join()

    X=np.random.rand(int(2.70e8),)
    P=[]
    P.append(mp.Process(target=funcA,args=(X,))) # FAIL
    P.append(mp.Process(target=funcB,args=(X,))) # FAIL
    for p in P:
        p.start()

    for p in P:
        p.join()

funcA 和 funcB 在顺序调用时似乎可以接受非常大的 numpy 数组。但是，如果将它们作为多进程调用，那么可以传递给函数的 numpy 数组的大小似乎有上限。我怎样才能最好地解决这个问题？

注意:

0) 我不想修改 X；只能从中读取；

1) 我在 64 位 Windows 7 Professional 上运行

Answer 1

问题可能出在向子进程的数据传输中。当必须使用只读对象时，我更喜欢利用 copy-on-write底层操作系统用来管理子进程内存的机制。不过不知道windows 7有没有用这个机制。当写时复制可用时，您可以访问父进程的区域而无需将它们复制到子进程中。仅当您以只读方式访问它们并且对象是在创建进程之前创建的时，此技巧才有效。

总而言之，一个可能的解决方案(至少对于 linux 机器而言)是这样的:

import multiprocessing as mp
import numpy as np

def funcA():
    print "A {}".format(X.shape)
    # do something with the global variable X
    print 'funcA OK'

def funcB():
    print "B {}".format(X.shape)
    # do something else with the global variable X
    print 'funcB OK'

if __name__=='__main__':
    X=np.random.rand(int(5.00e8),)
    funcA() # OK
    funcB() # OK

    X=np.random.rand(int(2.65e8),)
    P=[mp.Process(target=funcA), mp.Process(target=funcB)]
    for p in P:
        p.start()

    for p in P:
        p.join()

    X=np.random.rand(int(2.70e8),)
    P=[mp.Process(target=funcA), mp.Process(target=funcB)]
    for p in P:
        p.start()

    for p in P:
        p.join()

---

更新:在对与 Windows 的兼容性问题发表各种评论后，我草拟了一个独特的基于 native 内存映射的新解决方案。在这个解决方案中，我在文件上创建了一个 numpy 内存映射，它通过文件描述符共享，因此不需要将整个数组复制到子数组中。我发现这个解决方案比使用 multiprocessing.Array 快得多!

更新 2:以下代码已更新，以避免在内存映射随机化期间出现内存问题。

import multiprocessing as mp
import numpy as np
import tempfile

def funcA(X):
    print "A {}".format(X.shape)
    # do something with X
    print 'funcA OK'

def funcB(X):
    print "B {}".format(X.shape)
    # do something else with X
    print 'funcB OK'

if __name__=='__main__':
    dim = int(2.75e8)
    with tempfile.NamedTemporaryFile(dir='/tmp', delete=False) as tmpfile:
        X = np.memmap(tmpfile, shape=dim, dtype=np.float32)  # create the memory map
        # init the map by chunks of size 1e8
        max_chunk_size = int(1e8)
        for start_pos in range(0, dim, max_chunk_size):
            chunk_size = min(dim-start_pos, max_chunk_size)
            X[start_pos:start_pos+chunk_size] = np.random.rand(chunk_size,)
        P=[mp.Process(target=funcA, args=(X,)), mp.Process(target=funcB, args=(X,))]
        for p in P:
            p.start()

        for p in P:
            p.join()