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.

Quand on a compris, le code est très simple: on encadre la partie à examiner par une 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.

#!/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).