pyterminate

pyterminate

Jalankan kode pembersihan dengan andal setelah program dihentikan.

Daftar isi

• Mengapa ini ada?
• Apa yang bisa dilakukannya?
• Mulai cepat
• Tips, trik, dan catatan lainnya
  • Pendaftaran duplikat setelah forking
  • Metode mulai multiproses

Mengapa ini ada?

Saat ini ada dua modul bawaan untuk menangani perilaku terminasi dengan Python: atexit dan signal . Namun, menggunakannya secara langsung menyebabkan banyak kode boilerplate yang berulang, dan beberapa perilaku tidak jelas yang mudah salah secara tidak sengaja, itulah sebabnya saya menulis paket ini.

Modul atexit saat ini tidak mencukupi karena gagal menangani sinyal. Modul signal saat ini tidak mencukupi karena gagal menangani jalan keluar yang normal atau disebabkan oleh pengecualian.

Pendekatan yang umum akan mencakup kode yang sering diulang yang mendaftarkan suatu fungsi baik dengan atexit maupun pada sinyal yang diinginkan. Namun, terkadang diperlukan kehati-hatian ekstra untuk memastikan fungsi tersebut tidak berjalan dua kali (atau idempoten), dan pengendali sinyal yang telah terdaftar sebelumnya akan dipanggil.

Apa yang bisa dilakukannya?

Paket ini melakukan atau mengizinkan perilaku berikut:

• Daftarkan fungsi yang akan dipanggil pada penghentian program

  • Selalu dalam penghentian normal atau karena pengecualian: @ pyterminate .register
  • Dapat dikonfigurasi untuk sinyal apa pun yang diinginkan:
    @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))

• Memungkinkan beberapa fungsi untuk didaftarkan

• Akan menghubungi penangan sinyal terdaftar sebelumnya

• Mengizinkan kode keluar nol atau bukan nol pada sinyal yang ditangkap:
  @ pyterminate .register(successful_exit=True)

• Memungkinkan menekan atau membuang KeyboardInterrupt pada SIGINT :
  @ pyterminate .register(keyboard_interrupt_on_sigint=True)

  • Anda mungkin ingin membuang KeyboardInterrupt jika ada penanganan pengecualian tambahan yang ditentukan.

• Mengizinkan fungsi untuk dibatalkan pendaftarannya: pyterminate .unregister(func)

• Abaikan sinyal yang diminta saat fungsi terdaftar sedang dijalankan, pastikan tidak terganggu.

  • Penting untuk dicatat bahwa SIGKILL dan panggilan ke os._exit() tidak dapat diabaikan.

Mulai cepat

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

Tips, trik, dan catatan lainnya

Pendaftaran duplikat setelah forking

Karena pembuatan proses baru melalui forking menduplikasi seluruh proses, fungsi apa pun yang didaftarkan sebelumnya juga akan didaftarkan dalam proses fork. Ini merupakan konsekuensi nyata dari forking, namun penting untuk dipertimbangkan jika fungsi yang terdaftar mengakses sumber daya bersama. Untuk menghindari perilaku ini, Anda dapat membatalkan pendaftaran fungsi di awal proses bercabang, gerbang berdasarkan ID proses, atau menggunakan metode sinkronisasi lain yang sesuai.

Metode mulai multiproses

Saat memulai proses dengan modul multiprocessing Python, metode fork akan gagal memanggil fungsi terdaftar saat keluar, karena proses diakhiri dengan os._exit() secara internal, yang melewati semua pembersihan dan segera mematikan proses.

Salah satu cara untuk menyiasatinya adalah dengan menggunakan metode "spawn" start jika itu dapat diterima untuk aplikasi Anda. Metode lain adalah dengan mendaftarkan fungsi Anda ke sinyal yang ditentukan pengguna, dan menggabungkan kode proses Anda dalam blok coba-kecuali, sehingga meningkatkan sinyal yang ditentukan pengguna di bagian akhir. pyterminate menyediakan fungsionalitas ini dalam bentuk dekorator exit_with_signal , yang hanya membungkus fungsi yang didekorasi dalam blok try-finally, dan memunculkan sinyal yang diberikan. Contoh penggunaan:

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