Python多处理共享内存不起作用

Question

我正在尝试使用多处理共享数组。我想创建两个函数，每个函数都使用一个通用数据集。但只有一个函数修改了数组。

这是我的代码：

import numpy as np
import multiprocessing as mp
import time

arr = mp.Array('f', np.array([1,2,3,4]))

def testfunc1(arr):
    while True:
        arr = np.concatenate((arr, np.array([0])))
        print(arr)
        time.sleep(2)

def testfunc2(arr):
    while True:
        arr[0] *= 2
        time.sleep(2)
        print(arr)

proc1 = mp.Process(target=testfunc1, args=(arr,))
proc2 = mp.Process(target=testfunc2, args=(arr,))

proc1.start()
proc2.start()

结果如下：

[1. 2. 3. 4. 0.]
[1. 2. 3. 4. 0. 0.]
<SynchronizedArray wrapper for <multiprocessing.sharedctypes.c_float_Array_4 object at 0x7fcd865bfdd0>>
<SynchronizedArray wrapper for <multiprocessing.sharedctypes.c_float_Array_4 object at 0x7fcd865bfdd0>>
[1. 2. 3. 4. 0. 0. 0.]
<SynchronizedArray wrapper for <multiprocessing.sharedctypes.c_float_Array_4 object at 0x7fcd865bfdd0>>
[1. 2. 3. 4. 0. 0. 0. 0.]
<SynchronizedArray wrapper for <multiprocessing.sharedctypes.c_float_Array_4 object at 0x7fcd865bfdd0>>
[1. 2. 3. 4. 0. 0. 0. 0. 0.]

testfunc2 似乎不起作用。我做错了什么？

Answer 1

这里的问题是您尝试在不同的进程中写信给arr，但arr 不是您期望的arr。在处理具有不同内存边界的多个处理器时，您需要考虑不同。

查看this answer，了解如何解决此问题。

Answer 2

多处理导致单独的 Python 进程，因此它们不会共享内存。

我不知道我是否会建议它作为一种好的做法，但多处理确实有一个 shared_memory 类可供您探索：https://docs.python.org/3/library/multiprocessing.shared_memory.html

它看起来使用起来不太简单，在我看来有点像反模式，但它确实存在。

否则我会建议：

1. 使用队列在进程之间进行通信，并重组您的应用程序，使它们以这种方式工作。
2. 使用线程模块。请记住，您最终会遇到竞态条件，每次运行程序时程序的结果都会发生不同的变化（或错误）。

我强烈推荐 Raymond Hettinger 关于 Python 并发的主题演讲，您会更好地理解如何处理它：https://pybay.com/site_media/slides/raymond2017-keynote/index.html

Answer 3

你在正确的轨道上。虽然您应该努力避免在进程之间共享状态，但创建了 multiprocessing.Array 以便您可以在绝对需要时共享状态。在这里，我刚刚注释掉了您对数组执行任何操作的部分。

import numpy as np
import multiprocessing as mp
import time

arr = mp.Array('f', np.array([1,2,3,4]))

def testfunc1(arr):
    while True:
        # arr = np.concatenate((arr, np.array([0])))
        print(arr)
        time.sleep(2)

def testfunc2(arr):
    while True:
        # arr[0] *= 2
        time.sleep(2)
        print(arr)

proc1 = mp.Process(target=testfunc1, args=(arr,))
proc2 = mp.Process(target=testfunc2, args=(arr,))

proc1.start()
proc2.start()

这会打印出以下内容：

<SynchronizedArray wrapper for <multiprocessing.sharedctypes.c_float_Array_4 object at 0x7f82f890f340>>
<SynchronizedArray wrapper for <multiprocessing.sharedctypes.c_float_Array_4 object at 0x7f82f890f340>>
<SynchronizedArray wrapper for <multiprocessing.sharedctypes.c_float_Array_4 object at 0x7f82f890f340>>

Python 基本上是将 NumPy 数组转换为 SynchronizedArray 对象并在进程之间共享它。您的代码中的问题在于 testfunc2 您试图对 SynchronizedArray 对象执行 numpy 操作。在testfunc1 中，请注意您正在通过np.concatenate 将SynchronizedArray 隐式转换为numpy 对象。要解决这个问题，请像这样更改代码：

import numpy as np
import multiprocessing as mp
import time

arr = mp.Array('f', np.array([1,2,3,4]))

def testfunc1(arr):
    while True:
        arr = np.concatenate((arr, np.array([0])))
        print(arr)
        time.sleep(2)

def testfunc2(arr):
    arr = np.array(arr)
    while True:
        arr[0] = arr[0] * 2
        time.sleep(2)
        print(arr)

proc1 = mp.Process(target=testfunc1, args=(arr,))
proc2 = mp.Process(target=testfunc2, args=(arr,))

proc1.start()
proc2.start()

这会返回：

[1. 2. 3. 4. 0.]
[2. 2. 3. 4.]
[1. 2. 3. 4. 0. 0.]
[4. 2. 3. 4.]
[1. 2. 3. 4. 0. 0. 0.]
[8. 2. 3. 4.]
[1. 2. 3. 4. 0. 0. 0. 0.]
[16.  2.  3.  4.]
[1. 2. 3. 4. 0. 0. 0. 0. 0.]
[32.  2.  3.  4.]
[1. 2. 3. 4. 0. 0. 0. 0. 0. 0.]
[64.  2.  3.  4.]
[1. 2. 3. 4. 0. 0. 0. 0. 0. 0. 0.]

但是，这是您想要的吗？否则，我会查看其他两个建议改革您的问题的答案，以便您不需要在进程之间共享状态。