Python 中的`await` 是否屈服于事件循环？

Question

我想知道当我们await在异步 Python 代码中使用协程时究竟会发生什么，例如：

await send_message(string)

(1)send_message加入事件循环，调用协程放弃对事件循环的控制，或

(2) 我们直接跳进去 send_message

我读到的大多数解释都指向 (1)，因为它们将调用协程描述为exiting。但我自己的实验表明（2）是这种情况：我试图在调用者之后但在被调用者之前运行一个协程，但无法实现这一点。

Answer 1

免责声明：自从我来到这里寻找答案以来，可以进行更正（尤其是细节和正确的术语）。尽管如此，下面的研究指出了一个非常决定性的“要点”结论：

正确答案OP：没有，await（本身）也不会屈服于事件循环，yield产量事件循环，因此对情况作了说明：“（2）我们直接跳入send_message”。特别是，某些yield表达式是唯一可以实际切换异步任务的点（在确定可以暂停 Python 代码执行的精确位置方面）。

需要证明和证明：1) 通过理论/文档，2) 通过实现代码，3) 通过示例。

通过理论/文件

PEP 492：协程async和await语法

虽然 PEP 不依赖于任何特定的事件循环实现，但它只与yield用作调度程序信号的协程类型相关，表明协程将等待事件（例如 IO）完成。...

[ await] 使用yield from实现 [带有验证其参数的额外步骤。] ...

任何yield from调用链都以yield. 这是Future实现 s的基本机制。由于在内部，协程是一种特殊的生成器，每个生成器都await被yield等待调用链的某个地方挂起（请参阅 PEP 3156 以获取详细说明）。...

协程内部基于生成器，因此它们共享实现。与生成器对象类似，协程也有throw()、send()和close()方法。...

现有的基于生成器的协程背后的愿景和这个提议是让用户很容易看到代码可能被挂起的地方。

在上下文中，“让用户很容易看到代码可能被挂起的位置”似乎是指在同步代码yield中可以在允许其他代码运行的例程中“挂起”执行的地方，并且该原则现在扩展了完全适用于异步上下文，其中 a yield（如果它的值没有在正在运行的任务中消耗，而是传播到调度程序）是“给调度程序的信号”以切换任务。

更简洁地说：发电机产量控制在哪里？在yield. 协程（包括使用async和await语法的协程）是生成器，因此也是如此。

这不仅仅是一个类比，在实现中（见下文），任务“进入”和“退出”协程的实际机制并不是异步世界中任何新的、神奇的或独特的，而只是通过调用科罗的<generator>.send()方法。那是（据我所知）PEP 492 背后“愿景”的一部分：async并且await不会提供代码暂停的新机制，而只是将异步糖倒在 Python 已经深受喜爱且功能强大的生成器上。

和 PEP 3156：“asyncio”模块

在报告慢回调之前loop.slow_callback_duration attribute控制两个屈服点之间允许的最大执行时间[强调原文]。

也就是说，一个不间断的代码段（从异步的角度来看）被划分为两个连续yield点之间的代码段（其值达到运行Task级别（通过await/yield from隧道）而没有在其中消耗）。

和这个：

调度程序没有公共接口。您可以使用yield from future和与它进行交互yield from task。

反对意见：“那是 ' yield from'，但你试图争辩说任务只能在 ayield本身上切换！yield from而且yield是不同的东西，我的朋友，yield from它本身不会挂起代码！”

答：不矛盾。PEP 说您通过使用yield from future/task. 但正如上面 PEP 492 中所指出的，任何yield from(~aka await)链最终都会到达一个yield(“底海龟”)。特别是（见下文），在一些包装工作之后yield from future实际上也是yield如此future，这yield是另一个任务接管的实际“切换点”。但它是不正确的代码，直接yield一个Future上升到目前Task，因为你将绕过必要的包装。

