Asyncio run_in_executor: Run Blocking Code in Python

Python's asyncio framework relies on a single-threaded cooperative multitasking model, meaning that any synchronous, blocking function call will freeze the entire event loop and halt all concurrent tasks. To integrate legacy synchronous code without rewriting it from scratch, Python provides loop.run_in_executor(). This article explains how run_in_executor() works, how to offload both I/O-bound and CPU-bound operations to background worker pools, and how modern Python versions have simplified this integration pattern.

The Blocking Problem in Asyncio

The asyncio event loop executes asynchronous tasks sequentially on a single thread. When a coroutine yields control via await, the event loop proceeds to run other tasks. However, if a task invokes a standard synchronous function—such as time.sleep(), a blocking database driver, or a standard HTTP request using requests—the thread blocks entirely. As long as that synchronous operation executes, the event loop cannot process any other incoming events, timers, or network traffic.

How loop.run_in_executor() Works

The run_in_executor() method bridges synchronous and asynchronous execution by delegating the execution of a blocking callable to an external executor, usually an instance of concurrent.futures.ThreadPoolExecutor or concurrent.futures.ProcessPoolExecutor.

The method signature is:

await loop.run_in_executor(executor, func, *args)

When called, run_in_executor() returns an asyncio.Future that tracks the execution of the synchronous function in the background pool. When the worker finishes execution, the future resolves with the function's return value, resuming the awaiting coroutine without having stalled the main event loop.

Handling Blocking I/O with ThreadPoolExecutor

For synchronous operations dominated by network or disk I/O, a thread pool is the appropriate choice. Threads share memory and have low overhead compared to processes.

import asyncio
import time
import requests

def blocking_fetch(url: str) -> int:
    # A standard blocking HTTP request
    response = requests.get(url)
    return len(response.text)

async def main():
    loop = asyncio.get_running_loop()
    
    # Passing None uses the default ThreadPoolExecutor
    result = await loop.run_in_executor(
        None, 
        blocking_fetch, 
        "https://httpbin.org/get"
    )
    print(f"Fetched {result} characters.")

asyncio.run(main())

If the legacy function requires keyword arguments, wrap the target function with functools.partial:

from functools import partial

# Wrapping a function with keyword arguments
bound_function = partial(blocking_fetch_with_kwargs, timeout=10)
result = await loop.run_in_executor(None, bound_function, "https://example.com")

Handling CPU-Bound Tasks with ProcessPoolExecutor

Python's Global Interpreter Lock (GIL) limits multiple system threads from executing Python bytecode simultaneously. For intensive computations—such as data parsing, image processing, or cryptography—using a ThreadPoolExecutor will still block execution across other threads.

To overcome this, pass an explicit ProcessPoolExecutor to run_in_executor(). This spins up separate Python processes, each with its own interpreter and memory space:

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy_computation(n: int) -> int:
    count = 0
    for i in range(n):
        count += i * i
    return count

async def main():
    loop = asyncio.get_running_loop()
    
    # Use ProcessPoolExecutor to bypass the GIL
    with ProcessPoolExecutor() as executor:
        result = await loop.run_in_executor(
            executor, 
            cpu_heavy_computation, 
            10_000_000
        )
        print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

Note: Functions and arguments passed to a ProcessPoolExecutor must be picklable.

Modern Alternative: asyncio.to_thread()

In Python 3.9 and later, asyncio.to_thread() was introduced as a higher-level wrapper around loop.run_in_executor(None, ...). It removes boilerplate code and natively accepts keyword arguments:

import asyncio
import time

def blocking_task(seconds: int, message: str = "Done"):
    time.sleep(seconds)
    return message

async def main():
    # Directly offload to a thread with args and kwargs
    result = await asyncio.to_thread(blocking_task, 2, message="Finished")
    print(result)

asyncio.run(main())

Under the hood, asyncio.to_thread() accesses the running event loop and delegates execution to the default ThreadPoolExecutor, while automatically propagating context variables (contextvars), which run_in_executor() does not do by default.

Key Considerations