async-programming

非同期プログラミングのスキル

ファイル構成

• SKILL.md：中核となる原則、パターン、基本的なセキュリティ（このファイル）
• references/security-examples.md：競合状態とリソース安全性の例
• references/advanced-patterns.md：高度な非同期パターンと最適化

検証ゲート

ゲート 0.1：ドメイン専門知識の検証

• ステータス：合格
• 専門分野：asyncio、Tokio、競合状態、リソース管理、並行実行の安全性

ゲート 0.2：脆弱性の調査

• ステータス：合格（MEDIUM-RISK に対して 3 件以上の問題）
• 調査日：2025-11-20
• 問題：CVE-2024-12254（asyncio のメモリ）、Redis の競合状態（CVE-2023-28858/9）

ゲート 0.11：ファイル構成の決定

• 決定：分割構成（MEDIUM-RISK、メイン約 400 行＋参照資料）

---

1. 概要

リスクレベル：MEDIUM

根拠：非同期プログラミングでは、競合状態、リソースリーク、タイミングに基づく脆弱性が発生する可能性があります。外部からの攻撃に直接さらされるわけではありませんが、不適切な非同期コードは、データ破損、デッドロック、二重支払いなどのセキュリティ上重要な競合状態や、TOCTOU（time-of-check-time-of-use）を引き起こす可能性があります。

あなたは、Python（asyncio）および Rust（Tokio）の非同期プログラミングパターンに精通しています。競合状態がなく、リソースを適切に管理し、エラーを適切に処理する並行コードを作成します。

中核となる専門分野

• 競合状態の特定と防止
• 非同期リソース管理（接続、ロック、ファイル）
• 並行処理コンテキストでのエラー処理
• 非同期ワークロードのパフォーマンス最適化
• グレースフルシャットダウンとキャンセル

---

2. 中核となる原則

1. まず TDD：pytest-asyncio を使い、実装前に非同期テストを作成します
2. パフォーマンスを考慮：asyncio.gather とセマフォを使い、ブロッキング呼び出しを避けます
3. 競合状態を特定：await ポイントをまたいでアクセスされる共有状態を認識します
4. 共有状態を保護：ロック、アトミック操作、またはメッセージパッシングを使います
5. リソースを管理：キャンセル時にもクリーンアップが行われるようにします
6. エラーを処理：1 つのタスクの失敗によって、ほかのタスクが破損しないようにします
7. デッドロックを回避：ロックの順序を統一し、ロックにタイムアウトを設定します

判断フレームワーク

状況	アプローチ
変更可能な共有状態	asyncio.Lock または RwLock を使用
データベーストランザクション	アトミック操作、SELECT FOR UPDATE を使用
リソースのクリーンアップ	非同期コンテキストマネージャーを使用
タスクの調整	asyncio.Event、Queue、または Semaphore を使用
バックグラウンドタスク	タスクを追跡し、キャンセルを処理

---

3. 実装ワークフロー（TDD）

ステップ 1：まず失敗するテストを書く

import pytest
import asyncio

@pytest.mark.asyncio
async def test_concurrent_counter_safety():
    """Test counter maintains consistency under concurrent access."""
    counter = SafeCounter()  # Not implemented yet - will fail

    async def increment_many():
        for _ in range(100):
            await counter.increment()

    # Run 10 concurrent incrementers
    await asyncio.gather(*[increment_many() for _ in range(10)])

    # Must be exactly 1000 (no lost updates)
    assert await counter.get() == 1000

@pytest.mark.asyncio
async def test_resource_cleanup_on_cancellation():
    """Test resources are cleaned up even when task is cancelled."""
    cleanup_called = False

    async def task_with_resource():
        nonlocal cleanup_called
        async with managed_resource() as resource:  # Not implemented yet
            await asyncio.sleep(10)  # Long operation
        cleanup_called = True

    task = asyncio.create_task(task_with_resource())
    await asyncio.sleep(0.1)
    task.cancel()

    with pytest.raises(asyncio.CancelledError):
        await task

    assert cleanup_called  # Cleanup must happen

ステップ 2：テストに合格する最小限の実装を行う

import asyncio
from contextlib import asynccontextmanager

class SafeCounter:
    def __init__(self):
        self._value = 0
        self._lock = asyncio.Lock()

    async def increment(self) -> int:
        async with self._lock:
            self._value += 1
            return self._value

    async def get(self) -> int:
        async with self._lock:
            return self._value

@asynccontextmanager
async def managed_resource():
    resource = await acquire_resource()
    try:
        yield resource
    finally:
        await release_resource(resource)  # Always runs

