Outils pour utilisateurs

Outils du site


arret_thread

Comment arrêter un thread?

1- Méthode avec le déclenchement d'une exception


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)


2- Méthode avec sys.settrace

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:

  • soit l'arrêt génère une exception “raise” avec message, auquel cas on peut encore récupérer l'erreur avec un try: .. except: avant de quitter le thread
  • soit on veut vraiment arrêter le thread, auquel cas on génère une exception “SystemExit()” et le thread s'arrête (sans try:..except:).

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)


3- Méthode avec une classe dérivée de threading.Thread utilisée comme nouveau modèle


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

Option pour debugging

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é

arret_thread.txt · Dernière modification: 2008/04/12 08:45 de tyrtamos

Outils de la page