来自numpy数组的RawArray？答案

【问题标题】：RawArray from numpy array?来自numpy数组的RawArray？
【发布时间】：2014-12-05 19:15:33
【问题描述】：

我想在多个进程之间共享一个 numpy 数组。进程只读取数据，所以我想避免复制。如果我可以从multiprocessing.sharedctypes.RawArray 开始，然后使用numpy.frombuffer 创建一个numpy 数组，我知道该怎么做。但是如果我最初得到一个 numpy 数组怎么办？有没有办法用 numpy 数组的数据初始化 RawArray 而无需复制数据？还是有其他方法可以在不复制的情况下跨进程共享数据？

【问题讨论】：

我不确定这是否可能 - numpy 数组不会从共享内存中分配。我不认为您可以在不将数据实际复制到共享内存空间的情况下创建sharedctypes.RawArray。 The docs 请注意，您可以只将指向对象的指针存储在共享内存中，但它可能在第二个进程中无效，因为它指向另一个进程的地址空间。
@christianmbrodbeck this answer 解释了如何使用 Cython 和 OpenMP 通过使用共享内存的不同进程在同一个数组上工作
感谢@saullo-castro 的指点。由于在 OS X 上尝试此解决方案似乎存在一些障碍（据我所知，Xcode 不支持 openmp）它是否值得在大型阵列上执行相对简单的操作？不是每个 prange 语句都会产生创建子进程的开销吗？示例中的点积似乎并没有从中受益。（即我假设 prange 语句必须是最佳速度增益的最外层循环？）
@christianmbrodbeck 是的，prange 会产生一些开销，如果您创建 static threads，这些开销就会最小化。你是对的...in this exampleprange已经在最外层循环了
谢谢@saullo-castro，假设我的目标函数是您的示例中的dot()，并且我将多次从Python调用dot()，是否可以避免与相关的开销在每次调用中初始化子进程？或者这基本上需要在 Cython 中实现外循环？

标签： python arrays numpy multiprocessing

【解决方案1】：

我不确定这是否会在内部复制数据，但您可以传递平面数组：

a = numpy.random.randint(1,10,(4,4))
>>> a
array([[5, 6, 7, 7],
       [7, 9, 2, 8],
       [3, 4, 6, 4],
       [3, 1, 2, 2]])

b = RawArray(ctypes.c_long, a.flat)
>>> b[:]
[5, 6, 7, 7, 7, 9, 2, 8, 3, 4, 6, 4, 3, 1, 2, 2]

【讨论】：

不行，无论是从内存使用情况来看还是因为修改b不影响a...
最初未分配为“共享”的内存无法（至少不容易）共享。您需要将数据复制到共享内存块或使用@Jey 的方法使数组最初基于共享内存，因此始终共享。

【解决方案2】：

我也有你的一些要求：a）给定一个大的 numpy 数组，b）需要在一堆进程之间共享它 c）只读等。为此，我一直在使用类似的东西：

mynparray = #initialize a large array from a file
shrarr_base_ptr = RawArray(ctypes.c_double, len*rows*cols)
shrarr_ptr = np.frombuffer(shrarr_base_ptr)
shrarr_ptr = mynparray

在我的例子中，mynparray 是 3-D。至于实际的分享，我用了下面的样式，目前还可以。

    inq1 = Queue()
    inq2 = Queue()  
    outq = Queue()
    p1 = Process(target = myfunc1, args=(inq1, outq,))
    p1.start()
    inq1.put((shrarr_ptr, ))
    p2 = Process(target = myfunc2, args=(inq2, outq,))
    p2.start()
    inq2.put((shrarr_ptr,))
    inq1.close()
    inq2.close()
    inq1.join_thread()
    inq2.join_thread()
    ....

【讨论】：

在shrarr_ptr = mynparray 行中，您将原始numpy 数组分配给名称shrarr_ptr。当您稍后执行 inq1.put((shrarr_ptr,)) 时，您将通过 Queue 发送整个 numpy 数组...
不，RawArray 来自 sharedctypes，所以对象是在共享内存中创建并被继承的。我并没有实际发送整个阵列。此外，根据我的经验，通过队列发送如此大的对象将需要很长时间。这是我上面代码的inspiration。
@christianbrodbeck 是对的，shrarr_ptr = mynparray 行有问题。我认为它需要是shrarr_ptr[:] = mynparray[:]，以便将数据复制到新的共享内存中。
为了解决数据被传播回原始numpy数组的问题，这是不可能的。一种解决方案是更早分配基于RawArray 的ndarray，并将其用作创建原始ndarray 的out 参数（如果可能）。
我同意@christianbrodbeck。这个答案与 np.ndarray 一起使用，并且有两行不必要的代码。这可以通过print(id(mynparray) 和print(id(shrarr_base_ptr) 进行检查。

【解决方案3】：

据我所知，在将内存分配给特定进程后，无法将其声明为共享内存。类似的讨论可以在here和here (more suitable)找到。

让我快速勾勒出您提到的解决方法（从 RawArray 开始并获得 numpy.ndarray 的引用）。

import numpy as np
from multiprocessing.sharedctypes import RawArray
# option 1
raw_arr = RawArray(ctypes.c_int, 12)
# option 2 (set is up, similar to some existing np.ndarray np_arr2)
raw_arr = RawArray(
        np.ctypeslib.as_ctypes_type(np_arr2.dtype), len(np_arr2)
        )
np_arr = np.frombuffer(raw_arr, dtype=np.dtype(raw_arr))
# np_arr: numpy array with shared memory, can be processed by multiprocessing

如果您必须以numpy.ndarray 开头，则您别无选择，只能复制数据

import numpy as np
from multiprocessing.sharedctypes import RawArray

np_arr = np.zeros(shape=(3, 4), dtype=np.ubyte)
# option 1
tmp = np.ctypeslib.as_ctypes(np_arr)
raw_arr = RawArray(tmp._type_, tmp)
# option 2
raw_arr = RawArray(np.ctypeslib.as_ctypes_type(np_arr.dtype), np_arr.flatten())

print(raw_arr[:])

【讨论】：

这里不用np.frombuffer，直接用np.asarray，它也会自动找到合适的dtype。

【解决方案4】：

请注意，如果您打算使用 numpy 数组，可以完全省略 RawArray，并使用：

from multiprocessing.heap import BufferWrapper

def shared_array(shape, dtype):
    dt = np.dtype((dtype, shape))
    wrapper = BufferWrapper(dt.itemsize)
    mem = wrapper.create_memoryview()

    # workaround for bpo-41673 to keep `wrapper` alive
    ct = (ctypes.c_ubyte * dt.itemsize).from_buffer(mem)
    ct._owner = wrapper
    mem = memoryview(ct)

    return np.asarray(mem).view(dt)

这种方法的优点是它适用于 np.ctypeslib.as_ctypes_type 失败的情况。

【讨论】：