如何避免此酸洗错误，以及在Python中并行化此代码的最佳方法是什么？

Question

我有以下代码。

def main(): 
    (minI, maxI, iStep, minJ, maxJ, jStep, a, b, numProcessors) = sys.argv 
    for i in range(minI, maxI, iStep): 
    for j in range(minJ, maxJ, jStep): 
     p = multiprocessing.Process(target=functionA, args=(minI, minJ)) 
     p.start() 
     def functionB((a, b)): 
     subprocess.call('program1 %s %s %s %s %s %s' %(c, a, b, 'file1', 
      'file2', 'file3'), shell=True) 
     for d in ['a', 'b', 'c']: 
      subprocess.call('program2 %s %s %s %s %s' %(d, 'file4', 'file5', 
      'file6', 'file7'), shell=True) 
     abProduct = list(itertools.product(range(0, 10), range(0, 10))) 
     pool = multiprocessing.Pool(processes=numProcessors) 
     pool.map(functionB, abProduct) 

它会产生以下错误。

Exception in thread Thread-1: 
Traceback (most recent call last): 
    File "/usr/lib64/python2.6/threading.py", line 532, in __bootstrap_inner 
    self.run() 
    File "/usr/lib64/python2.6/threading.py", line 484, in run 
    self.__target(*self.__args, **self.__kwargs) 
    File "/usr/lib64/python2.6/multiprocessing/pool.py", line 255, in _handle_tasks 
    put(task) 
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function fa 
iled 

函数A的内容并不重要，也不会产生错误。当我尝试映射functionB时，该错误似乎发生。我如何删除这个错误，以及在Python 2.6中并行化这些代码的最佳方式是什么？

Answer 1

您最有可能看到此行为的原因是由于您定义池，对象和函数的顺序。 multiprocessing与使用线程不太一样。每个进程都会生成并加载环境的副本。如果您在作用域中创建可能无法用于进程的函数，或者在该池之前创建对象，则该池将失败。

首先，尝试创建你的大循环前一个池：

(minI, maxI, iStep, minJ, maxJ, jStep, a, b, numProcessors) = sys.argv 
pool = multiprocessing.Pool(processes=numProcessors) 
for i in range(minI, maxI, iStep): 
    ... 

然后，将你的目标可赎回动态环路外：

def functionB(a, b): 
    ... 

def main(): 
    ... 

考虑这个例子...

破碎

import multiprocessing 

def broken(): 
    vals = [1,2,3] 

    def test(x): 
     return x 

    pool = multiprocessing.Pool() 
    output = pool.map(test, vals) 
    print output 

broken() 
# PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed 

工作

import multiprocessing 

def test(x): 
    return x 

def working(): 
    vals = [1,2,3] 

    pool = multiprocessing.Pool() 
    output = pool.map(test, vals) 
    print output 

working() 
# [1, 2, 3]