pyterminate

pyterminate

Exécutez de manière fiable le code de nettoyage à la fin du programme.

Table des matières

• Pourquoi est-ce que cela existe ?
• Que peut-il faire ?
• Démarrage rapide
• Trucs, astuces et autres notes
  • Inscription en double après fork
  • Méthode de démarrage multitraitement

Pourquoi est-ce que cela existe ?

Il existe actuellement deux modules intégrés pour gérer le comportement de terminaison en Python : atexit et signal . Cependant, leur utilisation conduit directement à de nombreux codes passe-partout répétés et à certains comportements non évidents qui peuvent facilement se tromper accidentellement, c'est pourquoi j'ai écrit ce package.

Le module atexit est actuellement insuffisant car il ne parvient pas à gérer les signaux. Le module signal est actuellement insuffisant car il ne parvient pas à gérer les sorties normales ou provoquées par des exceptions.

Les approches typiques incluraient un code fréquemment répété enregistrant une fonction à la fois avec atexit et sur des signaux souhaités. Cependant, des précautions supplémentaires doivent parfois être prises pour garantir que la fonction ne s'exécute pas deux fois (ou qu'elle est idempotente) et qu'un gestionnaire de signal précédemment enregistré soit appelé.

Que peut-il faire ?

Ce package effectue ou autorise le comportement suivant :

• Enregistrez une fonction à appeler à la fin du programme

  • Toujours en terminaison normale ou provoquée par une exception : @ pyterminate .register
  • Configurable pour tous les signaux souhaités :
    @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))

• Permet d'enregistrer plusieurs fonctions

• Appellera les gestionnaires de signaux enregistrés précédemment

• Permet des codes de sortie nuls ou non nuls sur les signaux capturés :
  @ pyterminate .register(successful_exit=True)

• Permet de supprimer ou de lancer KeyboardInterrupt sur SIGINT :
  @ pyterminate .register(keyboard_interrupt_on_sigint=True)

  • Vous souhaiterez peut-être lancer un KeyboardInterrupt si une gestion des exceptions supplémentaire est définie.

• Permet de désenregistrer les fonctions : pyterminate .unregister(func)

• Ignorez les signaux demandés pendant l'exécution de la fonction enregistrée, en vous assurant qu'elle n'est pas interrompue.

  • Il est important de noter que SIGKILL et les appels à os._exit() ne peuvent pas être ignorés.

Démarrage rapide

python3 -m pip install pyterminate 

pyterminate @ pyterminate .register( args=(None,), kwargs={"b": 42}, signals=(signal.SIGINT, signal.SIGTERM), successful_exit=True, keyboard_interrupt_on_sigint=True ) def cleanup(*args, **kwargs): ... # or pyterminate .register(cleanup, ...)">

 import signal

import pyterminate


@ pyterminate . register (
    args = ( None ,),
    kwargs = { "b" : 42 },
    signals = ( signal . SIGINT , signal . SIGTERM ),
    successful_exit = True ,
    keyboard_interrupt_on_sigint = True
)
def cleanup ( * args , ** kwargs ):
    ...

# or

pyterminate . register ( cleanup , ...)

Trucs, astuces et autres notes

Inscription en double après fork

Étant donné que la création d'un nouveau processus via le fork duplique l'ensemble du processus, toutes les fonctions précédemment enregistrées seront également enregistrées dans le processus forké. Il s'agit d'une conséquence évidente du fork, mais il est important de considérer si les fonctions enregistrées accèdent à des ressources partagées. Pour éviter ce comportement, vous pouvez désenregistrer la fonction au début du processus forké, en fonction de l'ID du processus, ou utiliser toute autre méthode de synchronisation appropriée.

Méthode de démarrage multitraitement

Lors du démarrage de processus avec le module multiprocessing de Python, la méthode fork ne parviendra pas à appeler les fonctions enregistrées à la sortie, car le processus se termine avec os._exit() en interne, ce qui contourne tout nettoyage et tue immédiatement le processus.

Une façon de contourner ce problème consiste à utiliser la méthode de démarrage "spawn" si cela est acceptable pour votre application. Une autre méthode consiste à enregistrer votre fonction sur un signal défini par l'utilisateur et à envelopper votre code de processus dans un bloc try-sauf, en augmentant le signal défini par l'utilisateur à la fin. pyterminate fournit cette fonctionnalité sous la forme du décorateur exit_with_signal , qui enveloppe simplement la fonction décorée dans un bloc try-finally et déclenche le signal donné. Exemple d'utilisation :

pyterminate @ pyterminate .exit_with_signal(signal.SIGUSR1) def run_process(): @ pyterminate .register(signals=[signal.SIGUSR1, signal.SIGINT, signal.SIGTERM]) def cleanup(): ... ... if __name__ == "__main__" mp.set_start_method("fork") proc = mp.Process(target=run_process) proc.start() try: proc.join(timeout=300) except TimeoutError: proc.terminate() proc.join()">

 import multiprocessing as mp
import signal

import pyterminate


@ pyterminate . exit_with_signal ( signal . SIGUSR1 )
def run_process ():

    @ pyterminate . register ( signals = [ signal . SIGUSR1 , signal . SIGINT , signal . SIGTERM ])
    def cleanup ():
        ...

    ...


if __name__ == "__main__"
    mp . set_start_method ( "fork" )

    proc = mp . Process ( target = run_process )
    proc . start ()

    try :
        proc . join ( timeout = 300 )
    except TimeoutError :
        proc . terminate ()
        proc . join ()