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

Question

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

• 使用运行代码的线程，join 超时。如果超时 - 杀死线程。这在 Python 中不直接支持（使用私有 _Thread__stop 函数），所以这是不好的做法
• 使用 signal.SIGALRM - 但这种方法不适用于 Windows！
• 使用超时的子进程 - 但这太重了 - 如果我想经常启动可中断的任务怎么办，我不想为每个进程触发！

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

Answer 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
  

Answer 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 编程的新手，所以如果该解决方案存在基本错误，例如资源、时间、死锁问题，请回复））。

Answer 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()

Answer 4

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

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

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

Answer 5

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

import socket
socket.setdefaulttimeout(number)

Answer 6

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

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

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

Answer 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 的一部分。

Answer 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'

Answer 9

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

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