Timeout on a function call

Question

I m calling a function in Python which I know may stall and force me to restart the script    How do I call the function or what do I wrap it in so that if it takes longer than 5 seconds the script cancels it and does something else

User · Answer

Building on and and enhancing the answer by @piro , you can build a contextmanager. This allows for very readable code which will disable the alaram signal after a successful run (sets signal.alarm(0))

@contextmanager
def timeout(duration):
    def timeout_handler(signum, frame):
        raise Exception(f'block timedout after {duration} seconds')
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(duration)
    yield
    signal.alarm(0)

def sleeper(duration):
    time.sleep(duration)
    print('finished')

Example usage:

In [19]: with timeout(2):
    ...:     sleeper(1)
    ...:     
finished

In [20]: with timeout(2):
    ...:     sleeper(3)
    ...:         
---------------------------------------------------------------------------
Exception                                 Traceback (most recent call last)
<ipython-input-20-66c78858116f> in <module>()
      1 with timeout(2):
----> 2     sleeper(3)
      3 

<ipython-input-7-a75b966bf7ac> in sleeper(t)
      1 def sleeper(t):
----> 2     time.sleep(t)
      3     print('finished')
      4 

<ipython-input-18-533b9e684466> in timeout_handler(signum, frame)
      2 def timeout(duration):
      3     def timeout_handler(signum, frame):
----> 4         raise Exception(f'block timedout after {duration} seconds')
      5     signal.signal(signal.SIGALRM, timeout_handler)
      6     signal.alarm(duration)

Exception: block timedout after 2 seconds

User · Answer

There are a lot of suggestions  but none using concurrent futures  which I think is the most legible way to handle this   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 simple to read and maintain   We make a pool  submit a single process and then wait up to 5 seconds before raising a TimeoutError that you could catch and handle however you needed   Native to python 3 2  and backported to 2 7  pip install futures    Switching between threads and processes is as simple as replacing ProcessPoolExecutor with ThreadPoolExecutor   If you want to terminate the Process on timeout I would suggest looking into Pebble

User · Answer

Highlights  Raises TimeoutError uses exceptions to alert on timeout - can easily be modified Cross Platform  Windows  amp  Mac OS X Compatibility  Python 3 6   I also tested on python 2 7 and it works with small syntax adjustments   For full explanation and extension to parallel maps  see here https   flipdazed github io blog quant 20dev parallel-functions-with-timeouts Minimal Example  gt  gt  gt   killer call timeout 4      def bar x              import time            time sleep x             return x  gt  gt  gt  bar 10  Traceback  most recent call last           main   TimeoutError  function  bar  timed out after 4s  and as expected  gt  gt  gt  bar 2  2  Full code import multiprocessing as mp import multiprocessing queues as mpq import functools import dill  from typing import Tuple  Callable  Dict  Optional  Iterable  List  Any  class TimeoutError Exception        def   init   self  func  Callable  timeout  int           self t   timeout         self fname   func   name        def   str   self               return f quot function   self fname   timed out after  self t s quot    def  lemmiwinks func  Callable  args  Tuple  kwargs  Dict str  Any   q  mp Queue        quot  quot  quot lemmiwinks crawls into the unknown quot  quot  quot      q put dill loads func   args    kwargs     def killer call func  Callable   None  timeout  int   10  - gt  Callable       quot  quot  quot      Single function call with a timeout      Args          func  the function         timeout  The timeout in seconds      quot  quot  quot       if not isinstance timeout  int           raise ValueError f timeout needs to be an int  Got   timeout         if func is None          return functools partial killer call  timeout timeout        functools wraps killer call      def  inners  args    kwargs  - gt  Any          q worker   mp Queue           proc   mp Process target  lemmiwinks  args  dill dumps func   args  kwargs  q worker           proc start           try              return q worker get timeout timeout          except mpq Empty              raise TimeoutError func  timeout          finally              try                  proc terminate               except                  pass     return  inners  if   name         main          killer call timeout 4      def bar x           import time         time sleep x          return x      print bar 2       bar 10   Notes You will need to import inside the function because of the way dill works  This will also mean these functions may not be not compatible with doctest if there are imports inside your target functions  You will get an issue with   import   not found

