如何让多个并行进程更改单个共享 NumPy 数组？

Question

我已经在互联网上搜寻了答案，但我找不到任何适用于我的情况。我读过multiprocessing.Manager，尝试来回传递东西，但似乎都不能很好地与NumPy数组一起使用。我尝试过使用Pool，但是我的目标方法没有返回任何东西，它只是进行了更改到一个数组，所以我也不知道如何设置它。

现在我有：

def Multiprocess(self, sigmaI, sigmaX):
    cpus = mp.cpu_count()
    print('Number of cpu\'s to process WM: %d' % cpus)

    processes = [mp.Process(target = self.CreateMatrixMp, args = (sigmaI, sigmaX, i,)) for i in range(self.numPixels)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

目标函数CreateMatrixMp 获取传递的值，并在进行计算后将值附加到数组data。这个数组被声明为self.data = numpy.zeros(self.size, numpy.float64)。如果CreateMatrixMp 方法的详细信息有帮助，我也可以发布。

我尝试在上面运行进程的地方添加这个：

mgr = mp.Manager()
sharedData = mgr.Array(ctypes.c_numpy.float64, self.data)

然后将sharedData传递给CreateMatrixMp，可以在其中进行修改。一旦所有进程都运行完毕并且数组完成后，我只需执行self.data = sharedData。

但这不起作用（尽管我知道我没有正确设置它）。这应该如何使用 NumPy 数组来完成？我希望每个进程（将有数千个）附加到同一个数组。

非常感谢任何帮助。

Answer 1

欢迎来到多线程的黑暗世界。我认为您的大问题是 mgr.Array 将同步放在阵列周围。如果您快速生成数据，这将是一个瓶颈，因为进程将等待最后一个完成数组。如果每个进程都保留一个 num 数组的私有副本，它会更有效并且会有所帮助。输入所有数据后，请等待所有过程完成。然后您可以将所有数组组合到 self.data 中。这样，任何处理都不需要等待共享资源。此解决方案和您的解决方案都不能保证输出列表的顺序。我怀疑 self.numPixels 的顺序可能很重要。许多解决方案，但最简单的方法是按顺序输入数据并在完成后执行 self.data.sort(...) 。或者更快地，预先创建 self.data 并让进程将结果戳到正确的位置。 self.data 不需要是共享数据结构，因为进程永远不会改变任何共同点。如果数组映射到类 C 数组，则此方法有效。它不适用于链接列表等。希望这会有所帮助。询问您是否需要更多详细信息。