本頁面由 Cloud Translation API 翻譯而成。

在 Fuchsia Controller 中使用非同步 Python 的最佳做法

本頁會討論非同步 Python 的常見工作流程可能會發生錯誤

繼續操作前，請先參閱這篇 asyncio和官方文件 Fuchsia Controller 的教學課程。

背景

下列項目與非同步程式碼相關的一般問題，不一定是具體問題至 Python：

非同步程式碼不一定是多執行緒。如果選擇 Fuchsia Controller 相關程式碼，幾乎所有內容都是單一執行緒。
非同步程式碼 (也稱為執行程式) 內部執行就會產生預測結果

因此，可能會導致某些問題難以偵錯，例如如果有要永久執行的背景工作，但狀態例外狀況。如果您從未等待此工作，例外狀況只會顯示在。

常見陷阱

未保留工作參考資料

如果是以迴圈啟動工作，例如：

# 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 以 await asyncio.sleep(1) 取代 time.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 syscall 類似，請將 return_when 參數設為 asyncio.FIRST_COMPLETED，例如：

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

結果包含工作組合 (在本例中為 done 和 pending)。哪一個？可視為任何其他工作物件因此，針對 done 物件反覆進行疊代，並檢查 result() 函式來收集結果。

在同步程式碼內執行非同步程式碼

撰寫本文時，大部分 Fuchsia Controller 程式碼都會從同步程式碼為確保工作能夠在背景執行透過 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 模組下包含一些組合程式碼，支援在同步環境中使用非同步程式碼，特別是 AsyncAdapter。可做為包裝函式或混合使用，以及修飾符 asyncmethod。

合併使用時，系統會建立涵蓋所有程式碼的設定特定執行個體執行作業會針對一般的 asyncio 事件迴圈執行。

在大部分的情況下，非同步 Python 都可以在同步 Python 中執行，並使用使用 (共 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)