User · Answer

Highlights  Raises TimeoutError uses exceptions to alert on timeout - can easily be modified Cross Platform  Windows  amp  Mac OS X Compatibility  Python 3 6   I also tested on python 2 7 and it works with small syntax adjustments   For full explanation and extension to parallel maps  see here https   flipdazed github io blog quant 20dev parallel-functions-with-timeouts Minimal Example  gt  gt  gt   killer call timeout 4      def bar x              import time            time sleep x             return x  gt  gt  gt  bar 10  Traceback  most recent call last           main   TimeoutError  function  bar  timed out after 4s  and as expected  gt  gt  gt  bar 2  2  Full code import multiprocessing as mp import multiprocessing queues as mpq import functools import dill  from typing import Tuple  Callable  Dict  Optional  Iterable  List  Any  class TimeoutError Exception        def   init   self  func  Callable  timeout  int           self t   timeout         self fname   func   name        def   str   self               return f quot function   self fname   timed out after  self t s quot    def  lemmiwinks func  Callable  args  Tuple  kwargs  Dict str  Any   q  mp Queue        quot  quot  quot lemmiwinks crawls into the unknown quot  quot  quot      q put dill loads func   args    kwargs     def killer call func  Callable   None  timeout  int   10  - gt  Callable       quot  quot  quot      Single function call with a timeout      Args          func  the function         timeout  The timeout in seconds      quot  quot  quot       if not isinstance timeout  int           raise ValueError f timeout needs to be an int  Got   timeout         if func is None          return functools partial killer call  timeout timeout        functools wraps killer call      def  inners  args    kwargs  - gt  Any          q worker   mp Queue           proc   mp Process target  lemmiwinks  args  dill dumps func   args  kwargs  q worker           proc start           try              return q worker get timeout timeout          except mpq Empty              raise TimeoutError func  timeout          finally              try                  proc terminate               except                  pass     return  inners  if   name         main          killer call timeout 4      def bar x           import time         time sleep x          return x      print bar 2       bar 10   Notes You will need to import inside the function because of the way dill works  This will also mean these functions may not be not compatible with doctest if there are imports inside your target functions  You will get an issue with   import   not found

User · Answer

Here is a POSIX version that combines many of the previous answers to deliver following features   Subprocesses blocking the execution  Usage of the timeout function on class member functions  Strict requirement on time-to-terminate   Here is the code and some test cases  import threading import signal import os import time  class TerminateExecution Exception        quot  quot  quot      Exception to indicate that execution has exceeded the preset running time       quot  quot  quot    def quit function pid         Killing all subprocesses     os setpgrp       os killpg 0  signal SIGTERM         Killing the main thread     os kill pid  signal SIGTERM    def handle term signum  frame       raise TerminateExecution     def invoke with timeout timeout  fn   args    kwargs         Setting a sigterm handler and initiating a timer     old handler   signal signal signal SIGTERM  handle term      timer   threading Timer timeout  quit function  args  os getpid         terminate   False        Executing the function     timer start       try          result   fn  args    kwargs      except TerminateExecution          terminate   True     finally            Restoring original handler and cancel timer         signal signal signal SIGTERM  old handler          timer cancel        if terminate          raise BaseException  quot xxx quot        return result      Test cases def countdown n       print  countdown started   flush True      for i in range n  -1  -1           print i  end       flush True          time sleep 1      print  countdown finished       return 1337   def really long function        time sleep 10    def really long function2        os system  quot sleep 787 quot       Checking that we can run a function as expected  assert invoke with timeout 3  countdown  1     1337    Testing various scenarios t1   time time   try      print invoke with timeout 1  countdown  3       assert False  except BaseException      assert time time   - t1  lt  1 1      print  quot All good quot   time time   - t1   t1   time time   try      print invoke with timeout 1  really long function2       assert False  except BaseException      assert time time   - t1  lt  1 1      print  quot All good quot   time time   - t1    t1   time time   try      print invoke with timeout 1  really long function       assert False  except BaseException      assert time time   - t1  lt  1 1      print  quot All good quot   time time   - t1     Checking that classes are referenced and not   copied  as would be the case with multiprocessing    class X      def   init   self           self value   0      def set self  v           self value   v   x   X   invoke with timeout 2  x set  9  assert x value    9

