使用多处理填充类成员数据数组

Question

我有一个类，里面有一个浮点数组（它可以是 np.array 甚至是 python 列表，这并不重要）。

from multiprocessing import Pool, RawArray
import numpy as np

class ClassTest:
    def __init__(self, a_N):
        self.N = a_N
        self.arr = np.zeros(a_N, float)

数组不一定很大，但可能很大。然而，为了填充self.arr[i] 的每一个元素，需要大量的工作，即打开文件、读取和解析数据、计算复杂的函数等，因此，我想使用 threading 或 multiprocessing 模块。前者受python全局解释器锁的影响，基本上所有线程一个接一个地运行，而不是并行运行，因此不提供任何计算加速。后者似乎适合这项工作，但是，self.arr 然后需要以某种方式在进程之间共享。我试过这段代码（都是类的方法）。

    def fillAll(self):
        Np = 4
        Nc = self.N//Np
        pool = Pool(processes=Np)
        for i in range(self.Np):
            i0 = i*Nc
            i1 = i0 + Nc
            pool.apply_async(self.fillChunk, (i0, i1, ))
        pool.close()
        pool.join()

    def fillChunk(self, a_start, a_finish):
        for i in range(a_start, a_finish):
            self.arr[i] = computeOneCell(i)
  

computeOneCell(i) 是一个外部函数，它完成所有工作并返回一个浮点数。就我从 Internet 上的帮助中获得的信息而言，这段代码不起作用，因为每个进程都有自己的类实例副本，并且分叉进程完成的所有操作都不会影响父进程。我还尝试使用 RawArray，根据documentation，它被用作共享内存。但是，当我将 self.arr = np.zeros(a_N, float) 替换为 self.arr_X = RawArray('d', a_N) 时，代码不起作用。它不会崩溃，但也不会进入fillChunk 函数，就好像从未调用过pool.apply_async(self.fillChunk, (i0, i1, ))。

当不涉及类时，我有一个解决此问题的方法，即 RawArray 和 pool 在全局范围内声明，但我不知道如何使用类。

Answer 1

您真的不希望多个线程都对同一个数组进行更改。而且您不能真正让多个进程以简单的方式执行此操作。

我会做类似的事情

def fillAll(self):
    with Pool() as pool:
        for index, value in pool.imap(valueForCellAtIndex, range(self.N), chunksize=10):
            self.arr[index] = value

def valueForCellAtIndex(self, index):
    return index, computeOneCell(index)

您甚至可以使用imap_unordered 做得更好。通过同时返回来自fillChunk 的索引和值，您不需要按顺序返回结果。

您可能想尝试一下chunksize，看看什么能给您带来最好的结果。