python（flask）中函数的最大执行时间答案

【问题标题】：Max execution time for function in python (flask)python（flask）中函数的最大执行时间
【发布时间】：2013-09-02 21:46:45
【问题描述】：

我正在尝试限制函数在 python (flask) 中运行的时间。它看起来像这样：

def my_function:
    try:
        long_function_time()
    catch TimeOutException:
        return "Function took too long to execute"

def long_function_time:
    #stuff here

我最初尝试使用信号，但被告知这不是一个好方法，因为烧瓶在线程环境中运行。我希望最大执行时间灵活，以便我可以轻松更改它。

我目前使用的代码（有时不起作用，不知道为什么）：

class TimedOutExc(Exception):
  pass

def deadline(timeout, *args):
  def decorate(f):
    def handler(signum, frame):
      signal.alarm(0)
      raise TimedOutExc()

    def new_f(*args):

      signal.signal(signal.SIGALRM, handler)
      signal.alarm(timeout)
      return f(*args)

    new_f.__name__ = f.__name__
    return new_f
  return decorate

提前致谢！

【问题讨论】：

在this question 中有许多不同的解决方法可以中断长 Python 函数。
@Miguel：他们中的大多数似乎都使用信号？
您可能会发现this blog post 和各种cmets 很有趣。基本上，您尝试做的并不是一个容易解决的问题。您最好将长时间运行的代码分解成更小的块（也许使长期运行的函数成为生成器函数）。然后，在每个处理块之后测试超时。或者，重新评估您实际尝试做的事情；也许这个长时间运行的过程应该使用像 Celery 这样的异步任务队列库来完成。
这在我的情况下是不可能的@Mark。你不能分叉线程然后在 x 秒后杀死它吗？如果我知道足够多的 python，我可以自己尝试。
典型的经验法则是“永远不要从外部杀死线程”。这在理论上听起来不错，但在实践中，它最终会带来更多的麻烦而不是它的价值（资源没有正确分配）。

标签： python flask

【解决方案1】：

这适用于 CPython 2（使用 2.7 测试），显然不使用信号，而是使用 CPython 解释器循环。因此，您仍然必须在本机 I/O 调用和类似调用中处理非 Python 块（例如，您必须设置 socket.timeout）。此外，它可能会带来严重的运行时损失（您可以让它在每 *n* 次调用时检查超时以减轻一点）。

仍然有助于解决特定类别的问题（例如计算）。如果您希望它与分析器共存，则必须投入一些工作。

import sys
import time

class WatchdogTimeoutError(RuntimeError):
    """Raised in case of runtime limit violations."""

def sleeper(tick):
    """Endless loop."""
    while True:
        time.sleep(tick)

def watchdog(timeout, code, *args, **kwargs):
    "Time-limited execution."
    def tracer(frame, event, arg, start=time.time()):
        "Helper."
        now = time.time()
        if now > start + timeout:
            raise WatchdogTimeoutError(start, now)
        return tracer if event == "call" else None

    old_tracer = sys.gettrace()
    try:
        sys.settrace(tracer)
        code(*args, **kwargs)
    finally:
        sys.settrace(old_tracer)

def demo():
    """Show timeout executor."""
    try:
        watchdog(5, sleeper, 0.1)
    except WatchdogTimeoutError, exc:
        start, abort = exc.args
        print "Aborted after %.3f secs" % (abort - start,)
    else:
        print "Ended"

if __name__ == "__main__":
    demo()

【讨论】：

它与解释器挂钩，显然不会让事情变得更快。只需测量。

【解决方案2】：

尝试使用线程：

from multiprocessing import Pool, TimeoutError
from time import sleep

class TimedOutExc(Exception):
  pass

def f(x):
    sleep(2)
    return x*x

pool = Pool(processes=1)
result = pool.apply_async(f, (1,))
try:
    print(result.get(timeout=1))
except TimeoutError:
    print 'timeout'
    raise TimedOutExc()

【讨论】：