User · Answer

I had a need for nestable timed interrupts  which SIGALARM can t do  that won t get blocked by time sleep  which the thread-based approach can t do   I ended up copying and lightly modifying code from here  http   code activestate com recipes 577600-queue-for-managing-multiple-sigalrm-alarms-concurr   The code itself      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  gt  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  lt   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     and a usage example   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

User · Answer

We can use signals for the same  I think the below example will be useful for you  It is very simple compared to 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

User · Answer

You may use the signal package if you are running on UNIX  In  1   import signal    Register an handler for the timeout In  2   def handler signum  frame               print  quot Forever is over  quot               raise Exception  quot end of time quot               This function  may  run for an indetermined time    In  3   def loop forever                import time             while 1                  print  quot sec quot                   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 seconds after the call signal alarm 10   the handler is called  This raises an exception that you can intercept from the regular Python code  This module doesn t play well with threads  but then  who does   Note that since we raise an exception when timeout happens  it may end up caught and ignored inside the function  for example of one such function  def loop forever        while 1          print  sec           try              time sleep 10          except              continue

User · Answer

I ran across this thread when searching for a timeout call on unit tests   I didn t find anything simple in the answers or 3rd party packages so I wrote the decorator below you can drop right into code   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   Then it s as simple as this to timeout a test or any function you like    timeout 5 0     if execution takes longer than 5 seconds  raise a TimeoutError def test base regression self

User · Answer

Another solution with asyncio   If you want to cancel the background task and not just timeout on the running main code  then you need an explicit communication from main thread to ask the code of the task to cancel   like a threading Event   import asyncio import functools import multiprocessing from concurrent futures thread import ThreadPoolExecutor   class SingletonTimeOut      pool   None       classmethod     def run cls  to run  functools partial  timeout  float           pool   cls get pool           loop   cls get loop           try              task   loop run in executor pool  to run              return loop run until complete asyncio wait for task  timeout timeout           except asyncio TimeoutError as e              error type   type e    name    TODO             raise e       classmethod     def get pool cls           if cls pool is None              cls pool   ThreadPoolExecutor multiprocessing cpu count            return cls pool       classmethod     def get loop cls           try              return asyncio get event loop           except RuntimeError              asyncio set event loop asyncio new event loop                  print  quot NEW LOOP quot    str threading current thread   ident               return asyncio get event loop      ---------------  TIME OUT   float  0 2      seconds  def toto input items nb predictions       return 1  to run   functools partial toto                             input items 1                             nb predictions  quot a quot    results   SingletonTimeOut run to run  TIME OUT

User · Answer

You can use multiprocessing Process to do exactly that  Code import multiprocessing import time    bar def bar        for i in range 100           print  quot Tick quot          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  quot running    let s kill it    quot             Terminate - may not work if process is stuck for good         p terminate             OR Kill - will work for sure  no chance for process to finish nicely however           p kill            p join

User · Answer

I have a different proposal which is a pure function  with the same API as the threading suggestion  and seems to work fine  based on suggestions on this 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

User · Answer

The stopit package  found on pypi  seems to handle timeouts well   I like the  stopit threading timeoutable decorator  which adds a timeout parameter to the decorated function  which does what you expect  it stops the function   Check it out on pypi  https   pypi python org pypi stopit

User · Answer

