مهلة استدعاء دالة

Question

أقوم باستدعاء وظيفة في Python والتي أعلم أنها قد تتعطل وتجبرني على إعادة تشغيل البرنامج النصي.

كيف يمكنني استدعاء الوظيفة أو ما الذي أقوم بتغليفه بحيث إذا استغرق الأمر أكثر من 5 ثوانٍ فإن البرنامج النصي يلغيها ويفعل شيئًا آخر؟

Answer 1

يمكنك استخدام الإشارة الحزمة إذا كنت تعمل على UNIX:

In [1]: import signal

# Register an handler for the timeout
In [2]: def handler(signum, frame):
   ...:     print "Forever is over!"
   ...:     raise Exception("end of time")
   ...: 

# This function *may* run for an indetermined time...
In [3]: def loop_forever():
   ...:     import time
   ...:     while 1:
   ...:         print "sec"
   ...:         time.sleep(1)
   ...:         
   ...:         

# Register the signal function handler
In [4]: signal.signal(signal.SIGALRM, handler)
Out[4]: 0

# Define a timeout for your function
In [5]: signal.alarm(10)
Out[5]: 0

In [6]: try:
   ...:     loop_forever()
   ...: except Exception, exc: 
   ...:     print exc
   ....: 
sec
sec
sec
sec
sec
sec
sec
sec
Forever is over!
end of time

# Cancel the timer if the function returned before timeout
# (ok, mine won't but yours maybe will :)
In [7]: signal.alarm(0)
Out[7]: 0

10 ثواني بعد المكالمة alarm.alarm(10), ، يتم استدعاء المعالج.يؤدي هذا إلى ظهور استثناء يمكنك اعتراضه من كود بايثون العادي.

هذه الوحدة لا تعمل بشكل جيد مع المواضيع (ولكن بعد ذلك، من يفعل؟)

لاحظ أن نظرًا لأننا نثير استثناءً عند انتهاء المهلة، فقد يتم اكتشافه وتجاهله داخل الوظيفة، على سبيل المثال إحدى هذه الوظائف:

def loop_forever():
    while 1:
        print 'sec'
        try:
            time.sleep(10)
        except:
            continue

Answer 2

يمكنك استخدام multiprocessing.Process للقيام بذلك بالضبط.

شفرة

import multiprocessing
import time

# bar
def bar():
    for i in range(100):
        print "Tick"
        time.sleep(1)

if __name__ == '__main__':
    # Start bar as a process
    p = multiprocessing.Process(target=bar)
    p.start()

    # Wait for 10 seconds or until process finishes
    p.join(10)

    # If thread is still active
    if p.is_alive():
        print "running... let's kill it..."

        # Terminate
        p.terminate()
        p.join()

Answer 3

كيف يمكنني استدعاء الوظيفة أو ما الذي أقوم بتغليفه بحيث إذا استغرق الأمر أكثر من 5 ثوانٍ، يقوم البرنامج النصي بإلغائه؟

لقد نشرت أ جوهر الذي يحل هذا السؤال/المشكلة باستخدام مصمم الديكور و threading.Timer.ومن هنا مع انهيار.

الواردات والإعدادات للتوافق

تم اختباره مع بايثون 2 و 3.يجب أن يعمل أيضًا ضمن أنظمة Unix/Linux وWindows.

أولا الواردات.يحاول هذا الحفاظ على اتساق الكود بغض النظر عن إصدار بايثون:

from __future__ import print_function
import sys
import threading
from time import sleep
try:
    import thread
except ImportError:
    import _thread as thread

استخدم رمز الإصدار المستقل:

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

لقد قمنا الآن باستيراد وظائفنا من المكتبة القياسية.

exit_after مصمم ديكور

بعد ذلك نحتاج إلى وظيفة لإنهاء main() من موضوع الطفل:

