blocks|key|1423927|text|我有一个不同的建议，它是一个纯函数(具有与线程建议相同的API+)，并且似乎工作得很好(基于这个线程的建议)。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1423928|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|code-block|syntax|javascript|1423929|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

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)

<pre><code>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
</code></pre>

blocks|key|1638894|text|我们可以使用信号来做同样的事情。我认为下面的例子会对你有用。与线程相比，它非常简单。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1638895|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"|code-block|syntax|javascript|1638896|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

We can use signals for the same. I think the below example will be useful for you. It is very simple compared to threads. 

<pre><code>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"
</code></pre>

blocks|key|1638948|text|在pypi上找到的stopit包似乎很好地处理了超时。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1638949|我喜欢@stopit.threading_timeoutable装饰器，它向装饰性函数添加了一个timeout参数，这会如您所期望的那样停止该函数。|1638950|在pypi上查看：https://pypi.python.org/pypi/stopit|1638951|entityMap|0|LINK|mutability|MUTABLE|url|https://pypi.python.org/pypi/stopit^0|9|6|0|3|T|1C|7|0|9|Z|0|0^^$0|@$1|2|3|4|5|6|7|R|8|@$9|S|A|T|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|U|8|@$9|V|A|W|B|C]|$9|X|A|Y|B|C]]|D|@]|E|$]]|$1|H|3|I|5|6|7|Z|8|@]|D|@$9|10|A|11|1|12]]|E|$]]|$1|J|3|-4|5|6|7|13|8|@]|D|@]|E|$]]]|K|$L|$5|M|N|O|E|$P|Q]]]]

The <code>stopit</code> package, found on pypi, seems to handle timeouts well.

I like the <code>@stopit.threading_timeoutable</code> decorator, which adds a <code>timeout</code> parameter to the decorated function, which does what you expect, it stops the function.

Check it out on pypi: <a href="https://pypi.python.org/pypi/stopit" rel="noreferrer">https://pypi.python.org/pypi/stopit</a>

blocks|key|1639002|text|在单元测试中搜索超时调用时，我遇到了这个线程。我在答案或第三方包中找不到任何简单的东西，所以我写了下面的装饰器，你可以直接进入代码：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1639003|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|code-block|syntax|javascript|1639004|然后，让一个测试或任何你喜欢的函数超时就像下面这样简单：|1639005|@timeout(5.0)++#+if+execution+takes+longer+than+5+seconds,+raise+a+TimeoutError
def+test_base_regression(self):
++++...|1639006|entityMap^0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|N|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|O|8|@]|9|@]|A|$]]|$1|I|3|J|5|D|7|P|8|@]|9|@]|A|$E|F]]|$1|K|3|-4|5|6|7|Q|8|@]|9|@]|A|$]]]|L|$]]

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:

<pre><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
</code></pre>

Then it's as simple as this to timeout a test or any function you like:

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

blocks|key|1424100|text|#!/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)|type|code-block|depth|inlineStyleRanges|entityRanges|data|syntax|javascript|1424101|unstyled|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|G|8|@]|9|@]|A|$B|C]]|$1|D|3|-4|5|E|7|H|8|@]|9|@]|A|$]]]|F|$]]

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

blocks|key|1639092|text|有很多建议，但没有一个使用concurrent.futures，我认为这是处理这个问题最清晰的方法。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1639093|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)|code-block|syntax|javascript|1639094|非常容易阅读和维护。|1639095|我们创建一个池，提交一个进程，然后等待5秒，然后引发一个TimeoutError，您可以根据需要捕获和处理它。|1639096|原生于python+3.2%2B，并向后移植到2.7+(未来的pip安装)。|1639097|在线程和进程之间切换就像用ThreadPoolExecutor替换ProcessPoolExecutor一样简单。|offset|length|style|CODE|1639098|如果您想在超时时终止进程，我建议您查看Pebble。|1639099|entityMap|0|LINK|mutability|MUTABLE|url|http://pythonhosted.org/Pebble/^0|0|0|0|0|0|D|I|X|J|0|J|6|0|0^^$0|@$1|2|3|4|5|6|7|12|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|13|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|14|8|@]|9|@]|A|$]]|$1|I|3|J|5|6|7|15|8|@]|9|@]|A|$]]|$1|K|3|L|5|6|7|16|8|@]|9|@]|A|$]]|$1|M|3|N|5|6|7|17|8|@$O|18|P|19|Q|R]|$O|1A|P|1B|Q|R]]|9|@]|A|$]]|$1|S|3|T|5|6|7|1C|8|@]|9|@$O|1D|P|1E|1|1F]]|A|$]]|$1|U|3|-4|5|6|7|1G|8|@]|9|@]|A|$]]]|V|$W|$5|X|Y|Z|A|$10|11]]]]

