Python Asyncio Pitfalls

---
license: Apache-2.0
allowed-tools: Read,Write,Edit,Bash,Glob,Grep,WebSearch,WebFetch
name: python-asyncio-pitfalls
description: 'Use when debugging asyncio event-loop hangs, blocking calls inside coroutines, asyncio.gather vs TaskGroup choice, cancellation handling, structured concurrency, mixing sync and async, run_in_executor decisions, asyncio task lifetimes, "Task was destroyed but it is pending" warnings, or async context manager bugs. Triggers: event loop blocked by sync I/O, ConnectionResetError on cancellation, asyncio in libraries that also offer sync API, FastAPI/aiohttp performance regressions, queue.Queue used in async code. NOT for trio/anyio (different paradigm), threading without asyncio, or Python 2 sync patterns.'
metadata:
  category: AI & Machine Learning
  tags:
    - python
    - asyncio
    - concurrency
    - performance
    - taskgroup
    - structured-concurrency
  provenance:
    kind: first-party
    owners: [port-daddy]
  pairs-with:
    - skill: background-job-orchestrator
      reason: Celery/queue worker code is where sync-in-async mixing and missing backpressure bite hardest; that skill owns the queue architecture this one debugs inside.
    - skill: error-handling-patterns
      reason: Retry, circuit-breaker, and exception-hierarchy strategy sits above the asyncio cancellation/ExceptionGroup primitives covered here.
    - skill: websocket-streaming
      reason: Long-lived async streams are where cancellation propagation and unbounded-queue OOMs surface in practice.
  io-contract:
    kind: deliverable
    consumes:
      - kind: concurrency-bug-report
        format: markdown
        description: A symptom description -- loop hang, throughput collapse under load, "Task was destroyed but it is pending", partial fan-out results -- with the code path in question.
      - kind: asyncio-concurrency-plan
        format: json
        description: A structured concurrency design (Python version, fan-out primitive, blocking-call offload, timeout and cancellation posture) matching schemas/python-asyncio-pitfalls-plan.schema.json.
    produces:
      - kind: asyncio-diagnosis
        format: markdown
        description: The pinpointed pitfall (which trap from this catalog) and the corrected pattern, with the debug-mode evidence that confirms it.
      - kind: asyncio-plan-audit
        format: json
        description: Deterministic pass/fail audit of an asyncio-concurrency-plan against this skill's Quality Gates, produced by scripts/python_asyncio_pitfalls_audit.mjs.
---

# Python Asyncio Pitfalls

Asyncio is cooperative — one blocking call freezes the entire event loop. Most "asyncio is slow" stories are actually "we accidentally called a synchronous function inside a coroutine." This catalogs the traps.

## When to use

- Performance debug: a single slow request blocks all others.
- `RuntimeError: This event loop is already running` or "Task was destroyed but it is pending".
- Choosing between `asyncio.gather`, `asyncio.TaskGroup` (3.11+), and `asyncio.wait`.
- Cancellation isn't propagating; child tasks keep running.
- Mixing a sync library with async code without freezing the loop.

## Core capabilities

### Detect blocking calls in the event loop

```python
import asyncio

async def main():
    loop = asyncio.get_running_loop()
    loop.set_debug(True)         # warns when a coroutine takes too long
    # PYTHONASYNCIODEBUG=1 also enables debug mode at the env var level
```

Debug mode logs:
```
Executing <Task ... took 0.250 seconds
```

For production, instrument with `loop.slow_callback_duration = 0.1`. Anything over 100ms means a blocking call slipped in.

### TaskGroup — structured concurrency (Python 3.11+)

```python
async def fetch_all(urls: list[str]) -> list[str]:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(u)) for u in urls]
    return [t.result() for t in tasks]   # exceptions raised here
```

If any task raises, all siblings are cancelled and the exception (or `ExceptionGroup`) propagates out. This is the safe default.

`asyncio.gather` semantics:

```python
results = await asyncio.gather(fetch(a), fetch(b), return_exceptions=True)
# Without return_exceptions=True, the first exception cancels nothing — siblings keep running.
```

Use `gather` only when you want the older fire-and-collect-errors semantics; `TaskGroup` for everything else.

### Mixing sync code

```python
# WRONG — blocks the loop.
async def slow():
    return time.sleep(2)         # NOT awaitable; runs sync, blocks loop

# RIGHT — offload to a thread.
async def slow():
    return await asyncio.to_thread(time.sleep, 2)
```

`asyncio.to_thread` (3.9+) runs the function in a thread from the default executor. For CPU-bound work, use a `ProcessPoolExecutor` and `loop.run_in_executor`:

```python
import concurrent.futures
loop = asyncio.get_running_loop()
with concurrent.futures.ProcessPoolExecutor() as pool:
    result = await loop.run_in_executor(pool, expensive_pure_function, arg)
```

### Cancellation that actually works

```python
async def with_cleanup():
    try:
        await long_operation()
    except asyncio.CancelledError:
        await cleanup()
        raise                    # re-raise so the caller knows you were cancelled
    return result
```

Swallowing `CancelledError` is a common cause of "task was destroyed but it is pending" warnings. Always re-raise unless you're explicitly catching to ignore.

### Context propagation across tasks

```python
import contextvars

request_id = contextvars.ContextVar('request_id')

async def handler(rid: str):
    request_id.set(rid)
    # Spawned tasks inherit ContextVar values automatically.
    asyncio.create_task(log_async())
```

Use ContextVars for request-scoped data instead of thread-locals; threading.local doesn't work right in asyncio.

### Async iterators and generators

