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

我想在共享内存中使用一个 numpy 数组，以便与多处理模块一起使用。困难之处在于，它不仅仅是一个 ctype 数组，而是像一个 numpy 数组那样使用它。

from multiprocessing import Process, Array
import scipy


def f(a):
a[0] = -a[0]


if __name__ == '__main__':
# Create the array
N = int(10)
unshared_arr = scipy.rand(N)
arr = Array('d', unshared_arr)
print "Originally, the first two elements of arr = %s"%(arr[:2])


# Create, start, and finish the child processes
p = Process(target=f, args=(arr,))
p.start()
p.join()


# Printing out the changed values
print "Now, the first two elements of arr = %s"%arr[:2]

这样产生的产出如下:

Originally, the first two elements of arr = [0.3518653236697369, 0.517794725524976]
Now, the first two elements of arr = [-0.3518653236697369, 0.517794725524976]

数组可以以 ctype 的方式访问，例如 arr[i]是有意义的。但是，它不是一个数字数组，并且我不能执行诸如 -1*arr或 arr.sum()之类的操作。我认为解决方案是将 ctype 数组转换为 numpy 数组。然而(除了不能使这个工作) ，我不相信它会被共享了。

似乎有一个标准的解决方案，什么必须是一个共同的问题。

95926 次浏览
小开

Array对象有一个与之关联的 get_obj()方法，该方法返回表示缓冲区接口的 ctype 数组。我觉得下面这些应该可以。

from multiprocessing import Process, Array
import scipy
import numpy


def f(a):
a[0] = -a[0]


if __name__ == '__main__':
# Create the array
N = int(10)
unshared_arr = scipy.rand(N)
a = Array('d', unshared_arr)
print "Originally, the first two elements of arr = %s"%(a[:2])


# Create, start, and finish the child process
p = Process(target=f, args=(a,))
p.start()
p.join()


# Print out the changed values
print "Now, the first two elements of arr = %s"%a[:2]


b = numpy.frombuffer(a.get_obj())


b[0] = 10.0
print a[0]

在运行时，这将打印出 a的第一个元素现在是10.0，显示 a和 b只是同一内存中的两个视图。

为了确保它仍然是多处理器安全的，我相信你将不得不使用 acquire和 release方法存在于 Array对象，a，和它的内置锁，以确保它的所有安全访问(虽然我不是一个多处理器模块的专家)。

小开

添加到@unutbu’s (已经不可用了)和@Henry Gomersall 的回答中，你可以在需要的时候使用 shared_arr.get_lock()来同步访问:

shared_arr = mp.Array(ctypes.c_double, N)
# ...
def f(i): # could be anything numpy accepts as an index such another numpy array
with shared_arr.get_lock(): # synchronize access
arr = np.frombuffer(shared_arr.get_obj()) # no data copying
arr[i] = -arr[i]

例子

import ctypes
import logging
import multiprocessing as mp


from contextlib import closing


import numpy as np


info = mp.get_logger().info


def main():
logger = mp.log_to_stderr()
logger.setLevel(logging.INFO)


# create shared array
N, M = 100, 11
shared_arr = mp.Array(ctypes.c_double, N)
arr = tonumpyarray(shared_arr)


# fill with random values
arr[:] = np.random.uniform(size=N)
arr_orig = arr.copy()


# write to arr from different processes
with closing(mp.Pool(initializer=init, initargs=(shared_arr,))) as p:
# many processes access the same slice
stop_f = N // 10
p.map_async(f, [slice(stop_f)]*M)


# many processes access different slices of the same array
assert M % 2 # odd
step = N // 10
p.map_async(g, [slice(i, i + step) for i in range(stop_f, N, step)])
p.join()
assert np.allclose(((-1)**M)*tonumpyarray(shared_arr), arr_orig)


def init(shared_arr_):
global shared_arr
shared_arr = shared_arr_ # must be inherited, not passed as an argument


def tonumpyarray(mp_arr):
return np.frombuffer(mp_arr.get_obj())


def f(i):
"""synchronized."""
with shared_arr.get_lock(): # synchronize access
g(i)


def g(i):
"""no synchronization."""
info("start %s" % (i,))
arr = tonumpyarray(shared_arr)
arr[i] = -1 * arr[i]
info("end   %s" % (i,))


if __name__ == '__main__':
mp.freeze_support()
main()

如果您不需要同步访问或创建自己的锁，那么 mp.Array()是不必要的。在这种情况下，您可以使用 mp.sharedctypes.RawArray。

小开

您可以使用 sharedmem模块: https://bitbucket.org/cleemesser/numpy-sharedmem

下面是您的原始代码，这次使用的是行为类似 NumPy 数组的共享内存(请注意最后附加的调用 NumPy sum()函数的语句) :

from multiprocessing import Process
import sharedmem
import scipy


def f(a):
a[0] = -a[0]


