使用asyncio处理超时

免责声明:这是我第一次尝试该asyncio模块。

我使用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。经过研究后,我发现了多种取消任务并等待它们被实际取消的方法:

[task.cancel() for task in unfinished]

for task in unfinished:

with suppress(asyncio.CancelledError):

loop.run_until_complete(task)

[task.cancel() for task in unfinished]

loop.run_until_complete(asyncio.wait(unfinished))

# 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

像这样的答案中的某种while循环。似乎我的其他选择更好,但包括完整性。


到目前为止,选项1和2似乎都可以正常工作。每种选择都可能是“正确的”,但是随着asyncio这些年来的发展,网络上的示例和建议已经过时或相差很大。所以我的问题是

选项1和2之间有实际区别吗?我知道run_until_complete它将一直运行到将来完成为止,所以由于选项1以特定的顺序循环,所以我认为如果较早的任务需要更长的时间才能完成,它的行为可能会有所不同。我尝试查看asyncio源代码以了解其在后台asyncio.wait的任务/功能是否有效地完成了同样的事情,但这并不明显。

我认为,如果一项任务处于长时间运行的阻塞操作中间,它实际上可能不会立即取消吗?也许这仅取决于所使用的基础操作或库是否会立即引发CancelledError?为异步设计的库也许永远不会发生这种情况?

由于我试图在此处实现超时功能,因此我对此有些敏感。如果可能的话,这些事情可能需要很长时间才能取消,所以我会考虑致电cancel而不是等待它真正发生,或者设置一个很短的超时时间以等待取消完成。

是否有可能loop.run_until_complete(或者实际上是对的基础调用async.waitunfinished出于超时以外的原因返回值?如果是这样,我显然必须调整一下逻辑,但是从文档看来这是不可能的。

回答:

选项1和2之间有实际区别吗?

否。选项2看起来更好,效率可能略高,但是它们的最终效果是相同的。

我知道run_until_complete它将一直运行到将来完成为止,所以由于选项1以特定的顺序循环,所以我认为如果较早的任务需要更长的时间才能完成,它的行为可能会有所不同。

一开始似乎是这样,但实际上并非如此,因为loop.run_until_complete运行提交到循环的

任务,而不仅仅是运行作为参数传递的任务。它仅在提供的等待完成时

-这就是“运行直到完成”的意思。调用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 trun_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,这使它们可以独立运行。如果您像开始那样给它任务,那么将跳过此步骤。
  • 用于add_done_callback在期货交易完成时得到通知,届时它将恢复其呼叫者。

是的,它执行相同的操作,但是实现方式不同,因为它支持更多功能。

我认为,如果一项任务处于长时间运行的阻塞操作中间,它实际上可能不会立即取消吗?

在异步中,不应有“阻塞”操作,只有那些挂起的操作才应该被取消。例外是使用阻止附加到asyncio的代码run_in_executor,其中根本不会取消基础操作,但是asyncio协程将立即获取该异常。

也许这仅取决于所使用的基础操作或库是否会立即引发CancelledError?

库不 提高CancelledError,它在取消发生之前暂停的等待点 接收 它。对于图书馆而言,取消的效果是await

...中断其等待并立即引发CancelledError。除非被捕获,否则异常将通过函数传播并一直await调用顶级协程,协程的提升CancelledError将整个任务标记为已取消。行为良好的异步代码将执行此操作,可能会finally用于释放它们持有的OS级资源。当CancelledError被捕获时,代码可以选择不重新引发它,在这种情况下,取消被有效地忽略了。

是否有可能loop.run_until_complete(或者实际上是对的基础调用async.wait)由于超时以外的原因而返回未完成的值?

如果您使用的return_when=asyncio.ALL_COMPLETE是默认设置,那是不可能的。使用很有可能return_when=FIRST_COMPLETED,然后显然有可能独立于超时。

以上是 使用asyncio处理超时 的全部内容, 来源链接: utcz.com/qa/417000.html

回到顶部