pyterminate

pyterminate

프로그램 종료 시 정리 코드를 안정적으로 실행합니다.

목차

• 이것이 존재하는 이유는 무엇입니까?
• 그것은 무엇을 할 수 있습니까?
• 빠른 시작
• 팁, 요령 및 기타 참고 사항
  • Fork 후 중복 등록
  • 다중 처리 시작 방법

이것이 존재하는 이유는 무엇입니까?

현재 Python에서 종료 동작을 처리하기 위한 두 개의 내장 모듈이 있습니다: atexit 및 signal . 그러나 이를 직접 사용하면 반복되는 상용구 코드가 많이 발생하고 실수로 잘못되기 쉬운 일부 명확하지 않은 동작이 발생하므로 이 패키지를 작성했습니다.

atexit 모듈은 신호 처리에 실패하여 현재 부족합니다. signal 모듈은 정상 또는 예외로 인한 종료를 처리하지 못하기 때문에 현재 부족합니다.

일반적인 접근 방식에는 atexit 및 원하는 신호 모두에 함수를 등록하는 자주 반복되는 코드가 포함됩니다. 그러나 함수가 두 번 실행되지 않고(또는 멱등성이 있는지) 이전에 등록된 신호 처리기가 호출되는지 확인하기 위해 특별한 주의가 필요한 경우가 있습니다.

그것은 무엇을 할 수 있습니까?

이 패키지는 다음 동작을 수행하거나 허용합니다.

• 프로그램 종료 시 호출할 함수 등록

  • 항상 정상 종료 또는 예외로 인한 종료: @ pyterminate .register
  • 원하는 신호에 대해 구성 가능:
    @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))

• 여러 기능을 등록할 수 있습니다.

• 이전에 등록된 신호 처리기를 호출합니다.

• 캡처된 신호에 대해 0 또는 0이 아닌 종료 코드를 허용합니다.
  @ pyterminate .register(successful_exit=True)

• SIGINT 에서 KeyboardInterrupt 억제하거나 발생시키는 것을 허용합니다.
  @ pyterminate .register(keyboard_interrupt_on_sigint=True)

  • 추가 예외 처리가 정의되어 있는 경우 KeyboardInterrupt 발생시킬 수 있습니다.

• 함수 등록 취소를 허용합니다: pyterminate .unregister(func)

• 등록된 기능이 실행되는 동안 요청된 신호를 무시하여 중단되지 않도록 합니다.

  • SIGKILL 및 os._exit() 호출을 무시할 수 없다는 점에 유의하는 것이 중요합니다.

빠른 시작

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

팁, 요령 및 기타 참고 사항

Fork 후 중복 등록

Fork를 통해 새로운 프로세스를 생성하면 전체 프로세스가 중복되므로 이전에 등록된 기능도 Fork된 프로세스에 등록됩니다. 이는 포크로 인한 당연한 결과이지만, 등록된 함수가 공유 리소스에 액세스하는지 고려하는 것이 중요합니다. 이 동작을 방지하려면 포크된 프로세스 시작 시 함수 등록을 취소하거나, 프로세스 ID를 기반으로 게이트하거나, 적절한 다른 동기화 방법을 사용할 수 있습니다.

다중 처리 시작 방법

Python의 multiprocessing 모듈로 프로세스를 시작하면 프로세스가 내부적으로 os._exit() 로 종료되어 모든 정리를 우회하고 즉시 프로세스가 종료되므로 종료 시 fork 메서드가 등록된 함수를 호출하지 못합니다.

이 문제를 해결하는 한 가지 방법은 애플리케이션에 허용되는 경우 "spawn" 시작 방법을 사용하는 것입니다. 또 다른 방법은 함수를 사용자 정의 신호에 등록하고 프로세스 코드를 try-Exception 블록에 래핑하여 마지막에 사용자 정의 신호를 발생시키는 것입니다. pyterminate 단순히 try-finally 블록에 장식된 함수를 래핑하고 주어진 신호를 발생시키는 exit_with_signal 데코레이터의 형태로 이 기능을 제공합니다. 사용 예:

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