Python asyncio Event Loop and Thread Safety: When to Use run_in_executor

Originally published on Code Beneath.

The asyncio event loop is single-threaded. When you call a blocking function inside an async context, you freeze the entire loop, and all other coroutines starve. The standard solution is to push blocking work to a thread pool using run_in_executor, but many engineers misunderstand when to reach for it, how it actually works, and what performance tradeoffs exist. This article walks through a python asyncio run in executor blocking io example and shows you the internals, the gotchas, and when a pure async solution is genuinely better.

The Problem: Blocking Code in the Event Loop

Asyncio's event loop cannot pause a blocking call. If you call time.sleep(1) or perform a synchronous database query inside a coroutine, the loop itself hangs. Every other coroutine waiting for I/O stalls. Here is what this looks like:

import asyncio
import time

async def bad_example():
    print("Start:", time.time())
    time.sleep(2)  # Blocks the entire loop
    print("End:", time.time())
    return "done"

async def other_work():
    for i in range(5):
        print(f"Work {i} at {time.time()}")
        await asyncio.sleep(0.2)

async def main():
    # These should run concurrently
    task1 = asyncio.create_task(bad_example())
    task2 = asyncio.create_task(other_work())
    await asyncio.gather(task1, task2)

asyncio.run(main())

Run this. You will see that other_work does not print until after bad_example completes its 2 second sleep. The loop never yields. Now let's fix it with run_in_executor.

Understanding run_in_executor

The run_in_executor method pushes a blocking function call to a thread pool, allowing the event loop to continue. By default, it uses a ThreadPoolExecutor with a modest number of worker threads (typically equal to the number of CPU cores plus 5). The loop yields, other coroutines run, and when the thread finishes the blocking work, the result bubbles back as an awaitable.

import asyncio
import time

async def blocking_work():
    # This runs in a thread, not blocking the loop
    time.sleep(2)
    return "result from thread"

async def other_work():
    for i in range(5):
        print(f"Work {i} at {time.time()}")
        await asyncio.sleep(0.2)

async def main():
    loop = asyncio.get_event_loop()

    # Offload blocking_work to thread pool
    task1 = asyncio.create_task(
        loop.run_in_executor(None, blocking_work)
    )
    task2 = asyncio.create_task(other_work())

    results = await asyncio.gather(task1, task2)
    print("Results:", results)

asyncio.run(main())

Now other_work runs concurrently with the sleep in the thread. The loop does not block. Print timestamps to confirm timing interleaves.

Real-World Case: Database Queries

A common pattern is synchronous database libraries (psycopg2, pymysql, sqlite3) running inside async code. These are blocking, so they belong in an executor. Here is a realistic example with PostgreSQL:

import asyncio
import psycopg2
import time

def query_db(query, params=None):
    """Synchronous database call. Blocks."""
    conn = psycopg2.connect(
        host="localhost",
        database="testdb",
        user="postgres",
        password="password"
    )
    cursor = conn.cursor()
    try:
        cursor.execute(query, params or ())
        results = cursor.fetchall()
        return results
    finally:
        cursor.close()
        conn.close()

async def fetch_user(user_id):
    """Async wrapper around sync DB call."""
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,
        query_db,
        "SELECT * FROM users WHERE id = %s",
        (user_id,)
    )
    return result

async def main():
    start = time.time()

    # Fetch 5 users concurrently
    tasks = [fetch_user(i) for i in range(1, 6)]
    results = await asyncio.gather(*tasks)

    elapsed = time.time() - start
    print(f"Fetched {len(results)} users in {elapsed:.2f}s")
    for i, user in enumerate(results, 1):
        print(f"User {i}: {user}")

asyncio.run(main())

Without the executor, each query would block the loop sequentially, taking roughly 5 times as long. With the executor, multiple queries run in parallel threads.

Providing a Custom Executor

The default executor works fine for many cases, but you can configure a custom one to control thread count, naming, or other attributes:

import asyncio
import concurrent.futures
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_cpu_work(n):
    """Simulate CPU-bound work."""
    total = 0
    for i in range(n):
        total += i
    time.sleep(0.5)  # Simulate I/O
    return total

async def main():
    # Create a custom thread pool
    executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="worker")
    loop = asyncio.get_event_loop()
    loop.set_default_executor(executor)

    # Now all run_in_executor calls use this pool by default
    tasks = [
        asyncio.create_task(
            loop.run_in_executor(None, blocking_cpu_work, 1_000_000)
        )
        for _ in range(4)
    ]

    results = await asyncio.gather(*tasks)
    print("Results:", results)

    executor.shutdown(wait=True)

asyncio.run(main())

You can also pass a specific executor to a single call, bypassing the default.

When NOT to Use run_in_executor

The real question engineers often miss is: should you use run_in_executor at all, or should you use a truly async library instead? Here's the comparison:

Approach	Thread Overhead	Code Complexity	Best For
run_in_executor	High (threads are expensive)	Low (simple wrapping)	Legacy sync libs, CPU work
Async library (aiopg, motor)	None (pure async I/O)	Medium (new patterns)	New code, high concurrency
Sync library on sync event loop	None	Low	Scripts, CLI tools, non-concurrent code

If you're building a new system and can control your dependencies, prefer a native async database driver like aiopg or motor (for MongoDB). If you're integrating legacy code or a library that has no async variant, then run_in_executor is the right tool.

Performance Measurement: Executor vs. Async