def quit_function(fn_name):
    # print to stderr, unbuffered in Python 2.
    print('{0} took too long'.format(fn_name), file=sys.stderr)
    sys.stderr.flush() # Python 3 stderr is likely buffered.
    thread.interrupt_main() # raises KeyboardInterrupt

وهنا الديكور نفسه:

def exit_after(s):
    '''
    use as decorator to exit process if 
    function takes longer than s seconds
    '''
    def outer(fn):
        def inner(*args, **kwargs):
            timer = threading.Timer(s, quit_function, args=[fn.__name__])
            timer.start()
            try:
                result = fn(*args, **kwargs)
            finally:
                timer.cancel()
            return result
        return inner
    return outer

الاستخدام

وهذا الاستخدام يجيب بشكل مباشر على سؤالك حول الخروج بعد 5 ثوانٍ!:

@exit_after(5)
def countdown(n):
    print('countdown started', flush=True)
    for i in range(n, -1, -1):
        print(i, end=', ', flush=True)
        sleep(1)
    print('countdown finished')

العرض التوضيحي:

>>> countdown(3)
countdown started
3, 2, 1, 0, countdown finished
>>> countdown(10)
countdown started
10, 9, 8, 7, 6, countdown took too long
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in inner
  File "<stdin>", line 6, in countdown
KeyboardInterrupt

لن ينتهي استدعاء الوظيفة الثانية، وبدلاً من ذلك يجب إنهاء العملية بتتبع!

KeyboardInterrupt لا يوقف دائمًا خيط النوم

لاحظ أن وضع السكون لن ينقطع دائمًا عن طريق مقاطعة لوحة المفاتيح، في Python 2 على نظام التشغيل Windows، على سبيل المثال:

@exit_after(1)
def sleep10():
    sleep(10)
    print('slept 10 seconds')

>>> sleep10()
sleep10 took too long         # Note that it hangs here about 9 more seconds
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in inner
  File "<stdin>", line 3, in sleep10
KeyboardInterrupt

وليس من المحتمل مقاطعة التعليمات البرمجية التي تعمل في الامتدادات ما لم يتم التحقق منها بشكل صريح PyErr_CheckSignals(), ، يرى تم تجاهل Cython وPython وKeyboardInterrupt

أود أن أتجنب وضع الخيط في وضع السكون لأكثر من ثانية، على أي حال - وهذا يمثل دهرًا في وقت المعالج.

كيف يمكنني استدعاء الوظيفة أو ماذا أقوم بتغليفها بحيث إذا استغرق الأمر أكثر من 5 ثوانٍ يقوم البرنامج النصي بإلغائها ويفعل شيئا آخر؟

للقبض عليه والقيام بشيء آخر، يمكنك التقاط KeyboardInterrupt.

>>> try:
...     countdown(10)
... except KeyboardInterrupt:
...     print('do something else')
... 
countdown started
10, 9, 8, 7, 6, countdown took too long
do something else

Answer 4

ولدي اقتراح آخر وهو وظيفة نقية (مع نفس API مثل اقتراح خيوط)، ويبدو أن تعمل بشكل جيد (على أساس اقتراحات حول هذا الموضوع)

def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
    import signal

    class TimeoutError(Exception):
        pass

    def handler(signum, frame):
        raise TimeoutError()

    # set the timeout handler
    signal.signal(signal.SIGALRM, handler) 
    signal.alarm(timeout_duration)
    try:
        result = func(*args, **kwargs)
    except TimeoutError as exc:
        result = default
    finally:
        signal.alarm(0)

    return result

Answer 5

وركضت عبر هذا الموضوع عند البحث عن مكالمة مهلة على وحدة الاختبارات. لم أجد أي شيء بسيط في الأجوبة أو حزم 3rd الطرف حتى كتبت الديكور أدناه يمكنك إسقاط الحق إلى رمز:

import multiprocessing.pool
import functools

