此页面由 Cloud Translation API 翻译。

在 Fuchsia Controller 中使用异步 Python 的最佳实践

本页面尽可能介绍了异步 Python 的常见任务流，使用异步 Python 时会遇到的一些问题。

继续操作之前，请查看此官方文档，关于 asyncio 和 Fuchsia 控制器教程。

背景

以下各项通常与异步代码相关，不一定是特定的至 Python：

异步代码不一定是多线程的。如果与 Fuchsia 控制器相关的代码，几乎所有内容都是单线程的。
异步代码在循环（也称为执行器）内运行，将释放到相应的优先级

因此，这可能会使某些问题难以调试，例如如果某个后台任务打算永久运行，但又引发了异常。如果您从不等待此任务，日志。

常见误区

未保留对任务的引用

如果您在循环中启动任务，例如：

# Don't do this!
asyncio.get_running_loop().create_task(foo_bar())

而您没有保存对返回值的引用，则有可能 Python 垃圾回收器可能会随时取消和删除此任务。

请始终确保将创建的任务存储在变量中，例如：

# Always store a variable!
foo_bar_task = asyncio.get_running_loop().create_task(foo_bar())

然后，确保任务的生命周期与需要的内容相关。如果这是特定测试用例，请将任务存储为测试用例类的成员，在拆解期间取消它

将阻塞代码与异步代码混用

在测试代码时，某些实验通常会抛出 sleep 语句。切勿将 asyncio.sleep 与 time.sleep。前者返回一个协程，而后者会阻止整个循环。

例如：

import asyncio
import time


async def printer():
    """ Prints a thing."""
    print("HELLO")


async def main():
    task = asyncio.get_running_loop().create_task(printer())
     time.sleep(1) # THIS WILL BLOCK EXECUTION. 
    task.cancel()

if __name__ == "__main__":
    asyncio.run(main())

上面的代码不会输出任何内容。但是，如果您输入 HELLO，将 time.sleep(1) 替换为 await asyncio.sleep(1)，从而得到异步循环。

FIDL 服务器错误

鉴于异步编程的性质，您可以使用您的 FIDL 服务器实现中存在的错误而不知道，特别是 FIDL 服务器实现已连接到 Fuchsia 设备上的组件。

遗憾的是，这种情况难以调试，而且只能以由于 FIDL 服务器实现崩溃而导致设备上的 PEER_CLOSED 错误。

一种方法（用于调试）是向任务添加一个回调，用于检查任务中发生了异常，并向日志发送一些内容（使程序崩溃）。

例如：

def log_exception(task):
    try:
        _ = task.result()
    except Exception as e:
        # Can log the exception here for debugging.

task = asyncio.get_running_loop().create_task(foobar())
task.add_done_callback(log_exception)

常见任务流

同步任务

建议您先查看有关入门指南的 Python 文档页面可用的同步对象。请注意，这些对象并非线程安全对象。然而，其中一些示例值得参考。

正在等待多个任务

您可能需要等待多个任务完成。为此，使用 asyncio.wait。想做点什么类似于 select 系统调用，请将 return_when 参数设为 asyncio.FIRST_COMPLETED，例如：

done, pending = await asyncio.wait(
    [task1, task2, task3],
    return_when=asyncio.FIRST_COMPLETED
)

结果包含一个任务元组（在本例中为 done 和 pending）。哪个可以像对待任何其他任务对象一样处理。对于 done 对象，可以执行以下操作：遍历这些指标，并通过检查 result() 函数来收集结果。

在同步代码中运行异步代码

在编写本文时，大部分 Fuchsia 控制器代码都是从同步代码。为确保任务可以在后台运行，合理的做法是通过 asyncio.new_event_loop() 保留一个循环实例。

原因在于 asyncio 任务必须保持在单个循环中。这个 asyncio.Queue 等同步对象也是如此。

如果您交互的代码足够简单（运行单个 FIDL）方法），则可以使用：

asyncio.run_until_complete(...)

但是，如果您需要运行任务或处理同步，将内容封装在包含循环的类中。请务必确保都有用于关闭循环的拆解代码，例如：

class FidlHandlingClass():

    def __init__(self):
        self._loop = asyncio.new_event_loop()
        # Can launch tasks here.
        self._important_task = self._loop.create_task(foo())

    def __del__(self):
        self._important_task.close()
        self._loop.close()

您还可以使用装饰器扩展类，使循环隐式地用于所有公共函数中（尽管它们对调用方而言会显示为同步）例如：

from functools import wraps

class FidlInstance():

    def __init__(self, proxy_channel):
        self._loop = asyncio.new_event_loop()
        self.echo_proxy = fidl.fuchsia_developer_ffx.Echo.Client(proxy_channel)

    def __del__(self):
        self._loop.close()

    def _asyncmethod(f):
        @wraps(f)
        def wrapped(inst, *args, **kwargs):
            coro = f(inst, *args, **kwargs)
            inst._loop.run_until_complete(coro)
        return wrapped

    @_asyncmethod
    async def echo(self, string: str):
        return await self.echo_proxy.echo_string(value=string)

然后，可以使用以下代码调用上述类：

client_channel = ...
inst = FidlInstance(client_channel)
print(inst.echo("foobar"))

虽然此实例可在非异步上下文中运行，但它仍在运行异步代码。不过，这样会大大降低总体读写操作难度。

调整代码以支持异步 Python

FIDL 是一种围绕异步设计的语言。不过，像下面这样测试框架： Mobly 需要同步的 Python 函数来运行测试。如果您有主要用于同步测试和 Python，那么这适用于您。

fuchsia-controller 的 wrappers 模块下包含一些 mixin 代码，支持在同步上下文（特别是 AsyncAdapter）中使用异步代码，它可用作封装容器或 mixin，而装饰器 asyncmethod。

当彼此一起使用时，这会创建一个设置，其中所有代码在给定实例中运行的测试针对常见的 asyncio 事件循环执行。

在大多数情况下，您可以使用共 asyncio.run() 页。然而，您可能需要等待队列或异步任务检查状态，这变得非常难以用途。每次调用 asyncio.run() 都会创建一个全新的事件循环，异步数据结构的管理可能会带来麻烦，一次只能用于一个事件循环中。保留单个 asyncio 事件您可以避免遇到这些异常。

例如，假设您要编写一个 Mobly 测试用例，但要构建的类公开依赖于某种底层异步 Python，您可以像这样因此可以在 Mobly 中使用：

import asyncio
from mobly import asserts, base_test, test_runner
from fuchsia_controller_py.wrappers import AsyncAdapter, asyncmethod

# AsyncAdapter should be included first in the list of base classes.
class ClassYouWantToTest(AsyncAdapter, SomeBaseClass):

    def __init__(self):
        super().__init__()
        self.async_init()

    @asyncmethod
    async def async_init(self):
        self.queue: asyncio.Queue[int] = asyncio.Queue()
        await self.queue.put(1)

    @asyncmethod
    async def function_we_care_about(self) -> int:
        got = await self.queue.get()
        self.queue.task_done()
        return got


class ExampleTest(base_test.BaseTestClass):

    def test_case_example(self) -> None:
        """Example... doesn't really do much useful."""
        c = ClassYouWantToTest()
        asserts.assert_equal(c.function_we_care_about(), 1)

if __name__ == "__main__":
    test_runner.main()

方法 function_we_care_about 将被封装，以便在功能上执行 Mobly 测试用例时，系统会执行以下代码：

def function_we_care_about(self) -> None:
  coro = self._function_we_care_about_impl()
  self._mixin_asyncio_loop.run_until_complete(coro)