¿Cómo evito este error de decapado y cuál es la mejor forma de paralelizar este código en Python?

Question

Tengo el siguiente código.

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) 

Produce el siguiente error.

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 

Los contenidos de la función A no son importantes y no producen un error. El error parece ocurrir cuando trato de mapear la función B. ¿Cómo elimino este error y cuál es la mejor forma de paralelizar este código en Python 2.6?

Answer 1

La razón por la que es más probable que vea este comportamiento es por el orden en que define su grupo, objetos y funciones. multiprocessing no es lo mismo que usar hilos. Cada proceso generará y cargará una copia del entorno. Si crea funciones en ámbitos que pueden no estar disponibles para los procesos, o crea objetos antes del grupo, el grupo fallará.

En primer lugar, intente crear una piscina antes de su bucle grande:

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

A continuación, mueva el objetivo exigible fuera del bucle dinámico:

def functionB(a, b): 
    ... 

def main(): 
    ... 

Considere este ejemplo ...

roto

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 

trabajo

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]