run_in_executor and asyncio.to_thread

Article.

How do you call blocking code from async Python without stalling everything else? The short answer is “hand it to a thread”. The senior answer knows what that thread pool is, who else is using it, and the three ways it fails without raising an error: it queues, it ignores your timeout, and it drops your context.

What blocking costs

An event loop runs one coroutine at a time and switches only at an await. A sync call inside a coroutine never reaches one, so the whole process stops:

@app.get("/users/{id}")
async def get_user(id: int):
    user = db.query(User).get(id)        # sync driver: the loop is frozen
    return user

Measured on Python 3.14.5: a heartbeat task that wakes every 10 ms, while one handler waits 300 ms:

time.sleep in the coroutine: worst heartbeat delay 306 ms
await asyncio.to_thread:     worst heartbeat delay 10 ms

306 ms is every other request in the process waiting for one query. The 10 ms left with to_thread is the timer’s own granularity. If every request makes a 50 ms sync call on the loop, the process tops out at 20 requests a second.

The fix, and what it actually does

@app.get("/users/{id}")
async def get_user(id: int):
    return await asyncio.to_thread(load_user, id)

The mechanism

asyncio.to_thread is four lines in the standard library, and each one matters:

async def to_thread(func, /, *args, **kwargs):
    loop = events.get_running_loop()
    ctx = contextvars.copy_context()
    func_call = functools.partial(ctx.run, func, *args, **kwargs)
    return await loop.run_in_executor(None, func_call)

run_in_executor(None, ...) then does the rest:

coroutine         event loop             worker thread
  │ to_thread()       │                       │
  │─────────────────▶ │ default executor      │
  │                   │ (created lazily)      │
  │                   │ submit(func) ───────▶ │ func runs;
  │ awaits an         │                       │ loop is free
  │ asyncio.Future    │ ◀─ call_soon_threadsafe
  │ ◀──────────────── │ future resolves       │

Three facts fall out of that code, and every production failure below is one of them:

1. None means one shared pool per loop. The default executor is a ThreadPoolExecutor(thread_name_prefix="asyncio") with min(32, os.process_cpu_count() + 4) workers: 16 on a 12-CPU machine. Every to_thread call in the process, and every library that uses it, draws from it.
2. The thread is not the task. wrap_future links an asyncio future to the pool’s future. Cancelling the task cancels the asyncio side; Python has no way to stop a running thread.
3. Only to_thread copies the context. The copy_context line is to_thread’s; a bare run_in_executor call has none.

asyncio’s own DNS uses the same pool

loop.getaddrinfo and loop.getnameinfo, which every outbound connection made by asyncio.open_connection, aiohttp or httpx’s asyncio backend goes through, are run_in_executor(None, socket.getaddrinfo, ...). Saturate the default pool with slow sync calls and new connections stall on DNS: not your code’s DNS, the event loop’s.

Where it breaks in production

The pool queues, and nothing says so

48 calls of 200 ms on the 16-worker default pool, with a DNS lookup issued after the pool filled:

16 workers, 48 calls of 200 ms: finished at [0.2, 0.4, 0.6] s
getaddrinfo('localhost') while the pool was full: 546 ms

The calls finish in waves, and a lookup of localhost took half a second because it queued behind them. No exception, no log line: latency rises while CPU stays low. High p99 with idle CPU is the signature of an exhausted thread pool, and the cheapest diagnostic is counting live threads named asyncio_* against the pool size.

A timeout is not a cancellation

async def save_with_timeout(record):
    try:
        await asyncio.wait_for(asyncio.to_thread(slow_write, record), timeout=0.1)
    except TimeoutError:
        log.warning("write timed out")        # the write is still running

wait_for raised TimeoutError at 0.1 s
thread finished its write at 1.0 s
asyncio.run returned at 1.0 s

The caller was told the write failed; the write happened anyway. Three consequences:

• Retries duplicate work. A handler that retries on timeout now has two writes in flight. The sync work must be idempotent, or the timeout must live inside it (a driver-level statement timeout, a socket timeout).
• Timed-out calls still hold workers. Ten abandoned 30-second calls are ten fewer workers for everyone else, which feeds the queueing above.
• Shutdown waits for them. asyncio.run calls shutdown_default_executor with a 300-second timeout, so a stuck sync call turns a deploy’s graceful shutdown into a five-minute hang and a SIGKILL.

Context is lost in run_in_executor

request_id.set("req-42")
await loop.run_in_executor(None, request_id.get)   # '-'  (the default)
await asyncio.to_thread(request_id.get)            # 'req-42'

OpenTelemetry spans, structlog’s bound request id and any ContextVar your middleware set all go dark inside the thread. Prefer to_thread; when you need a specific executor, copy the context yourself:

ctx = contextvars.copy_context()
result = await loop.run_in_executor(db_pool, ctx.run, load_user, user_id)

Whatever you hand over now runs concurrently

Offloading turns one sequence of calls into several threads. A SQLAlchemy Session is not thread-safe, so one session shared by concurrent to_thread calls corrupts its identity map without an error. Give each offloaded call its own session (or connection), the same rule as for any threaded code.

Choosing the executor

A process pool needs picklable arguments, costs a process start per worker, and is reused, never created per call (ProcessPoolExecutor: what processes add):

cpu_pool = ProcessPoolExecutor(max_workers=4)

async def score(batch):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(cpu_pool, heavy_cpu_work, batch)

Dedicated pools are bulkheads

Give each slow dependency its own pool, sized to that dependency, so a stalled database cannot starve DNS or a different client:

# sized to the driver's connection pool: more threads would only queue there
db_pool = ThreadPoolExecutor(max_workers=20, thread_name_prefix="db")

async def load_user(user_id: int) -> User:
    loop = asyncio.get_running_loop()
    ctx = contextvars.copy_context()
    return await loop.run_in_executor(db_pool, ctx.run, _load_user_sync, user_id)

Twenty threads against a connection pool of ten is ten threads waiting on a lock inside the driver: the thread pool should match the resource behind it.

FastAPI and sync drivers

FastAPI runs a sync def endpoint, and sync dependencies, through AnyIO’s thread limiter, 40 tokens by default and separate from asyncio’s default executor. run_in_threadpool uses the same limiter (FastAPI and the event loop). So a FastAPI service has at least two thread pools, and a sync DB driver called both ways shares neither.

For a sync driver in an async service there are two honest positions:

• Bridge: keep the sync driver, offload every call through one sized pool, and accept a thread per in-flight query.
• Migrate: an async driver (asyncpg, psycopg 3 async, SQLAlchemy 2.0 AsyncSession) removes the pool and its three failure modes entirely.

The bridge is a migration step, not a destination; a service that is mostly offloaded sync calls would be simpler as a sync service with more workers.

The trade not taken

The tempting fix for exhaustion is a bigger default pool:

loop.set_default_executor(ThreadPoolExecutor(max_workers=200))

It makes the symptom go away and the problem worse. Two hundred threads against a database pool of twenty still queue, now inside the driver where it is harder to see; memory grows by a thread stack each; and every library sharing the default pool, DNS included, still competes with your slowest dependency. Sized, named pools per dependency keep the queue where you can measure it and the failure where it belongs.

Quick reference

result = await asyncio.to_thread(sync_fn, *args, **kwargs)     # default pool + context
result = await loop.run_in_executor(pool, ctx.run, sync_fn, *args)  # own pool
result = await loop.run_in_executor(cpu_pool, cpu_fn, data)    # CPU-bound
result = await run_in_threadpool(sync_fn, *args)               # FastAPI, AnyIO limiter

Related

• ProcessPoolExecutor: what processes add - process pools: start methods, pickling, broken pools
• Cancellation, timeouts and error handling in asyncio - what cancellation does and does not stop
• The GIL, and free-threaded Python - why threads do not parallelise CPU-bound Python
• concurrent.futures - the executor API shared by both pools
• FastAPI and the event loop - def vs async def endpoints and AnyIO’s limiter