ステップ 3：パターンに従ってリファクタリングする

パフォーマンスパターンを適用し、タイムアウトを追加して、エラー処理を改善します。

ステップ 4：完全な検証を実行する

# Run async tests
pytest tests/ -v --asyncio-mode=auto

# Check for blocking calls
python -m asyncio debug

# Run with concurrency stress test
pytest tests/ -v -n auto --asyncio-mode=auto

---

4. パフォーマンスパターン

パターン 1：並行処理に asyncio.gather を使用する

# BAD - Sequential execution
async def fetch_all_sequential(urls: list[str]) -> list[str]:
    results = []
    for url in urls:
        result = await fetch(url)  # Waits for each
        results.append(result)
    return results  # Total time: sum of all fetches

# GOOD - Concurrent execution
async def fetch_all_concurrent(urls: list[str]) -> list[str]:
    return await asyncio.gather(*[fetch(url) for url in urls])
    # Total time: max of all fetches

パターン 2：レート制限にセマフォを使用する

# BAD - Unbounded concurrency (may overwhelm server)
async def fetch_many(urls: list[str]):
    return await asyncio.gather(*[fetch(url) for url in urls])

# GOOD - Bounded concurrency with semaphore
async def fetch_many_limited(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_limit(url: str):
        async with semaphore:
            return await fetch(url)

    return await asyncio.gather(*[fetch_with_limit(url) for url in urls])

パターン 3：タスクグループ（Python 3.11 以降）

# BAD - Manual task tracking
async def process_items_manual(items):
    tasks = []
    for item in items:
        task = asyncio.create_task(process(item))
        tasks.append(task)
    return await asyncio.gather(*tasks)

# GOOD - Task groups with automatic cleanup
async def process_items_taskgroup(items):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(process(item)) for item in items]
    return [task.result() for task in tasks]
    # Automatic cancellation on any failure

パターン 4：イベントループを効率的に使用する

# BAD - Creating new event loop each time
def run_async_bad():
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(main())
    finally:
        loop.close()

# GOOD - Reuse running loop or use asyncio.run
def run_async_good():
    return asyncio.run(main())  # Handles loop lifecycle

# GOOD - For library code, get existing loop
async def library_function():
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    # Use the existing loop

パターン 5：ブロッキング呼び出しを避ける

# BAD - Blocks event loop
async def process_file_bad(path: str):
    with open(path) as f:  # Blocking I/O
        data = f.read()
    result = hashlib.sha256(data).hexdigest()  # CPU-bound blocks loop
    return result

# GOOD - Non-blocking with aiofiles and executor
import aiofiles

async def process_file_good(path: str):
    async with aiofiles.open(path, 'rb') as f:
        data = await f.read()

    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(
        None, lambda: hashlib.sha256(data).hexdigest()
    )
    return result

---

5. 技術基盤

推奨バージョン

コンポーネント	バージョン	備考
Python	3.11+	asyncio の改善、TaskGroup
Rust	1.75+	安定した非同期処理
Tokio	1.35+	非同期ランタイム
aioredis	redis-py を使用	メンテナンス性がより高い

主要ライブラリ

# Python async ecosystem
asyncio           # Core async
aiohttp           # HTTP client
asyncpg           # PostgreSQL
aiofiles          # File I/O
pytest-asyncio    # Testing

---

6. 実装パターン

パターン 1：ロックによる共有状態の保護

import asyncio

class SafeCounter:
    """Thread-safe counter for async contexts."""
    def __init__(self):
        self._value = 0
        self._lock = asyncio.Lock()

    async def increment(self) -> int:
        async with self._lock:
            self._value += 1
            return self._value

    async def get(self) -> int:
        async with self._lock:
            return self._value

パターン 2：アトミックなデータベース操作

from sqlalchemy.ext.asyncio import AsyncSession

async def transfer_safe(db: AsyncSession, from_id: int, to_id: int, amount: int):
    """Atomic transfer using row locks."""
    async with db.begin():
        stmt = (
            select(Account)
            .where(Account.id.in_([from_id, to_id]))
            .with_for_update()  # Lock rows
        )
        accounts = {a.id: a for a in (await db.execute(stmt)).scalars()}

        if accounts[from_id].balance < amount:
            raise ValueError("Insufficient funds")

        accounts[from_id].balance -= amount
        accounts[to_id].balance += amount

パターン 3：コンテキストマネージャーによるリソース管理

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_connection():
    """Ensure connection cleanup even on cancellation."""
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

パターン 4：グレースフルシャットダウン

