python - 正确使用 asyncio.Condition 的 wait_for() 方法

Question

我正在使用 Python 的 asyncio 编写一个项目模块，我想使用它的同步原语来同步我的任务。但是，它的行为似乎并不像我预期的那样。

从文档看来，Condition.wait_for()提供了一种方法，允许协程等待特定的用户定义条件评估为真。然而，在尝试使用该方法时，它的表现似乎出乎我的意料——我的条件只被检查一次，如果发现它是假的，等待的任务就会永远挂起，而不会再次检查。我在下面写了一个简短的例子来演示我正在尝试做的事情:

#!/usr/bin/env python

import asyncio

thing = False

setter_done = None
getter_done = None

async def main():

    setter_done = asyncio.Event()
    getter_done = asyncio.Event()

    setter = asyncio.ensure_future(set_thing())
    getter = asyncio.ensure_future(get_thing())

    #To avoid the loop exiting prematurely:
    await setter_done.wait()
    await getter_done.wait()

async def set_thing():

    global thing
    global setter_done

    thing = False
    #sleep for some arbitrary amount of time; simulate work happening
    await asyncio.sleep(10)
    thing = True

    print("Thing was set to True!")
    setter_done.set()

async def get_thing():

    global thing
    global getter_done

    def check_thing():
        print("Checking...")
        return thing

    c = asyncio.Condition()
    await c.acquire()
    await c.wait_for(check_thing)
    c.release()
    print("Thing was found to be true!")
    getter_done.set()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

我希望这会打印如下内容:

Checking...
Thing was set to True!
Checking...
Thing was found to be True!

相反，我得到:

Checking...
Thing was set to True!
... (hangs indefinitely)

Answer 1

我发布了一个包含大量评论的完整答案，以帮助解决类似问题的人。我已将代码示例更改为使用类而不是全局变量。它有点长，但我希望它不会太复杂。

Command 类基本上代表一个任务。它是异步的，所以它可以做很多事情。在我的例子中，我只创建了两个虚拟命令(读作“两个任务”)，一个暂停 5 秒，一个暂停 8 秒，然后我等待它们都在一个条件下结束。显然，条件并不是完成我所做工作的唯一方法，但为了与原始答案保持一致，我认为提供一个完整的示例很有趣。所以开始吧!

import asyncio
from typing import Set

class Command:

    """A command, an asynchronous task, imagine an asynchronous action."""

    async def run(self):
        """To be defined in sub-classes."""
        pass

    async def start(self, condition: asyncio.Condition,
            commands: Set['Command']):
        """
        Start the task, calling run asynchronously.

        This method also keeps track of the running commands.

        """
        commands.add(self)
        await self.run()
        commands.remove(self)

        # At this point, we should ask the condition to update
        # as the number of running commands might have reached 0.
        async with condition:
            condition.notify()

class Look(Command):

    """A subclass of a command, running a dummy task."""

    async def run(self):
        print("Before looking...")
        await asyncio.sleep(5)
        print("After looking")

class Scan(Command):

    """A subclass of a command, running a dummy task."""

    async def run(self):
        print("Before scanning...")
        await asyncio.sleep(8)
        print("After scanning")

async def main():
    """Our main coroutine, starting commands."""
    condition = asyncio.Condition()
    commands = set()
    commands.add(Look())
    commands.add(Scan())
    asyncio.gather(*(cmd.start(condition, commands) for cmd in commands))

    # Wait for the number of commands to reach 0
    async with condition:
        await condition.wait_for(lambda: len(commands) == 0)
        print("There's no running command now, exiting.")

asyncio.run(main())

所以在实践中(像往常一样，从末尾开始)，我们将 main 称为协程。在 main 中，我们创建了两个命令，Look 和 Scan，并调用它们的 start 方法。 start 方法在每个命令上定义，它基本上负责在命令运行前将命令本身写入集合中，并在运行后(即完全完成后)将其删除。然后它应该通知条件再次检查命令的长度。当没有剩余命令时，程序结束。如果你运行这个脚本(我用 Python 3.8 运行它)你应该看到类似的东西:

Before scanning...
Before looking...
After looking
After scanning
There's no running command now, exiting.

请注意，这两个命令同时开始(嗯，事实上，Look 稍早开始，但 Scan 仍然在 Look 之前开始 完成)。但是 Look 确实在 Scan 之前结束(大约 3 秒)。在两个命令都完成之前，我们的条件不会被检查。

是否可以使用事件或锁或信号量来代替？有可能，但我喜欢在那个例子中使用条件。您无需大量修改即可轻松完成更多任务。