多工序泡菜

Question

以下代码的目标仅仅是用0到9填充数组a：

import multiprocessing
import numpy as np
from joblib import Parallel, delayed

num_cores = multiprocessing.cpu_count()

inputs = range(10)
a = np.zeros(10)

def processInput():
    def testNested(t):
        a[t]= t
    Parallel(n_jobs=num_cores, backend="threading")(delayed(testNested)(t) for t in range(0, 10))

processInput()

当我试图调用函数中的多进程时，我会得到泡菜错误：

AttributeError: Can't pickle local object 'processInput.<locals>.testNested'

问题：有什么建议如何实现这个目标，在这种情况下我必须在其他功能中操作多进程吗？

Answer 1

按照错误信息，如文件所述，嵌套函数不能被腌制，您应该在模块的顶层定义worker函数。

Answer 2

在内置多处理包中显示了一些示例，我对第一个包做了一些修改，其中包括使用池子类：

class multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]]) 一个进程池对象，它控制可以向其提交作业的工作进程池。它支持带有超时和回调的异步结果，并具有并行映射实现。 processes是要使用的辅助进程数。如果进程是None，则使用os.cpu_count()返回的数字。 如果initializer不是None，那么每个工作进程在启动时都会调用initializer(*initargs)。 maxtasksperchild是一个工作进程在退出并替换为一个新的辅助进程之前可以完成的任务数，以便释放未使用的资源。默认的maxtasksperchild是None，这意味着工作进程将与池一样长。 context可用于指定用于启动辅助进程的上下文。通常，使用函数multiprocessing.Pool()或上下文对象的Pool()方法创建池。在这两种情况下，上下文都是适当设置的。 注意到，池对象的方法只应由创建池的进程调用。

因此，您实际上不需要导入和使用joblib，因为池子默认使用the number of CPUs in the system：

from multiprocessing import Pool

def f(x):
    return x

if __name__ == '__main__':
    with Pool() as p:
        print(p.map(f, list(range(10))))

然而，也可以通过编写“一条龙”来实现同样的结果：

print(list(range(10)))

我将池p映射为函数f()，以防您想要做比为索引赋值更复杂的事情。