How do I call the function or what do I wrap it in so that if it takes longer than 5 seconds the script cancels it    I posted a gist that solves this question problem with a decorator and a threading Timer  Here it is with a breakdown    Imports and setups for compatibility  It was tested with Python 2 and 3  It should also work under Unix Linux and Windows   First the imports  These attempt to keep the code consistent regardless of the Python version   from   future   import print function import sys import threading from time import sleep try      import thread except ImportError      import  thread as thread   Use version independent 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   Now we have imported our functionality from the standard library    exit after decorator  Next we need a function to terminate the main   from the child thread   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   And here is the decorator itself   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   Usage  And here s the usage that directly answers your question about exiting after 5 seconds     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    gt  gt  gt  countdown 3  countdown started 3  2  1  0  countdown finished  gt  gt  gt  countdown 10  countdown started 10  9  8  7  6  countdown took too long Traceback  most recent call last     File   lt stdin gt    line 1  in  lt module gt    File   lt stdin gt    line 11  in inner   File   lt stdin gt    line 6  in countdown KeyboardInterrupt   The second function call will not finish  instead the process should exit with a traceback   KeyboardInterrupt does not always stop a sleeping thread  Note that sleep will not always be interrupted by a keyboard interrupt  on Python 2 on Windows  e g     exit after 1  def sleep10        sleep 10      print  slept 10 seconds     gt  gt  gt  sleep10   sleep10 took too long           Note that it hangs here about 9 more seconds Traceback  most recent call last     File   lt stdin gt    line 1  in  lt module gt    File   lt stdin gt    line 11  in inner   File   lt stdin gt    line 3  in sleep10 KeyboardInterrupt   nor is it likely to interrupt code running in extensions unless it explicitly checks for PyErr CheckSignals    see  Cython  Python and KeyboardInterrupt ignored  I would avoid sleeping a thread more than a second  in any case - that s an eon in processor time      How do I call the function or what do I wrap it in so that if it takes longer than 5 seconds the script cancels it and does something else    To catch it and do something else  you can catch the KeyboardInterrupt    gt  gt  gt  try          countdown 10      except KeyboardInterrupt          print  do something else        countdown started 10  9  8  7  6  countdown took too long do something else

User · Answer

I ran across this thread when searching for a timeout call on unit tests   I didn t find anything simple in the answers or 3rd party packages so I wrote the decorator below you can drop right into code   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   Then it s as simple as this to timeout a test or any function you like    timeout 5 0     if execution takes longer than 5 seconds  raise a TimeoutError def test base regression self

User · Answer

How do I call the function or what do I wrap it in so that if it takes longer than 5 seconds the script cancels it    I posted a gist that solves this question problem with a decorator and a threading Timer  Here it is with a breakdown    Imports and setups for compatibility  It was tested with Python 2 and 3  It should also work under Unix Linux and Windows   First the imports  These attempt to keep the code consistent regardless of the Python version   from   future   import print function import sys import threading from time import sleep try      import thread except ImportError      import  thread as thread   Use version independent 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   Now we have imported our functionality from the standard library    exit after decorator  Next we need a function to terminate the main   from the child thread   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   And here is the decorator itself   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   Usage  And here s the usage that directly answers your question about exiting after 5 seconds     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    gt  gt  gt  countdown 3  countdown started 3  2  1  0  countdown finished  gt  gt  gt  countdown 10  countdown started 10  9  8  7  6  countdown took too long Traceback  most recent call last     File   lt stdin gt    line 1  in  lt module gt    File   lt stdin gt    line 11  in inner   File   lt stdin gt    line 6  in countdown KeyboardInterrupt   The second function call will not finish  instead the process should exit with a traceback   KeyboardInterrupt does not always stop a sleeping thread  Note that sleep will not always be interrupted by a keyboard interrupt  on Python 2 on Windows  e g     exit after 1  def sleep10        sleep 10      print  slept 10 seconds     gt  gt  gt  sleep10   sleep10 took too long           Note that it hangs here about 9 more seconds Traceback  most recent call last     File   lt stdin gt    line 1  in  lt module gt    File   lt stdin gt    line 11  in inner   File   lt stdin gt    line 3  in sleep10 KeyboardInterrupt   nor is it likely to interrupt code running in extensions unless it explicitly checks for PyErr CheckSignals    see  Cython  Python and KeyboardInterrupt ignored  I would avoid sleeping a thread more than a second  in any case - that s an eon in processor time      How do I call the function or what do I wrap it in so that if it takes longer than 5 seconds the script cancels it and does something else    To catch it and do something else  you can catch the KeyboardInterrupt    gt  gt  gt  try          countdown 10      except KeyboardInterrupt          print  do something else        countdown started 10  9  8  7  6  countdown took too long do something else

User · Answer

Great  easy to use and reliable PyPi project timeout-decorator  https   pypi org project timeout-decorator    installation   pip install timeout-decorator   Usage   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

User · Answer

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

User · Answer

