asyncio.sleep(0) 不会将控制权交给事件循环

Question

我有一个简单的异步设置，其中包括两个协程：light_job 和 Heavy_job。light_job 中途停止，heavy_job 开始。我希望 Heavy_job 在中间产生控制权并允许 light_job 完成，但 asyncio.sleep(0) 没有按我的预期工作。

这是设置：

import asyncio
import time

loop = asyncio.get_event_loop()


async def light_job():
    print("hello ")
    print(time.time())
    await asyncio.sleep(1)
    print(time.time())
    print("world!")


async def heavy_job():
    print("heavy start")
    time.sleep(3)
    print("heavy halt started")
    await asyncio.sleep(0)
    print("heavy halt ended")
    time.sleep(3)
    print("heavy done")

loop.run_until_complete(asyncio.gather(
    light_job(),
    heavy_job()
))

如果我运行此代码，则 light_job 将不会继续，直到 Heavy_job 完成后。这是输出：

hello 
1668793123.159075
haevy start
heavy halt started
heavy halt ended
heavy done
1668793129.1706061
world!

但如果我将 asyncio.sleep(0) 更改为 asyncio.sleep(0.0001)，代码将按预期工作：

hello 
1668793379.599066
heavy start
heavy halt started
1668793382.605899
world!
heavy halt ended
heavy done

根据文档和相关线程，我希望 asyncio.sleep(0) 与 asyncio.sleep(0.0001) 完全一样工作。这里出了什么事？

Answer 1

我认为这个话题还需要更多讨论。我打算将这篇文章作为 Daniel T 出色且非常聪明的答案的附录 - 这是一篇出色的作品。但 Dan Getz 的评论让我认为更多细节会有所帮助。

丹表示，没有通用的方法可以让步于另一项任务。这是正确的，因为不能保证任何其他任务已准备好运行，也不能保证各个任务的执行顺序。由于事件循环实现中的细节问题，示例程序未能满足预期，我将在下面讨论。

然而，有一些工具可以明确地同步不同任务之间的工作。asyncio.sleep()为此目的而依赖时间间隔可能是个坏主意。考虑以下程序，它使用一个asyncio.Event强制 light_job() 在 Heavy_job() 进入第二次睡眠延迟之前完成。这总是有效的，因为程序逻辑是明确的：

import asyncio
import time

event = asyncio.Event()

async def light_job():
    print("hello ")
    print(time.time())
    await asyncio.sleep(1)
    print(time.time())
    print("world!")
    event.set()


async def heavy_job():
    print("heavy start")
    time.sleep(3)
    print("heavy halt started")
    # await asyncio.sleep(0)
    await event.wait()
    print("heavy halt ended")
    time.sleep(3)
    print("heavy done")
    
async def main():
    await asyncio.gather(light_job(), heavy_job())

asyncio.run(main())

这种方法更简单，它避免使用 Event 甚至 Gather：

import asyncio
import time

async def light_job():
    print("hello ")
    print(time.time())
    await asyncio.sleep(1)
    print(time.time())
    print("world!")

async def heavy_job():
    light = asyncio.create_task(light_job())
    print("heavy start")
    time.sleep(3)
    print("heavy halt started")
    # await asyncio.sleep(0)
    await light
    print("heavy halt ended")
    time.sleep(3)
    print("heavy done")
    
async def main():
    await heavy_job()

asyncio.run(main())

至于原脚本为什么失败，可以在事件循环实现中找到解释。事件循环跟踪两件事：“就绪”项目列表，表示现在能够执行的任务；以及“计划”项目列表，表示正在等待某个时间间隔到期的任务。

每当事件循环经历一个周期时，它的第一步是检查计划项目列表，看看是否有任何项目准备好继续进行。它将这些项目中的任何一个附加到“就绪”列表中。然后它执行这个简单的循环来运行所有准备好的任务（我省略了一些诊断代码；这是来自Python3.10标准库模块base_events.py）。这里，_ready是一个双端队列。队列中的项目都有一个run方法，使任务向前迈出一步，或者换句话说，使任务从先前挂起的位置恢复（通常是await 表达式）。

    ntodo = len(self._ready)
    for i in range(ntodo):
        handle = self._ready.popleft()
        if handle._cancelled:
            continue
        else:
            handle._run()

