我希望能在多个进程之间共享一个numpy数组。这些进程只需要读取数据,因此我希望避免复制数据。如果我从multiprocessing.sharedctypes.RawArray
开始,并使用numpy.frombuffer
创建numpy数组,我知道如何做。但是如果我最初收到的就是一个numpy数组呢?有没有一种方法可以使用numpy数组的数据初始化RawArray而不复制数据?或者是否有其他方法可以在进程之间共享数据而不复制它?
我希望能在多个进程之间共享一个numpy数组。这些进程只需要读取数据,因此我希望避免复制数据。如果我从multiprocessing.sharedctypes.RawArray
开始,并使用numpy.frombuffer
创建numpy数组,我知道如何做。但是如果我最初收到的就是一个numpy数组呢?有没有一种方法可以使用numpy数组的数据初始化RawArray而不复制数据?或者是否有其他方法可以在进程之间共享数据而不复制它?
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。 - Eric我还有一些你的要求: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 是三维的。至于实际共享,我使用了以下样式,目前它可以正常工作。
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数组... - christianbrodbeckshrarr_ptr = mynparray
这行是有问题的。我认为应该改成 shrarr_ptr[:] = mynparray[:]
,这样才能把数据复制到新的共享内存中。 - coderforlifeRawArray
的ndarray,并将其用作创建原始ndarray的函数的out
参数(如果可能)。 - coderforlifeprint(id(mynparray)
和print(id(shrarr_base_ptr)
进行检查。 - Markus DutschkeRawArray
,并使用以下内容: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
失败的情况下正常工作。
我不确定这是否会在内部复制数据,但您可以传递平坦的数组:
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]
prange
会产生一些开销,如果您创建静态线程,则可以将其最小化。您是正确的... 在这个例子中,prange
已经在最外层循环中了。 - Saullo G. P. Castromultiprocessing.dummy
?Numpy在释放GIL(CPython的全局解释器锁)方面做得非常好,而Cython也具有nogil
功能。 - user2379410