已经回答了反对意见，并注意到了它的实际编码考虑，我希望从上面的引文中提出的观点仍然是：yieldPython 异步代码中的一个合适的最终是一件事，它以标准方式暂停代码执行，任何其他yield会做，现在进一步参与调度程序以实现可能的任务切换。

按实现代码

异步/futures.py

class Future:
...
    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.

    __iter__ = __await__  # make compatible with 'yield from'.

释义：这一行yield self是告诉正在运行的任务暂时离开并让其他任务运行，在某个时间self完成后回到这个任务。

asyncio世界上几乎所有的 awaitable都是围绕Future. 事件循环遗体完全盲目所有较高级别await awaitable表达式直到代码执行滑下到一个await future或yield from future然后（如这里所示）的呼叫yield self，其产生self然后被“捕获”由无比其他Task其中本协程堆栈从而运行下向任务发出信号以休息一下。

在上下文中，上述“代码在yield self内部挂起await future”规则的一个也是唯一的例外可能asyncio是诸如 in之类的裸 的潜在使用。由于该函数是本文评论中的讨论主题，让我们来看看。yieldasyncio.sleep(0)sleep

异步/任务.py

@types.coroutine
def __sleep0():
    """Skip one event loop run cycle.
    This is a private helper for 'asyncio.sleep()', used
    when the 'delay' is set to 0.  It uses a bare 'yield'
    expression (which Task.__step knows how to handle)
    instead of creating a Future object.
    """
    yield


async def sleep(delay, result=None, *, loop=None):
    """Coroutine that completes after a given time (in seconds)."""
    if delay <= 0:
        await __sleep0()
        return result

    if loop is None:
        loop = events.get_running_loop()
    else:
        warnings.warn("The loop argument is deprecated since Python 3.8, "
                      "and scheduled for removal in Python 3.10.",
                      DeprecationWarning, stacklevel=2)

    future = loop.create_future()
    h = loop.call_later(delay,
                        futures._set_result_unless_cancelled,
                        future, result)
    try:
        return await future

    finally:
        h.cancel()

注意：我们这里有两个有趣的案例，在这些案例中，控制权可以转移到调度程序：

(1) 裸yield入__sleep0（当通过 调用时await）。

(2)yield self立即内await future。

asyncio/tasks.py 中的关键行（出于我们的目的）是何时通过Task._step运行其顶级协程result = self._coro.send(None)并识别四个案例：

(1)result = None由 coro（同样是一个生成器）生成：任务“放弃对一个事件循环迭代的控制”。

(2)result = future是在 coro 中生成的，具有进一步的魔术成员字段证据，表明未来以适当的方式从 out of 产生Future.__iter__ == Future.__await__：任务放弃对事件循环的控制，直到未来完成。

(3) AStopIteration由 coro 引发，表明协程已完成（即作为生成器，它耗尽了所有的yields）：任务的最终结果（它本身就是 a Future）被设置为协程的返回值。

(4) 任何其他Exception发生：相应地set_exception设置任务。

取模细节，我们关注的重点是asyncio事件循环中的协程段最终通过coro.send(). 除了初始启动和最终终止之外，send()精确地从yield它生成的最后一个值到下一个值。

举例

import asyncio
import types

def task_print(s):
    print(f"{asyncio.current_task().get_name()}: {s}")

async def other_task(s):
    task_print(s)

class AwaitableCls:
    def __await__(self):
        task_print("    'Jumped straight into' another `await`; the act of `await awaitable` *itself* doesn't 'pause' anything")
        yield
        task_print("    We're back to our awaitable object because that other task completed")
        asyncio.create_task(other_task("The event loop gets control when `yield` points (from an iterable coroutine) propagate up to the `current_task` through a suitable chain of `await` or `yield from` statements"))

async def coro():
    task_print("  'Jumped straight into' coro; the `await` keyword itself does nothing to 'pause' the current_task")
    await AwaitableCls()
    task_print("  'Jumped straight back into' coro; we have another pending task, but leaving an `__await__` doesn't 'pause' the task any more than entering the `__await__` does")

