La bonne façon de limiter au maximum le nombre de threads en cours d'exécution à la fois?

J'aimerais créer un programme qui s'exécute de multiples threads, mais il se limite à une constante, le nombre prédéfini de l'exécution de tâches simultanées, comme ceci (mais sans risque de race condition):

import threading

def f(arg):
    global running
    running += 1
    print("Spawned a thread. running=%s, arg=%s" % (running, arg))
    for i in range(100000):
        pass
    running -= 1
    print("Done")

running = 0
while True:
    if running < 8:
        arg = get_task()
        threading.Thread(target=f, args=[arg]).start()

Ce qui est le plus sûr/le moyen le plus rapide pour mettre en œuvre cette?

Il semble que vous décrivez un pool de threads.
Ne pool de threads implique de stocker les références des fils? J'aimerais le garder aussi léger que possible.
docs.python.org/2/library/...
Juste édité la question - qui a remplacé le "CPU" à la "lumière", comme il convient à mon problème plus précisément.
Il n'importe pas vraiment si une référence à la thread est stocké ou pas, n'est ce pas? Huit références de l'objet ne sont pas va faire ou défaire votre programme.
J'ai eu la pensée que de les remplacer rapidement pourrait causer une grosse perte de performance.
BTW, j'ai mis à jour le code pour montrer que j'ai besoin de garder en tirant le fil des arguments.
Le remplacement d'un objet de référence, par rapport à la surcharge de commencer un nouveau thread, est extrêmement rapide. Avez-vous fait des points de référence, ou vous êtes tout simplement en essayant de deviner ce qui est "lumière" et ce qui ne l'est pas?
doublon potentiel de stackoverflow.com/questions/1787397/...

InformationsquelleAutor d33tah | 2013-10-14

33

Il semble que vous voulez mettre en œuvre le producteur/consommateur avec motif huit travailleurs. Python a un File classe et à cette fin, et il n'est pas thread-safe.

Chaque travailleur doit appeler get() sur la file d'attente pour récupérer une tâche. Cet appel va bloquer si aucune tâche n'est disponible, entraînant le travailleur à aller au ralenti jusqu'à ce que l'on devient disponible. Ensuite, le travailleur doit exécuter la tâche et enfin appeler task_done() sur la file d'attente.

Vous mettrais des tâches dans la file d'attente en appelant put() sur la file d'attente.

Depuis le thread principal, vous pouvez appeler join() sur la file d'attente à attendre jusqu'à ce que toutes les tâches en attente ont été achevés.

Cette approche a l'avantage que vous n'êtes pas à la création et à la destruction de threads, ce qui est coûteux. Les threads de travail sera exécuté en continu, mais sera endormi quand pas de tâches dans la file d'attente, à l'aide de zéro de temps PROCESSEUR.

(Liés à la documentation de la page est un exemple de ce modèle.)
- Fonctionne très bien, mais je me demandais comment signaler les threads que j'ai fini d'envoyer les tâches?
- Envoyer un " je suis terminé la tâche, en demandant au pool de threads terminer. N'importe quel thread qui reçoit une telle tâche requeues et puis se suicide.
- La file d'attente est certainement vous utilisez des threads à bon escient, mais ce n'est pas de limiter le nombre de threads en cours de création(et a commencé simultanément), une limite définie dans la piscine va juste attendre que la file d'attente pour finir ces certain nombre de tâches dans la file d'attente et de passer au prochain chiffre. si la taille des données est grande, la file d'attente est toujours tenue trop de la consommation de la mémoire de ralentir les performances du système.
- Une file d'attente n'est pas directement limiter les threads, mais il permet de limiter le fils facilement à l'aide d'une piscine, et le lien exemple montre exactement comment le faire. Et de stocker plus de données dans la file d'attente ne ralentit pas les performances du système, ou du moins pas plus que de les stocker dans une liste, c'est juste que certaines écluses autour d'un deque, qui ne prend pas plus d'espace de stockage que une liste.
- Vous n'avez qu'à envoyer le suicide de la tâche une fois que si vous faites ce que Martin a dit: "n'Importe quel thread qui reçoit une telle tâche requeues c' et puis se suicide."
- Ah, raté la partie; merci.
- La suggestion au sujet de l'envoi d'un " je suis terminé la tâche telle que n'importe quel thread qui se requeues un commet le suicide ne fonctionne pas très bien pour moi. J'ai une question de suivi à ce sujet ici: stackoverflow.com/q/45169559/1175080 .
- salut @cdhowie, je suis un peu en retard à la fête, mais de quelle façon cette approche explicitement limiter le nombre max de threads à seulement 8?
- Vous n'créer huit travailleurs.
InformationsquelleAutor cdhowie
17