这种情况的await asyncio.sleep(0)实现也不同于await asyncio.sleep(x)，其中x > 0。在第一种情况下，await 表达式生成 None 值。Task 对象只是将一个项目附加到“就绪”列表中。在第二种情况下，await 表达式执行一个loop.call_later函数调用，这会创建一个 Future。任务对象将一个项目附加到“计划”列表中。这是tasks.py中的实现asyncio.sleep：

@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):
    """Coroutine that completes after a given time (in seconds)."""
    if delay <= 0:
        await __sleep0()
        return result

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

因此，在原始帖子的示例脚本中，任务test将从其“就绪”列表中的两项开始：[light_job，heavy_job]。预定列表为空。Light_job 启动并命中await asyncio.sleep(1)，因此将一个项目附加到表示此时间延迟的“计划”列表中。现在，heavy_job 运行了三秒并命中了await asyncio.sleep(0)，因此一个项目被附加到“就绪”列表中，这表明该任务将立即继续进行。这是事件循环的一个完整周期的结束。即使此时就绪列表不为空，循环也会结束，因为零延迟的等待导致 Heavy_job 立即追加到就绪列表中。

在事件循环的下一个周期中，就绪列表有一个项目，它是在上一个周期中放置的：[heavy_job]。预定列表还有一项：[light_job]。事件循环检查计划列表并发现 light_job 现在已准备就绪，因此它将light_job附加到 read_list，现在看起来像这样：[heavy_job, light_job]。因此，代码逻辑本质上导致了任务顺序的切换。结果：heavy_job 连续运行两次，一次在第一个周期结束时，一次在第二个周期开始时。

await asyncio.sleep(0)这也解释了当您替换为时发生的情况await asyncio.sleep(0.0001)。在这种情况下，任务将被附加到计划列表而不是就绪列表中。然后ready=[]和scheduled=[light_job, Heavy_job]。在循环的下一个周期，两个任务都已准备就绪，但顺序将再次变为 [light_job, Heavy_job]。

这种机制对于客户端代码来说是不可见的，正如它应该的那样，但它在这个特定的脚本中产生了一个奇怪的结果。这是否应该被称为“错误”是一个有争议的问题。我认为 asyncio.sleep(0) 的实现方式与 asyncio.sleep(nonzero) 不同有很好的性能原因。

Answer 2

拨打asyncio.sleep(0)3次：

import asyncio
import time


async def light_job():
    print("hello ")
    print(time.time())
    await asyncio.sleep(1)
    print(time.time())
    print("world!")


async def heavy_job():
    print("heavy start")
    time.sleep(3)
    print("heavy halt started")
    for _ in range(3):
        await asyncio.sleep(0)
    print("heavy halt ended")
    time.sleep(3)
    print("heavy done")


async def test():
    await asyncio.gather(
        light_job(),
        heavy_job()
    )

asyncio.run(test())

这导致：

hello 
1668844526.157173
heavy start
heavy halt started
1668844529.1575627
world!
heavy halt ended
heavy done

查看“asyncio/base_events.py”，“_run_once”首先遍历挂起的计时器，然后在计算后运行它看到的所有内容。asyncio.sleep只能跳过事件循环的一次迭代。需要多次睡眠，因为安排一个未来，在通过添加回队列asyncio.sleep(1)来交回控制权之前需要一次额外的迭代，并且碰巧最后运行新排队的作业。light_joblight_jobasyncio

为了更清晰的图片，可以添加更多打印语句：

hello 
1668844526.157173
heavy start
heavy halt started
1668844529.1575627
world!
heavy halt ended
heavy done

