الصفحة الرئيسية>المتعلقة بالبرمجة>كود الذكاء الاصطناعي
تنزيل

pyterminate

تشغيل تعليمات برمجية للتنظيف بشكل موثوق عند إنهاء البرنامج.

جدول المحتويات

  • لماذا هذا موجود؟
  • ماذا يمكن أن تفعل؟
  • بداية سريعة
  • النصائح والحيل والملاحظات الأخرى
    • تسجيل مكرر بعد التفرع
    • طريقة البدء بالمعالجة المتعددة

لماذا هذا موجود؟

يوجد حاليًا وحدتان مدمجتان للتعامل مع سلوك الإنهاء في بايثون: 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() .

بداية سريعة

pyterminate">
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" إذا كان ذلك مقبولًا لتطبيقك. هناك طريقة أخرى تتمثل في تسجيل وظيفتك لإشارة محددة من قبل المستخدم، ولف رمز العملية الخاص بك في كتلة محاولة باستثناء، مما يؤدي إلى رفع الإشارة المحددة من قبل المستخدم في النهاية. يوفر pyterminate هذه الوظيفة على شكل ديكور 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 ()
يوسع
معلومات إضافية
تطبيقات ذات صلة
نوصي لك
أخبار ذات صلة الكل