如何在Python中限制函数调用的执行时间

Question

在我的代码中有一个与套接字相关的函数调用,该函数来自另一个模块,因此无法控制,问题是它偶尔会阻塞几个小时,这是完全不可接受的,我如何限制代码中的函数执行时间？我想解决方案必须使用另一个线程.

Answer 1

对@ rik.the.vik的答案的改进是使用该with语句为超时函数提供一些语法糖:

import signal
from contextlib import contextmanager

class TimeoutException(Exception): pass

@contextmanager
def time_limit(seconds):
    def signal_handler(signum, frame):
        raise TimeoutException("Timed out!")
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)


try:
    with time_limit(10):
        long_function_call()
except TimeoutException as e:
    print("Timed out!")

Answer 2

我不确定这可能是跨平台的,但使用信号和警报可能是一个很好的方式来看待这个.通过一些工作,您可以使其完全通用,并且可以在任何情况下使用.

http://docs.python.org/library/signal.html

所以你的代码看起来像这样.

import signal

def signal_handler(signum, frame):
    raise Exception("Timed out!")

signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(10)   # Ten seconds
try:
    long_function_call()
except Exception, msg:
    print "Timed out!"

Answer 3

这是限制函数运行时间的Linux/OSX方法.这是因为您不想使用线程,并希望您的程序等到函数结束或时间限制到期.

from multiprocessing import Process
from time import sleep

def f(time):
    sleep(time)


def run_with_limited_time(func, args, kwargs, time):
    """Runs a function with time limit

    :param func: The function to run
    :param args: The functions args, given as tuple
    :param kwargs: The functions keywords, given as dict
    :param time: The time limit in seconds
    :return: True if the function ended successfully. False if it was terminated.
    """
    p = Process(target=func, args=args, kwargs=kwargs)
    p.start()
    p.join(time)
    if p.is_alive():
        p.terminate()
        return False

    return True


if __name__ == '__main__':
    print run_with_limited_time(f, (1.5, ), {}, 2.5) # True
    print run_with_limited_time(f, (3.5, ), {}, 2.5) # False

Answer 4

这里：获得所需效果的简单方法：

https://pypi.org/project/func-timeout

这救了我的命。

现在举一个例子来说明它是如何工作的：假设您有一个巨大的要处理的项目列表，并且您正在对这些项目迭代您的函数。然而，由于某些奇怪的原因，您的函数卡在第 n 项上，而没有引发异常。您需要处理其他物品，越多越好。在这种情况下，您可以设置处理每个项目的超时：

import time
import func_timeout


def my_function(n):
    """Sleep for n seconds and return n squared."""
    print(f'Processing {n}')
    time.sleep(n)
    return n**2


def main_controller(max_wait_time, all_data):
    """
    Feed my_function with a list of items to process (all_data).

    However, if max_wait_time is exceeded, return the item and a fail info.
    """
    res = []
    for data in all_data:
        try:
            my_square = func_timeout.func_timeout(
                max_wait_time, my_function, args=[data]
                )
            res.append((my_square, 'processed'))
        except func_timeout.FunctionTimedOut:
            print('error')
            res.append((data, 'fail'))
            continue

    return res


timeout_time = 2.1  # my time limit
all_data = range(1, 10)  # the data to be processed

res = main_controller(timeout_time, all_data)
print(res)

Answer 5

我更喜欢上下文管理器方法,因为它允许在语句中执行多个python with time_limit语句.因为Windows系统没有SIGALARM,所以可以使用更便携,也许更直接的方法Timer

from contextlib import contextmanager
import threading
import _thread

class TimeoutException(Exception):
    def __init__(self, msg=''):
        self.msg = msg

@contextmanager
def time_limit(seconds, msg=''):
    timer = threading.Timer(seconds, lambda: _thread.interrupt_main())
    timer.start()
    try:
        yield
    except KeyboardInterrupt:
        raise TimeoutException("Timed out for operation {}".format(msg))
    finally:
        # if the action ends in specified time, timer is canceled
        timer.cancel()

import time
# ends after 5 seconds
with time_limit(5, 'sleep'):
    for i in range(10):
        time.sleep(1)

# this will actually end after 10 seconds
with time_limit(5, 'sleep'):
    time.sleep(10)

这里的关键技术是使用_thread.interrupt_main从计时器线程中断主线程.需要注意的是,主线程并不总是快速响应所KeyboardInterrupt引发的Timer.例如,time.sleep()调用系统函数,以便KeyboardInterrupt在sleep调用后处理.

Answer 6

在信号处理程序中执行此操作很危险:在引发异常时您可能位于异常处理程序中,并使事物处于损坏状态.例如,

def function_with_enforced_timeout():
  f = open_temporary_file()
  try:
   ...
  finally:
   here()
   unlink(f.filename)

如果在此处引发异常(),则永远不会删除临时文件.

这里的解决方案是推迟异步异常,直到代码不在异常处理代码(除外或最后一个块)内,但Python不这样做.

请注意,执行本机代码时不会中断任何操作; 它只会在函数返回时中断它,所以这可能对这种特殊情况没有帮助.(SIGALRM本身可能会中断阻塞的调用 - 但是套接字代码通常只是在EINTR之后重试.)

使用线程执行此操作是一个更好的主意,因为它比信号更便携.因为你正在启动一个工作线程并阻塞直到它完成,所以没有通常的并发担忧.不幸的是,没有办法异步地将异常传递给Python中的另一个线程(其他线程API可以做到这一点).在异常处理程序中发送异常也会遇到同样的问题,并且需要相同的修复.

Answer 7

在任何语言中，执行此操作的唯一“安全”方法是使用辅助进程来执行超时操作，否则您需要以一种能够自行安全超时的方式构建代码，例如通过检查循环或类似的时间流逝。如果无法更改方法，则线程是不够的。

为什么？因为当你这样做时，你冒着让事情处于糟糕状态的风险。如果线程只是在方法中被杀死，则所持有的锁等将仅被持有，而无法释放。

所以看进程方式，不要看线程方式。

Answer 8

You don't have to use threads. You can use another process to do the blocking work, for instance, maybe using the subprocess module. If you want to share data structures between different parts of your program then Twisted is a great library for giving yourself control of this, and I'd recommend it if you care about blocking and expect to have this trouble a lot. The bad news with Twisted is you have to rewrite your code to avoid any blocking, and there is a fair learning curve.

你可以使用线程来避免阻塞,但我认为这是最后的手段,因为它会让你暴露在一个痛苦的世界里.在考虑在生产中使用线程之前,请阅读一本关于并发性的好书,例如Jean Bacon的"并发系统".我与一群用线程真正冷却高性能东西的人合作,除非我们确实需要,否则我们不会将线程引入项目.