python - asyncio.as_completed 是否产生 Futures 或协程？

Question

来自 asyncio docs :

asyncio.as_completed(aws, *, loop=None, timeout=None)

Run awaitable objects in the aws set concurrently. Return an iterator of Future objects. Each Future object returned represents the earliest result from the set of the remaining awaitables.

我假设这些 Future 对象中的每一个都具有 asyncio.Future 中描述的方法:.cancelled()、.exception() 和 .result()。但似乎产生的元素只是协程，而不是 Future 对象。我错过了什么？

这似乎与 .as_completed() 的描述不符。如果我需要await，协程如何“完成”？

>>> import asyncio
>>> import aiohttp
>>> 
>>> async def get(session, url):
...     async with session.request('GET', url) as resp:
...         t = await resp.text()
...         return t
... 
>>> async def bulk_as_completed(urls):
...     async with aiohttp.ClientSession() as session:
...         aws = [get(session, url) for url in urls]
...         for future in asyncio.as_completed(aws):
...             for i in ('cancelled', 'exception', 'result'):
...                 print(hasattr(future, i))
...             print(type(future))
...             try:
...                 result = await future
...             except:
...                 pass
...             else:
...                 print(type(result))
...                 print()
... 
>>> 
>>> urls = (
...     'https://docs.python.org/3/library/asyncio-task.html',
...     'https://docs.python.org/3/library/select.html',
...     'https://docs.python.org/3/library/this-page-will-404.html',
... )
>>> 
>>> asyncio.run(bulk_as_completed(urls))
False
False
False
<class 'coroutine'>
<class 'str'>

False
False
False
<class 'coroutine'>
<class 'str'>

False
False
False
<class 'coroutine'>
<class 'str'>

---

最终，我关心这个的原因是因为我想让异常冒泡，就像它们在asyncio.gather(..., return_exceptions=True)中所做的那样。 考虑添加一个伪造的 URL，当 session.request() 被调用时会出现:

urls = (
    'https://docs.python.org/3/library/asyncio-task.html',
    'https://docs.python.org/3/library/select.html',
    'https://docs.python.org/3/library/this-page-will-404.html',

    # This URL will raise on session.request().  How can I propagate
    # that exception to the iterator of results?
    'https://asdfasdfasdf-does-not-exist-asdfasdfasdf.com'
)

我希望能够做的是这样的事情(使用 Future 对象的方法，但这些根本不是 Future 对象，这就是问题所在):

async def bulk_as_completed(urls):
    async with aiohttp.ClientSession() as session:
        aws = [get(session, url) for url in urls]
        for future in asyncio.as_completed(aws):
            if future.cancelled():
                res = futures.CancelledError()
            else:
                exc = future.exception()
                if exc is not None:
                    res = exc
                else:
                    res = future.result()
            # ...
            # [Do something with `res`]

Answer 1

What I would like to be able to do is something like this [...]

也许不那么方便，但您应该能够使用如下代码提取异常:

async def bulk_as_completed(urls):
    async with aiohttp.ClientSession() as session:
        aws = [get(session, url) for url in urls]
        for future in asyncio.as_completed(aws):
            try:
                res = await future
            except Exception as e:
                res = e
            # ...
            # [Do something with `res`]

This [yielding coroutines rather than futures] seems to defeat the description of .as_completed(). How is the coroutine "completed" if I need to await it?

不是。首次实现 asyncio.as_completed 时，异步迭代器并不存在。如果没有异步迭代，就无法在 futures 完成时返回它们，因此 as_completed 通过产生(立即)虚拟等待对象来伪造它，人们必须等待这些等待对象才能获得实际结果。

即使 as_completed 产生了实际的 futures，它对您的用例也无济于事，因为如果没有人等待它们，这些 futures 就不会完成。为了提供 as_completed 产生 completed future 的预期语义，as_completed 需要实现异步迭代，其等效于 __next__可以等待。

as_completed 的惊人行为之前已经提到过，我已经提交了 an issue通过提供异步迭代来修复它。实现后，您的原始代码只需将 for 更改为 async for。