Let's measure the actual overhead. Here's a test that simulates 100 small I/O operations:

import asyncio
import time
import random

# Simulated sync I/O
def sync_io_call(duration):
    """Blocks for a random short duration."""
    time.sleep(duration)
    return f"done in {duration:.3f}s"

# Truly async I/O
async def async_io_call(duration):
    """Non-blocking sleep."""
    await asyncio.sleep(duration)
    return f"done in {duration:.3f}s"

async def benchmark_executor():
    """Using run_in_executor."""
    loop = asyncio.get_event_loop()

    durations = [random.uniform(0.01, 0.05) for _ in range(100)]
    tasks = [
        loop.run_in_executor(None, sync_io_call, d)
        for d in durations
    ]

    start = time.time()
    await asyncio.gather(*tasks)
    return time.time() - start

async def benchmark_pure_async():
    """Using pure async."""
    durations = [random.uniform(0.01, 0.05) for _ in range(100)]
    tasks = [async_io_call(d) for d in durations]

    start = time.time()
    await asyncio.gather(*tasks)
    return time.time() - start

async def main():
    # Warm up
    await benchmark_executor()
    await benchmark_pure_async()

    # Measure
    executor_time = await benchmark_executor()
    async_time = await benchmark_pure_async()

    print(f"Executor time:   {executor_time:.3f}s")
    print(f"Pure async time: {async_time:.3f}s")
    print(f"Overhead ratio:  {executor_time / async_time:.2f}x")

asyncio.run(main())

Run this on your machine. You'll typically see the executor version takes 2-5x longer due to thread creation, context switching, and GIL contention. For a single blocking call, this doesn't matter. For thousands of concurrent I/O operations, it becomes a bottleneck.

Thread Pool Size and Queue Buildup

A subtle trap: if you spam run_in_executor with more tasks than your thread pool can handle, requests queue up in memory. Here's what happens:

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def slow_work(task_id):
    """Simulates slow blocking work."""
    print(f"Task {task_id} starting")
    time.sleep(2)
    print(f"Task {task_id} done")
    return task_id

async def main():
    # Small thread pool
    executor = ThreadPoolExecutor(max_workers=2)
    loop = asyncio.get_event_loop()

    # Submit 10 tasks to a pool of 2 threads
    tasks = [
        loop.run_in_executor(executor, slow_work, i)
        for i in range(10)
    ]

    start = time.time()
    results = await asyncio.gather(*tasks)
    elapsed = time.time() - start

    print(f"All done in {elapsed:.1f}s")
    print(f"Results: {results}")

    executor.shutdown()

asyncio.run(main())

This takes roughly 10 seconds (10 tasks x 2s / 2 threads = 10s), even though you submitted them all at once. The first two run immediately, then the queue blocks and waits. If your workload is truly I/O-bound, increase max_workers. If it's CPU-bound, you're limited by the number of cores.

Exception Handling in Executors

Exceptions raised in executor threads propagate to the awaiting coroutine. Handle them like any async operation:

import asyncio

def failing_work():
    raise ValueError("Something went wrong in the thread")

async def main():
    loop = asyncio.get_event_loop()

    try:
        result = await loop.run_in_executor(None, failing_work)
    except ValueError as e:
        print(f"Caught exception: {e}")

asyncio.run(main())

The exception is wrapped in a Future and unwrapped when you await it. This is identical to exception handling in regular async code.

Common Mistakes

1. Forgetting to pass a callable: run_in_executor(None, query_db(user_id)) runs the function immediately in the calling coroutine. Pass query_db and args separately: run_in_executor(None, query_db, user_id).

2. Using run_in_executor for CPU-bound work expecting speedup: threads on Python face GIL contention. For true parallelism on CPU tasks, use ProcessPoolExecutor (with caveats) or multiprocessing directly.

3. Assuming zero overhead: threads have real cost. Context switches, memory per thread, and synchronization all add up. If you can use a native async library, do so.

4. Not tuning max_workers for your workload: too few threads means queue buildup and poor concurrency. Too many wastes memory and CPU. Profile your specific use case.

FAQ

Can I use run_in_executor with ProcessPoolExecutor for CPU work?

Yes, but it's rarely better than alternatives. Process pools have even higher overhead than thread pools (pickling, IPC). For CPU-bound work, consider using a dedicated CPU worker process pool outside asyncio, or use libraries like Ray. For most web backends, you don't need CPU parallelism.

Does run_in_executor release the GIL?

Not fully. Pure Python code in threads still contends for the GIL. Native C extensions (like those in database drivers) release the GIL during I/O, which is why psycopg2 or pymysql in threads works reasonably well. But you don't get true parallelism for Python bytecode.

What's the difference between run_in_executor and asyncio.to_thread (Python 3.9+)?

asyncio.to_thread is a simpler wrapper around run_in_executor with automatic default executor management. Under the hood, it's identical. Use to_thread for new code if you're on 3.9+, it's cleaner.

Should I use run_in_executor for file I/O?

File I/O is slower than network I/O, so offloading to a thread is reasonable for large reads/writes. For small files or logs, the overhead may dominate. Async file libraries like aiofiles exist but are thin wrappers around thread pools internally.

Can I set a timeout on executor work?

Wrap the task in asyncio.wait_for: await asyncio.wait_for(loop.run_in_executor(...), timeout=5.0). If the timeout fires, the task is cancelled, but the thread keeps running in the background. This is a known limitation.