def timeout(max_timeout):
    """Timeout decorator, parameter in seconds."""
    def timeout_decorator(item):
        """Wrap the original function."""
        @functools.wraps(item)
        def func_wrapper(*args, **kwargs):
            """Closure for function."""
            pool = multiprocessing.pool.ThreadPool(processes=1)
            async_result = pool.apply_async(item, args, kwargs)
            # raises a TimeoutError if execution exceeds max_timeout
            return async_result.get(max_timeout)
        return func_wrapper
    return timeout_decorator

وبعد ذلك انها بسيطة مثل هذه انتهاء مهلة اختبار أو أي وظيفة تريد:

@timeout(5.0)  # if execution takes longer than 5 seconds, raise a TimeoutError
def test_base_regression(self):
    ...

Answer 6

وهناك الكثير من الاقتراحات، ولكن أيا باستخدام concurrent.futures، التي أعتقد أنها الطريقة الأكثر نظامية للتعامل مع هذا.

from concurrent.futures import ProcessPoolExecutor

# Warning: this does not terminate function if timeout
def timeout_five(fnc, *args, **kwargs):
    with ProcessPoolExecutor() as p:
        f = p.submit(fnc, *args, **kwargs)
        return f.result(timeout=5)

وسوبر بسيط لقراءة والمحافظة عليها.

ونحن جعل بركة، تقدم في عملية واحدة ثم انتظر حتى 5 ثوان قبل رفع TimeoutError التي يمكن أن قبض والتعامل مع ذلك كنت في حاجة.

والأصلية لالثعبان 3.2+ وbackported إلى 2.7 (PIP تثبيت العقود الآجلة).

والتبديل بين المواضيع وعمليات بسيطة مثل استبدال ProcessPoolExecutor مع ThreadPoolExecutor.

إذا أردت إنهاء عملية على مهلة أود أن أقترح النظر في بيبل .

Answer 7

وحزمة stopit، وجدت في pypi، ويبدو أن تعامل بشكل جيد مهلة.

وأنا أحب الديكور @stopit.threading_timeoutable، والتي تضيف معلمة timeout إلى وظيفة المزينة، الذي يفعل ما تتوقع، فإنه يتوقف على وظيفة.

والتحقق من ذلك على pypi: https://pypi.python.org/pypi/stopit

Answer 8

رائعة وسهلة الاستخدام وموثوقة بيبي مشروع timeout-decorator (https://pypi.org/project/timeout-decorator/)

تثبيت:

pip install timeout-decorator

الاستخدام:

import time
import timeout_decorator

@timeout_decorator.timeout(5)
def mytest():
    print "Start"
    for i in range(1,10):
        time.sleep(1)
        print "%d seconds have passed" % i

if __name__ == '__main__':
    mytest()

Answer 9

#!/usr/bin/python2
import sys, subprocess, threading
proc = subprocess.Popen(sys.argv[2:])
timer = threading.Timer(float(sys.argv[1]), proc.terminate)
timer.start()
proc.wait()
timer.cancel()
exit(proc.returncode)

Answer 10

وtimeout-decorator لا تعمل على نظام ويندوز كما، ويندوز لم تدعم signal أيضا.

إذا كنت تستخدم مهلة الديكور في نظام ويندوز ستحصل على التالي

AttributeError: module 'signal' has no attribute 'SIGALRM'

واقترح البعض استخدام use_signals=False لكنه لم يعمل بالنسبة لي.

والكاتبbitranox إنشاء الحزمة التالية:

pip install https://github.com/bitranox/wrapt-timeout-decorator/archive/master.zip

ورمز نموذج:

import time
from wrapt_timeout_decorator import *

@timeout(5)
def mytest(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))

def main():
    mytest('starting')


if __name__ == '__main__':
    main()

ويعطي الاستثناء التالي:

TimeoutError: Function mytest timed out after 5 seconds

Answer 11

وأنا مؤلف wrapt_timeout_decorator

