在 Python 中运行一些超时代码的正确方法

Posted 2023-02-25

【中文标题】在 Python 中运行一些超时代码的正确方法

【英文标题】：right way to run some code with timeout in Python

【发布时间】：2011-10-20 07:07:19

【问题描述】：

我在网上查找了一些 SO 讨论和 ActiveState 配方，用于运行一些超时的代码。看起来有一些常见的方法：

使用运行代码的线程，join 超时。如果超时 - 杀死线程。这在 Python 中不直接支持（使用私有 _Thread__stop 函数），所以这是

不好的做法

使用 signal.SIGALRM - 但这种方法

不适用于 Windows

！ 使用超时的子进程 - 但这

太重了

- 如果我想经常启动可中断的任务怎么办，我不想为每个进程触发！

那么，什么是正确的方法？我不是在问解决方法（例如使用 Twisted 和异步 IO），而是解决实际问题的实际方法 - 我有一些功能，我只想在超时的情况下运行它。如果超时，我想要控制权。我希望它可以在 Linux 和 Windows 上运行。

【问题讨论】：

杀死代码是一种不好的做法，但如果你想这样做，线程是最好的方法。

我假设您不能修改相关代码？

你能准确地说出你在做什么吗？你运行的代码是做什么的？

没有正确的方法，这取决于你要打断什么，它保持什么共享状态等等。

这取决于你的算法，如果它是迭代的，你可以让出控制权，如果是 IO，你可以select 等等...

【参考方案1】：

用'with'结构解决并合并解决方案 -

Timeout function if it takes too long to finish

这个线程效果更好。

import threading, time

class Exception_TIMEOUT(Exception):
    pass

class linwintimeout:

    def __init__(self, f, seconds=1.0, error_message='Timeout'):
        self.seconds = seconds
        self.thread = threading.Thread(target=f)
        self.thread.daemon = True
        self.error_message = error_message

    def handle_timeout(self):
        raise Exception_TIMEOUT(self.error_message)

    def __enter__(self):
        try:
            self.thread.start()
            self.thread.join(self.seconds)
        except Exception, te:
            raise te

    def __exit__(self, type, value, traceback):
        if self.thread.is_alive():
            return self.handle_timeout()

def function():
    while True:
        print "keep printing ...", time.sleep(1)

try:
    with linwintimeout(function, seconds=5.0, error_message='exceeded timeout of %s seconds' % 5.0):
        pass
except Exception_TIMEOUT, e:
    print "  attention !! execeeded timeout, giving up ... %s " % e

【参考方案2】：

我已经通过这种方式解决了这个问题： 对我来说效果很好（在 Windows 中而且一点也不重）我希望它对某人有用）

import threading
import time

class LongFunctionInside(object):
    lock_state = threading.Lock()
    working = False

    def long_function(self, timeout):

        self.working = True

        timeout_work = threading.Thread(name="thread_name", target=self.work_time, args=(timeout,))
        timeout_work.setDaemon(True)
        timeout_work.start()

        while True:  # endless/long work
            time.sleep(0.1)  # in this rate the CPU is almost not used
            if not self.working:  # if state is working == true still working
                break
        self.set_state(True)

    def work_time(self, sleep_time):  # thread function that just sleeping specified time,
    # in wake up it asking if function still working if it does set the secured variable work to false
        time.sleep(sleep_time)
        if self.working:
            self.set_state(False)

    def set_state(self, state):  # secured state change
        while True:
            self.lock_state.acquire()
            try:
                self.working = state
                break
            finally:
                self.lock_state.release()

lw = LongFunctionInside()
lw.long_function(10)

主要思想是创建一个线程，该线程将与“长时间工作”并行睡眠，并在唤醒（超时后）更改安全变量状态，长函数在其工作期间检查安全变量。 我是 Python 编程的新手，所以如果该解决方案存在基本错误，例如资源、时间、死锁问题，请回复））。

【参考方案3】：

我在 eventlet 库中找到了这个：

http://eventlet.net/doc/modules/timeout.html

from eventlet.timeout import Timeout

timeout = Timeout(seconds, exception)
try:
    ... # execution here is limited by timeout
finally:
    timeout.cancel()

【讨论】：

如果没有eventlet.sleep，此代码不起作用 - 它至少需要调用此函数一次。只有在此调用期间才能拦截代码执行。在这种情况下，我看不到此模块的含义 - 如果不每第二行注入 eventlet.sleep，我就无法将它用于我的代码。

