Prevention and Testing

Make hangs impossible by construction where you can, loud and diagnosable where you cannot, and caught by tests that fail in seconds instead of hanging CI for an hour.

Principles
Timeouts everywhere, with one deadline
Lock hierarchy enforced at runtime
Structured concurrency
Shutdown protocol
Watchdogs and heartbeats
Event loop lag monitor
Testing: fail by timeout, never hang
Deterministic deadlock reproductions
Stress tests that find new interleavings
pytest, pytest-asyncio and anyio setup
CI gates
Review checklist

Principles

Every wait is bounded. Locks, queues, events, futures, joins, network IO. A bounded wait converts a deadlock into an exception with a stack trace at the exact place it happened.
One lock order. Document levels; enforce them in tests. Never hold a lock across an await, IO, a callback into user code, or a call into another component that may lock.
Every task and thread has an owner that waits for it, cancels it, and sees its exceptions (TaskGroup, executors, supervised threads). No fire-and-forget.
Every wakeup is guaranteed. task_done, release, set_result, notify in finally; failures are propagated to waiters, not just logged.
Hangs are observable. Heartbeats, loop-lag metrics, and a stack-dump trigger installed at startup.

Timeouts everywhere, with one deadline

Per-call timeouts multiply (5 retries x 3 calls x 30 s = 7.5 minutes). Give each request or job one deadline and derive the remaining budget for every wait from it.

python

import asyncio
import time


class Deadline:
    def __init__(self, seconds: float) -> None:
        self.at = time.monotonic() + seconds

    def remaining(self, cap: float = float("inf")) -> float:
        left = self.at - time.monotonic()
        if left <= 0:
            raise TimeoutError("deadline exceeded")
        return min(left, cap)


async def retrieve(q: str) -> list:
    await asyncio.sleep(0.01)
    return [q]


async def generate(docs: list) -> str:
    await asyncio.sleep(0.5)                       # a slow model
    return "answer"


async def rag(q: str, budget_s: float) -> str:
    dl = Deadline(budget_s)
    docs = await asyncio.wait_for(retrieve(q), timeout=dl.remaining(cap=2.0))
    return await asyncio.wait_for(generate(docs), timeout=dl.remaining(cap=30.0))


async def main() -> None:
    print(await rag("what is rope?", budget_s=1.0))
    try:
        await rag("what is rope?", budget_s=0.2)
    except (TimeoutError, asyncio.TimeoutError):
        print("timed out within the 0.2 s budget")


asyncio.run(main())
# answer
# timed out within the 0.2 s budget

On 3.11+ async with asyncio.timeout(budget): around the whole operation does the same with less code; asyncio.timeout_at(loop.time() + budget) and cm.reschedule() adjust it. Threads: pass timeout= to acquire, get, put, wait, join, result; on expiry raise or degrade explicitly.

Lock hierarchy enforced at runtime

Static analysis (hang_scan.py DLK001) finds cycles inside one module. Across modules, assign levels and check them at runtime in tests (and optionally in staging): acquiring a lock whose level is not higher than every lock the thread already holds raises immediately, on the first run, instead of deadlocking on the millionth.

python

import threading


class LockOrderError(RuntimeError):
    pass


class OrderedLock:
    """A threading.Lock with a level; a thread may only acquire locks in increasing level order."""

    _held = threading.local()

    def __init__(self, name: str, level: int) -> None:
        self.name, self.level = name, level
        self._lock = threading.Lock()

    def _stack(self) -> list:
        if not hasattr(self._held, "stack"):
            self._held.stack = []
        return self._held.stack

    def __enter__(self) -> "OrderedLock":
        stack = self._stack()
        if stack and stack[-1].level >= self.level:
            raise LockOrderError(f"acquiring {self.name}(L{self.level}) while holding "
                                 f"{stack[-1].name}(L{stack[-1].level})")
        if not self._lock.acquire(timeout=30):
            raise TimeoutError(f"{self.name} not acquired within 30s")
        stack.append(self)
        return self

    def __exit__(self, *exc) -> None:
        self._stack().remove(self)
        self._lock.release()


model_lock = OrderedLock("model", 1)
tokenizer_lock = OrderedLock("tokenizer", 2)

with model_lock, tokenizer_lock:
    print("correct order")
try:
    with tokenizer_lock, model_lock:
        pass
except LockOrderError as err:
    print("caught:", err)
# correct order
# caught: acquiring model(L1) while holding tokenizer(L2)

