Korzystanie z puli wieloprocesowej pracowników

Question

Mam następujący kod napisany, aby mój leniwy drugi rdzeń procesora działał. Zasadniczo po prostu znajdź żądane pliki "morze" w hierarchii katalogów, a następnie wykonaj zestaw skryptów zewnętrznych, aby przetworzyć te binarne pliki "morze", aby utworzyć od 50 do 100 tekstów i plików binarnych. Ponieważ tytuł pytania sugeruje w sposób równoległy, aby zwiększyć szybkość przetwarzania.

To pytanie pochodzi z długiej dyskusji, którą prowadziliśmy na liście użytkowników IPython zatytułowanej "Cannot start ipcluster". Począwszy od moich eksperymentów na równoległych funkcjach przetwarzania IPython.

Problem polega na tym, że nie mogę poprawnie uruchomić tego kodu. Jeśli w folderach zawierających pliki "morze" znajdują się tylko pliki "morze", skrypt kończy wykonywanie bez pełnego wykonywania zewnętrznych skryptów. (Załóżmy, że mam uruchamiane 30-50 zewnętrznych skryptów, ale mój skrypt obsługujący wiele procesorów działa tylko po wykonaniu pierwszego skryptu w łańcuchu zewnętrznych skryptów.) Co ciekawe, jeśli uruchomię ten skrypt na już przetworzonym folderze (który jest "morskim" plikiem przetwarzane wcześniej, a pliki wyjściowe są już w tym folderze), a następnie działa, ale tym razem dostaję przyspieszenia na poziomie około 2,4 do 2,7X w odniesieniu do czasu przetwarzania liniowego. Nie jest to zbyt oczekiwane, ponieważ mam tylko procesor Core 2 Duo 2,5 Ghz w moim laptopie. Chociaż mam procesor GPU z obsługą CUDA, nie ma to nic wspólnego z moją obecną równoległą walką komputerową :)

Co według ciebie może być źródłem tego problemu?

Dziękujemy za wszystkie uwagi i sugestie.

#!/usr/bin/env python 

from multiprocessing import Pool 
from subprocess import call 
import os 


def find_sea_files(): 

    file_list, path_list = [], [] 
    init = os.getcwd() 

    for root, dirs, files in os.walk('.'): 
     dirs.sort() 
     for file in files: 
      if file.endswith('.sea'): 
       file_list.append(file) 
       os.chdir(root) 
       path_list.append(os.getcwd()) 
       os.chdir(init) 

    return file_list, path_list 


def process_all(pf): 
    os.chdir(pf[0]) 
    call(['postprocessing_saudi', pf[1]]) 


if __name__ == '__main__': 
    pool = Pool(processes=2)    # start 2 worker processes 
    files, paths = find_sea_files() 
    pathfile = [[paths[i],files[i]] for i in range(len(files))] 
    pool.map(process_all, pathfile) 

Answer 1

chciałbym zacząć coraz lepsze wyczucie tego, co dzieje się z procesem roboczym. W razie potrzeby moduł wieloprocesowy zawiera rejestrację dla podprocesów. Skoro masz uproszczony kod, aby zawęzić problem, chciałbym po prostu debugować z kilku instrukcji print, tak jak (lub można prettyprint tablicę pf):

def process_all(pf): 
    print "PID: ", os.getpid() 
    print "Script Dir: ", pf[0] 
    print "Script: ", pf[1] 
    os.chdir(pf[0]) 
    call(['postprocessing_saudi', pf[1]]) 


if __name__ == '__main__': 
    pool = Pool(processes=2) 
    files, paths = find_sea_files() 
    pathfile = [[paths[i],files[i]] for i in range(len(files))] 
    pool.map(process_all, pathfile, 1) # Ensure the chunk size is 1 
    pool.close() 
    pool.join() 

wersji Pythona, że ​​mam zrealizował to za pomocą 2.6.4.

Answer 2

Istnieje kilka rzeczy, które można myślę o:

1) Czy wydrukowana na pathfiles? Czy jesteś pewien, że wszystkie są właściwie generowane?

a) Pytam, ponieważ twój os.walk jest trochę interesujący; dirs.sort() powinien być w porządku, ale wydaje się dość niepotrzebny. os.chdir() w ogóle nie powinien być używany; przywrócenie musi być w porządku, ale ogólnie powinno się po prostu dołączać root do init.

2) Widziałem, jak wieloprocesorowość na python2.6 ma problemy z tworzeniem subportów z pul. (Miałem specjalnie skrypt używający wieloprocesowości do odradzania podprocesów, te podprocesy nie mogły poprawnie używać wieloprocesowości (pula była zamknięta)). Wypróbuj python2.5 z mportem przetwarzania mulitprocessing.

3) Wypróbuj moduł cloud.mp picloud (który owija proces wieloprocesowy, ale obsługuje tagi inaczej) i sprawdź, czy to działa.

Można by zrobić

cloud.mp.join(cloud.mp.map(process_all, pathfile)) 

(Zastrzeżenie: Jestem jednym z twórców PiCloud)