Uso del grupo de multiprocesamiento de trabajadores

Question

Tengo el siguiente código escrito para que funcione mi segundo núcleo de CPU. Lo que hace básicamente el código es encontrar primero los archivos "mar" deseados en la jerarquía del directorio y luego ejecutar el conjunto de scripts externos para procesar estos archivos binarios "mar" para producir de 50 a 100 archivos de texto y binarios en número. Como el título de la pregunta sugiere de forma paralela para aumentar la velocidad de procesamiento.

Esta pregunta tiene su origen en la larga discusión que hemos estado teniendo en la lista de usuarios de IPython titulada como "Cannot start ipcluster". Comenzando con mi experimentación en las funcionalidades de procesamiento paralelo de IPython.

El problema es que no puedo obtener este código funcionando correctamente. Si las carpetas que contienen archivos "mar" solo albergan archivos "mar", el guión finaliza su ejecución sin ejecutar completamente las secuencias de comandos externas. (Digamos que tengo entre 30 y 50 scripts externos para ejecutar, pero mi script habilitado para multiprocesamiento se agota solo después de ejecutar el primer script en esta cadena de scripts externa.) Curiosamente, si ejecuto este script en una carpeta ya procesada (que es archivos "mar") procesados ​​de antemano y los archivos de salida ya están en esa carpeta) luego se ejecuta, pero esta vez obtengo aceleraciones de 2.4 a 2.7X con respecto a los tiempos de procesamiento lineal. No es muy esperado ya que solo tengo una CPU Core 2 Duo 2.5 Ghz en mi computadora portátil. Aunque tengo una GPU con CUDA, no tiene nada que ver con mi actual lucha paralela en la informática :)

¿Cuál crees que podría ser la causa de este problema?

Gracias por todos los comentarios y sugerencias.

#!/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

Me gustaría empezar con conseguir una mejor idea de lo que está pasando con el proceso de trabajo. El módulo de multiprocesamiento viene con el registro de sus subprocesos si es necesario. Puesto que usted ha simplificado el código para reducir el problema, sólo quiero depurar con algunas declaraciones de impresión, al igual que (o puede prettyprint la matriz 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() 

La versión de Python que tengo Lo logró con 2.6.4.

Answer 2

Hay varias cosas que se me ocurren:

1) ¿Ha impreso el pathfiles? ¿Estás seguro de que todos están bien generados?

a) Pregunto porque os.walk es un poco interesante; el dirs.sort() debería estar bien, pero parece bastante innecesario. os.chdir() en general no debería ser usado; la restauración debería estar bien, pero en general solo debería agregar root a init.

2) He visto el multiproceso en python2.6 tener problemas para generar subporcesos de las agrupaciones. (Específicamente, un script usa multiprocesamiento para generar subprocesos. Esos subprocesos no podían usar correctamente el multiprocesamiento (el grupo bloqueado)). Pruebe python2.5 con el backport multiproceso.

3) Pruebe el módulo cloud.mp de picloud (que ajusta el multiprocesamiento, pero maneja las piscinas un poco diferente) y vea si eso funciona.

Harías

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

(Negación: Soy uno de los desarrolladores de PiCloud)