使用 asyncio 处理超时

免责声明：这是我第一次尝试asyncio module.

我在用着asyncio.wait通过以下方式尝试支持等待一组异步任务的所有结果的超时功能。这是一个更大的库的一部分，因此我省略了一些不相关的代码。

请注意，该库已经支持提交任务并使用 ThreadPoolExecutors 和 ProcessPoolExecutors 使用超时，因此我对使用这些选项的建议或关于为什么我这样做的问题并不真正感兴趣asyncio。上代码...

import asyncio
from contextlib import suppress

... 

class AsyncIOSubmit(Node):
    def get_results(self, futures, timeout=None):
        loop = asyncio.get_event_loop()
        finished, unfinished = loop.run_until_complete(
            asyncio.wait(futures, timeout=timeout)
        )
        if timeout and unfinished:
            # Code options in question would go here...see below.
            raise asyncio.TimeoutError

起初我并不担心超时时取消待处理的任务，但后来我收到了警告Task was destroyed but it is pending!程序退出时或loop.close。经过一番研究后，我发现了多种取消任务并等待它们实际被取消的方法：

选项1：

[task.cancel() for task in unfinished]
for task in unfinished:
    with suppress(asyncio.CancelledError):
        loop.run_until_complete(task)

选项2：

[task.cancel() for task in unfinished]
loop.run_until_complete(asyncio.wait(unfinished))

选项 3：

# Not really an option for me, since I'm not in an `async` method
# and don't want to make get_results an async method.
[task.cancel() for task in unfinished]
for task in unfinished:
    await task

选项 4：

某种 while 循环，如this https://stackoverflow.com/a/40022171/10682164回答。似乎我的其他选择更好，但包括完整性。

到目前为止，选项 1 和 2 似乎都运行良好。任何一个选项都可能是“正确的”，但是asyncio经过多年的发展，网上的例子和建议要么已经过时，要么变化很大。所以我的问题是...

问题1

选项 1 和选项 2 之间有任何实际差异吗？我知道run_until_complete将运行直到未来完成，因此由于选项 1 按特定顺序循环，我想如果较早的任务需要更长的时间才能实际完成，它的行为可能会有所不同。我尝试查看 asyncio 源代码来了解是否asyncio.wait只是在幕后有效地对其任务/未来做了同样的事情，但这并不明显。

问题2

我假设如果其中一个任务正在进行长时间运行的阻塞操作，它实际上可能不会立即取消？也许这仅仅取决于正在使用的底层操作或库是否会立即引发 CancelledError ？也许为 asyncio 设计的库永远不应该发生这种情况？

由于我试图在这里实现超时功能，因此我对此有些敏感。如果可能的话，这些事情可能需要很长时间才能取消，我会考虑打电话cancel并且不等待它实际发生，或者设置一个非常短的超时来等待取消完成。

问题3

是否可以loop.run_until_complete（或者实际上，底层的调用async.wait) 返回值unfinished除了超时之外还有其他原因吗？如果是这样，我显然必须稍微调整一下我的逻辑，但从docs https://docs.python.org/3/library/asyncio-task.html#asyncio.wait看来那是不可能的。

选项 1 和选项 2 之间有任何实际差异吗？

不会。选项 2 看起来更好，而且效率可能稍微更高，但它们的净效果是相同的。

I know run_until_complete将运行直到未来完成，因此由于选项 1 按特定顺序循环，我想如果较早的任务需要更长的时间才能实际完成，它的行为可能会有所不同。

乍一看似乎是这样，但实际上并非如此，因为loop.run_until_complete runs all提交给循环的任务，而不仅仅是作为参数传递的任务。它仅仅stops一旦提供的等待完成 - 这就是“运行直到完成”所指的。循环调用run_until_complete已经调度的任务就像下面的异步代码：

ts = [asyncio.create_task(asyncio.sleep(i)) for i in range(1, 11)]
# takes 10s, not 55s
for t in ts:
    await t

这在语义上又等同于以下线程代码：

ts = []
for i in range(1, 11):
    t = threading.Thread(target=time.sleep, args=(i,))
    t.start()
    ts.append(t)
# takes 10s, not 55s
for t in ts:
    t.join()

换句话说，await t and run_until_complete(t)阻塞直到t已完成，但允许其他一切 - 例如之前使用计划的任务asyncio.create_task()在那段时间也跑步。因此，总运行时间将等于最长任务的运行时间，而不是它们的总和。例如，如果第一个任务恰好需要很长时间，那么所有其他任务都会同时完成，并且他们的等待根本不会休眠。

所有这些仅适用于之前已安排的等待任务。如果您尝试将其应用于协程，它将不起作用：

# runs for 55s, as expected
for i in range(1, 11):
    await asyncio.sleep(i)

# also 55s - we didn't call create_task() so it's equivalent to the above
ts = [asyncio.sleep(i) for i in range(1, 11)]
for t in ts:
    await t

# also 55s
for i in range(1, 11):
   t = threading.Thread(target=time.sleep, args=(i,))
   t.start()
   t.join()

对于 asyncio 初学者来说，这通常是一个症结所在，他们编写与最后一个 asyncio 示例等效的代码，并期望它能够并行运行。

我尝试查看 asyncio 源代码来了解是否asyncio.wait只是在幕后有效地对其任务/未来做了同样的事情，但这并不明显。

asyncio.wait只是一个方便的 API，它做两件事：

• 将输入参数转换为实现的东西Future。对于协程来说，这意味着它将它们提交到事件循环，就像create_task，这使得它们能够独立运行。如果您像您一样一开始就给它任务，则会跳过此步骤。
• uses add_done_callback当 future 完成时收到通知，此时它恢复其调用者。

所以是的，它做同样的事情，但采用不同的实现，因为它支持更多的功能。

我假设如果其中一个任务正在进行长时间运行的阻塞操作，它实际上可能不会立即取消？

在 asyncio 中不应该有“阻塞”操作，只有那些挂起的操作，并且应该立即取消它们。例外情况是附加到 asyncio 上的阻塞代码run_in_executor，其中底层操作根本不会取消，但 asyncio 协程将立即收到异常。

也许这仅仅取决于正在使用的底层操作或库是否会立即引发 CancelledError ？

图书馆没有raise CancelledError, it receives它在等待点，在取消发生之前它恰好暂停。对于图书馆来说，取消的影响是await ...中断其等待并立即引发CancelledError。除非被捕获，否则异常将通过函数传播并await一直调用顶级协程，其引发CancelledError将整个任务标记为已取消。行为良好的 asyncio 代码将做到这一点，可能使用finally释放它们所持有的操作系​​统级资源。什么时候CancelledError被捕获，代码可以选择不重新引发它，在这种情况下，取消实际上被忽略。

是否有可能loop.run_until_complete（或者实际上，底层调用async.wait) 由于超时以外的原因返回未完成的值？

如果您正在使用return_when=asyncio.ALL_COMPLETE（默认），这应该是不可能的。这是很有可能的return_when=FIRST_COMPLETED，那么显然可以独立于超时。