ومعظم الحلول المقدمة هنا تعمل wunderfully تحت لينكس على أول وهلة - لأن لدينا شوكة () وإشارات () - ولكن على ويندوز الأمور تبدو مختلفة بعض الشيء. وعندما يتعلق الأمر subthreads على لينكس، أنت غير قادر على إشارات استخدام بعد الآن.

في أجل إنتاج عملية تحت ويندوز، فإنه يجب أن يكون pickable - والعديد من الوظائف زينت أو أساليب الفئة ليست

وهكذا تحتاج إلى استخدام بيكلر أفضل مثل الشبت وmultiprocess (لا مخلل ومتعدد المعالجة) - ولهذا السبب أنت غير قادر على استخدام ProcessPoolExecutor (أو فقط مع وظائف محدودة)

لالمهلة نفسها - تحتاج إلى تعريف ما يعني مهلة - لأن على ويندوز سوف يستغرق كبيرا (وليس للتحديد) وقت لتفرخ هذه العملية. وهذا يمكن أن تكون خادعة في مهلة قصيرة. لنفترض، وضع البيض تستغرق العملية حوالي 0.5 ثانية (بسهولة !!!). إذا كنت تعطي مهلة 0.2 ثانية ما يجب أن يحدث؟ يجب على الوقت وظيفة خارج بعد 0.5 + 0.2 ثانية (لذلك دعونا المدى طريقة ل0.2 ثانية)؟ أو ينبغي أن تسمى عملية تستغرق وقتا للخروج بعد 0.2 ثانية (في هذه الحالة، فإن وظيفة زينت سوف ALWAYS مهلة، لأنه في ذلك الوقت لم يتم حتى ولدت فيه)؟

والديكور متداخلة أيضا يمكن أن يكون سيئة وأنت غير قادر على إشارات استخدامها في subthread. إذا كنت ترغب في إنشاء عالمي حقا، عبر منصة الديكور، كل هذا يحتاج إلى أن تؤخذ بعين الاعتبار (واختبار).

وقضايا أخرى تمر استثناءات إلى المتصل، فضلا عن القضايا تسجيل (إذا ما استخدمت في وظيفة زينت - لا يتم اعتماد تسجيل الدخول إلى الملفات في عملية أخرى)

وحاولت أن تغطي جميع الحالات الحافة، قد نظر في حزمة wrapt_timeout_decorator، أو على الأقل اختبار الحلول لديك الخاصة مستوحاة من unittests المستخدمة هناك.

وAlexis Eggermont - للأسف أنا لا أميل لها ما يكفي من النقاط التعليق - ربما شخص آخر يمكن أن يخطر لك - أعتقد أنني تحل قضية استيراد لديك

Answer 12

ويمكننا استخدام إشارات لنفسه. أعتقد أن المثال أدناه تكون مفيدة بالنسبة لك. وهو بسيط جدا بالمقارنة مع المواضيع.

import signal

def timeout(signum, frame):
    raise myException

#this is an infinite loop, never ending under normal circumstances
def main():
    print 'Starting Main ',
    while 1:
        print 'in main ',

#SIGALRM is only usable on a unix platform
signal.signal(signal.SIGALRM, timeout)

#change 5 to however many seconds you need
signal.alarm(5)

try:
    main()
except myException:
    print "whoops"

Answer 13

كان لدي حاجة ل قابل للتعشيش المقاطعات الموقوتة (التي لا يستطيع SIGALARM القيام بها) والتي لن يتم حظرها بواسطة time.sleep (وهو ما لا يستطيع النهج القائم على الخيط القيام به).انتهى بي الأمر بنسخ الكود وتعديله بشكل طفيف من هنا: http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/

الكود نفسه:

#!/usr/bin/python

# lightly modified version of http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/


"""alarm.py: Permits multiple SIGALRM events to be queued.

Uses a `heapq` to store the objects to be called when an alarm signal is
raised, so that the next alarm is always at the top of the heap.
"""

import heapq
import signal
from time import time

__version__ = '$Revision: 2539 $'.split()[1]

alarmlist = []

