使用 python 多处理访问和修改二维数组

Question

我刚刚开始使用 python 多处理。 我的代码面临的问题可以总结如下。

基本上，我试图让一个二维数组被几个不同的进程访问和修改。

我在一年前的 StackOverflow 上发现了 Alam (How to create a shared 2D array in python multiprocessing) 提出的类似问题，但没有明确给出该问题的答案。

这是一个代码 sn-p，类似于 Alam 提供的代码，以帮助讨论。

import multiprocessing as mp
import numpy as np
import ctypes as c

n = 2
m = 3

def addData(array, process_number):
    n,m = np.shape(array)
    i=0
    
    for nn in range(n):
        for mm in range(m):
            array[nn][mm] += i
            i=i+1
            
    print("Array after process " + str(process_number))
    print(array)

if __name__=='__main__':

    mp_arr=mp.Array('i', n*m)
    arr = np.frombuffer(mp_arr.get_obj(),c.c_int)

    u = arr.reshape((n, m))
    
    print("Array at the inital state: ")
    print(u)

    p1=mp.Process(target=addData,args=(u,1))
    p2=mp.Process(target=addData,args=(u,2))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()
    
    print("Array at the final state: ")
    print(u)

这是输出： output_result

可以看到，二维数组的最终输出还是[[0,0,0], [0,0,0]] 我希望它是 [[0,2,4], [6, 8, 10]]。

Answer 1

当我在 Windows 下使用 Python 3.8.5 运行此程序时，我得到了与您相同的结果，但在 Linux 和 Python 3.9.7 下，它可以工作。无论如何，以下是似乎无处不在的技术。

我们从一个看起来更自然的numpy 数组开始，然后创建一个共享内存数组，然后使用共享内存作为其缓冲区重新创建numpy 数组。另一个区别是我们将共享内存的引用和numpy数组的形状传递给进程，并让它使用辅助函数to_numpy_array重新创建numpy数组：

import multiprocessing as mp
import numpy as np
import ctypes as c

n = 2
m = 3

def addData(shared_array, shape, lock, process_number):
    array = to_numpy_array(shared_array, shape)
    n,m = shape
    i=0

    for nn in range(n):
        for mm in range(m):
            with lock:
                array[nn][mm] += i
            i=i+1

    print("Array after process " + str(process_number))
    print(array, flush=True)

def to_shared_array(arr, ctype):
    shared_array = mp.Array(ctype, arr.size, lock=False)
    temp = np.frombuffer(shared_array, dtype=arr.dtype)
    temp[:] = arr.flatten(order='C')
    return shared_array

def to_numpy_array(shared_array, shape):
    '''Create a numpy array backed by a shared memory Array.'''
    arr = np.ctypeslib.as_array(shared_array)
    return arr.reshape(shape)

if __name__=='__main__':

    # Start with a numpy array!
    mp_arr = np.zeros((n, m), dtype=np.int32)
    shared_array = to_shared_array(mp_arr, c.c_int32)
    # you have to now use the shared array as the base
    u = to_numpy_array(shared_array, mp_arr.shape)

    print("Array at the inital state: ")
    print(u)

    lock = mp.Lock()

    p1=mp.Process(target=addData,args=(shared_array, mp_arr.shape, lock, 1))
    p2=mp.Process(target=addData,args=(shared_array, mp_arr.shape, lock, 2))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print("Array at the final state: ")
    print(u)

打印：

Array at the inital state:
[[0 0 0]
 [0 0 0]]
Array after process 2
[[0 1 2]
 [3 4 5]]
Array after process 1
[[ 0  2  4]
 [ 6  8 10]]
Array at the final state:
[[ 0  2  4]
 [ 6  8 10]]