一定时间后中断功能

Question

在 Python 中，举个玩具例子：

for x in range(0, 3):
    # Call function A(x)

如果函数 A 需要超过 5 秒的时间跳过它，我想继续 for 循环，这样我就不会卡住或浪费时间。

通过搜索，我意识到子进程或线程可能会有所帮助，但我不知道如何在这里实现它。

Answer 1

我认为创建一个新进程可能是矫枉过正。如果您在 Mac 或基于 Unix 的系统上，您应该能够使用 signal.SIGALRM 强制使耗时过长的函数超时。这将适用于因网络或其他您绝对无法通过修改函数来处理的问题而空闲的函数。我有一个在这个答案中使用它的例子：

Option for SSH to timeout after a short time? ClientAlive & ConnectTimeout don't seem to do what I need them to do

在这里编辑我的答案，虽然我不确定我应该这样做：

import signal

class TimeoutException(Exception):   # Custom exception class
    pass

def timeout_handler(signum, frame):   # Custom signal handler
    raise TimeoutException

# Change the behavior of SIGALRM
signal.signal(signal.SIGALRM, timeout_handler)

for i in range(3):
    # Start the timer. Once 5 seconds are over, a SIGALRM signal is sent.
    signal.alarm(5)    
    # This try/except loop ensures that 
    #   you'll catch TimeoutException when it's sent.
    try:
        A(i) # Whatever your function that might hang
    except TimeoutException:
        continue # continue the for loop if function A takes more than 5 second
    else:
        # Reset the alarm
        signal.alarm(0)

这基本上将计时器设置为 5 秒，然后尝试执行您的代码。如果在时间用完之前未能完成，则会发送一个 SIGALRM，我们将其捕获并转换为 TimeoutException。这会迫使您进入 except 块，您的程序可以在其中继续。

Answer 2

cmets 是正确的，您应该检查内部。这是一个潜在的解决方案。请注意，异步函数（例如通过使用线程）与此解决方案不同。这是同步的，这意味着它仍将串联运行。

import time

for x in range(0,3):
    someFunction()

def someFunction():
    start = time.time()
    while (time.time() - start < 5):
        # do your normal function

    return;

Answer 3

如果您可以拆分工作并经常检查，那几乎总是最好的解决方案。但有时这是不可能的——例如，也许你正在从一个缓慢的文件共享中读取一个文件，该文件每隔一段时间就会挂起 30 秒。要在内部处理这个问题，您必须围绕异步 I/O 循环重构整个程序。

如果你不需要跨平台，你可以在*nix（包括Mac和Linux）上使用信号，在Windows上使用APC等。但如果你需要跨平台，那就不行了.

因此，如果您确实需要同时执行此操作，您可以，有时您必须这样做。在这种情况下，您可能希望为此使用进程，而不是线程。你不能真正安全地杀死一个线程，但你可以杀死一个进程，它可以像你想要的那样安全。此外，如果线程因为受 CPU 限制而需要 5 秒以上，那么您不想与它争夺 GIL。

这里有两个基本选项。

---

首先，您可以将代码放在另一个脚本中并使用subprocess 运行它：

subprocess.check_call([sys.executable, 'other_script.py', arg, other_arg],
                      timeout=5)

由于这是通过正常的子进程通道，您可以使用的唯一通信是一些 argv 字符串、成功/失败返回值（实际上是一个小整数，但这并没有好多少），以及可选的大块输入的文本和输出的文本。

---

或者，您可以使用multiprocessing 来生成类似线程的子进程：

p = multiprocessing.Process(func, args)
p.start()
p.join(5)
if p.is_alive():
    p.terminate()

如您所见，这有点复杂，但在某些方面更好：

• 您可以传递任意 Python 对象（至少是任何可以腌制的对象），而不仅仅是字符串。
• 不必将目标代码放在完全独立的脚本中，您可以将其作为函数保留在同一个脚本中。
• 它更加灵活 - 例如，如果您以后需要传递进度更新，则可以很容易地在一个或两个方向上添加一个队列。

---

任何类型的并行性的大问题是共享可变数据 - 例如，让后台任务更新全局字典作为其工作的一部分（您的 cmets 说您正在尝试这样做）。使用线程，你可以摆脱它，但竞争条件会导致数据损坏，所以你必须非常小心锁定。对于子进程，您根本无法摆脱它。 （是的，您可以使用共享内存，正如Sharing state between processes 解释的那样，但这仅限于简单类型，如数字、固定数组和您知道如何定义为 C 结构的类型，它只会让您回到与线程。）

---

理想情况下，您可以安排好事情，这样您就不需要在进程运行时共享任何数据——您将dict 作为参数传入，然后返回dict。当您有一个以前同步的函数想要放在后台时，这通常很容易安排。

但是，比如说，部分结果总比没有结果好怎么办？在这种情况下，最简单的解决方案是将结果传递给队列。您可以使用显式队列来执行此操作，如 Exchanging objects between processes 中所述，但有一种更简单的方法。

如果您可以将整体流程分解为单独的任务，每个任务对应您想要添加到字典中的每个值（或值组），您可以将它们安排在 Pool 上，或者更好的是 @987654323 @。 （如果您使用的是 Python 2.x 或 3.1，请参阅 PyPI 上的后向端口 futures。）

假设你的慢函数看起来像这样：

def spam():
    global d
    for meat in get_all_meats():
        count = get_meat_count(meat)
        d.setdefault(meat, 0) += count

相反，您应该这样做：

def spam_one(meat):
    count = get_meat_count(meat)
    return meat, count

with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
    results = executor.map(spam_one, get_canned_meats(), timeout=5)
    for (meat, count) in results:
        d.setdefault(meat, 0) += count

在 5 秒内获得的结果会被添加到字典中；如果这还不是全部，则放弃其余的，并引发TimeoutError（您可以随意处理它——记录它，做一些快速的后备代码，等等）。

如果任务真的是独立的（就像在我的愚蠢的小例子中那样，但当然它们可能不在您的真实代码中，至少在没有重大重新设计的情况下不会），您可以通过以下方式免费并行化工作删除max_workers=1。然后，如果你在一台 8 核机器上运行它，它会启动 8 个工人，并给他们每人 1/8 的工作要做，事情会更快地完成。 （通常速度不是 8 倍，但通常是 3-6 倍，这仍然相当不错。）

Answer 4

这似乎是一个更好的主意（抱歉，我还不确定 Python 的名称）：

import signal

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

signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(3) # Three seconds
try:
    for x in range(0, 3):
        # Call function A(x)
except Exception, msg:
    print "Timeout!"
signal.alarm(0) # Reset

Answer 5

也许有人发现这个装饰器很有用，基于 TheSoundDefense 的回答：

import time
import signal

class TimeoutException(Exception):   # Custom exception class
    pass


def break_after(seconds=2):
    def timeout_handler(signum, frame):   # Custom signal handler
        raise TimeoutException
    def function(function):
        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(seconds)
            try:
                res = function(*args, **kwargs)
                signal.alarm(0)      # Clear alarm
                return res
            except TimeoutException:
                print u'Oops, timeout: %s sec reached.' % seconds, function.__name__, args, kwargs
            return
        return wrapper
    return function

测试：

@break_after(3)
def test(a, b, c):
    return time.sleep(10)

>>> test(1,2,3)
Oops, timeout: 3 sec reached. test (1, 2, 3) {}