函数调用超时
我正在 Python 中调用一个函数,我知道该函数可能会停止并迫使我重新启动脚本。
如何调用该函数或者将其包装在什么中,以便如果花费时间超过 5 秒,脚本就会取消该函数并执行其他操作?
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?
如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。
绑定邮箱获取回复消息
由于您还没有绑定你的真实邮箱,如果其他用户或者作者回复了您的评论,将不能在第一时间通知您!
发布评论
评论(24)
如果您在 UNIX 上运行,则可以使用 signal 包:
调用后 10 秒< code>signal.alarm(10),调用处理程序。 这会引发一个异常,您可以从常规 Python 代码中拦截该异常。
这个模块不能很好地与线程配合使用(但是,谁呢?)
请注意,因为我们在超时发生时引发异常,所以它最终可能会在函数内部被捕获并忽略,例如一个这样的函数:
You may use the signal package if you are running on UNIX:
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:
您可以使用 multiprocessing.Process 来做到这一点。
代码
You can use
multiprocessing.Process
to do exactly that.Code
我发布了一个 gist ,它用装饰器和 threading.Timer 解决了这个问题/问题。 这是有故障的。
兼容性的导入和设置
它已经使用 Python 2 和 3 进行了测试。它也应该在 Unix/Linux 和 Windows 下工作。
首先是进口。 这些尝试使代码保持一致,无论 Python 版本如何:
使用与版本无关的代码:
现在我们已经从标准库导入了我们的功能。
exit_after
装饰器接下来我们需要一个函数来终止子线程中的
main()
:这是装饰器本身:
用法
这是直接回答您的问题的用法5 秒后退出!:
演示:
第二个函数调用将不会完成,进程应该退出并带有回溯!
KeyboardInterrupt
并不总是停止睡眠线程请注意,在 Windows 上的 Python 2 上,睡眠并不总是会被键盘中断打断,例如:
也不可能中断扩展中运行的代码,除非它显式检查对于
PyErr_CheckSignals()
,请参阅 Cython、Python 和 KeyboardInterrupt 被忽略我会避免更多地休眠线程无论如何,不到一秒——这就是处理器时间的亿万年。
要捕获它并执行其他操作,您可以捕获键盘中断。
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:
Use version independent code:
Now we have imported our functionality from the standard library.
exit_after
decoratorNext we need a function to terminate the
main()
from the child thread:And here is the decorator itself:
Usage
And here's the usage that directly answers your question about exiting after 5 seconds!:
Demo:
The second function call will not finish, instead the process should exit with a traceback!
KeyboardInterrupt
does not always stop a sleeping threadNote that sleep will not always be interrupted by a keyboard interrupt, on Python 2 on Windows, e.g.:
nor is it likely to interrupt code running in extensions unless it explicitly checks for
PyErr_CheckSignals()
, see Cython, Python and KeyboardInterrupt ignoredI would avoid sleeping a thread more than a second, in any case - that's an eon in processor time.
To catch it and do something else, you can catch the KeyboardInterrupt.
我有一个不同的建议,它是一个纯函数(具有与线程建议相同的 API),并且似乎工作正常(基于此线程的建议)
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)
我在单元测试中搜索超时调用时遇到了这个线程。 我在答案或第 3 方包中没有找到任何简单的内容,因此我在下面编写了装饰器,您可以直接将其放入代码中:
然后就可以像这样简单地使测试或您喜欢的任何函数超时:
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:
Then it's as simple as this to timeout a test or any function you like:
在 pypi 上找到的
stopit
包似乎可以很好地处理超时。我喜欢
@stopit.threading_timeoutable
装饰器,它向装饰函数添加一个timeout
参数,它会执行您所期望的操作,它会停止该函数。在 pypi 上查看:https://pypi.python.org/pypi/stopit
The
stopit
package, found on pypi, seems to handle timeouts well.I like the
@stopit.threading_timeoutable
decorator, which adds atimeout
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
我是 wrapt_timeout_decorator 的作者。
乍一看,这里介绍的大多数解决方案在 Linux 下工作得很好 - 因为我们有
fork()
和signals()
- 但在 Windows 上,事情看起来有点不同。当谈到 Linux 上的子线程时,你不能再使用信号了。
为了在 Windows 下生成进程,它需要是可挑选的 - 而许多修饰函数或类方法则不是。
因此,您需要使用更好的pickler,例如dill和multiprocess(不是pickle和multiprocessing) - 这就是为什么您不能使用
ProcessPoolExecutor
(或仅具有有限的功能)。对于超时本身 - 您需要定义超时的含义 - 因为在 Windows 上,需要相当长的(且不可确定的)时间来生成进程。 这对于短超时来说可能会很棘手。 让我们假设,生成过程大约需要 0.5 秒(很容易!!!)。 如果你给出 0.2 秒的超时时间,会发生什么?
该函数是否应该在 0.5 + 0.2 秒后超时(因此让该方法运行 0.2 秒)?
或者被调用的进程应该在 0.2 秒后超时(在这种情况下,修饰函数将始终超时,因为在那段时间它甚至没有生成)?
嵌套装饰器也可能很糟糕,并且您不能在子线程中使用信号。 如果你想创建一个真正通用的、跨平台的装饰器,所有这些都需要考虑(并测试)。
其他问题是将异常传递回调用者,以及日志记录问题(如果在装饰函数中使用 - 不支持在另一个进程中记录到文件)
我试图涵盖所有边缘情况,您可能会查看包wrapt_timeout_decorator,或者至少测试受此处使用的单元测试启发的您自己的解决方案。
@Alexis Eggermont - 不幸的是我没有足够的观点来评论 - 也许其他人可以通知你 - 我想我解决了你的导入问题。
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()
andsignals()
- 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.
有很多建议,但没有使用并发。futures,我认为这是处理这个问题的最清晰的方法。
超级简单的阅读和维护。
我们创建一个池,提交一个进程,然后等待最多 5 秒,然后引发一个 TimeoutError,您可以根据需要捕获并处理该错误。
原生于 python 3.2+ 并向后移植到 2.7 (pip install futures)。
线程和进程之间的切换就像用
ThreadPoolExecutor
替换ProcessPoolExecutor
一样简单。如果您想在超时时终止进程,我建议您查看 Pebble。
There are a lot of suggestions, but none using concurrent.futures, which I think is the most legible way to handle this.
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
withThreadPoolExecutor
.If you want to terminate the Process on timeout I would suggest looking into Pebble.
在 @piro 的答案的基础上构建和增强,您可以构建一个上下文管理器。 这允许非常可读的代码,它将在成功运行后禁用警报信号(设置 signal.alarm(0))
示例用法:
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))
Example usage:
很棒、易于使用且可靠的 PyPi 项目 timeout-decorator (https://pypi.org/project/timeout-decorator/)
安装:
使用:
Great, easy to use and reliable PyPi project timeout-decorator (https://pypi.org/project/timeout-decorator/)
installation:
Usage:
timeout-decorator
不适用于 Windows 系统,因为 Windows 不能很好地支持signal
。如果您在 Windows 系统中使用 timeout-decorator,您将得到以下信息
Some suggest to use
use_signals=False
but did not work for me.作者 @bitranox 创建了以下包:
代码示例:
给出以下异常:
timeout-decorator
don't work on windows system as , windows didn't supportsignal
well.If you use timeout-decorator in windows system you will get the following
Some suggested to use
use_signals=False
but didn't worked for me.Author @bitranox created the following package:
Code Sample:
Gives the following exception:
为了防止它对任何人有帮助,在 @piro 的答案的基础上,我制作了一个函数装饰器:在
具有
20 秒
超时的函数上使用包装器将如下所示:Just in case it is helpful for anyone, building on the answer by @piro, I've made a function decorator:
Using the wrapper on a function with a
20 seconds
timeout would look something like:亮点
TimeoutError
使用异常来警告超时 - 可以轻松修改有关并行映射的完整说明和扩展,请参阅此处 https://flipdazed.github.io/blog/quant%20dev/parallel-functions-with-timeouts
最小示例
和预期
完整代码
注释
由于
dill
的工作方式,您将需要在函数内部导入。这也意味着如果目标函数中有导入,这些函数可能与 doctest 不兼容。 您将遇到找不到
__import__
的问题。Highlights
TimeoutError
uses exceptions to alert on timeout - can easily be modifiedFor full explanation and extension to parallel maps, see here https://flipdazed.github.io/blog/quant%20dev/parallel-functions-with-timeouts
Minimal Example
and as expected
Full code
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.我们可以使用信号来实现同样的目的。 我认为下面的例子对你有用。 与线程相比,它非常简单。
We can use signals for the same. I think the below example will be useful for you. It is very simple compared to threads.
asyncio 的另一个解决方案:
如果您想取消后台任务,而不仅仅是在运行的主代码上超时,那么您需要主线程的显式通信来要求任务的代码取消,例如 threading.Event()
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()
Tim Savannah 的 func_timeout 包 对我来说效果很好。
安装:
pip install func_timeout
使用方法:
The func_timeout package by Tim Savannah has worked well for me.
Installation:
pip install func_timeout
Usage:
这是一个简单易用的装饰器,它返回给定的默认值如果函数的执行时间到期,灵感来自 此问题的第一个答案:
Here is a simple and easy to use decorator that returns a given default if the execution time for the function expires, inspired from the first answer to this question:
我需要可嵌套定时中断(SIGALARM 无法做到),并且不会被 time.sleep 阻塞(基于线程的方法无法做到)。 我最终从这里复制并稍微修改了代码: http://code.activestate.com/recipes/577600-queue-for-managing-multiple-sigalrm-alarms-concurr/
代码本身:
以及使用示例:
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:
and a usage example:
我遇到了同样的问题,但我的情况是需要在子线程上工作,信号对我不起作用,所以我编写了一个python包:timeout-timer来解决这个问题,支持用作上下文或装饰器,使用信号或子线程模块触发超时中断:
查看更多:https://github.com/dozysun/timeout-定时器
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:
see more: https://github.com/dozysun/timeout-timer
这是一个简单的示例,运行一个超时方法,如果成功则检索其值。
Here is a simple example running one method with timeout and also retriev its value if successfull.
这是对给定的基于线程的解决方案的轻微改进。
下面的代码支持异常:
以 5 秒超时调用它:
Here is a slight improvement to the given thread-based solution.
The code below supports exceptions:
Invoking it with a 5 second timeout:
这是一个 POSIX 版本,它结合了许多以前的答案,以提供以下功能:
这是代码和一些测试用例:
Here is a POSIX version that combines many of the previous answers to deliver following features:
Here is the code and some test cases:
如果工作没有完成,我打算杀死进程,使用线程和进程来实现这一点。
作为上下文管理器的实际类
如何使用它
I intend to kill the process if job not done , using thread and process both to achieve this.
Actual class as a contextmanager
How to use it