Tiempo de espera en una llamada de función

Estoy llamando a una función en Python que sé que puede detenerse y forzarme a reiniciar el script.

¿Cómo llamo a la función o en qué la envuelvo para que, si demora más de 5 segundos, el script la cancele y haga otra cosa?

Puede utilizar el paquete de señal si está ejecutando en 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 segundos después de la llamada alarm.alarm(10) , se llama al controlador. Esto genera una excepción que puede interceptar desde el código de Python normal.

Este módulo no juega bien con hilos (pero entonces, ¿quién lo hace?)

Tenga en cuenta que dado que provocamos una excepción cuando se agota el tiempo de espera, puede terminar atrapado e ignorado dentro de la función, por ejemplo, de una de esas funciones:

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

Puede usar el proceso de multiprocessing.Process Para hacer exactamente eso.

Código

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

¿Cómo llamo a la función o en qué la envuelvo para que, si demora más de 5 segundos, el script la cancele?

Publiqué una idea general que resuelve esta pregunta / problema con un decorador y un threading.Timer . Aquí está con un desglose.

Importaciones y configuraciones para compatibilidad.

Se probó con Python 2 y 3. También debería funcionar en Unix / Linux y Windows.

Primero las importaciones. Estos intentan mantener el código consistente independientemente de la versión de Python:

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

Usa código de versión independiente:

 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 

Ahora hemos importado nuestra funcionalidad desde la librería estándar.

exit_after decorador

A continuación necesitamos una función para terminar el main() del subproceso secundario:

 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 

Y aquí está el propio decorador:

 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

¡Y aquí está el uso que responde directamente a tu pregunta sobre cómo salir después de 5 segundos !:

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

Manifestación:

 >>> 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 "", line 1, in  File "", line 11, in inner File "", line 6, in countdown KeyboardInterrupt 

La segunda llamada a la función no terminará, en su lugar, el proceso debería salir con un rastreo.

KeyboardInterrupt no siempre detiene un hilo durmiente

Tenga en cuenta que la suspensión no siempre será interrumpida por una interrupción del teclado, en Python 2 en Windows, por ejemplo:

 @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 "", line 1, in  File "", line 11, in inner File "", line 3, in sleep10 KeyboardInterrupt 

ni es probable que interrumpa el código que se ejecuta en las extensiones a menos que compruebe explícitamente que PyErr_CheckSignals() , vea Cython, Python y KeyboardInterrupt ignorado

Evitaría dormir un hilo más de un segundo, en cualquier caso, eso es un eon en tiempo de procesador.

¿Cómo llamo a la función o en qué la envuelvo para que, si demora más de 5 segundos, el script la cancele y haga otra cosa?

Para atraparlo y hacer otra cosa, puede atrapar el Interruptor de teclado.

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

Tengo una propuesta diferente que es una función pura (con la misma API que la sugerencia de subprocesamiento) y parece funcionar bien (basado en las sugerencias de este hilo)

 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 

Me encontré con este hilo al buscar una llamada de tiempo de espera en las pruebas unitarias. No encontré nada simple en las respuestas o en los paquetes de terceros, así que escribí el decorador que se encuentra a continuación.

 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 

Entonces es tan simple como el de un examen o cualquier función que te guste:

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

Hay muchas sugerencias, pero ninguna usa concurrent.futures, que creo que es la forma más legible de manejar esto.

 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) 

Súper simple de leer y mantener.

Hacemos un grupo, enviamos un solo proceso y luego esperamos hasta 5 segundos antes de generar un TimeoutError que pueda detectar y manejar como lo necesite.

Es nativo de Python 3.2+ y está respaldado en 2.7 (instalación de futuros).

Cambiar entre subprocesos y procesos es tan simple como reemplazar ProcessPoolExecutor con ThreadPoolExecutor .

Si desea finalizar el proceso en el tiempo de espera, le sugiero que busque en Pebble .

El paquete stopit , que se encuentra en pypi, parece manejar bien los tiempos de espera.

Me gusta el decorador @stopit.threading_timeoutable , que agrega un parámetro de timeout a la función decorada, que hace lo que se espera, detiene la función.

Compruébelo en pypi: https://pypi.python.org/pypi/stopit

Excelente, fácil de usar y confiable proyecto -decorador de tiempo de espera de PyPi ( https://pypi.org/project/timeout-decorator/ )

instalación :

 pip install timeout-decorator 

Uso

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

Tenía la necesidad de interrupciones temporizadas anidables (que SIGALARM no puede hacer) que no se bloquearán por time.sleep (que el enfoque basado en hilos no puede hacer). Terminé copiando y modificando ligeramente el código desde aquí: http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/

El código en sí:

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

y un ejemplo de uso:

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

Aquí hay una ligera mejora de la solución basada en hilos dada.

El código a continuación admite excepciones :

 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] 

Invocando con un tiempo de espera de 5 segundos:

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

Podemos utilizar señales para el mismo. Creo que el siguiente ejemplo será útil para usted. Es muy sencillo en comparación con los hilos.

 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" 

timeout-decorator no funciona en el sistema Windows, ya que Windows no es compatible con la signal .

Si utiliza el tiempo de espera-decorador en el sistema de Windows obtendrá los siguientes

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

Algunos sugirieron usar use_signals=False pero no funcionó para mí.

El autor @bitranox creó el siguiente paquete:

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

Ejemplo de código:

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

Da la siguiente excepción:

 TimeoutError: Function mytest timed out after 5 seconds