wielowątkowość w Pythonie czeka na zakończenie wszystkich wątków

Mogłem o to zapytać w podobnym kontekście, ale nie mogłem znaleźć odpowiedzi po około 20 minutach poszukiwań, więc zapytam.

Napisałem skrypt w Pythonie (powiedzmy: scriptA.py) i skrypt (powiedzmy scriptB.py)

W skrypcie B chcę wywołać skryptA wiele razy z różnymi argumentami, za każdym razem uruchomienie trwa około godziny (jest to ogromny skrypt, robi wiele rzeczy ... nie martw się o to) i chcę mieć możliwość uruchomienia scriptA ze wszystkimi różnymi argumentami jednocześnie, ale przed kontynuowaniem muszę poczekać, aż WSZYSTKIE zostaną wykonane; mój kod:

import subprocess

#setup
do_setup()

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

#finish
do_finish()

Chcę uruchomić wszystko subprocess.call()w tym samym czasie, a następnie poczekać, aż wszystkie skończą , jak mam to zrobić?

Próbowałem użyć wątków, jak na przykładzie tutaj :

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()

Ale nie sądzę, żeby to było słuszne.

Skąd mam wiedzieć, że wszyscy skończyli biegać przed pójściem do mojego do_finish()?

Na końcu skryptu należy użyć metody łączeniaThread obiektu.

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()

Zatem główny wątek będzie czekać aż t1, t2i t3zakończyć działanie.

Umieść wątki na liście, a następnie użyj metody 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()

W Pythonie3, od Pythona 3.2, jest nowe podejście do osiągnięcia tego samego wyniku, który osobiście wolę od tradycyjnego tworzenia wątku / start / dołączanie, pakiet concurrent.futures: https://docs.python.org/3/library/concurrent.futures .html

Użycie ThreadPoolExecutorkodu wyglądałoby tak:

from concurrent.futures.thread import ThreadPoolExecutor
import time

def call_script(ordinal, arg):
    print('Thread', ordinal, 'argument:', arg)
    time.sleep(2)
    print('Thread', ordinal, 'Finished')

args = ['argumentsA', 'argumentsB', 'argumentsC']

with ThreadPoolExecutor(max_workers=2) as executor:
    ordinal = 1
    for arg in args:
        executor.submit(call_script, ordinal, arg)
        ordinal += 1
print('All tasks has been finished')

Wynik poprzedniego kodu jest podobny do:

Thread 1 argument: argumentsA
Thread 2 argument: argumentsB
Thread 1 Finished
Thread 2 Finished
Thread 3 argument: argumentsC
Thread 3 Finished
All tasks has been finished

Jedną z zalet jest to, że można kontrolować przepływność, ustawiając maksymalną liczbę równoczesnych pracowników.

Preferuję rozumienie list na podstawie listy wejściowej:

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]

Możesz mieć klasę podobną do poniżej, z której możesz dodać 'n' liczbę funkcji lub skryptów console_scripts, które chcesz wykonać równolegle, i rozpocząć wykonywanie i poczekać na zakończenie wszystkich zadań.

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()

Z threading dokumentacji modułu

Istnieje obiekt „wątek główny”; odpowiada to początkowemu wątkowi kontroli w programie w języku Python. To nie jest wątek demona.

Istnieje możliwość, że zostaną utworzone „fikcyjne obiekty wątku”. Są to obiekty wątków odpowiadające „obcym wątkom”, które są wątkami kontroli uruchomionymi poza modułem wątkowym, na przykład bezpośrednio z kodu C. Fikcyjne obiekty wątku mają ograniczoną funkcjonalność; są zawsze uważani za żywych i demonicznych i nie można ich join()edytować. Nigdy nie są usuwane, ponieważ niemożliwe jest wykrycie zakończenia obcych wątków.

Tak więc, aby uchwycić te dwa przypadki, gdy nie jesteś zainteresowany utrzymywaniem listy utworzonych wątków:

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

Po czym:

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

Może coś takiego

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

Właśnie natknąłem się na ten sam problem, w którym musiałem poczekać na wszystkie wątki, które zostały utworzone za pomocą pętli for. Właśnie wypróbowałem następujący fragment kodu Może nie jest to idealne rozwiązanie, ale pomyślałem, że będzie to proste rozwiązanie testować:

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