Discussion :

[rambc]

Bonjour.

Je dois utiliser des bibliothèques que je n'ai pas programmées. Certaines fonctions peuvent prendre un temps rédhibitoire dans des cas défavorables. Je voudrais donc les utiliser avec un temps maximum d'exécution. Est-ce faisable ?

On peut se baser sur l'ECM suivant.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
import time
 
def infiniteLoop():
    for i in range (5000):
        time.sleep(60000)
 
infiniteLoop() # Comment stopper inifiteLoop() au bout de 15s par exemple ?
               # Sans toucher au code de infiniteLoop().

[josmiley]

en l'exécutant dans un subprocess ?

[tyrtamos]

Bonjour,

Comme j'avais le problème avec une calculatrice qui calculait dans un thread, je vais te donner ma solution: utiliser sys.settrace.

sys.settrace est une fonction de debugging, qui arrête chaque ligne de code Python pour insérer un test. Il faut donc que ce soit une fonction Python (et pas 'C'), et il ne faut pas qu'une seule ligne de code dure trop longtemps (c'est pour ça que j'ai modifié ta fonction: sleep(60000) durait trop longtemps).

Voilà un petit exemple:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
#!/usr/bin/python
# -*- coding: utf-8 -*-
from __future__ import division
# Python 2.7
 
import sys
import threading
from time import clock, sleep
 
###################################################
def infiniteLoop(x, y, z="toto"):
    for i in range (5000000):
        sleep(0.1)
 
###################################################
class Execute(threading.Thread):
 
    def __init__(self, tempsmax, fonc, *args, **kwargs):
        threading.Thread.__init__(self)
 
        self.tempsmax = tempsmax
        self.fonc = fonc
        self.args = args
        self.kwargs = kwargs        
 
        # initialise le drapeau qui permettra l'arrêt sur demande
        self.stopexec=False
 
    def run(self):
        """Lance l'exécution en tâche de fond"""
        try:
            # mise en place du traçage
            sys.settrace(self.trace)
            # initialisation du temps d'exécution
            self.temps = clock()
            # exécution de la fonction demandée avec ses arguments
            self.fonc(*self.args, **self.kwargs)
            # arrêt du traçage
            sys.settrace(None)
        except ValueError:
            # arrêt du traçage en cas d'exception
            sys.settrace(None)
 
    def trace(self, frame, event, arg):
        """méthode appelée à chaque ligne de code Python exécutée"""
        if event == 'line':
            if self.stopexec or (clock()-self.temps)>self.tempsmax:
                raise  ValueError (u"Arrêt demandé")
        return self.trace
 
    def stop(self):
        """ méthode appelée pour arrêter le thread sur demande"""
        self.stopexec = True
 
###################################################
if __name__ == "__main__":
 
    prog = Execute(15, infiniteLoop, 10, 20, z="titi")
    prog.daemon = True
    prog.start()
    t = clock()
    prog.join()
    print u"Arrêt au bout de", clock()-t, u"sec."

On lance le thread avec: le temps maxi, la fonction à exécuter et ses arguments. Si la fonction dure plus longtemps que le temps maxi, elle est arrêtée avec une exception. Mais on peut aussi l'arrêter de l'extérieur avec la méthode stop().

D'après mon expérience, le traçage par sys.settrace est très rapide.

En dehors de cette solution, il y a, bien entendu, la mise en processus et son arrêt "sauvage" de l'extérieur.

[edit] pour les threads, il y a aussi: threading.settrace(func), mais je ne l'ai jamais utilisé: peut-être faut-il aussi chercher dans cette direction?

[dividee]

D'après un code trouvé ici, qui permet de lever une exception dans un thread:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from functools import wraps
import threading
import time
import inspect
import ctypes
 
def _async_raise(tid, exctype):
    '''Raises an exception in the threads with id tid'''
    if not inspect.isclass(exctype):
        raise TypeError("Only types can be raised (not instances)")
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        # "if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, 0)
        raise SystemError("PyThreadState_SetAsyncExc failed")
 