There are a lot of suggestions, but none using concurrent.futures, which I think is the most legible way to handle this.

<pre><code>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)
</code></pre>

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 <code>ProcessPoolExecutor</code> with <code>ThreadPoolExecutor</code>.

If you want to terminate the Process on timeout I would suggest looking into <a href="http://pythonhosted.org/Pebble/" rel="noreferrer">Pebble</a>.

blocks|key|1639162|text|timeout-decorator不能在windows系统上工作，因为windows不能很好地支持signal。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1639163|如果你在windows系统中使用timeout-decorator，你会得到以下信息|1639164|AttributeError:+module+'signal'+has+no+attribute+'SIGALRM'|code-block|syntax|javascript|1639165|有些人建议使用use_signals=False，但对我来说不起作用。|1639166|Author+@bitranox创建了以下包：|1639167|pip+install+https://github.com/bitranox/wrapt-timeout-decorator/archive/master.zip|1639168|代码示例：|1639169|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()|1639170|给出了以下异常：|1639171|TimeoutError:+Function+mytest+timed+out+after+5+seconds|1639172|entityMap^0|0|H|1D|6|0|0|0|7|H|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|12|8|@$9|13|A|14|B|C]|$9|15|A|16|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|17|8|@]|D|@]|E|$]]|$1|H|3|I|5|J|7|18|8|@]|D|@]|E|$K|L]]|$1|M|3|N|5|6|7|19|8|@$9|1A|A|1B|B|C]]|D|@]|E|$]]|$1|O|3|P|5|6|7|1C|8|@]|D|@]|E|$]]|$1|Q|3|R|5|J|7|1D|8|@]|D|@]|E|$K|L]]|$1|S|3|T|5|6|7|1E|8|@]|D|@]|E|$]]|$1|U|3|V|5|J|7|1F|8|@]|D|@]|E|$K|L]]|$1|W|3|X|5|6|7|1G|8|@]|D|@]|E|$]]|$1|Y|3|Z|5|J|7|1H|8|@]|D|@]|E|$K|L]]|$1|10|3|-4|5|6|7|1I|8|@]|D|@]|E|$]]]|11|$]]

<code>timeout-decorator</code> don't work on windows system as , windows didn't support <code>signal</code> well.

If you use timeout-decorator in windows system you will get the following 

<pre><code>AttributeError: module 'signal' has no attribute 'SIGALRM'
</code></pre>

Some suggested to use <code>use_signals=False</code> but didn't worked for me.

Author @bitranox created the following package:

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

Code Sample: 

<pre><code>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()
</code></pre>

Gives the following exception:

<pre><code>TimeoutError: Function mytest timed out after 5 seconds
</code></pre>

blocks|key|1639277|text|我是wrapt_timeout_decorator的作者|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1639278|这里介绍的大多数解决方案乍一看在Linux下都工作得很好-因为我们有fork()和signals()+-但在windows上看起来有点不同。当涉及Linux上的子线程时，你不能再使用信号了。|1639279|为了在Windows下生成一个进程，它需要是可拾取的-而许多修饰函数或类方法是不可拾取的。|1639280|所以你需要使用更好的选择器，比如dill和多进程(而不是pickle和多进程)--这就是为什么你不能使用ProcessPoolExecutor+(或者只使用有限的功能)。|1639281|对于超时本身-您需要定义超时意味着什么-因为在Windows上，它将需要相当长的时间(并且不确定)来产生进程。在较短的超时时间内，这可能会很棘手。让我们假设，衍生过程大约需要0.5秒(很容易！)。如果设置了0.2秒的超时，会发生什么情况？函数是否应该在0.5+%2B+0.2秒后超时(所以让方法运行0.2秒)？或者被调用的进程是否应该在0.2秒后超时(在这种情况下，修饰的函数将始终超时，因为在这段时间内它甚至都没有生成)？|1639282|此外，嵌套的装饰器可能会很脏，并且您不能在子线程中使用信号。如果你想创建一个真正通用的，跨平台的装饰器，所有这些都需要被考虑(和测试)。|1639283|其他问题是将异常传递回调用者，以及日志问题(如果在修饰的函数中使用-不支持将日志记录到另一个进程中的文件)|1639284|我试图涵盖所有的边缘情况，你可以查看wrapt_timeout_decorator包，或者至少测试你自己的解决方案，灵感来自于那里使用的单元测试。|1639285|@Alexis+Eggermont+-不幸的是，我没有足够的评论-也许其他人可以通知你-我想我解决了你的导入问题。|1639286|entityMap^0|0|0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|T|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|U|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|V|8|@]|9|@]|A|$]]|$1|F|3|G|5|6|7|W|8|@]|9|@]|A|$]]|$1|H|3|I|5|6|7|X|8|@]|9|@]|A|$]]|$1|J|3|K|5|6|7|Y|8|@]|9|@]|A|$]]|$1|L|3|M|5|6|7|Z|8|@]|9|@]|A|$]]|$1|N|3|O|5|6|7|10|8|@]|9|@]|A|$]]|$1|P|3|Q|5|6|7|11|8|@]|9|@]|A|$]]|$1|R|3|-4|5|6|7|12|8|@]|9|@]|A|$]]]|S|$]]

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.