There are a lot of suggestions  but none using concurrent futures  which I think is the most legible way to handle this   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 simple to read and maintain   We make a pool  submit a single process and then wait up to 5 seconds before raising a TimeoutError that you could catch and handle however you needed   Native to python 3 2  and backported to 2 7  pip install futures    Switching between threads and processes is as simple as replacing ProcessPoolExecutor with ThreadPoolExecutor   If you want to terminate the Process on timeout I would suggest looking into Pebble

User · Answer

timeout-decorator don t work on windows system as   windows didn t support signal well   If you use timeout-decorator in windows system you will get the following   AttributeError  module  signal  has no attribute  SIGALRM    Some suggested to use use signals False but didn t worked for me   Author  bitranox created the following package   pip install https   github com bitranox wrapt-timeout-decorator archive master zip   Code Sample    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     Gives the following exception   TimeoutError  Function mytest timed out after 5 seconds

User · Answer

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

User · Answer

Here is a slight improvement to the given thread-based solution   The code below supports exceptions   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    Invoking it with a 5 second timeout   result   timeout remote calculate   myarg    timeout duration 5

User · Answer

I am the author of wrapt timeout decorator Most of the solutions presented here work wunderfully under Linux on the first glance - because we have fork   and signals   - but on windows the things look a bit different  And when it comes to subthreads on Linux  You cant use Signals anymore  In order to spawn a process under Windows  it needs to be picklable - and many decorated functions or Class methods are not  So You need to use a better pickler like dill and multiprocess  not pickle and multiprocessing  - thats why You cant use ProcessPoolExecutor  or only with limited functionality   For the timeout itself - You need to define what timeout means - because on Windows it will take considerable  and not determinable  time to spawn the process  This can be tricky on short timeouts  Lets assume  spawning the process takes about 0 5 seconds  easily       If You give a timeout of 0 2 seconds what should happen   Should the function time out after 0 5   0 2 seconds  so let the method run for 0 2 seconds   Or should the called process time out after 0 2 seconds  in that case  the decorated function will ALWAYS timeout  because in that time it is not even spawned    Also nested decorators can be nasty and You cant use Signals in a subthread  If You want to create a truly universal  cross-platform decorator  all this needs to be taken into consideration  and tested   Other issues are passing exceptions back to the caller  as well as logging issues  if used in the decorated function - logging to files in another process is NOT supported  I tried to cover all edge cases  You might look into the package wrapt timeout decorator  or at least test Your own solutions inspired by the unittests used there   Alexis Eggermont - unfortunately I dont have enough points to comment - maybe someone else can notify You - I think I solved Your import issue

User · Answer

I had a need for nestable timed interrupts  which SIGALARM can t do  that won t get blocked by time sleep  which the thread-based approach can t do   I ended up copying and lightly modifying code from here  http   code activestate com recipes 577600-queue-for-managing-multiple-sigalrm-alarms-concurr   The code itself      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  gt  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  lt   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     and a usage example   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

User · Answer

We can use signals for the same  I think the below example will be useful for you  It is very simple compared to 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

User · Answer

Here is a POSIX version that combines many of the previous answers to deliver following features   Subprocesses blocking the execution  Usage of the timeout function on class member functions  Strict requirement on time-to-terminate   Here is the code and some test cases  import threading import signal import os import time  class TerminateExecution Exception        quot  quot  quot      Exception to indicate that execution has exceeded the preset running time       quot  quot  quot    def quit function pid         Killing all subprocesses     os setpgrp       os killpg 0  signal SIGTERM         Killing the main thread     os kill pid  signal SIGTERM    def handle term signum  frame       raise TerminateExecution     def invoke with timeout timeout  fn   args    kwargs         Setting a sigterm handler and initiating a timer     old handler   signal signal signal SIGTERM  handle term      timer   threading Timer timeout  quit function  args  os getpid         terminate   False        Executing the function     timer start       try          result   fn  args    kwargs      except TerminateExecution          terminate   True     finally            Restoring original handler and cancel timer         signal signal signal SIGTERM  old handler          timer cancel        if terminate          raise BaseException  quot xxx quot        return result      Test cases def countdown n       print  countdown started   flush True      for i in range n  -1  -1           print i  end       flush True          time sleep 1      print  countdown finished       return 1337   def really long function        time sleep 10    def really long function2        os system  quot sleep 787 quot       Checking that we can run a function as expected  assert invoke with timeout 3  countdown  1     1337    Testing various scenarios t1   time time   try      print invoke with timeout 1  countdown  3       assert False  except BaseException      assert time time   - t1  lt  1 1      print  quot All good quot   time time   - t1   t1   time time   try      print invoke with timeout 1  really long function2       assert False  except BaseException      assert time time   - t1  lt  1 1      print  quot All good quot   time time   - t1    t1   time time   try      print invoke with timeout 1  really long function       assert False  except BaseException      assert time time   - t1  lt  1 1      print  quot All good quot   time time   - t1     Checking that classes are referenced and not   copied  as would be the case with multiprocessing    class X      def   init   self           self value   0      def set self  v           self value   v   x   X   invoke with timeout 2  x set  9  assert x value    9

