共享内存中用于多处理的大型 numpy 数组:这种方法有问题吗?

问题描述

多处理是一个很棒的工具，但使用大内存块并不是那么直接.您可以在每个进程中加载??块并将结果转储到磁盘上，但有时您需要将结果存储在内存中.最重要的是，使用花哨的 numpy 功能.

Multiprocessing is a wonderful tool but is not so straight forward to use large memory chunks with it. You can load chunks in each process and dump results on disk but sometimes you need to store the results in the memory. And on top, use the fancy numpy functionality.

我已经阅读/谷歌了很多，并想出了一些答案:

I have read/googled a lot and came up with some answers:

在共享内存中使用numpy数组进行多处理

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

Python 多处理全局 numpy 数组

如何如何在 python 子进程之间传递大型 numpy 数组而不保存到磁盘?

等等等等等等.

它们都有缺点:不太主流的库(sharedmem)；全局存储变量；不太容易阅读代码、管道等.

They all have drawbacks: Not-so-mainstream libraries (sharedmem); globally storing variables; not so easy to read code, pipes, etc etc.

我的目标是在我的工作人员中无缝使用 numpy，而不用担心转换和其他东西.

My goal was to seamlessly use numpy in my workers without worrying about conversions and stuff.

经过多次试验，我想出了 this.它适用于我的 ubuntu 16、python 3.6、16GB、8 核机器.与以前的方法相比，我做了很多捷径".没有全局共享状态，没有需要在 worker 内部转换为 numpy 的纯内存指针，作为进程参数传递的大型 numpy 数组等.

After much trials I came up with this. And it works on my ubuntu 16, python 3.6, 16GB, 8 core machine. I did a lot of "shortcuts" compared to previous approaches. No global shared state, no pure memory pointers that need to be converted to numpy inside workers, large numpy arrays passed as process arguments, etc.

Pastebin 链接上面，但我会在这里放几个片段.

Pastebin link above, but I will put few snippets here.

一些进口:

import numpy as np
import multiprocessing as mp
import multiprocessing.sharedctypes
import ctypes

分配一些共享内存并将其包装到一个 numpy 数组中:

Allocate some shared mem and wrap it into an numpy array:

def create_np_shared_array(shape, dtype, ctype)
     . . . . 
    shared_mem_chunck = mp.sharedctypes.RawArray(ctype, size)
    numpy_array_view = np.frombuffer(shared_mem_chunck, dtype).reshape(shape)
    return numpy_array_view

创建共享数组并在其中放入一些东西

Create shared array and put something in it

src = np.random.rand(*SHAPE).astype(np.float32)
src_shared = create_np_shared_array(SHAPE,np.float32,ctypes.c_float)
dst_shared = create_np_shared_array(SHAPE,np.float32,ctypes.c_float)
src_shared[:] = src[:]  # Some numpy ops accept an 'out' array where to store the results

产生进程:

p = mp.Process(target=lengthly_operation,args=(src_shared, dst_shared, k, k + STEP))
p.start()
p.join()

以下是一些结果(完整参考请参见 pastebin 代码):

Here are some results (see pastebin code for full reference):

Serial version: allocate mem 2.3741257190704346 exec: 17.092209577560425 total: 19.46633529663086 Succes: True
Parallel with trivial np: allocate mem 2.4535582065582275 spawn  process: 0.00015354156494140625 exec: 3.4581971168518066 total: 5.911908864974976 Succes: False
Parallel with shared mem np: allocate mem 4.535916328430176 (pure alloc:4.014216661453247 copy: 0.5216996669769287) spawn process: 0.00015664100646972656 exec: 3.6783478260040283 total: 8.214420795440674 Succes: True

我还做了一个 cProfile(为什么在分配共享内存时要多花 2 秒?)并意识到有一些对 tempfile.py、{ 的调用'_io.BufferedWriter' 对象的 'write' 方法}.

I also did a cProfile (why 2 extra seconds when allocating shared mem?) and realized that there are some calls to the tempfile.py, {method 'write' of '_io.BufferedWriter' objects}.

问题

我做错了吗?
(大型)阵列是否来回腌制而我没有获得任何加快速度的东西?请注意，第二次运行(使用常规 np 数组未通过正确性测试)
有没有办法进一步改进时序、代码清晰度等?(针对多处理范例)

备注

我不能使用进程池，因为 mem 必须在 fork 处继承，而不是作为参数发送.

问题描述

推荐答案