在Python 3.2中并行执行for循环

Question

可能重复:
如何并行化一个简单的python循环？

我是Python的新手(使用Python 3.2),我有一个关于并行化的问题.我有一个for循环,我希望在Python 3.2中使用"多处理"并行执行:

def computation:    
    global output

    for x in range(i,j):
        localResult = ... #perform some computation as a function of i and j
        output.append(localResult)

总的来说,我想在i = 0到j = 100的范围内执行此计算.因此,我想创建一些进程,每个进程使用总范围的子域调用函数"calculate".有关于如何做到这一点的任何想法？有没有比使用多处理更好的方法？

更具体地说,我想执行域分解,我有以下代码:

from multiprocessing import Pool

class testModule:

    def __init__(self):
        self

    def computation(self, args):
        start, end = args
        print('start: ', start, ' end: ', end)

testMod = testModule()
length = 100
np=4
p = Pool(processes=np)
p.map(yes tMod.computation, [(length, startPosition, length//np) for startPosition in    range(0, length, length//np)]) 

我收到一条提到PicklingError的错误消息.任何想法可能是这里的问题？

Answer 1

Joblib专门用于包装多处理,以实现简单的并行循环.我建议使用它而不是直接处理多处理.

简单的案例看起来像这样:

from joblib import Parallel, delayed
Parallel(n_jobs=2)(delayed(foo)(i**2) for i in range(10))  # n_jobs = number of processes

一旦理解,语法就很简单了.我们使用生成器语法,其中delayed用于调用函数foo,其参数包含在后面的括号中.

在您的情况下,您应该使用生成器语法重写for循环,或者定义另一个函数(即'worker'函数)来执行单循环迭代的操作,并将其放入对Parallel的调用的生成器语法中.

在后一种情况下,您可以执行以下操作:

Parallel(n_jobs=2)(delayed(foo)(parameters) for x in range(i,j))

foo您定义的函数在哪里处理for循环的主体.请注意,您不希望附加到列表,因为Parallel无论如何都返回一个列表.

Answer 2

在这种情况下,您可能希望定义一个简单的函数来执行计算和获取localResult.

def getLocalResult(args):
    """ Do whatever you want in this func.  
        The point is that it takes x,i,j and 
        returns localResult
    """
    x,i,j = args  #unpack args
    return doSomething(x,i,j)

现在在您的计算功能中,您只需创建一个工作池并映射本地结果:

import multiprocessing
def computation(np=4):
    """ np is number of processes to fork """
    p = multiprocessing.Pool(np)
    output = p.map(getLocalResults, [(x,i,j) for x in range(i,j)] )
    return output

我已经删除了全局,因为它是不必要的(通常不需要全局变量).在您的通话程序中,您应该做output.extend(computation(np=4))或类似的事情.

编辑

这是代码的"工作"示例:

from multiprocessing import Pool

def computation(args):
    length, startPosition, npoints = args
    print(args)

length = 100
np=4
p = Pool(processes=np)
p.map(computation, [(startPosition,startPosition+length//np, length//np) for startPosition in  range(0, length, length//np)])

请注意,由于您使用实例方法作为函数,因此您所使用的不起作用.多处理启动新进程并在进程之间发送信息pickle,因此,只能使用可以被pickle的对象.请注意,无论如何使用实例方法确实没有意义.每个进程都是父进程的副本,因此对进程中发生的状态的任何更改都不会传播回父进程.