User · Answer

Building on and and enhancing the answer by @piro , you can build a contextmanager. This allows for very readable code which will disable the alaram signal after a successful run (sets signal.alarm(0))

@contextmanager
def timeout(duration):
    def timeout_handler(signum, frame):
        raise Exception(f'block timedout after {duration} seconds')
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(duration)
    yield
    signal.alarm(0)

def sleeper(duration):
    time.sleep(duration)
    print('finished')

Example usage:

In [19]: with timeout(2):
    ...:     sleeper(1)
    ...:     
finished

In [20]: with timeout(2):
    ...:     sleeper(3)
    ...:         
---------------------------------------------------------------------------
Exception                                 Traceback (most recent call last)
<ipython-input-20-66c78858116f> in <module>()
      1 with timeout(2):
----> 2     sleeper(3)
      3 

<ipython-input-7-a75b966bf7ac> in sleeper(t)
      1 def sleeper(t):
----> 2     time.sleep(t)
      3     print('finished')
      4 

<ipython-input-18-533b9e684466> in timeout_handler(signum, frame)
      2 def timeout(duration):
      3     def timeout_handler(signum, frame):
----> 4         raise Exception(f'block timedout after {duration} seconds')
      5     signal.signal(signal.SIGALRM, timeout_handler)
      6     signal.alarm(duration)

Exception: block timedout after 2 seconds

User · Answer

timeout-decorator don t work on windows system as   windows didn t support signal well   If you use timeout-decorator in windows system you will get the following   AttributeError  module  signal  has no attribute  SIGALRM    Some suggested to use use signals False but didn t worked for me   Author  bitranox created the following package   pip install https   github com bitranox wrapt-timeout-decorator archive master zip   Code Sample    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     Gives the following exception   TimeoutError  Function mytest timed out after 5 seconds

User · Answer

You can use multiprocessing Process to do exactly that  Code import multiprocessing import time    bar def bar        for i in range 100           print  quot Tick quot          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  quot running    let s kill it    quot             Terminate - may not work if process is stuck for good         p terminate             OR Kill - will work for sure  no chance for process to finish nicely however           p kill            p join

User · Answer

You may use the signal package if you are running on UNIX  In  1   import signal    Register an handler for the timeout In  2   def handler signum  frame               print  quot Forever is over  quot               raise Exception  quot end of time quot               This function  may  run for an indetermined time    In  3   def loop forever                import time             while 1                  print  quot sec quot                   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 seconds after the call signal alarm 10   the handler is called  This raises an exception that you can intercept from the regular Python code  This module doesn t play well with threads  but then  who does   Note that since we raise an exception when timeout happens  it may end up caught and ignored inside the function  for example of one such function  def loop forever        while 1          print  sec           try              time sleep 10          except              continue

User · Answer

