非同期パターン
bca.analyze は CPU バウンドです。処理の中身は tree-sitter によるパースとメトリクスの各パスで、どちらも PyO3 の Python::detach を介して Rust 側で GIL を解放します。したがって正規の非同期パターンは asyncio.to_thread です：
async def analyze_async(path: Path) -> FuncSpaceDict | None:
"""Run ``bca.analyze(path)`` on the default thread executor."""
return await asyncio.to_thread(bca.analyze, path)
async def analyze_all(
paths: Iterable[Path],
) -> list[FuncSpaceDict | BaseException | None]:
"""Fan ``analyze_async`` out across ``paths`` with ``asyncio.gather``.
``return_exceptions=True`` matters here: ``bca.analyze`` runs
inside ``asyncio.to_thread`` and Python threads cannot be
cancelled. If one call raises and gather re-raises with
``return_exceptions=False``, the surviving threads keep running
in the default executor, producing results that are silently
discarded. With ``return_exceptions=True`` every thread's
result (success OR exception) lands in the returned list so
the caller can dispatch per-file.
"""
return await asyncio.gather(
*(analyze_async(p) for p in paths),
return_exceptions=True,
)
ネイティブ async ではなく to_thread を使う理由
bca.analyze は同期的な Rust コードに支えられた同期的な Python 関数で、内部に await 境界はありません。asyncio.to_thread でラップすると：
- 呼び出しをデフォルトのスレッドプールにスケジュールします。
- パースとメトリクスのパスが走る間、他のコルーチンを進行させます。
- 完了時に結果を呼び出し元のコルーチンへ返します。
Rust 側が重い処理の間 GIL を解放するため、複数の to_thread(bca.analyze, ...) 呼び出しは本当に並列に実行されます — これは協調的な I/O 多重化ではなく、スレッドプールのサイズを上限とする実際のマルチコア活用です。
カスタムエグゼキューター
ワーカー数をより厳密に制限するには、専用に用意したエグゼキューターを to_thread に渡します：
import asyncio
from concurrent.futures import ThreadPoolExecutor
import big_code_analysis as bca
async def analyze_many(paths: list[str]) -> list[object]:
loop = asyncio.get_running_loop()
with ThreadPoolExecutor(max_workers=8) as pool:
return await asyncio.gather(
*(loop.run_in_executor(pool, bca.analyze, p) for p in paths)
)
純粋に CPU バウンドな処理では、8 コアのマシンで 8 ワーカーが無理のない上限です。これ以上増やすとマシンを過剰に割り当てることになり、スループットをコンテキストスイッチのオーバーヘッドと引き換えにしてしまいます。
結果のストリーミング
asyncio.as_completed を使うと、最初の解析が完了した時点から結果の消費を始められます。ファイルごとの処理コストが大きくばらつく場合（5 KB のファイルと 500 KB の生成バンドルなど）に有用です：
import asyncio
import big_code_analysis as bca
async def first_failure(paths: list[str]) -> str | None:
"""Return the path of the first file with cyclomatic > 50."""
tasks = [asyncio.create_task(asyncio.to_thread(bca.analyze, p)) for p in paths]
try:
for coro in asyncio.as_completed(tasks):
result = await coro
if result is None:
continue
if result["metrics"]["cyclomatic"]["sum"] > 50:
return result["name"]
finally:
for t in tasks:
t.cancel()
return None
finally ブロックでのキャンセルが重要です。as_completed は呼び出し元が早期リターンしても保留中のタスクを自動でキャンセルしないため、リークしたタスクが非同期関数の復帰後もスレッドプール上で動き続けることがあります。
アンチパターン：コルーチン内での bca.analyze の直接呼び出し
# これはやってはいけません。
async def bad(path: str) -> dict | None:
return bca.analyze(path) # 呼び出しのたびにイベントループをブロックします
async def は本体を非同期にするわけではありません。to_thread や明示的なエグゼキューターなしでは、bca.analyze を呼び出すすべてのコルーチンがパースの全時間にわたってイベントループを停止させます — I/O、タイマー、キューを待つ他のタスクはすべて、パースが返るまで凍結します。to_thread ラッパーは 1 行で書けて、応答性のあるサーバーとシングルスレッドのサーバーの違いを生みます。
analyze_batch が適している場面
静的で有限のパスのリストを処理していて結果のストリーミングが不要なら、bca.analyze_batch の方が gather(*to_thread(...)) よりシンプルです。呼び出し元スレッド上で逐次実行されますが、ファイルごとのエラーで例外を送出することはありません。イベントループの応答性を保つには、analyze_batch の呼び出し全体を asyncio.to_thread でラップしてください：
import asyncio
import big_code_analysis as bca
async def batch(paths: list[str]) -> list[object]:
return await asyncio.to_thread(bca.analyze_batch, paths)
これは gather のファイル単位の並列性を、analyze_batch のよりシンプルなエラーモデルと引き換えにします。並列性と型付き OSError の振り分けの両方が欲しい場合は gather を、非同期呼び出しを 1 つにまとめて例外を送出しない契約が欲しい場合は to_thread(analyze_batch, paths) を選んでください。