@types.coroutine
def iterable_coro(context):
    task_print(f"`{context} iterable_coro`: pre-yield")
    yield None # None or a Future object are the only legitimate yields to the task in asyncio
    task_print(f"`{context} iterable_coro`: post-yield")

async def original_task():
    asyncio.create_task(other_task("Aha, but a (suitably unconsumed) *`yield`* DOES 'pause' the current_task allowing the event scheduler to `_wakeup` another task"))

    task_print("Original task")
    await coro()
    task_print("'Jumped straight out of' coro. Leaving a coro, as with leaving/entering any awaitable, doesn't give control to the event loop")
    res = await iterable_coro("await")
    assert res is None
    asyncio.create_task(other_task("This doesn't run until the very end because the generated None following the creation of this task is consumed by the `for` loop"))
    for y in iterable_coro("for y in"):
        task_print(f"But 'ordinary' `yield` points (those which are consumed by the `current_task` itself) behave as ordinary without relinquishing control at the async/task-level; `y={y}`")
    task_print("Done with original task")

asyncio.get_event_loop().run_until_complete(original_task())

在 python3.8 中运行产生

任务 1：原始任务

任务 1：“直接跳入”coro；该await关键字本身并没有什么“暂停”的current_task

任务 1：“直接跳入”另一个await；await awaitable 本身的行为不会“暂停”任何事情

任务 2：啊哈，但是（适当地未消耗）yield确实“暂停”了 current_task，允许事件调度程序执行_wakeup另一个任务

任务 1：我们回到我们的可等待对象，因为其他任务已完成

任务 1：“直接跳回”coro；我们还有另一个待处理的任务，但是离开__await__并不会“暂停”任务，就像输入__await__一样

任务 1：“直接跳出”coro。离开 coro，就像离开/进入任何可等待的一样，不会控制事件循环

任务 1 await iterable_coro：：预屈服

任务 3：当yield点（来自可迭代协程）current_task通过合适的awaitoryield from语句链传播到时，事件循环获得控制

任务 1: await iterable_coro: 后屈服

任务 1 for y in iterable_coro：：预屈服

任务 1：但是“普通”yield点（那些被current_task自身消耗的点）表现得和普通点一样，不会放弃异步/任务级别的控制；y=None

任务 1: for y in iterable_coro: 后屈服

任务 1：完成原始任务

任务 4：这直到最后才运行，因为在创建此任务之后生成的 None 被for循环消耗

事实上，像下面这样的练习可以帮助人们将async/的功能await与“事件循环”等概念分离开来。前者有利于后者的良好实现和用法，但您可以使用async和await就像特殊语法的生成器一样，而无需任何“循环”（无论是否asyncio）：

import types # no asyncio, nor any other loop framework

async def f1():
    print(1)
    print(await f2(),'= await f2()')
    return 8

@types.coroutine
def f2():
    print(2)
    print((yield 3),'= yield 3')
    return 7

class F3:
   def __await__(self):
        print(4)
        print((yield 5),'= yield 5')
        print(10)
        return 11

task1 = f1()
task2 = F3().__await__()
""" You could say calls to send() represent our
   "manual task management" in this script.
"""
print(task1.send(None), '= task1.send(None)')
print(task2.send(None), '= task2.send(None)')
try:
    print(task1.send(6), 'try task1.send(6)')
except StopIteration as e:
    print(e.value, '= except task1.send(6)')
try:
    print(task2.send(9), 'try task2.send(9)')
except StopIteration as e:
    print(e.value, '= except task2.send(9)')

产生

1

2

3 = task1.send(无)

4

5 = task2.send(无)

6 = 产量 3

7 = 等待 f2()

8 = 除了 task1.send(6)

9 = 产量 5

10

11 = 除了 task2.send(9)