Another solution with asyncio   If you want to cancel the background task and not just timeout on the running main code  then you need an explicit communication from main thread to ask the code of the task to cancel   like a threading Event   import asyncio import functools import multiprocessing from concurrent futures thread import ThreadPoolExecutor   class SingletonTimeOut      pool   None       classmethod     def run cls  to run  functools partial  timeout  float           pool   cls get pool           loop   cls get loop           try              task   loop run in executor pool  to run              return loop run until complete asyncio wait for task  timeout timeout           except asyncio TimeoutError as e              error type   type e    name    TODO             raise e       classmethod     def get pool cls           if cls pool is None              cls pool   ThreadPoolExecutor multiprocessing cpu count            return cls pool       classmethod     def get loop cls           try              return asyncio get event loop           except RuntimeError              asyncio set event loop asyncio new event loop                  print  quot NEW LOOP quot    str threading current thread   ident               return asyncio get event loop      ---------------  TIME OUT   float  0 2      seconds  def toto input items nb predictions       return 1  to run   functools partial toto                             input items 1                             nb predictions  quot a quot    results   SingletonTimeOut run to run  TIME OUT

User · Answer

I have a different proposal which is a pure function  with the same API as the threading suggestion  and seems to work fine  based on suggestions on this 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

User · Answer

I am the author of wrapt timeout decorator Most of the solutions presented here work wunderfully under Linux on the first glance - because we have fork   and signals   - but on windows the things look a bit different  And when it comes to subthreads on Linux  You cant use Signals anymore  In order to spawn a process under Windows  it needs to be picklable - and many decorated functions or Class methods are not  So You need to use a better pickler like dill and multiprocess  not pickle and multiprocessing  - thats why You cant use ProcessPoolExecutor  or only with limited functionality   For the timeout itself - You need to define what timeout means - because on Windows it will take considerable  and not determinable  time to spawn the process  This can be tricky on short timeouts  Lets assume  spawning the process takes about 0 5 seconds  easily       If You give a timeout of 0 2 seconds what should happen   Should the function time out after 0 5   0 2 seconds  so let the method run for 0 2 seconds   Or should the called process time out after 0 2 seconds  in that case  the decorated function will ALWAYS timeout  because in that time it is not even spawned    Also nested decorators can be nasty and You cant use Signals in a subthread  If You want to create a truly universal  cross-platform decorator  all this needs to be taken into consideration  and tested   Other issues are passing exceptions back to the caller  as well as logging issues  if used in the decorated function - logging to files in another process is NOT supported  I tried to cover all edge cases  You might look into the package wrapt timeout decorator  or at least test Your own solutions inspired by the unittests used there   Alexis Eggermont - unfortunately I dont have enough points to comment - maybe someone else can notify You - I think I solved Your import issue

User · Answer

The stopit package  found on pypi  seems to handle timeouts well   I like the  stopit threading timeoutable decorator  which adds a timeout parameter to the decorated function  which does what you expect  it stops the function   Check it out on pypi  https   pypi python org pypi stopit

User · Answer

You may use the signal package if you are running on UNIX  In  1   import signal    Register an handler for the timeout In  2   def handler signum  frame               print  quot Forever is over  quot               raise Exception  quot end of time quot               This function  may  run for an indetermined time    In  3   def loop forever                import time             while 1                  print  quot sec quot                   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 seconds after the call signal alarm 10   the handler is called  This raises an exception that you can intercept from the regular Python code  This module doesn t play well with threads  but then  who does   Note that since we raise an exception when timeout happens  it may end up caught and ignored inside the function  for example of one such function  def loop forever        while 1          print  sec           try              time sleep 10          except              continue

User · Answer

Great  easy to use and reliable PyPi project timeout-decorator  https   pypi org project timeout-decorator    installation   pip install timeout-decorator   Usage   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

User · Answer

Here is a slight improvement to the given thread-based solution   The code below supports exceptions   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    Invoking it with a 5 second timeout   result   timeout remote calculate   myarg    timeout duration 5

User · Answer

You may use the signal package if you are running on UNIX  In  1   import signal    Register an handler for the timeout In  2   def handler signum  frame               print  quot Forever is over  quot               raise Exception  quot end of time quot               This function  may  run for an indetermined time    In  3   def loop forever                import time             while 1                  print  quot sec quot                   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 seconds after the call signal alarm 10   the handler is called  This raises an exception that you can intercept from the regular Python code  This module doesn t play well with threads  but then  who does   Note that since we raise an exception when timeout happens  it may end up caught and ignored inside the function  for example of one such function  def loop forever        while 1          print  sec           try              time sleep 10          except              continue

[python] Timeout on a function call

Examples related to python

Examples related to multithreading

Examples related to timeout

Examples related to python-multithreading