stop - python multithreading espera hasta que todos los hilos terminen

programa de hilos en python

Question

Esto puede haberse preguntado en un contexto similar, pero no pude encontrar una respuesta después de unos 20 minutos de búsqueda, así que lo preguntaré.

He escrito un script de Python (digamos: scriptA.py) y un script (digamos scriptB.py)

En scriptB quiero llamar a scriptA varias veces con diferentes argumentos, cada vez toma alrededor de una hora en ejecutarse (es un script enorme, hace muchas cosas ... no se preocupe) y quiero poder ejecutar el script. scriptA con todos los diferentes argumentos al mismo tiempo, pero tengo que esperar hasta que TODOS hayan terminado antes de continuar; mi código:

import subprocess

#setup
do_setup()

#run scriptA
subprocess.call(scriptA + argumentsA)
subprocess.call(scriptA + argumentsB)
subprocess.call(scriptA + argumentsC)

#finish
do_finish()

Quiero ejecutar todo el subprocess.call() al mismo tiempo, y luego esperar a que todo esté listo, ¿cómo debo hacer esto?

Intenté usar el enhebrado como el ejemplo here :

from threading import Thread
import subprocess

def call_script(args)
    subprocess.call(args)

#run scriptA   
t1 = Thread(target=call_script, args=(scriptA + argumentsA))
t2 = Thread(target=call_script, args=(scriptA + argumentsB))
t3 = Thread(target=call_script, args=(scriptA + argumentsC))
t1.start()
t2.start()
t3.start()

Pero no creo que esto sea correcto.

¿Cómo sé que todos han terminado de ejecutarse antes de ir a mi do_finish() ?

Answer 1

Coloque los hilos en una lista y luego use el método Join

 threads = []

 t = Thread(...)
 threads.append(t)

 ...repeat as often as necessary...

 # Start all threads
 for x in threads:
     x.start()

 # Wait for all of them to finish
 for x in threads:
     x.join()

Answer 2

De la documentación del módulo de threading

Hay un objeto "hilo principal"; esto corresponde al hilo de control inicial en el programa Python. No es un hilo demoníaco.

Existe la posibilidad de que se creen "objetos ficticios". Estos son objetos de subprocesos correspondientes a "hilos de origen", que son hilos de control que se inician fuera del módulo de subprocesamiento, como por ejemplo directamente desde el código C. Los objetos de subprocesos tienen una funcionalidad limitada; siempre se consideran vivos y demoníacos, y no se pueden join() ed. Nunca se eliminan, ya que es imposible detectar la terminación de los hilos extraterrestres.

Entonces, para atrapar esos dos casos cuando no esté interesado en mantener una lista de los hilos que crea:

import threading as thrd


def alter_data(data, index):
    data[index] *= 2


data = [0, 2, 6, 20]

for i, value in enumerate(data):
    thrd.Thread(target=alter_data, args=[data, i]).start()

for thread in thrd.enumerate():
    if thread.daemon:
        continue
    try:
        thread.join()
    except RuntimeError as err:
        if ''cannot join current thread'' in err.args[0]:
            # catchs main thread
            continue
        else:
            raise

Después de lo cual:

>>> print(data)
[0, 4, 12, 40]

Answer 3

Debe utilizar el método de join del objeto Thread al final del script.

t1 = Thread(target=call_script, args=(scriptA + argumentsA))
t2 = Thread(target=call_script, args=(scriptA + argumentsB))
t3 = Thread(target=call_script, args=(scriptA + argumentsC))

t1.start()
t2.start()
t3.start()

t1.join()
t2.join()
t3.join()

Por lo tanto, el hilo principal esperará hasta que t1 , t2 y t3 finalicen la ejecución.

Answer 4

En Python3, desde Python 3.2 hay un nuevo enfoque para alcanzar el mismo resultado, que personalmente prefiero al paquete tradicional de creación de hilos / inicio / unión, concurrent.futures : https://docs.python.org/3/library/concurrent.futures.html

Usando un ThreadPoolExecutor el código sería:

from concurrent.futures.thread import ThreadPoolExecutor

def call_script(arg)
    subprocess.call(scriptA + arg)

args = [argumentsA, argumentsB, argumentsC]
with ThreadPoolExecutor(max_workers=2) as executor:
    for arg in args:
        executor.submit(call_script, arg)
print(''All tasks has been finished'')

Una de las ventajas es que puede controlar la configuración de rendimiento de los trabajadores simultáneos máximos.

Answer 5

Me encontré con el mismo problema en el que necesitaba esperar a todos los hilos creados con el ciclo for. Simplemente probé la siguiente pieza de código. Puede que no sea la solución perfecta, pero pensé que sería una solución simple. Probar:

for t in threading.enumerate():
    try:
        t.join()
    except RuntimeError as err:
        if ''cannot join current thread'' in err:
            continue
        else:
            raise

Answer 6

Prefiero usar la lista de comprensión basada en una lista de entrada:

inputs = [scriptA + argumentsA, scriptA + argumentsB, ...]
threads = [Thread(target=call_script, args=(i)) for i in inputs]
[t.start() for t in threads]
[t.join() for t in threads]

Answer 7

Puede tener una clase similar a la siguiente a partir de la cual puede agregar ''n'' número de funciones o console_scripts que desea ejecutar en paralelo passion e iniciar la ejecución y esperar a que se completen todos los trabajos.

from multiprocessing import Process

class ProcessParallel(object):
    """
    To Process the  functions parallely

    """    
    def __init__(self, *jobs):
        """
        """
        self.jobs = jobs
        self.processes = []

    def fork_processes(self):
        """
        Creates the process objects for given function deligates
        """
        for job in self.jobs:
            proc  = Process(target=job)
            self.processes.append(proc)

    def start_all(self):
        """
        Starts the functions process all together.
        """
        for proc in self.processes:
            proc.start()

    def join_all(self):
        """
        Waits untill all the functions executed.
        """
        for proc in self.processes:
            proc.join()


def two_sum(a=2, b=2):
    return a + b

def multiply(a=2, b=2):
    return a * b


#How to run:
if __name__ == ''__main__'':
    #note: two_sum, multiply can be replace with any python console scripts which
    #you wanted to run parallel..
    procs =  ProcessParallel(two_sum, multiply)
    #Add all the process in list
    procs.fork_processes()
    #starts  process execution 
    procs.start_all()
    #wait until all the process got executed
    procs.join_all()

Answer 8

Tal vez, algo como

for t in threading.enumerate():
    if t.daemon:
        t.join()