Le principe est simple: dans la boucle de la méthode run, on teste une variable booléenne initialisée à False. Quand elle passe à True grâce à l'appel de la méthode stopthread(), une exception est déclenchée par raise qui arrête run, et donc le thread. Dans l'exemple donné, l'arrêt est demandé au bout de 5 secondes.
#!/usr/bin/python # -*- coding: utf-8 -*- # importations fonctionnellement nécessaires import threading import sys # importation seulement nécessaire pour l'exemple import time ############################################################################## class Monthread(threading.Thread): def __init__(self): threading.Thread.__init__(self) self.arret=False # =variable pour porter l'ordre d'arrêter le thread def run(self): try: # ... while True: if self.arret: raise ValueError ("arrêt demandé") # ... # ... except: # ... # récupération et impression du message d'erreur print u"%s" % sys.exc_info()[1] # ... # fin de run, donc fin du thread def stopthread(self): self.arret=True ############################################################################## print "Lancement du thread" app=Monthread() app.start() tps=time.time() while app.isAlive(): if time.time()-tps > 5: # arrêt du thread au bout de 5 secondes app.stopthread() time.sleep(0.1) print "fin du thread" time.sleep(2)
L'instruction sys.settrace() est utilisée dans le debugging. Son principe est de détourner l'exécution à chaque ligne d'instruction exécutée en lançant une fonction qui peut faire des tests.
J'utilisais surtout la méthode précédente, mais elle ne s'applique pas pour arrêter, par exemple, un eval() parce qu'il n'y a pas de boucle pour insérer un test.
Quand on a compris le principe, le code est très simple: on encadre la partie à surveiller par la mise en place d'une fonction de traçage et sa désactivation. La fonction de traçage comporte un test sur un drapeau d'arrêt qui, s'il est True, va interrompre le déroulement normal du thread par le lancement d'une exception.
De l'extérieur du thread, il suffit d'appeler sa méthode stopthread() pour déclencher son arrêt.
On a encore 2 solutions:
Attention: ce traçage ne fonctionne que dans les instructions des fonctions appelées, et pas dans les instructions qui seraient directement placées dans la méthode run.
Et cette méthode n'est pas compatible avec certains outils de développements qui ont déjà une fonction trace pour debugging (comme IDLE).
#!/usr/bin/python # -*- coding: utf-8 -*- from __future__ import division # importations fonctionnellement nécessaires import threading import sys # importations seulement nécessaires pour l'exemple import time from math import * ############################################################################## def bidon(): i=1L while i<10000000000L: x=sqrt(i) i+=1 return x ############################################################################## class Monthread(threading.Thread): def __init__(self): threading.Thread.__init__(self) # création de la variable qui portera l'ordre d'arrêter le thread self.arretthread=False def run(self): try: # met en place le traçage par la fonction testarret() sys.settrace(self.trace) #... #... # ce traçage ne fonctionne que sur les fonctions appelées self.result=bidon() #... #... # arrêt normal du traçage sys.settrace(None) except: # arrêt du traçage en cas d'exception sys.settrace(None) # ... # récupération du message d'erreur self.result = u"%s" % sys.exc_info()[1] # ... print self.result # fin de run, donc fin du thread def trace(self, frame, event, arg): if event=='line': if self.arretthread: raise ValueError ("Arrêt du thread demandé") # autre option possible: raise SystemExit() return self.trace def stopthread(self): self.arretthread=True ############################################################################## app=Monthread() app.start() tps=time.clock() while app.isAlive(): # arrêt du thread au bout de 5 secondes if (time.clock()-tps) > 5.0: app.stopthread() # tempo pour laisser la priorité au thread time.sleep(0.1)
On commence par créer un nouveau module appelé ici “threadatuer” (fichier threadatuer.py) comportant une nouvelle classe dérivée de threading.Thread dotée d'une méthode “tuer()”:
#!/usr/bin/python # -*- coding: utf-8 -*- # source: inspiré de: http://mail.python.org/pipermail/python-list/2004-May/260937.html import sys import threading ############################################################################## class Threadatuer(threading.Thread): """Sous-classe de threading.Thread, avec une methode tuer()""" def __init__(self, *args, **keywords): threading.Thread.__init__(self, *args, **keywords) self.atuer = False def start(self): self.run_sav = self.run self.run = self.run2 threading.Thread.start(self) def run2(self): sys.settrace(self.trace) self.run_sav() self.run = self.run_sav def trace(self, frame, event, arg): if self.atuer: if event == 'line': raise SystemExit() return self.trace def tuer(self): self.atuer = True
Exemple d'utilisation: après importation du module threadatuer, on utilise désormais threadatuer.Threadatuer au lieu de threading.Thread pour créer les classes thread.
Ce qui fait qu'on peut désormais tuer le thread actif à n'importe quel moment simplement en appelant sa nouvelle méthode .tuer().
Dans l'exemple ci-dessous, on tue le thread au bout de 5 secondes.
#!/usr/bin/python # -*- coding: utf-8 -*- # importation fonctionnellement nécessaire import threadatuer # importation seulement nécessaire pour l'exemple import time ############################################################################## class Monthread(threadatuer.Threadatuer): def __init__(self): threadatuer.Threadatuer.__init__(self) def run(self): # ... k=0 while True: print "k= ",k k+=1 # ... # fin de run, donc fin du thread ############################################################################## print "lancement du thread" app=Monthread() app.start() tps=time.clock() while app.isAlive(): if (time.clock()-tps) > 5.0: # arrêt du thread au bout de 5 secondes app.tuer() print "fin du thread" time.sleep(2)
Attention: ce n'est pas compatible avec les outils de développement qui utilisent déjà leur propre trace (comme IDLE).
J'ai vérifié que ça fonctionnait sur Windows XP et sur Linux (opensuse 10.3).
Avec les 2 méthodes précédentes basées sur la fonction sys.settrace(), si le fonctionnement n'est pas celui attendu, il y a une possibilité d'ajouter à la fonction de traçage des instructions pour examiner la liste des instructions interrompues.
Voilà le code proposé. J'ai seulement repris la fonction trace. Il est de plus nécessaire d'importer le module linecache.
Je l'ai emprunté à l'excellent article ici: http://www.dalkescientific.com/writings/diary/archive/2005/04/20/tracing_python_code.html
import linecache def trace(self, frame, event, arg): if event=='line': lineno = frame.f_lineno filename = frame.f_globals["__file__"] if filename == "<stdin>": filename = "traceit.py" if (filename.endswith(".pyc") or filename.endswith(".pyo")): filename = filename[:-1] name = frame.f_globals["__name__"] line = linecache.getline(filename, lineno) print "%s:%s: %s" % (name, lineno, line.rstrip()) if self.arretthread: raise ValueError ("Arrêt du thread demandé") # autre option possible: raise SystemExit() return self.trace
Bien sûr, le “print” va écrire dans la console, même si le programme en exécution est en graphique avec tkinter.
Par exemple, dans la méthode numéro 2, l'utilisation de cette option donnera les sorties console suivantes:
__main__:16: i=1L __main__:17: while i<10000000000L: __main__:18: x=sqrt(i) __main__:19: i+=1 __main__:18: x=sqrt(i) __main__:19: i+=1 .... __main__:18: x=sqrt(i) __main__:19: i+=1 Arrêt du thread demandé