本頁會討論非同步 Python 的常見工作流程,以及使用非同步 Python 時可能出現的錯誤。
繼續操作之前,請先參閱這份 asyncio
的官方說明文件,以及 Fuchsia Controller 的教學課程。
背景
下列項目大致上涉及非同步程式碼,不一定適用於 Python:
- 非同步程式碼不一定是多執行緒。如果是與 Fuuchsia 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())
上述程式碼不會列印任何內容。不過,如果將 time.sleep(1)
替換成 await asyncio.sleep(1)
,該欄位就會輸出 HELLO
,這樣會載入非同步迴圈。
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 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"))
雖然這個執行個體可以在非非同步環境中執行,但仍會執行非同步程式碼。但這樣可以讓整體讀取及寫入更容易。