使用 Python 多处理共享对象数组

Question

对于这个问题，我参考example in Python docs 讨论“使用SharedMemory 类和NumPy 数组，从两个不同的Python shell 访问相同的numpy.ndarray”。

我想实现的一个主要变化是操作类对象数组而不是整数值，如下所示。

import numpy as np
from multiprocessing import shared_memory    

# a simplistic class example
class A(): 
    def __init__(self, x): 
        self.x = x

# numpy array of class objects 
a = np.array([A(1), A(2), A(3)])       

# create a shared memory instance
shm = shared_memory.SharedMemory(create=True, size=a.nbytes, name='psm_test0')

# numpy array backed by shared memory
b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)                                    

# copy the original data into shared memory
b[:] = a[:]                                  

print(b)                                            

# array([<__main__.Foo object at 0x7fac56cd1190>,
#       <__main__.Foo object at 0x7fac56cd1970>,
#       <__main__.Foo object at 0x7fac56cd19a0>], dtype=object)

现在，在不同的 shell 中，我们附加到共享内存空间并尝试操作数组的内容。

import numpy as np
from multiprocessing import shared_memory

# attach to the existing shared space
existing_shm = shared_memory.SharedMemory(name='psm_test0')

c = np.ndarray((3,), dtype=object, buffer=existing_shm.buf)

即使在我们能够操作c 之前，打印它也会导致分段错误。事实上，我不能指望观察到尚未写入模块的行为，所以我的问题是可以如何处理共享的对象数组？

我目前正在挑选列表，但受保护的读/写会增加相当多的开销。我也尝试过使用Namespace，这很慢，因为不允许索引写入。另一个想法可能是在 ShareableList 中使用共享 Ctypes 结构，但我不知道从哪里开始。

此外还有一个设计方面：似乎shared_memory 中有一个open bug 可能会影响我的实现，其中我有多个进程处理数组的不同元素。

是否有一种更具可扩展性的方式在多个进程之间共享大量对象列表，以便在任何给定时间所有正在运行的进程都与列表中的唯一对象/元素进行交互？

更新：在这一点上，我也将接受部分答案，讨论这是否可以用 Python 实现。

Answer 1

所以，我做了一些研究 (Shared Memory Objects in Multiprocessing) 并提出了一些想法：

传递 numpy 字节数组

序列化对象，然后将它们作为字节字符串保存到 numpy 数组中。这里的问题是

1. 需要将数据类型从'psm_test0' 的创建者传递给'psm_test0' 的任何消费者。不过，这可以通过另一个共享内存来完成。

2. pickle 和unpickle 本质上类似于deepcopy，即它实际上复制了基础数据。

“主”进程的代码如下：

import pickle
from multiprocessing import shared_memory
import numpy as np


# a simplistic class example
class A():
    def __init__(self, x):
        self.x = x

    def pickle(self):
        return pickle.dumps(self)

    @classmethod
    def unpickle(self, bts):
        return pickle.loads(bts)


if __name__ == '__main__':
    # Test pickling procedure
    a = A(1)
    print(A.unpickle(a.pickle()).x)
    # >>> 1

    # numpy array of byte strings
    a_arr = np.array([A(1).pickle(), A(2).pickle(), A('This is a really long test string which should exceed 42 bytes').pickle()])

    # create a shared memory instance
    shm = shared_memory.SharedMemory(
        create=True,
        size=a_arr.nbytes,
        name='psm_test0'
    )

    # numpy array backed by shared memory
    b_arr = np.ndarray(a_arr.shape, dtype=a_arr.dtype, buffer=shm.buf)

    # copy the original data into shared memory
    b_arr[:] = a_arr[:]

    print(b_arr.dtype)
    # 'S105'

为了消费者

import numpy as np
from multiprocessing import shared_memory
from test import A

# attach to the existing shared space
existing_shm = shared_memory.SharedMemory(name='psm_test0')

c = np.ndarray((3,), dtype='S105', buffer=existing_shm.buf)

# Test data transfer
arr = [a.x for a in list(map(A.unpickle, c))]
print(arr)
# [1, 2, ...]

我想说你有几种前进的方式：

1. 使用简单的数据类型。

2. 使用 C api 实现一些东西，但我不能真正帮助你。

3. 使用Rust

4. 使用Mangers。您可能会失去一些性能（不过我希望看到一个真正的基准测试），但是您可以获得一个相对安全且简单的共享对象接口。

5. 使用Redis，它也有 Python 绑定...