使用多处理将数组添加到全局数组

Question

我有一个全局 NumPy 数组 ys_final 并定义了一个生成数组 ys 的函数。 ys 数组将根据输入参数生成，我想将这些 ys 数组添加到全局数组中，即ys_final = ys_final + ys。
添加的顺序无关紧要，所以我想使用多处理库中的 Pool.apply_async() 但我无法写入全局数组。参考代码为：

import multiprocessing as mp

ys_final = np.zeros(len)
def ys_genrator(i):
    #code to generate ys array
    return ys
pool = mp.Pool(mp.cpu_count())
for i in range(3954):
    ys_final = ys_final + pool.apply_async(ys_genrator, args=(i)).get()
pool.close()
pool.join()

上面的代码块一直在运行，没有任何反应。我也试过 *mp.Process ，但我仍然面临同样的问题。在那里我定义了一个直接添加到全局数组的目标函数，但它也无法正常工作，因为该块会一直运行。参考：

def func(i):
    #code to generate ys
    global ys_final
    ys_final = ys_final + ys

for i in range(3954):
    p = mp.Process(target=func, args=(i,))
    p.start()
    p.join()

任何建议都会很有帮助。

编辑： 我的 ys_genrator 是一个线性插值函数。基于参数 i 是二维图像中行的索引，该函数创建一个插值幅度数组，该数组将与图像中的所有插值幅度叠加，因此ys 需要被添加到ys_final 变量len是插值数组的长度，所有行都一样。

供参考，ys_genrator(i) 的简单版本如下：

def ys_genrator(i):
    ys = np.ones(10)*i
    return ys

Answer 1

几点：

1. pool.apply_async(ys_genrator, args=(i)) 必须是 pool.apply_async(ys_genrator, args=(i,))。注意i 后面的逗号。
2. pool.apply_async(ys_genrator, args=(i,)).get() 完全等同于 pool.apply(ys.genrator, args=(i,))。也就是说，你会因为你立即调用get 而阻塞，你绝对不会有并行。您需要对pool.apply_async 进行所有调用并保存返回的AsyncResult 实例，然后才在这些实例上调用get。
3. 如果您在 Windows 下运行，您将遇到问题。创建新进程的代码必须在由if __name__ == '__main__': 管理的块内
4. 如果您在 Jupyter Notebook 或 iPython 之类的系统下运行，则会遇到问题。工作函数 ys_genrator 需要在外部文件中并导入。
5. 使用apply_async 提交大量任务效率低下。您最好使用imap 或imap_unordered，其中任务以“块”的形式提交，您可以在结果可用时一一处理。但您必须选择一个“合适的”chunksize 参数。
6. 如果您在 Windows 下运行，您在全局级别拥有的任何代码，例如 ys_final = np.zeros(len)，都将由每个子进程执行，如果子进程不需要“看到”这个变量，这可能是一种浪费。如果他们确实需要查看此变量，请注意池中的每个进程都将使用其自己的变量副本，因此最好是只读用法。即使这样，如果变量很大，也可能非常浪费存储。有多种方法可以在进程之间共享这样的变量，但您是否需要（您甚至还没有定义变量len）并不十分清楚。所以很难给你改进的代码。但是，您的工作函数似乎不需要“看到”ys_final，因此我将尝试改进解决方案。
7. 但请注意，如果您的函数ys_genrator 非常简单，则使用多处理将不会获得任何收益，因为在创建处理池和将参数从一个进程传递到另一个进程时都会产生开销。此外，如果 ys_genrator 正在使用 numpy，这也可能是问题的根源，因为 numpy 对自己的某些功能使用多处理，您最好不要将 numpy 与您自己的多处理混合使用。

import multiprocessing as mp
import numpy as np

SIZE = 3

def ys_genrator(i):
    #code to generate ys array
    # for this dummy example all SIZE entries will end up with the same result:
    ys = [i] * SIZE # for example: [1, 1, 1]
    return ys

def compute_chunksize(poolsize, iterable_size):
    chunksize, remainder = divmod(iterable_size, 4 * poolsize)
    if remainder:
        chunksize += 1
    return chunksize

if __name__ == '__main__':
    ys_final = np.zeros(SIZE)
    n_iterations = 3954
    poolsize = min(mp.cpu_count(), n_iterations)
    chunksize = compute_chunksize(poolsize, n_iterations)
    print('poolsize =', poolsize, 'chunksize =', chunksize)
    pool = mp.Pool(poolsize)
    for result in pool.imap_unordered(ys_genrator, range(n_iterations), chunksize):
        ys_final += result
    print(ys_final)

打印：

poolsize = 8 chunksize = 124
[7815081. 7815081. 7815081.]

更新

你也可以使用：

    for result in pool.map(ys_genrator, range(n_iterations)):
        ys_final += result

问题是，当您使用方法map 时，该方法希望根据可迭代参数的大小计算一个有效的chunksize 参数（参见上面我的compute_chunksize 函数，这基本上就是pool.map将使用）。但要做到这一点，必须首先将可迭代对象转换为列表以获取其大小。如果n_iterations 非常大，这不是很有效，尽管对于 3954 的大小来说这可能不是一个主要问题。不过，在这种情况下，您最好使用我的 compute_chunksize 函数，因为您知道可迭代，然后将 chunksize 参数显式传递给 map，就像我在代码中使用 imap_unordered 所做的那样。