timeit 内的 Python 函数调用超时答案

【问题标题】：Timeout on a Python function call inside timeittimeit 内的 Python 函数调用超时
【发布时间】：2013-12-20 01:54:22
【问题描述】：

我有一个函数，我们称之为 my_function(*args, **kwargs)，根据传递给它的参数，它需要 30 秒到很多小时（天）。我有一个不同参数的列表，我想知道给定参数的函数需要多长时间。

我对使用timeit 还是很陌生，但我已经学到了足够的知识来做这部分；但是，这对我的目的来说不是很有效。任何需要超过 4 小时才能完成的论点都被认为是棘手的，尽管它们都能够在“有限”时间内解决。由于某些（可能是大多数）参数会导致运行时间超过 20 多个小时，因此我正在寻找一种在 4 小时后终止测试的方法，这样我就不必在弄清楚之后浪费时间了这是棘手的。

我查看了Timeout on a Python function call 和Stop code after time period，这可能是一个足够接近重复的问题，但我无法将这些答案与timeit 整合在一起，因此记录的时间少于 4 小时就像他们应该的那样，而长时间运行会返回大于 4 小时的有效时间。

最好的方法是什么？

编辑：我遇到的问题之一是我看到的答案包含func(*args,**kwargs)，而timeit 函数看起来像这样：

timeit.Timer('my_function(*args, **kwargs)', setup=setup).timeit(1)

我不知道如何处理这个表格。

编辑：我在下面使用线程提供的原始答案实际上并没有终止线程。使用以下函数运行它可以很容易地显示出来。

def foo(x):
    for i in xrange(1, x + 1):
        sleep(1)
        print i
    return x

使用涉及multiprocessing.Pool 的代码，实际上有一个terminate()，允许这样做。

【问题讨论】：

您为什么决定使用timeit？你不能只运行一次超时，看看它是否超时？
我需要知道如果它没有超时，它会运行多长时间。因此，如果它运行 3.5 小时，它不会超时，但我必须再次运行它才能知道它确实运行了 3.5 小时。

标签： python python-2.7 timeout python-multithreading timeit

【解决方案1】：

基于在Timeout function using threading in python does not work 中找到的答案。如果您在 foo(x) 上尝试一下，它确实会停止计数，这与我之前的答案不同。

import multiprocessing as mp
import timeit

def timeout(func, args=(), kwargs=None, TIMEOUT=10, default=None, err=.05):

    if hasattr(args, "__iter__") and not isinstance(args, basestring):
        args = args
    else:
        args = [args]    
    kwargs = {} if kwargs is None else kwargs

    pool = mp.Pool(processes = 1)

    try:
        result = pool.apply_async(func, args=args, kwds=kwargs)
        val = result.get(timeout = TIMEOUT * (1 + err))
    except mp.TimeoutError:
        pool.terminate()
        return default
    else:
        pool.close()
        pool.join()
        return val

def Timeit(command, setup=''):
    return timeit.Timer(command, setup=setup).timeit(1)

def timeit_timeout(command, setup='', TIMEOUT=10, default=None, err=.05):
    return timeout(Timeit, args=command, kwargs={'setup':setup},
                   TIMEOUT=TIMEOUT, default=default, err=err)

【讨论】：

【解决方案2】：

经过一番折腾，我有了一个基于Timeout function using threading 的初步答案。我仍然很想听听任何有更好想法的人的意见，因为我还是新手。

def timeout(func, args=None, kwargs=None, TIMEOUT=10, default=None, err=.05):
    if args is None:
        args = []
    elif hasattr(args, "__iter__") and not isinstance(args, basestring):
        args = args
    else:
        args = [args]

    kwargs = {} if kwargs is None else kwargs

    import threading
    class InterruptableThread(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)
            self.result = None

        def run(self):
            try:
                self.result = func(*args, **kwargs)
            except:
                self.result = default

    it = InterruptableThread()
    it.start()
    it.join(TIMEOUT* (1 + err))
    if it.isAlive():
        return default
    else:
        return it.result

def timeit_timeout(command, setup='', TIMEOUT=10, default=None, err=.05):
    import timeit
    f = lambda: timeit.Timer(command, setup=setup).timeit(1)
    return timeout(f,TIMEOUT=TIMEOUT, default=default, err=err) 

if __name__ == '__main__':    
    from time import sleep
    setup = 'gc.enable()\nfrom time import sleep'
    for n in range(1,15):
        command = 'sleep({})'.format(n)
        print timeit_timeout(command, setup=setup, default=float('Inf'))

【讨论】：