blocks|key|1639321|text|使用asyncio的另一个解决方案：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1639322|如果您想取消后台任务，而不仅仅是运行的主代码超时，那么您需要一个来自主线程的显式通信来请求要取消的任务的代码，就像threading.Event()|1639323|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("NEW+LOOP"+%2B+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="a")

results+=+SingletonTimeOut.run(to_run,+TIME_OUT)|code-block|syntax|javascript|1639324|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|L|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|M|8|@]|9|@]|A|$G|H]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

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()
<pre class="lang-py prettyprint-override"><code>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)

</code></pre>

blocks|key|1639387|text|在和的基础上，并通过@piro增强答案，您可以构建一个contextmanager。这允许非常易读的代码在成功运行后禁用报警信号(设置signal.alarm(0))|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1639388|from+contextlib+import+contextmanager
import+signal
import+time

@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')|code-block|syntax|javascript|1639389|示例用法：|1639390|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|1639391|entityMap^0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|N|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|O|8|@]|9|@]|A|$]]|$1|I|3|J|5|D|7|P|8|@]|9|@]|A|$E|F]]|$1|K|3|-4|5|6|7|Q|8|@]|9|@]|A|$]]]|L|$]]

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))
<pre><code>from contextlib import contextmanager
import signal
import time

@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')
</code></pre>
Example usage:
<pre><code>In [19]: with timeout(2):
 ...: sleeper(1)
 ...: 
finished

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

&lt;ipython-input-7-a75b966bf7ac&gt; in sleeper(t)
 1 def sleeper(t):
----&gt; 2 time.sleep(t)
 3 print('finished')
 4 

&lt;ipython-input-18-533b9e684466&gt; in timeout_handler(signum, frame)
 2 def timeout(duration):
 3 def timeout_handler(signum, frame):
----&gt; 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
</code></pre>

blocks|key|1639478|text|如果它对任何人有帮助，基于@piro的答案，我做了一个函数装饰器：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1639479|import+time
import+signal
from+functools+import+wraps


def+timeout(timeout_secs:+int):
++++def+wrapper(func):
++++++++@wraps(func)
++++++++def+time_limited(*args,+**kwargs):
++++++++++++#+Register+an+handler+for+the+timeout
++++++++++++def+handler(signum,+frame):
++++++++++++++++raise+Exception(f"Timeout+for+function+'{func.__name__}'")

++++++++++++#+Register+the+signal+function+handler
++++++++++++signal.signal(signal.SIGALRM,+handler)

++++++++++++#+Define+a+timeout+for+your+function
++++++++++++signal.alarm(timeout_secs)

++++++++++++result+=+None
++++++++++++try:
++++++++++++++++result+=+func(*args,+**kwargs)
++++++++++++except+Exception+as+exc:
++++++++++++++++raise+exc
++++++++++++finally:
++++++++++++++++#+disable+the+signal+alarm
++++++++++++++++signal.alarm(0)

++++++++++++return+result

++++++++return+time_limited

++++return+wrapper|code-block|syntax|javascript|1639480|在带有20+seconds超时的函数上使用包装器将如下所示：|offset|length|style|CODE|1639481|++++@timeout(20)
++++def+my_slow_or_never_ending_function(name):
++++++++while+True:
++++++++++++time.sleep(1)
++++++++++++print(f"Yet+another+second+passed+{name}...")