然后在“asyncio/base_events.py”的“def _run_once(self):”中添加断点。在第 1842 行开始处添加一个打印“loop start”的断点，即“sched_count =”。在第 1910 行末尾添加另一个，即“handle = None”，打印“循环结束”。然后在第 1897 行运行每个任务之前添加一个，即“if self._debug:”，评估并打印“_format_handle(handle)”。事件的先后顺序已揭晓：

loop start
<Task pending name='Task-1' coro=<test() running at /home/home/PycharmProjects/sandbox/notsync.py:34> cb=[_run_until_complete_cb() at /usr/lib/python3.11/asyncio/base_events.py:180]>
loop end
loop start
<Task pending name='Task-2' coro=<light_job() running at /home/home/PycharmProjects/sandbox/notsync.py:5> cb=[gather.<locals>._done_callback() at /usr/lib/python3.11/asyncio/tasks.py:759]>
hello 
1668844827.5052986
<Task pending name='Task-3' coro=<heavy_job() running at /home/home/PycharmProjects/sandbox/notsync.py:13> cb=[gather.<locals>._done_callback() at /usr/lib/python3.11/asyncio/tasks.py:759]>
heavy start
heavy halt started
loop end
loop start
<Task pending name='Task-3' coro=<heavy_job() running at /home/home/PycharmProjects/sandbox/notsync.py:18> cb=[gather.<locals>._done_callback() at /usr/lib/python3.11/asyncio/tasks.py:759]>
after 1 sleep
<TimerHandle when=37442.097934711 _set_result_unless_cancelled(<Future pendi...ask_wakeup()]>, None) at /usr/lib/python3.11/asyncio/futures.py:317>
loop end
loop start
<Task pending name='Task-3' coro=<heavy_job() running at /home/home/PycharmProjects/sandbox/notsync.py:23> cb=[gather.<locals>._done_callback() at /usr/lib/python3.11/asyncio/tasks.py:759]>
after 2 sleeps
<Task pending name='Task-2' coro=<light_job() running at /home/home/PycharmProjects/sandbox/notsync.py:8> wait_for=<Future finished result=None> cb=[gather.<locals>._done_callback() at /usr/lib/python3.11/asyncio/tasks.py:759]>
1668844830.9250844
world!
loop end
loop start
<Task pending name='Task-3' coro=<heavy_job() running at /home/home/PycharmProjects/sandbox/notsync.py:27> cb=[gather.<locals>._done_callback() at /usr/lib/python3.11/asyncio/tasks.py:759]>
heavy halt ended
heavy done
<Handle gather.<locals>._done_callback(<Task finishe...> result=None>) at /usr/lib/python3.11/asyncio/tasks.py:759>
loop end
loop start
<Handle gather.<locals>._done_callback(<Task finishe...> result=None>) at /usr/lib/python3.11/asyncio/tasks.py:759>
loop end
loop start
<Task pending name='Task-1' coro=<test() running at /home/home/PycharmProjects/sandbox/notsync.py:35> wait_for=<_GatheringFuture finished result=[None, None]> cb=[_run_until_complete_cb() at /usr/lib/python3.11/asyncio/base_events.py:180]>
loop end
loop start
<Handle _run_until_complete_cb(<Task finishe...> result=None>) at /usr/lib/python3.11/asyncio/base_events.py:180>
loop end
loop start
<Task pending name='Task-4' coro=<BaseEventLoop.shutdown_asyncgens() running at /usr/lib/python3.11/asyncio/base_events.py:539> cb=[_run_until_complete_cb() at /usr/lib/python3.11/asyncio/base_events.py:180]>
loop end
loop start
<Handle _run_until_complete_cb(<Task finishe...> result=None>) at /usr/lib/python3.11/asyncio/base_events.py:180>
loop end
loop start
<Task pending name='Task-5' coro=<BaseEventLoop.shutdown_default_executor() running at /usr/lib/python3.11/asyncio/base_events.py:564> cb=[_run_until_complete_cb() at /usr/lib/python3.11/asyncio/base_events.py:180]>
loop end
loop start
<Handle _run_until_complete_cb(<Task finishe...> result=None>) at /usr/lib/python3.11/asyncio/base_events.py:180>
loop end