3つの非同期検索を並行実行し、最初の有効な結果が見つかったら残りを止めたい場面があります。
asyncio.TaskGroupは通常、子タスクがすべて終了するまで待つため、そのままでは不要な検索も継続します。
Python 3.15で追加されたTaskGroup.cancel()なら、例外を正常終了の合図として使わずにグループを終了できます。
ただし、キャンセルを子タスクが決めるか、監督側が決めるかで、結果の採用責任と失敗処理の境界が変わります。
TaskGroupが最後まで待ってしまう問題
たとえば、複数の検索先に同じ問い合わせを投げ、最初に条件を満たした結果だけを採用する処理を考えます。
検索先Aは有効な結果を早く返します。検索先Bは結果がありません。検索先Cは応答に時間がかかります。
普通にTaskGroupで起動すると、Aが成功してもCの終了を待ちます。
import asyncio
async def search(name, delay, value):
await asyncio.sleep(delay)
print(f"{name}: {value}")
return value
async def main():
async with asyncio.TaskGroup() as tg:
tasks = [
tg.create_task(search("A", 0.1, "found")),
tg.create_task(search("B", 0.2, None)),
tg.create_task(search("C", 1.0, "other")),
]
print([task.result() for task in tasks])
asyncio.run(main())
これはTaskGroupの不具合ではありません。
async withを抜けるとき、所属する子タスクの終了を保証するのが基本動作です。子タスクが失敗した場合は残りをキャンセルし、終了を待ってから例外を報告します。
しかし「最初の有効な結果を採用した時点で検索を終える」という要件では、全件完了が成功条件ではありません。
従来、このような正常な早期終了を実現するには、終了専用の例外を投げる方法が公式ドキュメントでも示されていました。
class SearchFinished(Exception):
pass
async def stop_group():
raise SearchFinished()
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(search("A", 0.1, "found"))
tg.create_task(search("C", 1.0, "other"))
tg.create_task(stop_group())
except* SearchFinished:
pass
この短い例では、検索結果を確認せず即座に終了します。終了専用例外の仕組みだけを示しています。
実用上は有効な結果を検出してから例外を投げる必要があります。
問題は、業務上の正常終了を例外として表現する点です。
TaskGroupは通常の例外を子タスクの失敗として扱い、ExceptionGroupにまとめます。そのため、終了用の例外だけをexcept*で取り除く処理が必要になります。
さらに、本当の検索エラーまで握りつぶさないようにしなければなりません。
TaskGroup.cancel()で正常な早期終了を表す
Python 3.15ではTaskGroup.cancel()が追加されました。
公式ドキュメントでは、グループの目的を達成したときなどに使う、例外を伴わない早期終了として定義されています。
動作上の要点は次の3つです。
- まだ終了していない所属タスクへキャンセルを要求する。
- グループ本体を実行している親タスクにもキャンセルを要求する。
- グループ自身による正常な終了では、
CancelledErrorを外側へ伝播させずに抜ける。
特に2番目を見落としやすいです。
tg.cancel()は単に子タスクを順番にcancel()するための短縮記法ではありません。async withの本体が待機中なら、その待機も中断の対象になります。
次のコードはPython 3.15用です。追加パッケージは不要です。
import asyncio
async def search(name, delay, value, tg, results):
task = asyncio.current_task()
try:
await asyncio.sleep(delay)
print(f"{name}: result={value}")
if value is not None:
results.append((name, value))
tg.cancel()
except asyncio.CancelledError:
print(f"{name}: cancelled={task.cancelling()}")
raise
finally:
print(f"{name}: finally={task.cancelling()}")
async def main():
results = []
async with asyncio.TaskGroup() as tg:
for name, delay, value in [
("A", 0.1, "found"),
("B", 0.3, None),
("C", 1.0, "other"),
]:
tg.create_task(search(name, delay, value, tg, results))
print("selected:", results)
asyncio.run(main())
通常のイベントループ設定では、次のような出力になります。
A: result=found
A: finally=0
B: cancelled=1
B: finally=1
C: cancelled=1
C: finally=1
selected: [('A', 'found')]
BとCの終了順は保証されません。
Aは結果を保存してからtg.cancel()を呼びます。BとCにはCancelledErrorが送られ、finallyが実行されます。
ここでCancelledErrorを捕まえて再送出しているのは意図的です。
キャンセルは即時にタスクを強制終了する操作ではありません。通常は次の中断可能な地点でCancelledErrorが発生します。
finallyはリソース解放のために使い、キャンセルそのものは原則として再送出します。
Task.cancelling()は、そのタスクに対するキャンセル要求数を返します。キャンセル済みかどうかを表す単純な真偽値ではありません。
また、Aのfinally=0は、キャンセル要求を受けずに正常終了したことを示しています。
このコードでは結果を保存してからキャンセルするため、async withの外側で結果を参照できます。
ただし、複数の検索が同じタイミングで成功する場合、結果が必ず1件になるわけではありません。cancel()は排他的な結果確定操作ではないからです。
キャンセルを決める場所が責務を変える
子タスクからtg.cancel()する設計は、検索先が独立しており、どれか1つの成功だけで処理を終了できる場合には簡潔です。
一方、複数の結果を比較したい場合は困ります。
たとえば、検索結果に有効期限や信頼度があり、最初に返ってきた結果でも採用できない可能性があるとします。
この場合、検索処理そのものにグループ全体の終了権限を持たせると、判定条件が各子タスクへ分散します。
私は、結果を作る処理と、結果を採用する処理を分けるほうが、後から条件を変更しやすいと考えています。
監督側で採用を決める実装は次のようになります。
import asyncio
async def search(name, delay, value, queue):
try:
await asyncio.sleep(delay)
await queue.put((name, value))
finally:
task = asyncio.current_task()
print(f"{name}: finally={task.cancelling()}")
async def main():
queue = asyncio.Queue()
selected = None
async with asyncio.TaskGroup() as tg:
for name, delay, value in [
("A", 0.1, "found"),
("B", 0.3, None),
("C", 1.0, "other"),
]:
tg.create_task(search(name, delay, value, queue))
for _ in range(3):
name, value = await queue.get()
if value is not None:
selected = (name, value)
tg.cancel()
break
print("selected:", selected)
asyncio.run(main())
この実装では、子タスクは結果をQueueへ送るだけです。
監督側が結果を受け取り、採用条件を確認してからグループを終了します。
有効な結果がなければ3件すべてを受信し、通常どおりグループを抜けます。
子タスクにTaskGroupの参照を渡す必要もありません。
ただし、このコードでは検索が必ず結果を送るか、例外で終了することを前提としています。実際の外部API呼び出しでは、タイムアウトや無期限の待機を別途設計する必要があります。
また、監督側がqueue.get()で待機している最中に別のタスクがtg.cancel()を呼んだ場合、その待機自体もキャンセルされます。
監督側が呼ぶ設計なら、その呼び出しの直後にbreakしているため、正常終了の境界がコード上で明確になります。
CancelledErrorとExceptionGroupを混同しない
TaskGroup.cancel()が正常な早期終了を表せるようになっても、子タスクの失敗まで正常終了になるわけではありません。
確認すべき経路は3つあります。
| 状況 | 子タスク | グループの外側 |
|---|---|---|
tg.cancel()による早期終了 |
未完了タスクへキャンセル要求 | 通常は正常に抜ける |
子タスクがValueErrorなどで失敗 |
残りをキャンセル |
ExceptionGroupなどを送出 |
| 外部から親タスクをキャンセル | 子タスクにもキャンセルが伝わる | 原則としてCancelledErrorが伝播 |
最初の2つは特に区別が必要です。
次のコードでは、子タスクが通常の例外で失敗します。
import asyncio
async def fail():
await asyncio.sleep(0.1)
raise ValueError("search failed")
async def wait():
try:
await asyncio.sleep(10)
finally:
print("wait: cleanup")
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(fail())
tg.create_task(wait())
except* ValueError as group:
print("errors:", len(group.exceptions))
asyncio.run(main())
通常の実行では、待機タスクのfinallyが実行された後に、errors: 1が表示されます。
失敗したタスクが発生すると、TaskGroupは残りをキャンセルして終了を待ち、非キャンセル例外をグループ化します。
tg.cancel()を使った正常終了とは異なります。
さらに注意したいのが、外部キャンセルとの競合です。
たとえば、上位のリクエスト処理がキャンセルされたのと同時に、検索結果が見つかってtg.cancel()が呼ばれる可能性があります。
Python 3.15のドキュメントでも、TaskGroup内部のキャンセルと外部からのキャンセルを混同しないように扱うこと、Task.cancelling()のカウントを保持することが説明されています。
したがって、async withを正常に抜けたという事実だけを、あらゆる外部キャンセルが存在しなかった証拠にはできません。
外部キャンセルを正常な検索成功へ変換するようなexcept asyncio.CancelledError: passも避けるべきです。
CancelledErrorはBaseExceptionの直接のサブクラスであり、通常のexcept Exceptionには捕まりません。これも意図された設計です。
また、キャンセル中のfinallyで別の例外を発生させると、その例外がグループの終了結果に影響する可能性があります。
クリーンアップ処理は必要ですが、そこでの失敗を黙って無視する設計にもすべきではありません。
結局、TaskGroup.cancel()をどこで呼ぶかは、検索の成功を誰が確定するかで決めるのが自然です。単一の成功条件で十分なら子タスクから呼べます。採用条件や優先順位が変わるなら監督側へ寄せます。ただし、キャンセルに非協力的な処理やブロッキングI/Oは、このAPIだけでは停止を保証できません。実際の検索処理へ適用する前に、同時成功、検索失敗、外部キャンセルが重なった場合の動作をテストする必要があります。