Ich rufe eine Funktion in Python auf, von der ich weiß, dass sie möglicherweise blockiert und mich zwingt, das Skript neu zu starten.

Wie rufe ich die Funktion auf oder was verpacke ich sie, damit das Skript sie abbricht und etwas anderes tut, wenn es länger als 5 Sekunden dauert?

Sie können das Signalpaket verwenden , wenn Sie unter UNIX arbeiten:

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 Sekunden nach dem Aufruf alarm.alarm(10)wird der Handler aufgerufen. Dies löst eine Ausnahme aus, die Sie vom regulären Python-Code abfangen können.

Dieses Modul spielt nicht gut mit Threads (aber wer dann?)

Beachten Sie, dass eine Ausnahme, wenn eine Zeitüberschreitung auftritt, möglicherweise innerhalb der Funktion abgefangen und ignoriert wird, z. B. bei einer solchen Funktion:

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

Sie können multiprocessing.Processgenau das verwenden.

Code

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

Wie rufe ich die Funktion auf oder was verpacke ich sie, damit das Skript sie abbricht, wenn sie länger als 5 Sekunden dauert?

Ich habe einen Kern gepostet , der diese Frage / dieses Problem mit einem Dekorateur und einem löst threading.Timer. Hier ist es mit einer Panne.

Importe und Setups aus Kompatibilitätsgründen

Es wurde mit Python 2 und 3 getestet. Es sollte auch unter Unix / Linux und Windows funktionieren.

Zuerst die Importe. Diese versuchen, den Code unabhängig von der Python-Version konsistent zu halten:

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

Verwenden Sie den versionunabhängigen Code:

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

Jetzt haben wir unsere Funktionalität aus der Standardbibliothek importiert.

exit_after Dekorateur

Als nächstes benötigen wir eine Funktion, um den main()vom untergeordneten Thread zu beenden :

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

Und hier ist der Dekorateur selbst:

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

Verwendung

Und hier ist die Verwendung, die Ihre Frage zum Beenden nach 5 Sekunden direkt beantwortet!:

@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')

Demo:

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

Der zweite Funktionsaufruf wird nicht beendet, stattdessen sollte der Prozess mit einem Traceback beendet werden!

KeyboardInterrupt stoppt nicht immer einen schlafenden Faden

Beachten Sie, dass der Ruhezustand unter Python 2 unter Windows nicht immer durch einen Tastaturinterrupt unterbrochen wird, z.

@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

Es ist auch nicht wahrscheinlich, dass Code, der in Erweiterungen ausgeführt wird, unterbrochen wird, es sei denn, er prüft explizit PyErr_CheckSignals(), ob Cython, Python und KeyboardInterrupt ignoriert werden

Ich würde es auf jeden Fall vermeiden, einen Thread länger als eine Sekunde zu schlafen - das ist ein Äon in der Prozessorzeit.

Wie rufe ich die Funktion auf oder was verpacke ich sie, damit das Skript sie abbricht und etwas anderes tut , wenn es länger als 5 Sekunden dauert?

Um es zu fangen und etwas anderes zu tun, können Sie den KeyboardInterrupt fangen.

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

Ich habe einen anderen Vorschlag, der eine reine Funktion ist (mit der gleichen API wie der Threading-Vorschlag) und anscheinend gut funktioniert (basierend auf Vorschlägen zu diesem Thread).

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

Ich bin auf diesen Thread gestoßen, als ich bei Unit-Tests nach einem Timeout-Aufruf gesucht habe. Ich habe in den Antworten oder Paketen von Drittanbietern nichts Einfaches gefunden, also habe ich den Dekorateur unten geschrieben, den Sie direkt in den Code einfügen können:

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

Dann ist es so einfach, einen Test oder eine beliebige Funktion zu deaktivieren:

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

Das stopitauf pypi gefundene Paket scheint Timeouts gut zu verarbeiten.

Ich mag den @stopit.threading_timeoutableDekorator, der timeoutder dekorierten Funktion einen Parameter hinzufügt , der das tut, was Sie erwarten, er stoppt die Funktion.

Überprüfen Sie es auf pypi: https://pypi.python.org/pypi/stopit

Es gibt viele Vorschläge, aber keine, die concurrent.futures verwenden. Ich denke, dies ist der am besten lesbare Weg, um damit umzugehen.

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)

Super einfach zu lesen und zu warten.

Wir erstellen einen Pool, senden einen einzelnen Prozess und warten dann bis zu 5 Sekunden, bevor wir einen TimeoutError auslösen, den Sie nach Bedarf abfangen und verarbeiten können.

