asyncio.to_thread()で実行した同期関数は、呼び出し元のタスクをキャンセルしても動き続けることがあります。
Task.cancel()が取り消すのは非同期タスクの待機であり、実行中のワーカースレッドではありません。
同期関数も止めたいなら、停止要求を別経路で伝える必要があります。
threading.Eventを使い、キャンセル後にスレッドの終了まで確認する形を試します。
cancel()しても同期関数のログが続く
たとえば、同期ライブラリをasyncioから呼び出す場面を考えます。
外部APIへのアクセスやファイル処理など、同期関数をそのまま呼ぶとイベントループをブロックする処理があります。
asyncio.to_thread()は、そうした関数を別スレッドで実行するための仕組みです。
ただし、呼び出し元をキャンセルしたときの動作には注意が必要です。
次のコードを実行します。
動作確認の対象はPython 3.14系です。外部ライブラリは使いません。
import asyncio
import time
def blocking_work():
for i in range(5):
print(f"worker: {i}", flush=True)
time.sleep(0.5)
print("worker: finished", flush=True)
async def main():
task = asyncio.create_task(
asyncio.to_thread(blocking_work)
)
await asyncio.sleep(0.7)
print("main: cancel", flush=True)
task.cancel()
try:
await task
except asyncio.CancelledError:
print("main: cancelled", flush=True)
await asyncio.sleep(2.0)
asyncio.run(main())
実行すると、たとえば次のようになります。
worker: 0
worker: 1
main: cancel
main: cancelled
worker: 2
worker: 3
worker: 4
worker: finished
ログの順序はスケジューリングによって多少変わります。
注目したいのは、main: cancelledが出力された後もworkerのログが続くことです。
task.cancel()は成功しています。await taskもCancelledErrorを受け取っています。
それなのに同期関数は最後まで実行されます。
これはasyncio.to_thread()の不具合ではありません。
キャンセルされた非同期タスクと、実行中の同期関数は別の実行単位だからです。
キャンセルされるのは待機側であってスレッドではない
asyncio.to_thread()は、同期関数をイベントループのデフォルトエグゼキューター経由で実行します。
内部的にはloop.run_in_executor()を利用する構成です。
処理を分けると、次のようになります。
- 非同期タスクが
to_thread()を実行する - 同期関数がエグゼキューターへ投入される
- ワーカースレッドが同期関数を実行する
- 非同期タスクは実行結果を待つ
-
cancel()によって待機側へキャンセルが伝わる
問題は5番目です。
Task.cancel()は、タスクにキャンセルを要求します。タスクが待機中であれば、その待機先にもキャンセルが伝播します。
しかし、ワーカースレッドで実行中のPython関数を強制的に終了させる操作ではありません。
エグゼキューター側のFuture.cancel()にも同様の制約があります。まだ実行されていない処理ならキャンセルできる場合がありますが、すでに実行中の処理はキャンセルできません。
つまり、Futureのキャンセルと、実行中の関数の停止は同義ではありません。
ここを混同すると、タイムアウト設計でも問題が起きます。
たとえば次のコードです。
try:
async with asyncio.timeout(1.0):
await asyncio.to_thread(blocking_work)
except TimeoutError:
print("timeout")
asyncio.timeout()は期限に達すると現在のタスクをキャンセルし、コンテキストマネージャーの外側でTimeoutErrorとして扱えるようにします。
しかし、実行中のblocking_work()が停止するわけではありません。
呼び出し元ではタイムアウトを処理できても、裏側では同期関数が動き続ける可能性があります。
その同期関数がファイルを書き換えていたり、外部サービスへ更新要求を送っていたりすれば、タイムアウト後にも副作用が発生し得ます。
また、キャンセルされたタスクの結果を待たなくなっても、スレッドが使用しているリソースが直ちに解放されるわけではありません。
実行中の処理が増え続ければ、エグゼキューターのワーカーが占有され、後続処理の実行にも影響します。
asyncio側のタイムアウトだけで、同期処理の実行時間まで制限できたと考えるのは危険です。
threading.Eventで停止要求を伝える
実行中の同期関数を止めたい場合は、同期関数自身が停止要求を確認する仕組みにします。
標準ライブラリのthreading.Eventが使えます。
Eventはスレッド間で共有できるフラグです。
set()で停止を要求し、同期関数側はis_set()やwait()で状態を確認します。
先ほどのコードを修正します。
import asyncio
import threading
def blocking_work(stop: threading.Event):
for i in range(5):
if stop.is_set():
print("worker: stopped", flush=True)
return
print(f"worker: {i}", flush=True)
if stop.wait(0.5):
print("worker: stopped", flush=True)
return
print("worker: finished", flush=True)
async def main():
stop = threading.Event()
finished = threading.Event()
def run_worker():
try:
blocking_work(stop)
finally:
finished.set()
task = asyncio.create_task(
asyncio.to_thread(run_worker)
)
await asyncio.sleep(0.7)
print("main: cancel", flush=True)
task.cancel()
stop.set()
try:
await task
except asyncio.CancelledError:
print("main: cancelled", flush=True)
await asyncio.to_thread(finished.wait)
print("main: worker joined", flush=True)
asyncio.run(main())
実行例は次のとおりです。
worker: 0
worker: 1
main: cancel
main: cancelled
worker: stopped
main: worker joined
ここでは停止要求と終了確認を分けています。
stopは同期関数に停止を要求するためのフラグです。
finishedは同期関数を呼び出すラッパーのfinallyでセットします。
これにより、正常終了した場合も例外で終了した場合も、ラッパーの処理が終わることを通知できます。
finished.wait()はブロッキング操作なので、イベントループ上で直接呼ばず、再びasyncio.to_thread()へ渡しています。
ここで重要なのは、キャンセル済みのtaskを再度awaitしても、ワーカースレッドの終了待ちにはならないことです。
そのタスクはすでにキャンセル状態です。
終了確認には、別の同期手段が必要になります。
なお、表示しているworker joinedは説明用のログです。実際にthreading.Thread.join()を呼んでいるわけではありません。
このコードが確認しているのは、ワーカー関数を呼び出したラッパーがfinallyに到達したことです。
エグゼキューターが管理するスレッドそのものの終了とは区別してください。
協調的キャンセルが効かない境界
threading.Eventを渡せば、あらゆる同期関数を停止できるわけではありません。
協調的キャンセルは、同期関数が停止要求を確認できることを前提にしています。
たとえば、次のような処理です。
def blocking_work(stop):
while not stop.is_set():
call_blocking_library()
call_blocking_library()が長時間戻ってこない場合、停止要求はすぐには反映されません。
フラグを確認するのは、同期関数がライブラリ呼び出しから戻った後だからです。
この場合、停止フラグだけでは不十分です。
ライブラリが独自のタイムアウトやキャンセル機構を提供しているなら、それを利用する必要があります。
ネットワークI/Oであれば接続タイムアウトや読み取りタイムアウト、データベース操作であればドライバー側のタイムアウトなどが候補になります。
ただし、タイムアウトが例外を返すことと、外部で実行された操作が取り消されることも別です。
更新処理を伴う場合は、結果が不明な状態をどう扱うかまで考える必要があります。
もう一つ注意したいのは、停止要求を送った後の待機です。
修正版の次の部分は、ワーカーが終了するまで待ちます。
await asyncio.to_thread(finished.wait)
ワーカーが停止要求を無視した場合、この待機も終わりません。
終了確認まで無制限に待てないシステムでは、待機側にも期限を設ける設計が必要です。
ただし、その期限を超えてもワーカーが停止したことにはなりません。
停止できなかった処理を記録し、後続の更新処理と競合させないなど、別の対処が必要になります。
また、今回のサンプルではtask.cancel()とstop.set()を同じ場所で呼んでいます。
実際のアプリケーションでは、外側のタスクがキャンセルされることもあります。その場合は、呼び出し側のfinallyなどで停止要求を送る構成を検討します。
停止要求を送る処理自体が、通常の完了経路だけに置かれていないか確認したいところです。
なお、今回のfinishedは終了通知専用です。ワーカーで発生した例外を呼び出し元へ伝える仕組みではありません。
例外の取得まで必要なら、実行結果と終了通知を別々に管理する設計が必要です。
キャンセル要求、処理の終了、結果の確定は、それぞれ別の状態として扱うべきです。
asyncio.to_thread()は、同期関数をイベントループの外で実行するための手段です。実行中の同期関数を強制停止するためのAPIではありません。停止フラグを確認できないライブラリや、外部で副作用を発生させる処理では、この方法だけでは完結しません。次に確認したいのは、利用する同期ライブラリがどこで制御を返し、どのタイミングなら安全に停止できるかです。