Làm cách nào để tránh lỗi này, và cách tốt nhất để song song mã này bằng Python là gì?

Question

Tôi có mã sau đây.

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) 

Nó tạo ra lỗi sau.

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 

Nội dung chức năngA không quan trọng và không tạo ra lỗi. Lỗi này dường như xảy ra khi tôi cố gắng ánh xạ hàmB. Làm thế nào để loại bỏ lỗi này, và cách tốt nhất để song song mã này trong Python 2.6 là gì?

Answer 1

Lý do bạn rất có thể thấy hành vi này là do thứ tự mà bạn xác định hồ bơi, đối tượng và chức năng của mình. multiprocessing không hoàn toàn giống với việc sử dụng chuỗi. Mỗi quá trình sẽ sinh ra và tải một bản sao của môi trường. Nếu bạn tạo các hàm trong phạm vi có thể không có sẵn cho các quy trình hoặc tạo đối tượng trước hồ bơi, thì hồ bơi sẽ không thành công.

Đầu tiên, hãy thử tạo một hồ bơi trước khi vòng lặp lớn của bạn:

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

Sau đó, di chuyển mục tiêu của bạn có thể được gọi bên ngoài vòng lặp động:

def functionB(a, b): 
    ... 

def main(): 
    ... 

Hãy xem xét ví dụ này ...

bị hỏng

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 

làm việc

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]