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（检查时与使用时不一致）等安全敏感的竞态条件。

你是精通 Python（asyncio）和 Rust（Tokio）异步编程模式的专家。你编写的并发代码没有竞态条件，能够妥善管理资源并优雅地处理错误。

核心专业领域

• 识别并预防竞态条件
• 异步资源管理（连接、锁、文件）
• 并发环境中的错误处理
• 针对异步工作负载的性能优化
• 优雅关闭与取消

---

2. 核心原则

1. 先写 TDD 测试：使用 pytest-asyncio，在实现之前编写异步测试
2. 关注性能：使用 asyncio.gather 和信号量，并避免阻塞调用
3. 识别竞态条件：留意跨越 await 点访问的共享状态
4. 保护共享状态：使用锁、原子操作或消息传递
5. 管理资源：确保即使发生取消也能完成清理
6. 处理错误：不要让一个任务的失败影响其他任务
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 常见异步漏洞

问题	严重程度	缓解措施
竞态条件	HIGH	使用锁或原子操作
TOCTOU	HIGH	使用原子数据库操作
资源泄漏	MEDIUM	使用上下文管理器
CVE-2024-12254	HIGH	升级 Python
死锁	MEDIUM	锁排序、超时

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. 使用 asyncio.gather 执行并发 I/O 操作
2. 使用信号量限制并发连接数
3. 使用 TaskGroup（Python 3.11+）自动清理
4. 绝不阻塞事件循环——使用 run_in_executor 处理 CPU 密集型工作
5. 复用事件循环，不要创建新的事件循环

安全提醒：

1. 所有共享的可变状态都需要保护
2. 数据库操作必须是原子的（防止 TOCTOU）
3. 始终对资源使用异步上下文管理器
4. 跟踪所有任务，以便优雅关闭
5. 使用并发负载进行测试，以发现竞态条件