Timeout su una chiamata di funzione

Question

Sto chiamando una funzione in Python che conosco potrebbe bloccarsi e costringermi a riavviare lo script.

Come posso chiamare la funzione o come inserirla in modo tale che se impiega più di 5 secondi lo script la annulla e fa qualcos'altro?

Answer 1

È possibile utilizzare il pacchetto signal se si esegue 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 secondi dopo la chiamata alarm.alarm(10), viene chiamato il gestore. Ciò solleva un'eccezione che puoi intercettare dal normale codice Python.

Questo modulo non funziona bene con i thread (ma poi, chi lo fa?)

Tieni presente che poiché solleviamo un'eccezione quando si verifica il timeout, potrebbe essere catturato e ignorato all'interno della funzione, ad esempio di una di queste funzioni:

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

Answer 2

Puoi usare multiprocessing.Process per fare esattamente questo.

Codice

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

  

Come posso chiamare la funzione o come inserirla in modo tale che se impiega più di 5 secondi lo script la annulla?

Ho pubblicato un gist che risolve questa domanda / problema con un decoratore e un threading.Timer . Eccolo con un guasto.

Importazioni e configurazioni per la compatibilità

È stato testato con Python 2 e 3. Dovrebbe funzionare anche su Unix / Linux e Windows.

Prima le importazioni. Questi tentano di mantenere coerente il codice indipendentemente dalla versione di Python:

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

Usa il codice indipendente dalla versione:

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

Ora abbiamo importato le nostre funzionalità dalla libreria standard.

exit_after decoratore

Successivamente abbiamo bisogno di una funzione per terminare il main() dal thread figlio:

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

Ed ecco lo stesso decoratore:

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

Uso

Ed ecco l'uso che risponde direttamente alla tua domanda sull'uscita dopo 5 secondi !:

@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

La seconda chiamata di funzione non finirà, invece il processo dovrebbe uscire con un traceback!

KeyboardInterrupt non sempre interrompe un thread in sospeso

Nota che la sospensione non sarà sempre interrotta da un'interruzione della tastiera, su Python 2 su Windows, ad es .:

@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

né è probabile che interrompa l'esecuzione del codice nelle estensioni a meno che non controlli esplicitamente la presenza di PyErr_CheckSignals(), vedere Cython, Python e KeyboardInterrupt ignorati

Eviterei di dormire un thread più di un secondo, in ogni caso - è un eone nel tempo del processore.

  

Come posso chiamare la funzione o come inserirla in modo tale che se impiega più di 5 secondi lo script la annulla e fa qualcos'altro?

Per catturarlo e fare qualcos'altro, puoi catturare 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

Ho una proposta diversa che è una funzione pura (con la stessa API del suggerimento per il threading) e sembra funzionare bene (sulla base di suggerimenti su questo 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

Answer 5

Mi sono imbattuto in questo thread durante la ricerca di una chiamata di timeout sui test unitari. Non ho trovato nulla di semplice nelle risposte o nei pacchetti di terze parti, quindi ho scritto il decoratore di seguito che puoi inserire nel codice:

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

Quindi è semplice come timeout un test o qualsiasi funzione che ti piace:

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

Answer 6

Ci sono molti suggerimenti, ma nessuno usa concurrent.futures, che penso sia il modo più leggibile per gestirlo.

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 semplice da leggere e mantenere.

Realizziamo un pool, inviamo un singolo processo e quindi aspettiamo fino a 5 secondi prima di generare un TimeoutError che potresti catturare e gestire come necessario.

Originario di Python 3.2+ e backportato su 2.7 (installazione futures su pip).

Il passaggio tra thread e processi è semplice come sostituire ProcessPoolExecutor con ThreadPoolExecutor.

Se si desidera interrompere il processo al timeout, suggerirei di esaminare Pebble .

Answer 7

Il pacchetto stopit, trovato su pypi, sembra gestire bene i timeout.

Mi piace il @stopit.threading_timeoutable decoratore, che aggiunge un parametro timeout alla funzione decorata, che fa quello che ti aspetti, interrompe la funzione.

Dai un'occhiata su pypi: https://pypi.python.org/pypi/stopit

Answer 8

Progetto PyPi fantastico, facile da usare e affidabile timeout-decorator ( https://pypi.org/project/timeout-decorator/ )

Installazione :

pip install timeout-decorator

Utilizzo :

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 non funziona sul sistema Windows poiché Windows non supportava signal bene.

Se usi timeout-decorator nel sistema di Windows otterrai quanto segue

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

Alcuni hanno suggerito di usare use_signals=False ma non ha funzionato per me.

L'autore @bitranox ha creato il seguente pacchetto:

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

Esempio di codice:

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

Fornisce la seguente eccezione:

TimeoutError: Function mytest timed out after 5 seconds

Answer 11

Sono l'autore di wrapt_timeout_decorator

La maggior parte delle soluzioni presentate qui funzionano in modo meraviglioso sotto Linux al primo sguardo - perché abbiamo fork () e signal () - ma su Windows le cose sembrano un po 'diverse. E quando si tratta di sottotread su Linux, non puoi più usare i segnali.

Per generare un processo in Windows, deve essere selezionabile e molte funzioni decorate o metodi Class non lo sono.

Quindi devi usare un miglior pickler come aneto e multiprocesso (non pickle e multiprocessing) - ecco perché non puoi usare ProcessPoolExecutor (o solo con funzionalità limitate).

Per il timeout stesso - Devi definire cosa significa timeout - perché su Windows ci vorrà un tempo considerevole (e non determinabile) per generare il processo. Questo può essere complicato con brevi timeout. Supponiamo che la generazione del processo richieda circa 0,5 secondi (facilmente !!!). Se si concede un timeout di 0,2 secondi, cosa dovrebbe accadere? La funzione dovrebbe scadere dopo 0,5 + 0,2 secondi (quindi lasciare funzionare il metodo per 0,2 secondi)? O dovrebbe il timeout del processo chiamato dopo 0,2 secondi (in quel caso, la funzione decorata SEMPRE timeout, perché in quel momento non viene nemmeno generata)?

Anche i decoratori nidificati possono essere cattivi e non puoi usare i segnali in un sottotread. Se vuoi creare un decoratore veramente universale e multipiattaforma, tutto ciò deve essere preso in considerazione (e testato).

Altri problemi stanno restituendo al chiamante le eccezioni, nonché i problemi di registrazione (se utilizzato nella funzione decorata - NON è supportata la registrazione di file in un altro processo)

Ho provato a coprire tutti i casi limite, potresti esaminare il pacchetto wrapt_timeout_decorator o almeno testare le tue soluzioni ispirate agli unittest utilizzati lì.

@Alexis Eggermont - purtroppo non ho abbastanza punti per commentare - forse qualcun altro può avvisarti - Penso di aver risolto il tuo problema di importazione.

Answer 12

Possiamo usare segnali per lo stesso. Penso che l'esempio che segue sarà utile per te. È molto semplice rispetto ai thread.

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

Avevo bisogno di nidificabili interruzioni temporizzate (che SIGALARM non può fare) che non verranno bloccate da time.sleep (che l'approccio basato su thread non può fare). Ho finito per copiare e modificare leggermente il codice da qui: http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/

Il codice stesso:

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

e un esempio di utilizzo:

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

Ecco un leggero miglioramento rispetto alla soluzione basata su thread fornita.

Il codice seguente supporta eccezioni :

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]

Richiamandolo con un timeout di 5 secondi:

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