sémaphore est une variable ou d'un type abstrait de données qui est utilisée pour contrôler l'accès à une ressource commune par plusieurs processus en même temps une système comme un multiprogramming système d'exploitation, ce qui peut vous aider ici.
```
threadLimiter = threading.BoundedSemaphore(maximumNumberOfThreads)

class MyThread(threading.Thread):

    def run(self):
        threadLimiter.acquire()
        try:
            self.Executemycode()
        finally:
            threadLimiter.release()

    def Executemycode(self):
        print(" Hello World!") 
        # <your code here>
```
De cette façon, vous pouvez facilement limiter le nombre de threads qui seront exécutées en parallèle au cours de l'exécution du programme. Variable "maximumNumberOfThreads' peut être utilisé pour définir une limite supérieure à la valeur maximum de threads.

crédits
- semble très similaire à stackoverflow.com/a/5991741/288875
- Ajout des crédits 🙂
- Merci! C'est exactement ce que je voulais!
InformationsquelleAutor Hammad Haleem
5

Il serait beaucoup plus facile à mettre en œuvre la présente comme un fil de la piscine ou à l'exécuteur testamentaire, à l'aide multiprocessing.dummy.Pool, ou concurrent.futures.ThreadPoolExecutor (ou, si l'utilisation de Python 2.x, le backport terme). Par exemple:
```
import concurrent

def f(arg):
    print("Started a task. running=%s, arg=%s" % (running, arg))
    for i in range(100000):
        pass
    print("Done")

with concurrent.futures.ThreadPoolExecutor(8) as executor:
    while True:
        arg = get_task()
        executor.submit(f, arg)
```
Bien sûr, si vous pouvez modifier le modèle d'extraction get_task à un push-modèle get_tasks que, par exemple, les rendements des tâches une à la fois, c'est encore plus simple:
```
with concurrent.futures.ThreadPoolExecutor(8) as executor:
    for arg in get_tasks():
        executor.submit(f, arg)
```
Lorsque vous exécutez des tâches (par exemple, get_task soulève une exception, ou get_tasks fonctionne à sec), ce sera automatiquement dire que l'exécuteur de s'arrêter après il draine la file d'attente, attendre qu'il s'arrête, et nettoyer tout.
- comment est-ce réalisable seulement avec threading et Queue ?
- Si vous avez besoin de comprendre comment créer un pool de threads correctement, vous pouvez regarder le multiprocessing et concurrent.futures modules le code source, ils ne sont pas si compliqué que ça. Ou n'importe quel nombre de tutoriels. Mais pourquoi voulez-vous construire votre propre piscine de mise en œuvre lorsque vous avez déjà une bonne parfaitement l'un dans l'stdlib?
- uttmost merci pour le lien, je n'ai pas passé beaucoup de temps avec le multitraitement et simultanées.les contrats à terme qui est pourquoi la question s'est posée, j'ai eu un coup d'oeil, de sorte que le filetage est à la base de ces deux modules...
- Non, multiprocessing simule les discussions avec les processus enfants, puis génère des fonctionnalités supplémentaires (comme les piscines, explicite données partagées, etc.) sur le dessus de cela, et aussi (dans multiprocessing.dummy) fournit les mêmes fonctionnalités supplémentaires pour les threads. (Pas idéal stdlib design, mais des raisons historiques...) futures s'exécute au-dessus de threading ou multiprocessing (selon l'exécuteur vous utilisez), qui fournit la même interface à l'autre.
- est le ThreadPoolExecutor ne fonctionne pas avec la méthode d'instance comme self.xxxx ?
- Cette interface est la plus belle, mais est-il un bon moyen de détecter les exceptions sur les travailleurs de la fonction? stackoverflow.com/questions/33448329/...
InformationsquelleAutor abarnert
3

J'ai vu que le plus souvent écrit comme:
```
threads = [threading.Thread(target=f) for _ in range(8)]
for thread in threads:
    thread.start()
...
for thread in threads:
    thread.join()
```
Si vous voulez maintenir fixe la taille du pool de threads en cours d'exécution que les processus de courte durée des tâches que de demander de nouveaux travaux, envisager une solution construite autour de Files d'attente, comme "Comment faire pour attendre jusqu'à ce que seulement le premier thread est terminé en Python".
- J'ai mis à jour la question pour montrer que ce sont plutôt léger court-séjour tâches et j'ai besoin de garder en emparer.
InformationsquelleAutor Kirk Strauser
1