++++try:
++++++++results+=+my_slow_or_never_ending_function("Yooo!")
++++except+Exception+as+e:
++++++++print(f"ERROR:+{e}")|1639482|entityMap^0|0|0|3|A|0|0^^$0|@$1|2|3|4|5|6|7|Q|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|R|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|S|8|@$I|T|J|U|K|L]]|9|@]|A|$]]|$1|M|3|N|5|D|7|V|8|@]|9|@]|A|$E|F]]|$1|O|3|-4|5|6|7|W|8|@]|9|@]|A|$]]]|P|$]]

Just in case it is helpful for anyone, building on the answer by @piro, I've made a function decorator:
<pre class="lang-py prettyprint-override"><code>import time
import signal
from functools import wraps


def timeout(timeout_secs: int):
 def wrapper(func):
 @wraps(func)
 def time_limited(*args, **kwargs):
 # Register an handler for the timeout
 def handler(signum, frame):
 raise Exception(f&quot;Timeout for function '{func.__name__}'&quot;)

 # Register the signal function handler
 signal.signal(signal.SIGALRM, handler)

 # Define a timeout for your function
 signal.alarm(timeout_secs)

 result = None
 try:
 result = func(*args, **kwargs)
 except Exception as exc:
 raise exc
 finally:
 # disable the signal alarm
 signal.alarm(0)

 return result

 return time_limited

 return wrapper
</code></pre>
Using the wrapper on a function with a <code>20 seconds</code> timeout would look something like:
<pre><code> @timeout(20)
 def my_slow_or_never_ending_function(name):
 while True:
 time.sleep(1)
 print(f&quot;Yet another second passed {name}...&quot;)

 try:
 results = my_slow_or_never_ending_function(&quot;Yooo!&quot;)
 except Exception as e:
 print(f&quot;ERROR: {e}&quot;)
</code></pre>

blocks|key|1639596|text|我也遇到过同样的问题，但我的情况是需要在子线程上工作，信号对我不起作用，所以我写了一个python包:+timeout-timer来解决这个问题，支持用作上下文或装饰器，使用信号或子线程模块来触发超时中断：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1639597|from+timeout_timer+import+timeout,+TimeoutInterrupt

class+TimeoutInterruptNested(TimeoutInterrupt):
++++pass

def+test_timeout_nested_loop_both_timeout(timer="thread"):
++++cnt+=+0
++++try:
++++++++with+timeout(5,+timer=timer):
++++++++++++try:
++++++++++++++++with+timeout(2,+timer=timer,+exception=TimeoutInterruptNested):
++++++++++++++++++++sleep(2)
++++++++++++except+TimeoutInterruptNested:
++++++++++++++++cnt+%2B=+1
++++++++++++time.sleep(10)
++++except+TimeoutInterrupt:
++++++++cnt+%2B=+1
++++assert+cnt+==+2|code-block|syntax|javascript|1639598|更多信息请访问：https://github.com/dozysun/timeout-timer|offset|length|1639599|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/dozysun/timeout-timer^0|0|0|8|14|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|T|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|U|8|@]|9|@$I|V|J|W|1|X]]|A|$]]|$1|K|3|-4|5|6|7|Y|8|@]|9|@]|A|$]]]|L|$M|$5|N|O|P|A|$Q|R]]]]

I have face the same problem but my situation is need work on sub thread, signal didn't work for me, so I wrote a python package: timeout-timer to solve this problem, support for use as context or decorator, use signal or sub thread module to trigger a timeout interrupt:
<pre><code>from timeout_timer import timeout, TimeoutInterrupt

class TimeoutInterruptNested(TimeoutInterrupt):
 pass

def test_timeout_nested_loop_both_timeout(timer=&quot;thread&quot;):
 cnt = 0
 try:
 with timeout(5, timer=timer):
 try:
 with timeout(2, timer=timer, exception=TimeoutInterruptNested):
 sleep(2)
 except TimeoutInterruptNested:
 cnt += 1
 time.sleep(10)
 except TimeoutInterrupt:
 cnt += 1
 assert cnt == 2
</code></pre>
see more: <a href="https://github.com/dozysun/timeout-timer" rel="nofollow noreferrer">https://github.com/dozysun/timeout-timer</a>

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?

Timeout on a function call

我正在调用Python中的一个函数，我知道该函数可能会停止并迫使我重新启动脚本。我如何调用这个函数，或者我用什么包装它，这样如果花费的时间超过5秒，脚本就会取消它并执行其他操作？

问函数调用超时
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问函数调用超时EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问函数调用超时
EN