asyncio 的默认调度程序何时公平？

Question

据我了解，asyncio.gather 旨在并发运行其参数，并且当协程执行等待表达式时，它为事件循环提供了调度其他任务的机会。考虑到这一点，我惊讶地发现以下 sn-p 忽略了 asyncio.gather 的输入之一。

import asyncio                                                             
  
async def aprint(s):
    print(s)

async def forever(s):
    while True:
        await aprint(s)

async def main():
    await asyncio.gather(forever('a'), forever('b'))

asyncio.run(main())

据我了解，会发生以下情况：

1. asyncio.run(main()) 对事件循环进行任何必要的全局初始化并安排 main() 执行。
2. main() 安排 asyncio.gather(...) 执行并等待其结果
3. asyncio.gather 调度 forever('a') 和 forever('b') 的执行
4. 无论哪一个先执行，它们都会立即等待 aprint() 并让调度程序有机会在需要时运行另一个协程（例如，如果我们从 'a' 开始，那么我们就有机会开始尝试评估 'b' ，它应该已经被安排执行）。
5. 在输出中，我们将看到一行行，每行都包含“a”或“b”，并且调度程序应该足够公平，以便我们在足够长的时间段内至少看到每个行。李>

实际上，这不是我观察到的。相反，整个程序相当于while True: print('a')。我发现非常有趣的是，即使是对代码的微小更改似乎也重新引入了公平性。例如，如果我们改为使用以下代码，那么我们会在输出中得到大致相等的 'a' 和 'b' 组合。

async def forever(s):
    while True:
        await aprint(s)
        await asyncio.sleep(1.)

验证它似乎与我们在无限循环中花费多长时间没有任何关系，我发现以下更改也提供了公平性。

async def forever(s):
    while True:
        await aprint(s)
        await asyncio.sleep(0.)

有谁知道为什么会发生这种不公平以及如何避免这种情况？我想当有疑问时，我可以主动在任何地方添加一个空的 sleep 语句，并希望就足够了，但是对于我来说，为什么原始代码没有按预期运行是非常不明显的。

如果 asyncio 似乎经历了很多 API 更改，那么我在 Ubuntu 机器上使用 Python 3.8.4 的香草安装。

Answer 1

4. 无论哪个先执行，它们都会立即 await aprint() 并让调度程序有机会在需要时运行另一个协程

这部分是一个常见的误解。 Python 的await 并不意味着“将控制权交给事件循环”，它的意思是“开始执行可等待对象，允许它与它一起暂停我们”。所以是的，如果等待的对象选择挂起，当前的协程也将挂起，等待它的协程也将挂起，依此类推，一直到事件循环。但是如果等待的对象没有选择暂停，就像aprint 的情况一样，等待它的协程也不会。这有时是错误的来源，如 here 或 here 所示。

有谁知道为什么会发生这种不公平以及如何避免它？

幸运的是，这种效果在不与外界真正交流的玩具示例中最为明显。尽管您可以通过将await asyncio.sleep(0) 添加到战略位置来修复它们（甚至有文档证明它会强制进行上下文切换），但您可能会在生产代码中使用shouldn't。

一个真正的程序将依赖于来自外部世界的输入，无论是来自网络、本地数据库或由另一个线程或进程填充的工作队列的数据。实际数据很少会以如此快的速度到达而使程序的其余部分饿死，如果是这样，饿死很可能是暂时的，因为程序最终会由于其输出端的背压而暂停。在极少数情况下，程序从一个源接收数据的速度比处理它的速度快，但仍需要观察来自另一个源的数据，您可能会遇到饥饿问题，但这可以通过强制上下文切换来解决 if 它曾经被证明会发生。 （我还没有听说有人在生产中遇到过。）

除了上面提到的错误之外，更常见的情况是协程调用了 CPU 密集型或遗留阻塞代码，并最终占用了事件循环。此类情况应通过将 CPU/阻塞部分传递给run_in_executor 来处理。

Answer 2

我想提请注意PEP 492，上面写着：

await 与 yield from 类似，暂停执行 [...] 协程，直到 [...] awaitable 完成并返回结果数据。

它使用yield from 实现，并带有验证其参数的额外步骤。

任何yield from 调用链都以yield 结束。这是Futures 如何实现的基本机制。因为，在内部，协程是一种特殊的生成器，每个await 都被yield 挂起在await 调用链中的某处（请参阅PEP 3156 以获得详细说明） .

但在您的情况下，async def aprint() 没有 yield，也就是说，它不调用任何 事件函数，如 I/O 或只是 await sleep(0)，如果我们看一下源代码，只做yield:

@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
...

因此，由于永远存在while True:，我们可以说，您创建了一个不以yield 结尾的yield from 链。