Jak uniknąć tego błędu piklowania i jaki jest najlepszy sposób na zrównoleglenie tego kodu w Pythonie?

Question

Mam następujący kod.

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) 

Powoduje następujący błąd.

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 

Zawartość funkcji A jest nieistotna i nie powoduje błędów. Błąd pojawia się, gdy próbuję mapować funkcjęB. Jak mogę usunąć ten błąd i jaki jest najlepszy sposób na zrównoleglenie tego kodu w Pythonie 2.6?

Answer 1

Powodem, dla którego najprawdopodobniej widzisz takie zachowanie, jest kolejność definiowania puli, obiektów i funkcji. multiprocessing to nie to samo, co używanie wątków. Każdy proces odrodzi się i załaduje kopię środowiska. Jeśli utworzysz funkcje w zakresach, które mogą nie być dostępne dla procesów lub utworzysz obiekty przed pulą, pula zawiedzie.

pierwsze, spróbuj utworzyć jeden basen przed dużym pętli:

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

Następnie przenieść swój cel wywoływalnym poza dynamicznej pętli:

def functionB(a, b): 
    ... 

def main(): 
    ... 

Rozważmy następujący przykład ...

zerwana

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 

pracy

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]