タイムアウトを短くしても、裏で動く同期ツールの寿命は縮まらない。
自分はLLMエージェントの同期ツールを asyncio.to_thread に渡す形を確認していて、タスクの cancelled() が True なのに同期関数が残るところで引っかかった。同じ更新処理をすぐ再試行するのは危ない。先の呼び出しがまだ書き込みに進む可能性がある。
今回は外部APIを呼ばず、CPython 3.14.8で再現した。Python側で何を打ち切っているかの確認。
Pythonのwait_forで50msだけ待つ
次を timeout_thread.py に保存し、python3 timeout_thread.py で実行する。同期関数の開始通知を待ってから、50msのタイムアウトを設定した。
import asyncio
import threading
async def main():
loop = asyncio.get_running_loop()
started, finished = asyncio.Event(), asyncio.Event()
release = threading.Event()
def sync_tool():
loop.call_soon_threadsafe(started.set)
release.wait()
loop.call_soon_threadsafe(finished.set)
task = asyncio.create_task(asyncio.to_thread(sync_tool))
await started.wait()
try:
await asyncio.wait_for(task, timeout=0.05)
except TimeoutError:
print('task.cancelled():', task.cancelled())
print('同期処理は完了した？', finished.is_set())
finally:
release.set()
await finished.wait()
print('待機を打ち切った後に完了:', finished.is_set())
asyncio.run(main())
手元の出力。
task.cancelled(): True
同期処理は完了した？ False
待機を打ち切った後に完了: True
release.wait() で同期関数を止めておき、タイムアウト後の release.set() で先へ進めている。最後の通知が届いたので、キャンセル後も同期関数が続いたと分かる。sleep の秒数を調整するより、順序を追いやすかった。
ここでの50msは待機に設定した値。実測した応答時間ではない。
キャンセルの行き先をCPythonで追う
手元のCPythonソースを読んだ。asyncio/threads.py の to_thread は run_in_executor を待ち、既定のスレッドプールへ同期関数を渡す。
wait_for のキャンセルは元のFutureにも伝わる。ただ、concurrent/futures/_base.py の Future.cancel() は、状態が RUNNING なら False を返す実装だ。
待つ側のキャンセルと、実行する側の停止は別々に扱う。未実行の仕事ならキャンセルできる場合もあるので、再現では開始通知を挟んだ。
同時実行数の枠も早く空く
asyncio.Semaphore(1) の内側で to_thread を待つ構成も確認した。1件目の待機をタイムアウトさせて2件目を投入すると、同期関数が2件同時に動いた。
キャンセルで async with を抜け、セマフォの枠を返す。1件目のスレッドは残るので、制限値1と実際の稼働数がずれる。待機を早く切るほど、次の仕事が流れ込むんだよね。
対策は、止めたいものに合わせて置く。
| 対象 | 実装で確認するところ |
|---|---|
| 呼び出し元の待機 |
wait_for の期限 |
| 同期SDKの通信 | SDKやHTTPクライアントの通信タイムアウトと再試行設定 |
| 稼働中の同期処理数 | 同期関数の入口で枠を取り、関数が抜けるまで保持する |
枠待ちの仕事は残り得るので、取得直後に締切を確認し、期限切れなら更新を始めずに戻る。通信タイムアウトで戻っても、送信済みの更新が相手側で完了する可能性は残る。
ツール呼び出しでは、待機を打ち切った時刻と実処理の終了時刻を別々に記録したい。更新を再試行するなら、相手側で重複を防ぐ仕組みも要る。同期ツールが増えるほど、この時刻の差が稼働数の見積もりに効いてくると見ている。まずは cancelled() の横に、実処理の完了通知を一つ置くところから。