Tiempo de espera en una llamada de función
https://stackoverflow.com/questions/492519
-
20-08-2019 - |
italiano
english
français
española
中国
日本の
العربية
Deutsch
한국어
Português
RussianPregunta
Llamo a una función en Python que sé que puede detenerse y obligarme a reiniciar el script.
¿Cómo llamo a la función o en qué la envuelvo para que, si tarda más de 5 segundos, el script la cancela y hace algo más?
Solución
Puede usar el paquete signal 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 del código Python normal.
Este módulo no funciona bien con hilos (pero, ¿quién lo hace?)
Tenga en cuenta que dado que activamos una excepción cuando se agota el tiempo de espera, puede terminar atrapada e ignorada dentro de la función, por ejemplo, una de esas funciones:
def loop_forever():
while 1:
print 'sec'
try:
time.sleep(10)
except:
continue
Otros consejos
Puede usar 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 tarda más de 5 segundos, el script la cancela?
Publiqué un gist que resuelve esta pregunta / problema con un decorador y un threading.Timer . Aquí está con un desglose.
Importaciones y configuraciones para compatibilidad
Fue probado 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
Usar código independiente de la versión:
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 de la biblioteca estándar.
exit_after decorador
A continuación, necesitamos una función para terminar el main() del hilo 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 decorador en sí:
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 su pregunta sobre 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')
Demostració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 "<stdin>", line 1, in <module>
File "<stdin>", line 11, in inner
File "<stdin>", line 6, in countdown
KeyboardInterrupt
¡La segunda llamada de función no finalizará, en su lugar, el proceso debería salir con un rastreo!
KeyboardInterrupt no siempre detiene un hilo dormido
Tenga en cuenta que el sueño no siempre será interrumpido 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 "<stdin>", line 1, in <module>
File "<stdin>", line 11, in inner
File "<stdin>", line 3, in sleep10
KeyboardInterrupt
ni es probable que interrumpa el código que se ejecuta en extensiones a menos que compruebe explícitamente PyErr_CheckSignals(), consulte Cython, Python y KeyboardInterrupt ignorado
Evitaría dormir un hilo más de un segundo, en cualquier caso, eso es un eón en tiempo de procesador.
¿Cómo llamo a la función o en qué la envuelvo para que, si tarda más de 5 segundos, el script la cancela y hace algo más?
Para atraparlo y hacer otra cosa, puede atrapar el KeyboardInterrupt.
>>> 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 subprocesos) y parece funcionar bien (según 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 los paquetes de terceros, así que escribí el decorador a continuación, puede colocarlo directamente en el código:
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 agotar el tiempo de espera de una prueba o cualquier función que desee:
@timeout(5.0) # if execution takes longer than 5 seconds, raise a TimeoutError
def test_base_regression(self):
...
Hay muchas sugerencias, pero ninguna con 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.
Creamos 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.
Nativo de python 3.2+ y con respaldo a 2.7 (futuros de instalación de pip).
Cambiar entre hilos y procesos es tan simple como reemplazar ProcessPoolExecutor con ThreadPoolExecutor.
Si desea finalizar el proceso en el tiempo de espera, le sugiero que busque 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 timeout a la función decorada, que hace lo que espera, detiene la función.
Compruébelo en pypi: https://pypi.python.org/pypi/stopit
Excelente, fácil de usar y confiable PyPi proyecto timeout-decorator ( 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)
timeout-decorator no funciona en el sistema de Windows, ya que Windows no era compatible con signal bien.
Si usa timeout-decorator en el sistema de Windows obtendrá lo siguiente
AttributeError: module 'signal' has no attribute 'SIGALRM'
Algunos sugirieron usar use_signals=False pero no funcionaron para mí.
El autor @bitranox creó el siguiente paquete:
pip install https://github.com/bitranox/wrapt-timeout-decorator/archive/master.zip
Muestra 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
Soy el autor de wrapt_timeout_decorator
La mayoría de las soluciones presentadas aquí funcionan maravillosamente bajo Linux a primera vista, porque tenemos fork () y señales (), pero en Windows las cosas se ven un poco diferentes. Y cuando se trata de subprocesos en Linux, ya no puede usar señales.
Para generar un proceso en Windows, debe ser seleccionable, y muchas funciones decoradas o métodos de clase no lo son.
Por lo tanto, debe usar un mejor selector como eneldo y multiproceso (no encurtido y multiprocesamiento), es por eso que no puede usar ProcessPoolExecutor (o solo con una funcionalidad limitada).
Para el tiempo de espera en sí mismo: debe definir qué significa el tiempo de espera, porque en Windows tomará un tiempo considerable (y no determinable) para generar el proceso. Esto puede ser complicado en tiempos de espera cortos. Supongamos que generar el proceso lleva unos 0,5 segundos (¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡!!!). Si le das un tiempo de espera de 0.2 segundos, ¿qué debería pasar? ¿Debería la función agotar el tiempo de espera después de 0.5 + 0.2 segundos (así que deje que el método se ejecute durante 0.2 segundos)? ¿O debería agotar el tiempo del proceso llamado después de 0.2 segundos (en ese caso, la función decorada SIEMPRE expira, porque en ese tiempo ni siquiera se genera)?
También los decoradores anidados pueden ser desagradables y no se pueden usar señales en un subproceso. Si desea crear un decorador multiplataforma verdaderamente universal, todo esto debe tenerse en cuenta (y probarse).
Otros problemas son pasar excepciones a la persona que llama, así como problemas de registro (si se usa en la función decorada, el registro a archivos en otro proceso NO es compatible)
Traté de cubrir todos los casos extremos. Puede consultar el paquete wrapt_timeout_decorator, o al menos probar sus propias soluciones inspiradas en las pruebas unitarias utilizadas allí.
@Alexis Eggermont - desafortunadamente no tengo suficientes puntos para comentar - tal vez alguien más pueda notificarte - Creo que resolví tu problema de importación.
Podemos usar señales para lo mismo. Creo que el siguiente ejemplo te será útil. Es muy simple 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"
Necesitaba interrupciones temporizadas anidables (que SIGALARM no puede hacer) que no se bloqueen con 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 en la solución basada en hilos dada.
El siguiente código 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]
Invocarlo con un tiempo de espera de 5 segundos:
result = timeout(remote_calculate, (myarg,), timeout_duration=5)
