pyterminate

pyterminate

Execute o código de limpeza de forma confiável após o encerramento do programa.

Índice

• Por que isso existe?
• O que isso pode fazer?
• Início rápido
• Dicas, truques e outras notas
  • Registro duplicado após bifurcação
  • Método de início de multiprocessamento

Por que isso existe?

Atualmente existem dois módulos integrados para lidar com o comportamento de encerramento em Python: atexit e signal . No entanto, usá-los diretamente leva a muitos códigos clichê repetidos e a alguns comportamentos não óbvios que podem ser fáceis de errar acidentalmente, e é por isso que escrevi este pacote.

O módulo atexit é atualmente insuficiente, pois não consegue lidar com sinais. O módulo signal é atualmente insuficiente, pois não consegue lidar com saídas normais ou causadas por exceção.

Abordagens típicas incluiriam código frequentemente repetido registrando uma função tanto com atexit quanto em sinais desejados. No entanto, às vezes é necessário tomar cuidado extra para garantir que a função não seja executada duas vezes (ou seja idempotente) e que um manipulador de sinal registrado anteriormente seja chamado.

O que isso pode fazer?

Este pacote faz ou permite o seguinte comportamento:

• Registre uma função a ser chamada no encerramento do programa

  • Sempre em encerramento normal ou causado por exceção: @ pyterminate .register
  • Configurável para quaisquer sinais desejados:
    @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))

• Permite registrar múltiplas funções

• Chamará manipuladores de sinal registrados anteriormente

• Permite códigos de saída zero ou diferentes de zero em sinais capturados:
  @ pyterminate .register(successful_exit=True)

• Permite suprimir ou lançar KeyboardInterrupt no SIGINT :
  @ pyterminate .register(keyboard_interrupt_on_sigint=True)

  • Você pode lançar um KeyboardInterrupt se houver tratamento de exceção adicional definido.

• Permite que funções sejam canceladas: pyterminate .unregister(func)

• Ignore os sinais solicitados enquanto a função registrada estiver em execução, garantindo que não seja interrompida.

  • É importante observar que SIGKILL e chamadas para os._exit() não podem ser ignoradas.

Início rápido

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

Dicas, truques e outras notas

Registro duplicado após bifurcação

Como a criação de um novo processo por meio de bifurcação duplica todo o processo, quaisquer funções registradas anteriormente também serão registradas no processo bifurcado. Esta é uma consequência óbvia da bifurcação, mas é importante considerar se as funções registradas estão acessando recursos compartilhados. Para evitar esse comportamento, você pode cancelar o registro da função no início do processo bifurcado, bloquear com base no ID do processo ou usar qualquer outro método de sincronização apropriado.

Método de início de multiprocessamento

Ao iniciar processos com o módulo de multiprocessing do Python, o método fork falhará ao chamar funções registradas na saída, pois o processo é finalizado com os._exit() internamente, o que ignora toda a limpeza e mata imediatamente o processo.

Uma maneira de contornar isso é usar o método de início "spawn" se for aceitável para sua aplicação. Outro método é registrar sua função em um sinal definido pelo usuário e agrupar o código do processo no bloco try-except, gerando o sinal definido pelo usuário no final. pyterminate fornece essa funcionalidade na forma do decorador exit_with_signal , que simplesmente envolve a função decorada em um bloco try-finally e gera o sinal fornecido. Exemplo de uso:

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