【参考方案4】：

如果您正在运行您希望在设定的时间后死掉的代码，那么您应该正确编写它，以便对关闭没有任何负面影响，无论它是线程还是子进程。带撤消的命令模式在这里会很有用。

所以，这真的取决于你杀死它时线程在做什么。如果它只是处理数字，谁在乎你是否杀死它。如果它与文件系统交互并且你杀死它，那么也许你真的应该重新考虑你的策略。

在线程方面，Python 支持什么？守护进程线程和连接。如果您在守护进程仍处于活动状态时加入了守护进程，为什么 python 会让主线程退出？因为它理解使用守护线程的人将（希望）以一种线程死亡时无关紧要的方式编写代码。在这种情况下，给一个连接一个超时，然后让 main 死掉，从而使用它的任何守护线程，是完全可以接受的。

【参考方案5】：

如果是网络相关的可以试试：

import socket
socket.setdefaulttimeout(number)

【参考方案6】：

对此真的，老实说完全通用的解决方案不存在。您必须为给定的域使用正确的解决方案。

如果你想要完全控制的代码超时，你必须编写它来配合。这样的代码必须能够以某种方式分解成小块，就像在事件驱动系统中一样。如果您可以确保没有任何东西持有锁太久，您也可以通过线程来做到这一点，但正确处理锁实际上非常困难。

如果您因为担心代码失控而想要超时（例如，如果您担心用户会要求您的计算器计算 9**(9**9)），您需要在另一个进程中运行它.这是充分隔离它的唯一简单方法。在您的事件系统甚至不同的线程中运行它是不够的。也可以将事情分成类似于其他解决方案的小块，但需要非常小心处理并且通常不值得；无论如何，这不允许您执行与运行 Python 代码完全相同的操作。

【参考方案7】：

另一种方式是使用faulthandler:

import time
import faulthandler


faulthandler.enable()


try:
    faulthandler.dump_tracebacks_later(3)
    time.sleep(10)
finally:
    faulthandler.cancel_dump_tracebacks_later()

注意：faulthandler 模块是 python3.3 中 stdlib 的一部分。

【讨论】：

@zaharpopov：是的，确实如此；如果这不是您想要的 IMO，您最好的选择是使用 subprocess 。

dump_tracebacks_later 基于信号，因此在 Windows 上不可用

【参考方案8】：

您可能正在寻找的是multiprocessing 模块。如果subprocess 太重，那么这也可能不适合您的需要。

import time
import multiprocessing

def do_this_other_thing_that_may_take_too_long(duration):
    time.sleep(duration)
    return 'done after sleeping 0 seconds.'.format(duration)

pool = multiprocessing.Pool(1)
print 'starting....'
res = pool.apply_async(do_this_other_thing_that_may_take_too_long, [8])
for timeout in range(1, 10):
    try:
        print '0: 1'.format(duration, res.get(timeout))
    except multiprocessing.TimeoutError:
        print '0: timed out'.format(duration) 

print 'end'

【讨论】：

这使得函数在子进程中运行，对吧？当超时错误时，子进程被杀死？

如果您希望能够杀死任意代码，包括 C 扩展，并进行一些类似的清理，您需要使用进程。如果您认为它们太慢而无法分叉，那么请让一两个预生成的孩子在您需要时等待执行。 multiprocessing 似乎就是这样做的，所以我用一个赞成票来代替我的答案。

@LaC：但是进程的问题是 - 你不能让它预先生成，因为当超时发生时你杀死它！那么预产卵的好处在哪里呢？

@zaharpopov 这取决于你的用例——你希望多久杀死一次子任务？我正在使用改编自这个答案的代码。在我的用例中，我需要在每几千次运行中杀死 1 个 - 不使用池 真的 很慢。如果杀戮是正常现象，那么无论你做什么，它都会很慢。

【参考方案9】：

对于“正常”的 Python 代码，在 C 扩展或 I/O 等待中不会逗留较长时间，您可以通过使用sys.settrace() 设置跟踪函数来实现您的目标，该函数在达到超时时中止正在运行的代码.

这是否足够取决于您运行的代码的合作程度或恶意程度。如果它表现良好，跟踪功能就足够了。

【讨论】：

您排除了可以处理这些问题的解决方案。

另外，我认为使用sys.settrace 会使代码非常慢，不是吗？