```python
async def stream_chunks(reader: asyncio.StreamReader):
    while not reader.at_eof():
        chunk = await reader.read(4096)
        if not chunk:
            break
        yield chunk

async for chunk in stream_chunks(reader):
    process(chunk)
```

`async for` is the right way to consume async iterators; pulling them with `__anext__` directly is rarely correct.

### Timeouts

```python
# 3.11+ — preferred
async with asyncio.timeout(5):
    result = await long_call()

# Older — wait_for
result = await asyncio.wait_for(long_call(), timeout=5)
```

`asyncio.timeout` integrates with TaskGroup; `wait_for` doesn't and can cause a double-cancel surprise.

### Queues

`asyncio.Queue` is the right choice; `queue.Queue` is sync and would block.

```python
queue: asyncio.Queue[Job] = asyncio.Queue(maxsize=100)

async def producer():
    while item := await source():
        await queue.put(item)

async def consumer():
    while True:
        job = await queue.get()
        try:
            await process(job)
        finally:
            queue.task_done()

await queue.join()   # wait until all submitted jobs marked done
```

`maxsize` provides backpressure — without it, a fast producer can OOM you.

### Async context managers and locks

```python
lock = asyncio.Lock()

async with lock:
    await critical_section()
```

Don't share an `asyncio.Lock` across event loops; each loop has its own.

### FastAPI / aiohttp specifics

- Use `async def` route handlers when you have async I/O. Sync handlers run in a threadpool — fine, but you pay the thread overhead.
- Connection pools (`httpx.AsyncClient`, asyncpg pool) belong at app startup, shared across requests. Don't construct per-request.
- Background tasks: `BackgroundTasks` (FastAPI) for fire-and-forget; `asyncio.create_task` if you need to track.

## Anti-patterns

### Calling sync I/O inside a coroutine

**Symptom:** Throughput craters under load; one slow request blocks everything.
**Diagnosis:** `requests.get`, `time.sleep`, `psycopg2.execute` — all sync, all block the loop.
**Fix:** Switch to async libraries (`httpx.AsyncClient`, `asyncio.sleep`, `asyncpg`). For sync libraries you can't replace, `asyncio.to_thread`.

### Forgetting to await a coroutine

**Symptom:** `coroutine 'fn' was never awaited` warning. No error, no return value.
**Diagnosis:** `result = fn()` instead of `result = await fn()`.
**Fix:** Add the await. Linters (ruff, pylint) flag this; turn the rule on.

### `asyncio.gather` without `return_exceptions=True` when you want to wait for all

**Symptom:** Partial work; some tasks ran, others mysteriously didn't.
**Diagnosis:** First exception propagates and the siblings are NOT cancelled — they keep running detached.
**Fix:** Use `TaskGroup` (cancels siblings cleanly) or `gather(*coros, return_exceptions=True)` followed by manual handling.

### Swallowing `CancelledError`

**Symptom:** "Task was destroyed but it is pending" warning at shutdown.
**Diagnosis:** A task caught CancelledError without re-raising; the runtime never finished cleanup.
**Fix:** Always re-raise after cleanup. Use try/finally for cleanup that must run regardless.

### Sharing an event loop across threads

**Symptom:** `RuntimeError: This event loop is already running` or sporadic data corruption.
**Diagnosis:** Calling `loop.run_until_complete` from a thread other than the loop's owner.
**Fix:** Use `asyncio.run_coroutine_threadsafe(coro, loop)` from non-loop threads.

### Unbounded `asyncio.Queue`

**Symptom:** OOM under burst load.
**Diagnosis:** Producer faster than consumer; queue grows unbounded.
**Fix:** Set `maxsize` on the queue. Producer awaits `put` and naturally blocks.

## Quality gates

- [ ] No sync I/O calls inside async code paths (linted via ruff `ASYNC1*` rules).
- [ ] `TaskGroup` used for fan-out where any failure should cancel siblings.
- [ ] `asyncio.timeout(...)` (or `wait_for`) wraps every external call.
- [ ] Connection pools constructed at startup; shared per process.
- [ ] `CancelledError` always re-raised after cleanup.
- [ ] `asyncio.Queue(maxsize=…)` has explicit backpressure.
- [ ] `loop.set_debug(True)` enabled in dev; `slow_callback_duration` alerts in prod.
- [ ] ContextVars used for request-scoped state.

## Deterministic Audit

Before committing to an asyncio design (or reviewing another agent's), write it as a JSON
`asyncio-concurrency-plan` matching `schemas/python-asyncio-pitfalls-plan.schema.json` and
run it through the deterministic auditor:

```bash
node scripts/python_asyncio_pitfalls_audit.mjs --input examples/sample-input.json
```

`auditPythonAsyncioPitfalls(plan)` (in `scripts/python_asyncio_pitfalls_audit.mjs`) turns
this catalog's traps and Quality Gates into machine-checkable rules over structured fields —
no keyword matching: blocking calls with no `to_thread`/executor offload, `gather` where
sibling cancellation is required, `TaskGroup`/`asyncio.timeout` on a pre-3.11 interpreter,
`queue.Queue` inside async code, unbounded `asyncio.Queue`, swallowed `CancelledError`,
missing external-call timeouts, per-request connection pools, and thread-local request
state. It returns `{ pass, score, findings, recommendations }`;
`examples/sample-input.json` is a clean 3.12 TaskGroup fan-out design that audits
`pass: true`. Version history: `CHANGELOG.md`.

## NOT for

- **trio / anyio** — different structured-concurrency primitives; different mental model.
- **threading without asyncio** — multiprocessing/threading have different deadlock patterns.
- **Python 2 sync patterns** — different language era.
- **gevent / eventlet** — monkey-patching greenlets, separate ecosystem.