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

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

1> Josh Lee..：

---

对@ 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!")

---

而不是最后一次'Timed out!' 你可能打算写msg

2> rik.the.vik..：

---

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

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!"

---

此外,这不会禁用警报

---

signal.alarm和signal.SIGALRM仅在Unix上可用。

3> Ariel Cabib..：

---

这是限制函数运行时间的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

---

在我的Linux笔记本电脑上运行完美,在Windows上不起作用,并且几乎不能在OSX上运行.不是你的错,只是编程的精彩世界.

4> user2283347..：

---

我更喜欢上下文管理器方法,因为它允许在语句中执行多个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调用后处理.

5> Glenn Maynar..：

---

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

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

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

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

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

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

6> Dickon Reed..：

---

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的"并发系统".我与一群用线程真正冷却高性能东西的人合作,除非我们确实需要,否则我们不会将线程引入项目.