__new_alarm = lambda t, f, a, k: (t + time(), f, a, k)
__next_alarm = lambda: int(round(alarmlist[0][0] - time())) if alarmlist else None
__set_alarm = lambda: signal.alarm(max(__next_alarm(), 1))


class TimeoutError(Exception):
    def __init__(self, message, id_=None):
        self.message = message
        self.id_ = id_


class Timeout:
    ''' id_ allows for nested timeouts. '''
    def __init__(self, id_=None, seconds=1, error_message='Timeout'):
        self.seconds = seconds
        self.error_message = error_message
        self.id_ = id_
    def handle_timeout(self):
        raise TimeoutError(self.error_message, self.id_)
    def __enter__(self):
        self.this_alarm = alarm(self.seconds, self.handle_timeout)
    def __exit__(self, type, value, traceback):
        try:
            cancel(self.this_alarm) 
        except ValueError:
            pass


def __clear_alarm():
    """Clear an existing alarm.

    If the alarm signal was set to a callable other than our own, queue the
    previous alarm settings.
    """
    oldsec = signal.alarm(0)
    oldfunc = signal.signal(signal.SIGALRM, __alarm_handler)
    if oldsec > 0 and oldfunc != __alarm_handler:
        heapq.heappush(alarmlist, (__new_alarm(oldsec, oldfunc, [], {})))


def __alarm_handler(*zargs):
    """Handle an alarm by calling any due heap entries and resetting the alarm.

    Note that multiple heap entries might get called, especially if calling an
    entry takes a lot of time.
    """
    try:
        nextt = __next_alarm()
        while nextt is not None and nextt <= 0:
            (tm, func, args, keys) = heapq.heappop(alarmlist)
            func(*args, **keys)
            nextt = __next_alarm()
    finally:
        if alarmlist: __set_alarm()


def alarm(sec, func, *args, **keys):
    """Set an alarm.

    When the alarm is raised in `sec` seconds, the handler will call `func`,
    passing `args` and `keys`. Return the heap entry (which is just a big
    tuple), so that it can be cancelled by calling `cancel()`.
    """
    __clear_alarm()
    try:
        newalarm = __new_alarm(sec, func, args, keys)
        heapq.heappush(alarmlist, newalarm)
        return newalarm
    finally:
        __set_alarm()


def cancel(alarm):
    """Cancel an alarm by passing the heap entry returned by `alarm()`.

    It is an error to try to cancel an alarm which has already occurred.
    """
    __clear_alarm()
    try:
        alarmlist.remove(alarm)
        heapq.heapify(alarmlist)
    finally:
        if alarmlist: __set_alarm()

ومثال الاستخدام:

import alarm
from time import sleep

try:
    with alarm.Timeout(id_='a', seconds=5):
        try:
            with alarm.Timeout(id_='b', seconds=2):
                sleep(3)
        except alarm.TimeoutError as e:
            print 'raised', e.id_
        sleep(30)
except alarm.TimeoutError as e:
    print 'raised', e.id_
else:
    print 'nope.'

Answer 14

وهنا هو تحسن طفيف على الحل القائم على موضوع معين.

ورمز أدناه الدعم على الاستثناءات : ل

def runFunctionCatchExceptions(func, *args, **kwargs):
    try:
        result = func(*args, **kwargs)
    except Exception, message:
        return ["exception", message]

    return ["RESULT", result]


def runFunctionWithTimeout(func, args=(), kwargs={}, timeout_duration=10, default=None):
    import threading
    class InterruptableThread(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)
            self.result = default
        def run(self):
            self.result = runFunctionCatchExceptions(func, *args, **kwargs)
    it = InterruptableThread()
    it.start()
    it.join(timeout_duration)
    if it.isAlive():
        return default

    if it.result[0] == "exception":
        raise it.result[1]

    return it.result[1]

واستدعاء ذلك مع 5 المهلة الثانية:

result = timeout(remote_calculate, (myarg,), timeout_duration=5)