def timeout(maxtime):
    def deco(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            t = threading.Thread(target=f, args=args, kwargs=kwargs)
            t.start()
            def kill_thread():
                while t.is_alive():
                    _async_raise(t.ident, SystemExit)
                    time.sleep(0.1)
            timer = threading.Timer(maxtime, kill_thread)
            timer.start()
            t.join()
        return wrapper
    return deco
 
@timeout(5)
def infiniteLoop(l):
    while len(l)>1:
        del l[0]
 
l = range(1000000)
t0 = time.clock()
infiniteLoop(l)
print "Elapsed: %.1f s (looped %d times)" % (time.clock() - t0, l[0])

C'est certainement pas 100% robuste, mais cela semble fonctionner...

[EDIT:] Version 2; le code initial était mauvais. Attention au sleep, qui risque d'avaler l'exception...

[tyrtamos]

Bonjour,

Et voilà un exemple avec le module multiprocessing, dont le fonctionnement s'inspire de threading, en beaucoup plus puissant (et compliqué...).

Avec multiprocessing.Pool, on a une solution simple pour recevoir le résultat d'une fonction sans modifier son code (ce qui n'est pas le cas avec Pipe, Queue, Manager, ...).

Si le temps maxi est dépassé, on génère ici une exception "multiprocessing.TimeoutError" que l'appelant doit gérer.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#!/usr/bin/python
# -*- coding: utf-8 -*-
from __future__ import division
# Python 2.7
 
import time
import multiprocessing
 
#############################################################################
def calcul(x, y):
    return x*y
 
#############################################################################
def infiniteLoop():
    for i in range (5000000):
        time.sleep(0.1)
 
#############################################################################
def exectimeout(timeout, fonc, *args, **kwargs):
    """exécute fonc(args, kwargs), avec le temps maxi timeout"""
    pool = multiprocessing.Pool(processes=1) # crée 1 processus
    result = pool.apply_async(func=fonc, *args, **kwargs) # eval. asynchrone
    try:
        return result.get(timeout=timeout) 
    except multiprocessing.TimeoutError:
        raise # renvoi de l'exception à l'appelant
 
#############################################################################
if __name__ == '__main__':
 
    # essai avec une fonction AVEC résultat
    t = time.clock()
    try:
        result = exectimeout(15, calcul, args=(2,3))
        print u"Résultat:", result
    except multiprocessing.TimeoutError:
        result = None
        print u"Arrêt au bout de %.3f sec." % (time.clock()-t,)
 
    # essai avec une fonction SANS résultat (la solution précédente marche aussi!)
    t = time.clock()
    try:
        exectimeout(15, infiniteLoop)
    except multiprocessing.TimeoutError:
        print u"Arrêt au bout de %.3f sec." % (time.clock()-t,)

[dividee]

Bonjour tyrtamos,

Il y a un problème, il me semble, avec ta solution: le processus qui effectue le travail continue de tourner en asynchrone; il n'est pas interrompu par le timeout; c'est seulement l'attente du get qui est interrompue. En fait, j'avais écrit une première version avec des threads qui avait le même problème, mais j'ai supposé qu'ici on préférait "tuer" les calculs trop longs, plutôt que de les laisser continuer en tâche de fond pour rien.

Cela dit, ce problème peut être corrigé en tuant le processus, et c'est sans doute plus fiable que de tuer un thread.

[tyrtamos]

Bonjour dividee,

Effectivement, j'avais repris un ancien code. J'essayais alors, tant avec les threads qu'avec les processus, de les utiliser sur un flux de calculs sans les arrêter pour gagner sur le temps passé à les démarrer et les arrêter.

Voilà le code qui arrête le processus:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
def exectimeout(timeout, fonc, *args, **kwargs):
    """exécute fonc(args, kwargs), avec le temps maxi timeout"""
    pool = multiprocessing.Pool(processes=1) # crée 1 processus
    poolexec = pool.apply_async(func=fonc, *args, **kwargs) # eval. asynchrone
    try:
        result = poolexec.get(timeout=timeout)
        pool.terminate()
        return result
    except multiprocessing.TimeoutError:
        pool.terminate() 
        raise # renvoi de l'exception à l'appelant

[rambc]

Bonjour et merci pour ces propositions. Comme le module que j'utilise est basé
sur des bibliothèques C, l'utilisation de thread tombe à point.