multihilos - Los procesos de multiprocesamiento Python duermen después de un tiempo

multihilos en python

Question

Tengo un script que se ejecuta a través de un directorio y busca todos los archivos con una terminación dada (es decir, .xml) para cadenas determinadas y las reemplaza. Para lograr esto, utilicé la biblioteca de multiprocesamiento de python.

Como ejemplo, estoy usando 1100 archivos .xml con alrededor de 200MB de datos. El tiempo de ejecución completo es de 8 minutos en mi MBP ''15 15 ".

Pero después de unos minutos, el proceso para el proceso se va a dormir, lo que veo en "arriba" (aquí después de 7m ...).

salida superior

PID COMMAND %CPU TIME #TH #WQ #PORT MEM PURG CMPR PGRP PPID STATE BOOSTS %CPU_ME %CPU_OTHRS
1007 Python 0.0 07:03.51 1 0 7 5196K 0B 0B 998 998 sleeping *0[1] 0.00000 0.00000
1006 Python 99.8 07:29.07 1/1 0 7 4840K 0B 0B 998 998 running *0[1] 0.00000 0.00000
1005 Python 0.0 02:10.02 1 0 7 4380K 0B 0B 998 998 sleeping *0[1] 0.00000 0.00000
1004 Python 0.0 04:24.44 1 0 7 4624K 0B 0B 998 998 sleeping *0[1] 0.00000 0.00000
1003 Python 0.0 04:25.34 1 0 7 4572K 0B 0B 998 998 sleeping *0[1] 0.00000 0.00000
1002 Python 0.0 04:53.40 1 0 7 4612K 0B 0B 998 998 sleeping *0[1] 0.00000 0.00000

Entonces, ahora solo un proceso está haciendo todo el trabajo mientras los otros se durmieron después de 4 minutos.

Fragmento de código

# set cpu pool to cores in computer
pool_size = multiprocessing.cpu_count()
# create pool
pool = multiprocessing.Pool(processes=pool_size)
# give pool function and input data - here for each file in file_list
pool_outputs = pool.map(check_file, file_list)
# if no more tasks are available: close all
pool.close()
pool.join()

Entonces, ¿por qué todos los procesos están durmiendo?

Mi suposición: la lista de archivos está separada para todos los trabajadores en el grupo (la misma cantidad cada uno) y algunos simplemente tienen "suerte" de obtener los archivos pequeños, y por lo tanto terminar antes. ¿Esto puede ser cierto? Estaba pensando que funciona más como una cola para que cada trabajador obtenga un nuevo archivo cuando haya terminado, hasta que la lista esté vacía.

Answer 1

Como @ Felipe-Lema señaló que es un RTFM clásico.

Repasé la parte del script mencionada utilizando una Cola de multiprocesamiento en lugar de un Pool y mejoré el tiempo de ejecución:

def check_files(file_list):
"""Checks and replaces lines in files
@param file_list: list of files to search
@return counter: number of occurrence """
# as much workers as CPUs are available (HT included)
workers = multiprocessing.cpu_count()
# create two queues: one for files, one for results
work_queue = Queue()
done_queue = Queue()
processes = []
# add every file to work queue
for filename in file_list:
work_queue.put(filename)
# start processes
for w in xrange(workers):
p = Process(target=worker, args=(work_queue, done_queue))
p.start()
processes.append(p)
work_queue.put(''STOP'')
# wait until all processes finished
for p in processes:
p.join()
done_queue.put(''STOP'')
# beautify results and return them
results = []
for status in iter(done_queue.get, ''STOP''):
if status is not None:
results.append(status)
return results