if __name__ == '__main__':
# Create the array
N = int(10)
unshared_arr = scipy.rand(N)
arr = sharedmem.empty(N)
arr[:] = unshared_arr.copy()
print "Originally, the first two elements of arr = %s"%(arr[:2])


# Create, start, and finish the child process
p = Process(target=f, args=(arr,))
p.start()
p.join()


# Print out the changed values
print "Now, the first two elements of arr = %s"%arr[:2]


# Perform some NumPy operation
print arr.sum()
小开

我编写了一个小的 python 模块，它使用 POSIX 共享内存在 python 解释器之间共享 numpy 数组。也许你会发现它很方便。

Https://pypi.python.org/pypi/sharedarray

它是这样运作的:

import numpy as np
import SharedArray as sa


# Create an array in shared memory
a = sa.create("test1", 10)


# Attach it as a different array. This can be done from another
# python interpreter as long as it runs on the same computer.
b = sa.attach("test1")


# See how they are actually sharing the same memory block
a[0] = 42
print(b[0])


# Destroying a does not affect b.
del a
print(b[0])


# See how "test1" is still present in shared memory even though we
# destroyed the array a.
sa.list()


# Now destroy the array "test1" from memory.
sa.delete("test1")


# The array b is not affected, but once you destroy it then the
# data are lost.
print(b[0])
小开

虽然已经给出的答案是好的，但是只要满足两个条件，这个问题就有一个更容易的解决办法:

  1. 您使用的是 POSIX 兼容操作系统(例如 Linux、 Mac OSX) ; 以及
  2. 您的子进程需要对共享数组使用 只读访问。

在这种情况下，您不需要显式地使变量共享，因为子进程将使用 fork 创建。分叉子节点自动共享父节点的内存空间。在 Python 多处理的上下文中，这意味着它共享所有 模组级别变量; 注意，这个 不成立用于显式传递给子进程或者调用 multiprocessing.Pool左右的函数的参数。

举个简单的例子:

import multiprocessing
import numpy as np


# will hold the (implicitly mem-shared) data
data_array = None


# child worker function
def job_handler(num):
# built-in id() returns unique memory ID of a variable
return id(data_array), np.sum(data_array)


def launch_jobs(data, num_jobs=5, num_worker=4):
global data_array
data_array = data


pool = multiprocessing.Pool(num_worker)
return pool.map(job_handler, range(num_jobs))


# create some random data and execute the child jobs
mem_ids, sumvals = zip(*launch_jobs(np.random.rand(10)))


# this will print 'True' on POSIX OS, since the data was shared
print(np.all(np.asarray(mem_ids) == id(data_array)))
小开

有了 Python 3.8 + ，就有了 multiprocessing.shared_memory标准库:

# np_sharing.py
from multiprocessing import Process
from multiprocessing.managers import SharedMemoryManager
from multiprocessing.shared_memory import SharedMemory
from typing import Tuple


import numpy as np




def create_np_array_from_shared_mem(
shared_mem: SharedMemory, shared_data_dtype: np.dtype, shared_data_shape: Tuple[int, ...]
) -> np.ndarray:
arr = np.frombuffer(shared_mem.buf, dtype=shared_data_dtype)
arr = arr.reshape(shared_data_shape)
return arr




def child_process(
shared_mem: SharedMemory, shared_data_dtype: np.dtype, shared_data_shape: Tuple[int, ...]
):
"""Logic to be executed by the child process"""
arr = create_np_array_from_shared_mem(shared_mem, shared_data_dtype, shared_data_shape)
arr[0, 0] = -arr[0, 0]  # modify the array backed by shared memory




def main():
"""Logic to be executed by the parent process"""


# Data to be shared:
data_to_share = np.random.rand(10, 10)


SHARED_DATA_DTYPE = data_to_share.dtype
SHARED_DATA_SHAPE = data_to_share.shape
SHARED_DATA_NBYTES = data_to_share.nbytes


with SharedMemoryManager() as smm:
shared_mem = smm.SharedMemory(size=SHARED_DATA_NBYTES)


arr = create_np_array_from_shared_mem(shared_mem, SHARED_DATA_DTYPE, SHARED_DATA_SHAPE)
arr[:] = data_to_share  # load the data into shared memory


print(f"The [0,0] element of arr is {arr[0,0]}")  # before


# Run child process:
p = Process(target=child_process, args=(shared_mem, SHARED_DATA_DTYPE, SHARED_DATA_SHAPE))
p.start()
p.join()


print(f"The [0,0] element of arr is {arr[0,0]}")  # after


del arr  # delete np array so the shared memory can be deallocated




if __name__ == "__main__":
main()

运行脚本:

$ python3.10 np_sharing.py
The [0,0] element of arr is 0.262091705529628
The [0,0] element of arr is -0.262091705529628

由于不同进程中的数组共享相同的底层内存缓冲区，因此应用标准的竞争条件。


提交答案