Multiprocessing sur une fonction avec un timeout

Version imprimable

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:

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

07/05/2021, 14h20
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

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:

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

07/05/2021, 18h46
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

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

Code:

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

07/05/2021, 20h28
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 :aie:.
07/05/2021, 23h31
wiztricks

Salut,

On ne peut pas intégrer n'importe quelle bidouille dans un cadre sans connaître les limites qu'il impose. C'est comme çà.

- W
09/05/2021, 20h37
radjiv97441

J'ai oublié de relancer restart.txt (relancer le serveur)