Discussion :

[radjiv97441]

Bonjour étant une fonction test à appeler (peu importe le contenu), comment puis-je arrêter son exécution s'il excède 1 seconde.
Le test que j'ai réalisé, n'affiche pas la valeur de x. J'aimerais que son exécution s’exécute pendant une seconde au maximum.

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
 
import multiprocessing
import time
 
def mafonction():
   x=1
   while x<2:
       print(x)
 
if __name__ == '__main__':
    print("début:")
    p = multiprocessing.Process(target=mafonction)
    p.start()
    time.sleep(1)
    p.terminate()
    p.join()
    print(":Fin")

[wiztricks]

Salut,

Logiquement votre code devrait fonctionner et il affiche x chez moi.
Reste à connaître votre environnement système, la version de Python,...


- W

[radjiv97441]

non ça ne marche pas : je suis sur spyder (installé il y a quelques jour) pour les tests et
sur monn serveur linux (python 3.8) en ligne

J'ai essayé ce code qui semble marché:

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
 
import ctypes
import threading
import time
 
def f(u):
    x=u
    while u<10:
        u=u-1
    return u
 
def async_raise(thread_obj, exception):
    target_tid = thread_obj.ident
    if target_tid not in {thread.ident for thread in threading.enumerate()}:
        return ["?","ok"]
    else:
        affected_count = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(target_tid), ctypes.py_object(exception))
        if affected_count == 0:
            return ["?","ok"]
        elif affected_count > 1:
            ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(target_tid), ctypes.c_long(0))
            return ["?","no"]
        else:
            return ["?","no"]
 
def executeprocess():
    t = threading.Thread(target=f,args=(2,))
    t.start() 
    time.sleep(0.1) #laisse tourner 100 ms
    if (t.isAlive()):time.sleep(3) #laisse tourner 3s
    if (t.isAlive()):
            return async_raise(t, SystemExit)
    else: return ["?","ok"]
print(executeprocess())

Le seul soucis c'est que je n'arrive pas à extraire la valeur de retour, quoi mettre à "?"
Il y a peut-être plus simple avec la bibliothèque asyncio mais je n'arrive pas à l'utiliser

[wiztricks]

Salut,

Il faut attendre que quelqu'un prenne le temps de tester sur une version de Linux/Python semblable. Techniquement, je ne vois pas ce qui empêcherait ce code de fonctionner et il fonctionne sur mes machines Windows et Linux.

Après si vous remplacer multiprocessing par thread et que vous ajoutez une dose d'asyncio par dessus, vous changez de problème et vous devriez ouvrir une nouvelle discussion.

Ceci dit contrôler l'exécution d'une fonction avec un timeout est un sujet où vous avez plein d'articles sur Internet. Et si vous utilisez linux, les signaux suffisent.

- W

[radjiv97441]

ah, voilà : merci pour le mot clé: signal

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
 
import signal
import time
 
def timeout_handler(num, stack):
    raise Exception("LOOP")
 
def long_function():
    x=5
    while x>2:
        x=x+1
    return x
def test():
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(2)
    result=""
    erreur="no"
    try:
        result=long_function()
        erreur="ok"
    except Exception:
        pass
    finally:
        signal.alarm(0)
        return [result,erreur]
print(test())

[radjiv97441]

Mince sur le serveur Linux Python le signal se déclenche sur un exemple d'essai sur le terminal, mais pas dans mon programme complexe appelé depuis passanger_wsgi.py (environnement virtuel) via le net. Pas d'erreur dans le log non plus. Pour un gros calcul il y va jusqu'au bout (plus 2 min) et pour une boucle infinie il plante .