Come evitare questo errore di pickling e qual è il modo migliore per parallelizzare questo codice in Python?

Question

Ho il seguente codice.

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 il seguente errore.

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 

Il contenuto della funzione A non è importante e non genera un errore. L'errore sembra verificarsi quando provo a mappare la funzioneB. Come rimuovo questo errore e qual è il modo migliore per parallelizzare questo codice in Python 2.6?

Answer 1

Il motivo per cui è più probabile vedere questo comportamento è a causa dell'ordine in cui è definito il pool, gli oggetti e le funzioni. multiprocessing non è esattamente come usare i thread. Ogni processo genera spawn e carica una copia dell'ambiente. Se si creano funzioni in ambiti che potrebbero non essere disponibili per i processi o creare oggetti prima del pool, il pool avrà esito negativo.

In primo luogo, provare a creare un pool prima del vostro grande anello:

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

Quindi, spostare il bersaglio callable al di fuori del ciclo dinamico:

def functionB(a, b): 
    ... 

def main(): 
    ... 

Si consideri l'esempio ...

rotto

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 

lavoro

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]