Nativ in Python 3.2+ und zurückportiert auf 2.7 (Pip Install Futures).

Das Wechseln zwischen Threads und Prozessen ist so einfach wie das Ersetzen ProcessPoolExecutordurch ThreadPoolExecutor.

Wenn Sie den Prozess mit Zeitüberschreitung beenden möchten, würde ich empfehlen, sich mit Pebble zu befassen .

Großartiger, benutzerfreundlicher und zuverlässiger Timeout-Dekorator für PyPi- Projekte ( https://pypi.org/project/timeout-decorator/ )

Installation :

pip install timeout-decorator

Verwendung :

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

Ich bin der Autor von wrapt_timeout_decorator

Die meisten der hier vorgestellten Lösungen funktionieren auf den ersten Blick wunderbar unter Linux - weil wir Fork () und Signale () haben -, aber unter Windows sehen die Dinge etwas anders aus. Und wenn es um Subthreads unter Linux geht, können Sie keine Signale mehr verwenden.

Um einen Prozess unter Windows zu erzeugen, muss er auswählbar sein - und viele dekorierte Funktionen oder Klassenmethoden nicht.

Sie müssen also einen besseren Pickler wie Dill und Multiprocess verwenden (nicht Pickle und Multiprocessing) - deshalb können Sie ProcessPoolExecutor nicht verwenden (oder nur mit eingeschränkter Funktionalität).

Für das Timeout selbst - Sie müssen definieren, was Timeout bedeutet -, da unter Windows eine beträchtliche (und nicht bestimmbare) Zeit benötigt wird, um den Prozess zu erzeugen. Dies kann bei kurzen Zeitüberschreitungen schwierig sein. Nehmen wir an, das Laichen des Prozesses dauert ungefähr 0,5 Sekunden (leicht !!!). Wenn Sie eine Zeitüberschreitung von 0,2 Sekunden angeben, was soll passieren? Sollte die Funktion nach 0,5 + 0,2 Sekunden ablaufen (lassen Sie die Methode also 0,2 Sekunden lang laufen)? Oder sollte der aufgerufene Prozess nach 0,2 Sekunden eine Zeitüberschreitung aufweisen (in diesem Fall wird die dekorierte Funktion IMMER eine Zeitüberschreitung aufweisen, da sie in dieser Zeit nicht einmal erzeugt wird)?

Auch verschachtelte Dekorateure können böse sein und Sie können keine Signale in einem Unterfaden verwenden. Wenn Sie einen wirklich universellen, plattformübergreifenden Dekorateur erstellen möchten, muss dies alles berücksichtigt (und getestet) werden.

Andere Probleme sind die Rückgabe von Ausnahmen an den Aufrufer sowie Protokollierungsprobleme (falls in der dekorierten Funktion verwendet - die Protokollierung in Dateien in einem anderen Prozess wird NICHT unterstützt)

Ich habe versucht, alle Randfälle abzudecken. Sie könnten in das Paket wrapt_timeout_decorator schauen oder zumindest Ihre eigenen Lösungen testen, die von den dort verwendeten Unittests inspiriert sind.

@Alexis Eggermont - leider habe ich nicht genug Punkte zum Kommentieren - vielleicht kann dich jemand anderes benachrichtigen - ich glaube, ich habe dein Importproblem gelöst.

timeout-decoratorfunktioniert nicht auf Windows-System, da Windows nicht signalgut unterstützt.

Wenn Sie Timeout-Decorator im Windows-System verwenden, erhalten Sie Folgendes

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

Einige schlugen vor, zu verwenden use_signals=False, arbeiteten aber nicht für mich.

Der Autor @bitranox hat das folgende Paket erstellt:

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

Codebeispiel:

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

Gibt die folgende Ausnahme:

TimeoutError: Function mytest timed out after 5 seconds

Wir können Signale dafür verwenden. Ich denke, das folgende Beispiel wird für Sie nützlich sein. Es ist sehr einfach im Vergleich zu Threads.

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"

#!/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)

Ich brauchte verschachtelbare zeitgesteuerte Interrupts (was SIGALARM nicht kann), die nicht durch time.sleep blockiert werden (was der threadbasierte Ansatz nicht kann). Am Ende habe ich Code von hier kopiert und leicht modifiziert: http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/

Der Code selbst:

#!/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()

und ein Anwendungsbeispiel:

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.'

Hier ist eine leichte Verbesserung der gegebenen threadbasierten Lösung.

Der folgende Code unterstützt Ausnahmen :

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]

Aufrufen mit einer Zeitüberschreitung von 5 Sekunden:

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