pyterminate

pyterminate

Надежно запускайте код очистки после завершения программы.

Оглавление

• Почему это существует?
• Что он может сделать?
• Быстрый старт
• Советы, рекомендации и другие примечания
  • Двойная регистрация после форка
  • Метод запуска многопроцессорной обработки

Почему это существует?

В настоящее время в Python существует два встроенных модуля для обработки поведения завершения: atexit и signal . Однако их непосредственное использование приводит к большому количеству повторяющегося шаблонного кода и некоторым неочевидным поведениям, в которых легко случайно ошибиться, поэтому я написал этот пакет.

Модуль atexit в настоящее время недостаточен, поскольку он не может обрабатывать сигналы. signal модуль в настоящее время недостаточен, поскольку он не может обрабатывать обычные выходы или выходы, вызванные исключениями.

Типичные подходы включают часто повторяющийся код, регистрирующий функцию как с помощью atexit , так и по желаемым сигналам. Однако иногда необходимо проявлять особую осторожность, чтобы гарантировать, что функция не запускается дважды (или не является идемпотентной) и чтобы вызывался ранее зарегистрированный обработчик сигнала.

Что он может сделать?

Эти пакеты выполняют или допускают следующее поведение:

• Зарегистрируйте функцию, которая будет вызываться при завершении программы

  • Всегда при обычном завершении или завершении по причине исключения: @ pyterminate .register
  • Настраивается для любых желаемых сигналов:
    @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))

• Позволяет зарегистрировать несколько функций

• Вызовет предыдущие зарегистрированные обработчики сигналов.

• Разрешает нулевые или ненулевые коды выхода для захваченных сигналов:
  @ pyterminate .register(successful_exit=True)

• Позволяет подавлять или выдавать KeyboardInterrupt на SIGINT :
  @ 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 , ...)

Советы, рекомендации и другие примечания

Двойная регистрация после форка

Поскольку создание нового процесса посредством разветвления дублирует весь процесс, любые ранее зарегистрированные функции также будут зарегистрированы в разветвленном процессе. Это очевидное следствие разветвления, но его важно учитывать, если зарегистрированные функции получают доступ к общим ресурсам. Чтобы избежать такого поведения, вы можете отменить регистрацию функции в начале разветвленного процесса, использовать шлюз на основе идентификатора процесса или использовать любой другой подходящий метод синхронизации.

Метод запуска многопроцессорной обработки

При запуске процессов с помощью multiprocessing модуля Python метод fork не сможет вызвать зарегистрированные функции при выходе, поскольку процесс завершается внутренней функцией os._exit() , которая обходит всю очистку и немедленно завершает процесс.

Один из способов обойти эту проблему — использовать метод запуска "spawn" если это приемлемо для вашего приложения. Другой метод — зарегистрировать вашу функцию для определяемого пользователем сигнала и обернуть код процесса в блок try-кроме, поднимая в конце определяемый пользователем сигнал. pyterminate предоставляет эту функциональность в форме декоратора exit_with_signal , который просто оборачивает декорированную функцию в блок try-finally и вызывает данный сигнал. Пример использования:

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