pyterminate

pyterminate

Führen Sie den Bereinigungscode nach Programmbeendigung zuverlässig aus.

Inhaltsverzeichnis

• Warum gibt es das?
• Was kann es tun?
• Schnellstart
• Tipps, Tricks und andere Hinweise
  • Doppelte Registrierung nach der Verzweigung
  • Multiprocessing-Startmethode

Warum gibt es das?

Derzeit gibt es in Python zwei integrierte Module zur Handhabung des Beendigungsverhaltens: atexit und signal . Ihre direkte Verwendung führt jedoch zu einer Menge wiederholtem Standardcode und einigen nicht offensichtlichen Verhaltensweisen, die leicht versehentlich falsch gemacht werden können, weshalb ich dieses Paket geschrieben habe.

Das atexit Modul reicht derzeit nicht aus, da es keine Signale verarbeiten kann. Das signal reicht derzeit nicht aus, da es normale oder durch Ausnahmen verursachte Exits nicht verarbeiten kann.

Typische Ansätze umfassen die häufig wiederholte Coderegistrierung einer Funktion sowohl bei atexit als auch bei gewünschten Signalen. Allerdings muss manchmal besonders darauf geachtet werden, sicherzustellen, dass die Funktion nicht zweimal ausgeführt wird (oder idempotent ist) und dass ein zuvor registrierter Signalhandler aufgerufen wird.

Was kann es tun?

Dieses Paket führt oder ermöglicht das folgende Verhalten:

• Registrieren Sie eine Funktion, die bei Programmbeendigung aufgerufen werden soll

  • Immer bei normaler oder ausnahmebedingter Beendigung: @ pyterminate .register
  • Konfigurierbar für beliebige Signale:
    @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))

• Ermöglicht die Registrierung mehrerer Funktionen

• Ruft zuvor registrierte Signalhandler auf

• Ermöglicht Null- oder Nicht-Null-Exit-Codes für erfasste Signale:
  @ pyterminate .register(successful_exit=True)

• Ermöglicht das Unterdrücken oder Auslösen von KeyboardInterrupt bei SIGINT :
  @ pyterminate .register(keyboard_interrupt_on_sigint=True)

  • Möglicherweise möchten Sie einen KeyboardInterrupt auslösen, wenn eine zusätzliche Ausnahmebehandlung definiert ist.

• Ermöglicht die Aufhebung der Registrierung von Funktionen: pyterminate .unregister(func)

• Ignorieren Sie angeforderte Signale, während die registrierte Funktion ausgeführt wird, und stellen Sie sicher, dass sie nicht unterbrochen wird.

  • Es ist wichtig zu beachten, dass SIGKILL und Aufrufe von os._exit() nicht ignoriert werden können.

Schnellstart

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 , ...)

Tipps, Tricks und andere Hinweise

Doppelte Registrierung nach der Verzweigung

Da beim Erstellen eines neuen Prozesses durch Forking der gesamte Prozess dupliziert wird, werden alle zuvor registrierten Funktionen auch im geforkten Prozess registriert. Dies ist eine offensichtliche Folge der Verzweigung, aber es ist wichtig zu berücksichtigen, ob die registrierten Funktionen auf gemeinsam genutzte Ressourcen zugreifen. Um dieses Verhalten zu vermeiden, können Sie die Registrierung der Funktion zu Beginn des gespaltenen Prozesses aufheben, ein Gate basierend auf der Prozess-ID erstellen oder eine andere geeignete Synchronisierungsmethode verwenden.

Multiprocessing-Startmethode

Beim Starten von Prozessen mit multiprocessing Modul von Python kann die fork -Methode beim Beenden keine registrierten Funktionen aufrufen, da der Prozess intern mit os._exit() beendet wird, was alle Bereinigungen umgeht und den Prozess sofort beendet.

Eine Möglichkeit, dies zu umgehen, ist die Verwendung der Startmethode "spawn" sofern dies für Ihre Anwendung akzeptabel ist. Eine andere Methode besteht darin, Ihre Funktion bei einem benutzerdefinierten Signal zu registrieren und Ihren Prozesscode in einen Try-Except-Block einzuschließen, wodurch am Ende das benutzerdefinierte Signal ausgelöst wird. pyterminate stellt diese Funktionalität in Form des Dekorators exit_with_signal bereit, der die dekorierte Funktion einfach in einen try-finally-Block einschließt und das gegebene Signal auslöst. Beispielverwendung:

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