Terminer un programme python multithread

Question

Comment faire réagir un programme python multithread à l'événement Ctrl+C ?

Edita: Le code est le suivant :

import threading
current = 0

class MyThread(threading.Thread):
    def __init__(self, total):
        threading.Thread.__init__(self)
        self.total = total

    def stop(self):
        self._Thread__stop()

    def run(self):
        global current
        while current<self.total:
            lock = threading.Lock()
            lock.acquire()
            current+=1
            lock.release()
            print current

if __name__=='__main__':

    threads = []
    thread_count = 10
    total = 10000
    for i in range(0, thread_count):
        t = MyThread(total)
        t.setDaemon(True)
        threads.append(t)
    for i in range(0, thread_count):
        threads[i].start()

J'ai essayé de supprimer join() sur tous les threads mais cela ne fonctionne toujours pas. Est-ce parce que le segment de verrouillage se trouve à l'intérieur de la procédure run() de chaque thread ?

Edita: Le code ci-dessus est censé fonctionner, mais il s'interrompt toujours lorsque la variable courante se situe dans la plage 5 000-6 000 et génère les erreurs suivantes

Exception in thread Thread-4 (most likely raised during interpreter shutdown):
Traceback (most recent call last):
  File "/usr/lib/python2.5/threading.py", line 486, in __bootstrap_inner
  File "test.py", line 20, in run
<type 'exceptions.TypeError'>: unsupported operand type(s) for +=: 'NoneType' and 'int'
Exception in thread Thread-2 (most likely raised during interpreter shutdown):
Traceback (most recent call last):
  File "/usr/lib/python2.5/threading.py", line 486, in __bootstrap_inner
  File "test.py", line 22, in run

Answer 1

Faire de chaque thread, à l'exception du thread principal, un démon ( t.daemon = True en 2.6 ou plus, t.setDaemon(True) en 2.6 ou moins, pour chaque objet thread t avant de le démarrer). De cette façon, lorsque le thread principal reçoit l'interruption du clavier, s'il ne l'attrape pas ou s'il l'attrape mais décide de se terminer quand même, le processus entier se terminera. Voir les docs .

éditer : après avoir vu le code de l'OP (non posté à l'origine) et l'affirmation selon laquelle "il ne fonctionne pas", il semble que je doive ajouter... :

Bien entendu, si vous souhaitez que votre thread principal reste réactif (par exemple, à la commande C), ne l'embourbez pas dans des appels bloquants, tels que join de discussion -- et surtout pas d'une manière totalement inutile les appels de blocage, tels que join ingurgiter démon fils. Par exemple, il suffit de modifier la boucle finale dans le fil principal par rapport à la boucle actuelle (inutile et nuisible) :

for i in range(0, thread_count):
    threads[i].join()

à quelque chose de plus raisonnable comme :

while threading.active_count() > 0:
    time.sleep(0.1)

si votre système principal n'a rien de mieux à faire que d'attendre que tous les threads se terminent d'eux-mêmes ou qu'un contrôle-C (ou un autre signal) soit reçu.

Bien sûr, il existe de nombreux autres modèles utilisables si vous préférez que vos threads ne se terminent pas brusquement (comme peuvent le faire les threads démoniaques) -- à moins que ils sont eux aussi embourbés à jamais dans des appels inconditionnellement bloquants, des blocages, etc;-).

Answer 2

Il existe deux méthodes principales, l'une propre et l'autre facile.

La méthode la plus simple est d'attraper KeyboardInterrupt dans votre thread principal, et de définir un drapeau que vos threads d'arrière-plan peuvent vérifier afin de savoir s'ils doivent sortir ; voici une version simple/légèrement maladroite utilisant un global :

exitapp = False
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        exitapp = True
        raise

def threadCode(...):
    while not exitapp:
        # do work here, watch for exitapp to be True

La méthode la plus simple mais la plus compliquée consiste à attraper KeyboardInterrupt et à appeler os._exit(), ce qui met fin à tous les threads immédiatement.

Answer 3

A Travailleur pourrait vous être utile :

#!/usr/bin/env python

import sys, time
from threading import *
from collections import deque

class Worker(object):
    def __init__(self, concurrent=1):
        self.concurrent = concurrent
        self.queue = deque([])
        self.threads = []
        self.keep_interrupt = False

    def _retain_threads(self):
        while len(self.threads) < self.concurrent:
            t = Thread(target=self._run, args=[self])
            t.setDaemon(True)
            t.start()
            self.threads.append(t)

    def _run(self, *args):
        while self.queue and not self.keep_interrupt:
            func, args, kargs = self.queue.popleft()
            func(*args, **kargs)

    def add_task(self, func, *args, **kargs):
        self.queue.append((func, args, kargs))

    def start(self, block=False):
        self._retain_threads()

        if block:
            try:
                while self.threads:
                    self.threads = [t.join(1) or t for t in self.threads if t.isAlive()]
                    if self.queue:
                        self._retain_threads()
            except KeyboardInterrupt:
                self.keep_interrupt = True
                print "alive threads: %d; outstanding tasks: %d" % (len(self.threads), len(self.queue))
                print "terminating..."

# example
print "starting..."
worker = Worker(concurrent=50)

def do_work():
    print "item %d done." % len(items)
    time.sleep(3)

def main():
    for i in xrange(1000):
        worker.add_task(do_work)
    worker.start(True)

main()
print "done."

# to keep shell alive
sys.stdin.readlines()

Answer 4

Je préfère opter pour le code proposé dans cet article de blog :

def main(args):

    threads = []
    for i in range(10):
        t = Worker()
        threads.append(t)
        t.start()

    while len(threads) > 0:
        try:
            # Join all threads using a timeout so it doesn't block
            # Filter out threads which have been joined or are None
            threads = [t.join(1000) for t in threads if t is not None and t.isAlive()]
        except KeyboardInterrupt:
            print "Ctrl-c received! Sending kill to threads..."
            for t in threads:
                t.kill_received = True

Ce que j'ai changé, c'est le t.joindre de t.join(1) a t.join(1000) . Le nombre réel de secondes n'a pas d'importance, à moins que vous ne spécifiiez un délai d'attente, le thread principal restera réactif à Ctrl+C. L'exception sur Interruption du clavier rend la gestion des signaux plus explicite.

Answer 5

Si vous créez un fil comme suit - myThread = Thread(target = function) - puis faire myThread.start(); myThread.join() . Lorsque CTRL-C est lancé, le thread principal ne se termine pas parce qu'il est en attente de ce thread bloquant. myThread.join() appel. Pour remédier à ce problème, il suffit d'ajouter un délai d'attente à l'appel .join(). Ce délai peut être aussi long que vous le souhaitez. Si vous voulez qu'il attende indéfiniment, il suffit de mettre un délai très long, comme 99999. C'est aussi une bonne pratique de faire myThread.daemon = True de sorte que tous les threads se terminent lorsque le thread principal (non-daemon) se termine.