numpy vs.多处理和mmap

Question

我正在使用Python的multiprocessing模块并行处理大型numpy数组。阵列numpy.load(mmap_mode='r')在主进程中使用内存映射。之后，multiprocessing.Pool()分叉该过程（我想）。

一切似乎都正常，除了我得到类似以下内容的行：

AttributeError("'NoneType' object has no attribute 'tell'",)
  in `<bound method memmap.__del__ of
       memmap([ 0.57735026,  0.57735026,  0.57735026,  0.        ,  0.        ,        0.        ,  0.        ,  0.        ,  0.        ,  0.        ,        0.        ,  0.        ], dtype=float32)>`
     ignored

在单元测试日志中。尽管如此，测试仍然可以通过。

知道那里发生了什么吗？

使用Python 2.7.2，OS X，NumPy 1.6.1。

更新：

经过一些调试后，我将原因找出到使用此内存映射的numpy数组（的一小部分）作为Pool.imap调用输入的代码路径。

显然，“问题”是multiprocessing.Pool.imap将其输入传递给新流程的方式：它使用泡菜。这不适用于mmaped
numpy数组，并且内部的中断会导致错误。

我发现罗伯特·科恩（Robert Kern）的答复似乎解决了同一问题。他建议为imap输入何时来自内存映射的数组创建一个特殊的代码路径：在生成的过程中手动对同一数组进行内存映射。

这将是如此复杂和丑陋，以至于我宁愿忍受该错误和额外的内存副本。还有其他方法可以更轻松地修改现有代码吗？

Answer 1

我通常的方法（如果可以使用额外的内存副本）是在一个进程中完成所有IO，然后将其发送到工作线程池中。要将内存映射数组的切片加载到内存中，只需这样做x = np.array(data[yourslice])（data[yourslice].copy()实际上并没有这样做，这可能会引起一些混乱。）。

首先，让我们生成一些测试数据：

import numpy as np
np.random.random(10000).tofile('data.dat')

您可以使用以下方式重现错误：

import numpy as np
import multiprocessing

def main():
    data = np.memmap('data.dat', dtype=np.float, mode='r')
    pool = multiprocessing.Pool()
    results = pool.imap(calculation, chunks(data))
    results = np.fromiter(results, dtype=np.float)

def chunks(data, chunksize=100):
    """Overly-simple chunker..."""
    intervals = range(0, data.size, chunksize) + [None]
    for start, stop in zip(intervals[:-1], intervals[1:]):
        yield data[start:stop]

def calculation(chunk):
    """Dummy calculation."""
    return chunk.mean() - chunk.std()

if __name__ == '__main__':
    main()

而且，如果只是改用让步np.array(data[start:stop])，则可以解决问题：

import numpy as np
import multiprocessing

def main():
    data = np.memmap('data.dat', dtype=np.float, mode='r')
    pool = multiprocessing.Pool()
    results = pool.imap(calculation, chunks(data))
    results = np.fromiter(results, dtype=np.float)

def chunks(data, chunksize=100):
    """Overly-simple chunker..."""
    intervals = range(0, data.size, chunksize) + [None]
    for start, stop in zip(intervals[:-1], intervals[1:]):
        yield np.array(data[start:stop])

def calculation(chunk):
    """Dummy calculation."""
    return chunk.mean() - chunk.std()

if __name__ == '__main__':
    main()

当然，这确实为每个块制作了一个额外的内存副本。

从长远来看，您可能会发现从映射文件切换到HDF之类的操作会更容易。如果您的数据是多维的，则尤其如此。（我建议h5py，但是pyTables如果您的数据是“类似表格的”，那就很好了。）

祝你好运，无论如何！