对 coroutine.__await__() 调用 next 为什么会导致重复进入 Future.__await__()

[abersheeran]

import types
from typing import Any, Awaitable


@types.coroutine
def _async_yield(obj):
    return (yield obj)


async def gather(*aws: Awaitable[Any], return_exceptions: bool = False):
    result_array: list[Any] = [None for _ in range(len(aws))]
    done, pending = [], [*aws]
    while pending:
        for i, awaitable in enumerate(aws):
            if awaitable in done:
                continue
            try:
                next(awaitable.__await__())
                await _async_yield(None)
            except StopIteration as exc:
                result_array[i] = exc.value
                done.append(awaitable)
                pending.remove(awaitable)
            except BaseException as exc:
                if not return_exceptions:
                    raise
                result_array[i] = exc
                done.append(awaitable)
                pending.remove(awaitable)
    return result_array


async def test_wait():
    async def a():
        return 1

    async def b():
        import asyncio

        await asyncio.sleep(0)
        return 2

    async def c():
        import asyncio

        await asyncio.sleep(1)
        return 3

    return await gather(a(), b(), c(), asyncio.create_task(c()))


if __name__ == "__main__":
    import asyncio

    print(asyncio.run(test_wait()))

报错所处的位置：

    # asyncio/futures.py line 281
    def __await__(self):
        if not self.done():
            self._asyncio_future_blocking = True
            yield self  # This tells Task to wait for completion.
        if not self.done():
            raise RuntimeError("await wasn't used with future")
        return self.result()  # May raise too.

在 asyncio/tasks.py 的 Task.__step 找到了答案。此贴完结。

[veoco]

不能在同步 for 里调用异步代码

[ruanimal]

因为 Python 是无栈协程？

[Kobayashi]

原因大概可以解释。看样子 awaitable.__await__() 返回了 future. 而 Future.__iter__ = Future.__await__. 而 Future result 未被设置时，Future.__await__() 返回自己。

从你自定义的 gather() 逻辑推测，你想绕过 asyncio 的任务运行控制，自己严格控制过个 Task 交替运行？比如任务 A, B, C 分别拆分为步骤 A1, A2, A3, B1, B2 ...，你想确保 A1, B1, C1, A2, ... 顺序？

建议先把 asyncio 源码读完理解 loop 如何全局调控任务、Task 封装 coroutine 起什么作用，以及 Future 又是什么，之后回过头来想这个事情。

异步就是在等一个任务的时候，去做另外一个任务。asyncio 事件循环默认不实现 A1, B1, C1 有序完全没有问题。
假设实现了这样的机制，如果 A1 运行需要等待一个新任务 X ，而 B1, C1 要等待 A1 。现在事件循环中所有任务都在等 X ，这还算异步吗？
另外，这里强调的是事件循环默认行为不能这么做，不然可能引起阻塞。但 asyncio 确实提供了任务间依赖的机制：Event, Lock, Condition ... 其原理都是 Future ，而 Future 本质上就是一个信号，任务 X 开始时返回一个信号给 A ，A 拿到信号后把自己后续步骤执行作为信号处理函数挂上去，任务 X 完成时触发信号，运行跳回 A1 。

总之，要自定义实现任务有序，1 ）局部加入步骤有序，要么用 Event, Condition, Lock, Semaphore ，或者直接使用其更底层 Future 信号机制在步骤间建立机制。2 ）全局的话不行，但作为一个基础包，asyncio 不能让事件循环处理任务步骤时有序，这样就不是异步了。自己玩玩不发包怎么搞都行。

[Sinksky]

因为 asyncio.sleep 会跑到 asyncio 的事件循环里面？

[abersheeran]

@Kobayashi loop 是怎么感知到 Future 已经可以读结果的？是通过 isfuture 判断类型再去读状态吗？我没找到这部分代码。

[Kobayashi]

@abersheeran 不感知，loop 不直接处理 Future 。
异步里协程不是主动运行，而是把自己交给 loop ，loop 负责调控所有待运行任务列表，它不管协程返回什么值。

Task 是对于下边协程的封装，Task._step() 调用协程的每一步，并对每一步的返回值做出响应。假设有任务 A ，它在调用协程时需要等待任务 X ，任务 X 先返回 future 给 A ，TaskA._step() 判断拿到了 future ，则把后续运行动作（还是 ._step()）注为 future 的回调。等 X 完成后，它会设置 future 值，触发回调调用回到 A 。
一个任务等待另一个任务不经过 loop ，就是利用信号挂起后续执行。

[abersheeran]

@Kobayashi 不感知是怎么结束 await 等待的？