import asyncio, signal

class GracefulApp:
    def __init__(self):
        self.shutdown_event = asyncio.Event()
        self.tasks: set[asyncio.Task] = set()

    async def run(self):
        loop = asyncio.get_event_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(sig, self.shutdown_event.set)

        self.tasks.add(asyncio.create_task(self.worker()))
        await self.shutdown_event.wait()

        for task in self.tasks:
            task.cancel()
        await asyncio.gather(*self.tasks, return_exceptions=True)

---

7. セキュリティ基準

7.1 よくある非同期処理の脆弱性

問題	深刻度	対策
競合状態	高	ロックまたはアトミック操作を使用する
TOCTOU	高	アトミックなデータベース操作を使用する
リソースリーク	中	コンテキストマネージャーを使用する
CVE-2024-12254	高	Python をアップグレードする
デッドロック	中	ロックの順序付けとタイムアウトを設定する

7.2 競合状態の検出

# RACE CONDITION - read/await/write pattern
class UserSession:
    async def update(self, key, value):
        current = self.data.get(key, 0)  # Read
        await validate(value)             # Await = context switch
        self.data[key] = current + value  # Write stale value

# FIXED - validate outside lock, atomic update inside
class SafeUserSession:
    async def update(self, key, value):
        await validate(value)
        async with self._lock:
            self.data[key] = self.data.get(key, 0) + value

---

8. よくある間違いとアンチパターン

アンチパターン 1：保護されていない共有状態

# NEVER - race condition on cache
async def get_or_fetch(self, key):
    if key not in self.data:
        self.data[key] = await fetch(key)
    return self.data[key]

# ALWAYS - lock protection
async def get_or_fetch(self, key):
    async with self._lock:
        if key not in self.data:
            self.data[key] = await fetch(key)
        return self.data[key]

アンチパターン 2：タスクを起動して結果を待たない

# NEVER - task may be garbage collected
asyncio.create_task(background_work())

# ALWAYS - track tasks
task = asyncio.create_task(background_work())
self.tasks.add(task)
task.add_done_callback(self.tasks.discard)

アンチパターン 3：イベントループをブロックする

# NEVER - blocks all async tasks
time.sleep(5)

# ALWAYS - use async
await asyncio.sleep(5)
result = await loop.run_in_executor(None, cpu_bound_func)

---

9. 実装前チェックリスト

フェーズ 1：コードを書く前

• 競合状態が発生するケースについて、失敗するテストを書く
• キャンセル時のリソース解放をテストする
• 共有される変更可能な状態をすべて特定する
• デッドロックを避けるため、ロックの階層を計画する
• 適切な並行処理の上限を決定する

フェーズ 2：実装中

• 共有状態をすべてロックで保護する
• リソースには非同期コンテキストマネージャーを使用する
• 並行処理には asyncio.gather を使用する
• レート制限にはセマフォを適用する
• CPU 負荷の高い処理にはエグゼキューターを使用する
• 作成したタスクをすべて追跡する

フェーズ 3：コミット前

• 非同期テストがすべて成功する：pytest --asyncio-mode=auto
• イベントループ上でブロッキング呼び出しを行わない
• すべての外部処理にタイムアウトを設定する
• キャンセルに対応した正常なシャットダウンを実装する
• 競合状態のテストでスレッドセーフであることを検証する
• ロックの順序が一貫している（デッドロックの可能性がない）

---

10. まとめ

目指すべき非同期コードは、次の特性を備えています。

• テスト駆動：pytest-asyncio を使い、まず非同期テストを書く
• 競合なし：共有状態を保護し、アトミック操作を使用する
• リソース安全：コンテキストマネージャーを使い、適切にリソースを解放する
• 高パフォーマンス：asyncio.gather とセマフォを使い、ブロッキングを避ける
• 堅牢：エラーに対処し、キャンセルに対応する

パフォーマンスに関する重要なルール：

1. 並行 I/O 操作には asyncio.gather を使用する
2. 並行接続数の制限にはセマフォを適用する
3. 自動的なクリーンアップには TaskGroup（Python 3.11 以降）を使用する
4. イベントループをブロックしない。CPU 処理には run_in_executor を使用する
5. イベントループは再利用し、新しく作成しない

セキュリティに関する注意：

1. 共有される変更可能な状態はすべて保護する
2. データベース操作はアトミックにする（TOCTOU の防止）
3. リソースには必ず非同期コンテキストマネージャーを使用する
4. 正常なシャットダウンに備えて、すべてのタスクを追跡する
5. 並行負荷をかけてテストし、競合状態を見つける