The asyncio version keeps the held stack in a contextvars.ContextVar (each task has its own context; child tasks start with a copy of the parent's) and wraps asyncio.Lock.

Structured concurrency

Unowned tasks are where lost exceptions and stuck waiters hide. Rules:

3.11+: async with asyncio.TaskGroup() as tg:; a failing child cancels its siblings, and the block does not exit until every child finished. Put a timeout around the block, or the group waits for its slowest child forever.
3.9/3.10: keep a set of tasks, asyncio.wait(tasks, return_when=FIRST_EXCEPTION), then cancel and gather(..., return_exceptions=True) the rest.
Background tasks that must outlive a request belong to an application-level supervisor created in the lifespan (FastAPI) and cancelled at shutdown.
Threads: executors (with ThreadPoolExecutor() as pool:), or a context manager that starts, stops and joins (see threading-and-multiprocessing-deadlocks.md).

python

import asyncio


async def embed_batch(i: int) -> int:
    await asyncio.sleep(0.01 * i)
    if i == 3:
        raise ValueError("batch 3: invalid UTF-8")
    return i


async def main() -> None:
    done = []
    try:
        async with asyncio.timeout(5):                     # 3.11+
            async with asyncio.TaskGroup() as tg:
                for i in range(6):
                    tg.create_task(embed_batch(i), name=f"batch-{i}").add_done_callback(
                        lambda t: done.append(t.get_name()) if not t.cancelled() and not t.exception() else None)
    except* ValueError as eg:
        print("failed:", eg.exceptions[0], "| completed before cancel:", sorted(done))


asyncio.run(main())  # failed: batch 3: invalid UTF-8 | completed before cancel: ['batch-0', 'batch-1', 'batch-2']

Shutdown protocol

Every long-lived worker needs a documented way to stop:

Worker	Stop signal	Wait
asyncio task	`task.cancel()` (re-raise `CancelledError` after cleanup)	`await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 10)`
asyncio queue consumer	`q.shutdown()` (3.13+) or one sentinel per consumer	`await q.join()` with a timeout, then cancel
thread	`threading.Event` checked via `q.get(timeout=...)` / `stop.wait(interval)`	`t.join(timeout)`, then log `is_alive()`
thread queue consumer	sentinel per worker, or `q.shutdown()` (3.13+)	`join(timeout)`
executor	`shutdown(wait=True, cancel_futures=True)` (3.9+)	bounded by your tasks' own timeouts
process	`terminate()` after a grace period, `kill()` after another	`join(timeout)`; drain its queues first

Handle SIGTERM by setting the stop signal (never by doing the shutdown inside the handler) and give the whole shutdown a deadline; if it is exceeded, dump stacks (dump_all()) and exit non-zero.

Watchdogs and heartbeats

An in-process watchdog thread watches a progress counter; if it does not move for stall_s, it dumps all threads and tasks and optionally aborts so the orchestrator restarts the process.

python

import os
import sys
import threading
import time


class ProgressWatchdog:
    def __init__(self, stall_s: float, on_stall) -> None:
        self.stall_s, self.on_stall = stall_s, on_stall
        self._last = time.monotonic()
        self._stop = threading.Event()
        self._thread = threading.Thread(target=self._run, name="progress-watchdog", daemon=True)

    def beat(self) -> None:                       # call after each processed item / step
        self._last = time.monotonic()

    def __enter__(self) -> "ProgressWatchdog":
        self._thread.start()
        return self

    def __exit__(self, *exc) -> None:
        self._stop.set()
        self._thread.join(timeout=2)

    def _run(self) -> None:
        while not self._stop.wait(self.stall_s / 4):
            idle = time.monotonic() - self._last
            if idle >= self.stall_s:
                self.on_stall(idle)
                self._last = time.monotonic()      # one report per stall period


def report(idle: float) -> None:
    import faulthandler
    print(f"no progress for {idle:.1f}s; dumping stacks", file=sys.stderr)
    faulthandler.dump_traceback(file=sys.stderr, all_threads=True)
    # in production: dump_all() from hang_watchdog for asyncio tasks too; os.abort() to get restarted


stalls = []
with ProgressWatchdog(stall_s=0.2, on_stall=stalls.append) as wd:
    for item in range(3):
        time.sleep(0.05)
        wd.beat()
    time.sleep(0.35)                               # a stall
print("stalls detected:", len(stalls), "pid", os.getpid() > 0)  # stalls detected: 1 pid True

Externally, hang_watchdog.py --idle 120 does the same for any script: progress is output on stdout/stderr or a call to heartbeat(). In Kubernetes, a liveness probe that checks "last progress timestamp is recent" (not just "the HTTP server answers") restarts stalled workers; dump stacks before failing the probe.

Event loop lag monitor

python

import asyncio
import time


async def lag_monitor(interval: float, samples: list, n: int) -> None:
    loop = asyncio.get_running_loop()
    for _ in range(n):
        start = loop.time()
        await asyncio.sleep(interval)
        samples.append(loop.time() - start - interval)   # how late the loop woke us


async def main() -> None:
    samples: list = []
    mon = asyncio.create_task(lag_monitor(0.01, samples, 10))
    await asyncio.sleep(0.03)
    time.sleep(0.2)                                       # simulated sync call on the loop
    await mon
    print(f"max loop lag: {max(samples):.2f}s")


asyncio.run(main())  # max loop lag: 0.20s (approximately)

Export the lag as a metric; alert above ~100 ms. A lag spike pinpoints the time window; debug mode or a stack dump during the spike names the call.

Testing: fail by timeout, never hang

A deadlock test must terminate on the buggy code. Put a timeout at three levels:

Inside the test: asyncio.wait_for/asyncio.timeout around the scenario, join(timeout) on threads.
Per test: pytest-timeout (@pytest.mark.timeout(10) or --timeout=60), which dumps stacks and fails the test; pytest's built-in faulthandler_timeout dumps stacks of slow tests (does not fail them).
Per job: the CI step timeout, with faulthandler.dump_traceback_later(..., exit=True) or hang_watchdog.py --kill-after so you get stacks instead of "job cancelled".

Thread-based scenarios cannot be cancelled: run them in a daemon thread and check is_alive() after a join timeout, so a deadlocked thread does not keep the test process alive.

python

import threading
import unittest


def run_bounded(fn, timeout: float) -> None:
    """Run fn in a daemon thread; fail (do not hang) if it does not finish in time."""
    errors = []

    def target():
        try:
            fn()
        except BaseException as err:  # re-raised in the test thread below
            errors.append(err)

    t = threading.Thread(target=target, daemon=True, name="scenario")
    t.start()
    t.join(timeout)
    if t.is_alive():
        raise AssertionError(f"scenario did not finish within {timeout}s (deadlock?)")
    if errors:
        raise errors[0]


class Cache:
    def __init__(self):
        self.lock = threading.Lock()
        self.data = {}

    def put(self, k, v):
        with self.lock:
            self.data[k] = v
            self._evict_locked()

    def _evict_locked(self):
        while len(self.data) > 2:
            self.data.pop(next(iter(self.data)))


class CacheTest(unittest.TestCase):
    def test_put_does_not_self_deadlock(self):
        c = Cache()
        run_bounded(lambda: [c.put(i, i) for i in range(5)], timeout=2)
        self.assertEqual(sorted(c.data), [3, 4])


if __name__ == "__main__":
    unittest.main(argv=["x", "-v"], exit=False)  # test_put_does_not_self_deadlock ... ok

Deterministic deadlock reproductions

Do not rely on sleep() to create the bad interleaving: force it with events/barriers, so the test reproduces the deadlock on every run before the fix and proves its absence after.

python

import asyncio


async def scenario(fixed: bool) -> str:
    index_lock, memory_lock = asyncio.Lock(), asyncio.Lock()
    upsert_holds_first, summarise_holds_first = asyncio.Event(), asyncio.Event()

    async def wait_briefly(ev: asyncio.Event) -> None:
        try:                                             # let the other task take its first lock, if it can
            await asyncio.wait_for(ev.wait(), timeout=0.1)
        except asyncio.TimeoutError:
            pass

    async def upsert() -> None:
        async with index_lock:
            upsert_holds_first.set()
            await wait_briefly(summarise_holds_first)
            async with memory_lock:
                pass

    async def summarise() -> None:
        first, second = (index_lock, memory_lock) if fixed else (memory_lock, index_lock)
        async with first:
            summarise_holds_first.set()
            await wait_briefly(upsert_holds_first)
            async with second:
                pass

    try:
        await asyncio.wait_for(asyncio.gather(upsert(), summarise()), timeout=1)
        return "ok"
    except asyncio.TimeoutError:
        return "deadlock"


print(asyncio.run(scenario(fixed=False)), asyncio.run(scenario(fixed=True)))  # deadlock ok

The same technique works for threads with threading.Barrier(2, timeout=...): the barrier itself raises BrokenBarrierError instead of hanging if the other party never arrives.

Stress tests that find new interleavings

When you do not know the bad interleaving yet:

Run the concurrent scenario hundreds of times with randomised small sleeps/await asyncio.sleep(0) injected at yield points, each run bounded by a timeout.
For threads, shrink the GIL switch interval to force more context switches: sys.setswitchinterval(1e-6) (default 0.005 s), and test on the free-threaded build (3.13t/3.14t) where threads really run in parallel.
Run with PYTHONASYNCIODEBUG=1 / -X dev so blocking calls and never-awaited coroutines show up.
Wrap the whole stress job in hang_watchdog.py --kill-after so a hit produces a stack dump.

pytest, pytest-asyncio and anyio setup

python

# conftest.py / test_agent.py  (requires pytest, pytest-asyncio, pytest-timeout)
import asyncio

import pytest


@pytest.mark.timeout(10)                 # pytest-timeout: fail + dump stacks after 10 s
@pytest.mark.asyncio                     # pytest-asyncio; with anyio use @pytest.mark.anyio
async def test_agent_step_does_not_hang_on_dead_tool():
    async def dead_tool():
        await asyncio.Event().wait()

    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(dead_tool(), timeout=0.1)

ini

# pytest.ini
[pytest]
timeout = 120
faulthandler_timeout = 60
asyncio_mode = strict

pytest-timeout's default signal method raises in the main thread (POSIX); the thread method dumps all stacks and hard-exits, which also works when the main thread is stuck in C code. Use --timeout-method=thread in CI for deadlock-prone suites.

CI gates

bash

python scripts/hang_scan.py src/ --strict --ignore DLK011,DLK015      # static: high/medium rules block merges
python scripts/hang_watchdog.py --kill-after 1800 --dump-file hang-dump.txt -m pytest -x tests/

Upload hang-dump.txt as a CI artifact on failure: the next "the job timed out" ticket arrives with every thread and task stack attached.

Review checklist

Each lock has a documented level; nested acquisitions go upward; no lock held across await/IO
No method that takes a lock calls another public method that takes the same lock
No threading primitives, time.sleep, .result(), .join() or sync clients in async def
asyncio.run only at entry points; no run_until_complete inside coroutines
Each get() matched by task_done() in finally; each acquire() by release() in finally
Every create_future() resolved on all paths, with timeouts on the await
Every task/thread/process owned, stoppable, and joined with a timeout
Nothing submitted to an executor waits on the same executor
Multiprocessing: spawn/forkserver, drain before join, timeouts on results
Signal handlers only set flags
Tests for concurrency paths have per-test timeouts and deterministic interleavings

Contents

Principles

Timeouts everywhere, with one deadline

Lock hierarchy enforced at runtime

Structured concurrency

Shutdown protocol

Watchdogs and heartbeats

Event loop lag monitor

Testing: fail by timeout, never hang

Deterministic deadlock reproductions

Stress tests that find new interleavings

pytest, pytest-asyncio and anyio setup

CI gates

Review checklist

python-asyncio-pattern-advisor

python-beginner-concept-tutor

python-comprehension-refactorer

python-context-manager-writer

python-dataclass-converter

python-exception-handling-designer

Prevention and Testing

#Contents

#Principles

#Timeouts everywhere, with one deadline

#Lock hierarchy enforced at runtime

#Structured concurrency

#Shutdown protocol

#Watchdogs and heartbeats

#Event loop lag monitor

#Testing: fail by timeout, never hang

#Deterministic deadlock reproductions

#Stress tests that find new interleavings

#pytest, pytest-asyncio and anyio setup

#CI gates

#Review checklist

python-asyncio-pattern-advisor

python-beginner-concept-tutor

python-comprehension-refactorer

python-context-manager-writer

python-dataclass-converter

python-exception-handling-designer

Contents

Principles

Timeouts everywhere, with one deadline

Lock hierarchy enforced at runtime

Structured concurrency

Shutdown protocol

Watchdogs and heartbeats

Event loop lag monitor

Testing: fail by timeout, never hang

Deterministic deadlock reproductions

Stress tests that find new interleavings

pytest, pytest-asyncio and anyio setup

CI gates

Review checklist