J'ai rencontré ce même problème et a passé des jours (2 jours pour être précis) arriver à la bonne solution à l'aide d'une file d'attente. J'ai perdu un jour de descendre le ThreadPoolExecutor chemin, car il n'y a aucun moyen de limiter le nombre de threads que chose lance! Je nourris une liste de 5000 fichiers à copier et le code non réactif, une fois qu'il a obtenu jusqu'à environ 1500 simultanées des copies de fichiers, exécution de tous à la fois. Le max_workers paramètre sur la ThreadPoolExecutor contrôle uniquement la façon dont beaucoup de travailleurs sont en rotation fils pas combien de fils se lancé.

Ok, de toute façon, voici un exemple très simple de l'utilisation d'une File d'attente pour cette:
```
import threading, time, random
from queue import Queue

jobs = Queue()

def do_stuff(q):
    while not q.empty():
        value = q.get()
        time.sleep(random.randint(1, 10))
        print(value)
        q.task_done()

for i in range(10):
    jobs.put(i)

for i in range(3):
    worker = threading.Thread(target=do_stuff, args=(jobs,))
    worker.start()

print("waiting for queue to complete", jobs.qsize(), "tasks")
jobs.join()
print("all done")
```
- Je ne pense pas que vous avez besoin de la time.sleep(random.randint(1, 10)) depuis le q.get() doit bloquer jusqu'à ce qu'il y a quelque chose dans la file d'attente
- Grâce Vasilis, cette fois.le sommeil n'est là que pour que, lorsque vous exécutez l'exemple, vous pouvez voir la file d'attente des tâches d'exécution de la commande pour illustrer le fait que les trois fils sont de l'exécution des tâches dont chacune a un inconnu de temps pour terminer. C'est exactement l'exemple que j'ai construit pour s'assurer que j'avais la logique de droit avant de l'appliquer à une file d'attente de milliers de tâches de copie de fichier.
- Vous Pouvez limiter le nombre de threads qu'il lance à la fois comme suit: ThreadPoolExecutor(max_workers=10) ou 20 ou 30 etc
- Divij, Le max_workers paramètre sur la ThreadPoolExecutor contrôle uniquement la façon dont beaucoup de travailleurs sont en rotation fils pas combien de fils se lancé. Si vous le réglez sur 1, puis vous obtenez à thread unique de la performance. Si vous le réglez à 2 et vous avez une file d'attente si plusieurs milliers de tâches longues, ces deux travailleurs commencent à tourner jusqu'threads et ne vous arrêtez pas jusqu'à ce qu'ils ont filé un thread pour chaque élément. Si ces tâches sont en concurrence pour les mêmes ressources comme la mémoire, de stockage, ou d'un réseau, vous avez un gros problème sur vos mains.
InformationsquelleAutor Paul Jacobs

-1

Pour appliquer limitation sur fil de la création, de suivre cet exemple (il fonctionne vraiment):

import threading
import time


def some_process(thread_num):
    count = 0
    while count < 5:
        time.sleep(0.5)
        count += 1
        print "%s: %s" % (thread_num, time.ctime(time.time()))
        print 'number of alive threads:{}'.format(threading.active_count())


def create_thread():
    try:
        for i in range(1, 555):  # trying to spawn 555 threads.
            thread = threading.Thread(target=some_process, args=(i,))
            thread.start()

            if threading.active_count() == 100:  # set maximum threads.
                thread.join()

            print threading.active_count()  # number of alive threads.

    except Exception as e:
        print "Error: unable to start thread {}".format(e)


if __name__ == '__main__':
    create_thread()

Ou:

Une autre façon de définir un nombre de thread vérificateur de mutex/verrouillage comme l'exemple ci-dessous:

import threading
import time


def some_process(thread_num):
    count = 0
    while count < 5:
        time.sleep(0.5)
        count += 1
        # print "%s: %s" % (thread_num, time.ctime(time.time()))
        print 'number of alive threads:{}'.format(threading.active_count())


def create_thread2(number_of_desire_thread ):
    try:
        for i in range(1, 555):
            thread = threading.Thread(target=some_process, args=(i,)).start()

            while number_of_desire_thread <= threading.active_count():
                '''mutex for avoiding to additional thread creation.'''
                pass

            print 'unlock'
            print threading.active_count()  # number of alive threads.

    except Exception as e:
        print "Error: unable to start thread {}".format(e)


if __name__ == '__main__':
    create_thread2(100)

InformationsquelleAutor Benyamin Jafari

Vous devez vous connecter pour publier un commentaire.