Python Concurrency: asyncio vs threading vs multiprocessing

Python gives you three distinct ways to run work concurrently: asyncio, threading, and multiprocessing. Each one solves a different class of problem, and choosing the wrong tool can leave your program slower than a plain for loop. This guide puts all three side by side so you can make an informed decision the next time you need speed.

The Global Interpreter Lock (GIL)

Before comparing anything, you need to understand the GIL. CPython’s Global Interpreter Lock is a mutex that allows only one thread to execute Python bytecode at a time. This single constraint explains almost every performance difference between the three models.

• Threading is subject to the GIL. Threads can interleave I/O waits, but they cannot execute Python bytecodes in true parallel.
• Multiprocessing sidesteps the GIL by spawning separate processes, each with its own interpreter and memory space.
• asyncio runs in a single thread and uses cooperative scheduling. It never fights the GIL because only one coroutine runs at a time anyway.

Understanding this gives you a simple rule of thumb: use threading or asyncio for I/O-bound work, use multiprocessing for CPU-bound work.

A Baseline: Sequential Execution

Let’s start with a simple I/O-bound task and a CPU-bound task so we can benchmark later.

import time
import math

# I/O-bound: simulate network requests
def fetch_url(url: str) -> int:
    """Simulate a network request that takes 0.5 seconds."""
    time.sleep(0.5)
    return len(url)

# CPU-bound: compute-heavy work
def compute_primes(limit: int) -> int:
    """Count primes up to limit using trial division."""
    count = 0
    for num in range(2, limit):
        if all(num % i != 0 for i in range(2, int(math.sqrt(num)) + 1)):
            count += 1
    return count

Running 10 sequential fetches takes about 5 seconds. Running 4 sequential prime computations takes whatever your CPU needs, multiplied by 4. These are the baselines we will beat.

Threading: Shared Memory, Interleaved Execution

The threading module lets you run functions in separate OS threads. Because of the GIL, this only helps when threads spend most of their time waiting on I/O.

import threading
import time

def fetch_url(url: str, results: dict, index: int) -> None:
    time.sleep(0.5)  # simulate I/O wait
    results[index] = len(url)

urls = [f"https://example.com/page/{i}" for i in range(10)]
results: dict[int, int] = {}

start = time.perf_counter()

threads = []
for i, url in enumerate(urls):
    t = threading.Thread(target=fetch_url, args=(url, results, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

elapsed = time.perf_counter() - start
print(f"Threading: {elapsed:.2f}s for {len(urls)} requests")
# Threading: ~0.50s for 10 requests

All 10 threads sleep concurrently, so the total time drops from 5 seconds to roughly 0.5 seconds.

Using ThreadPoolExecutor

For cleaner code, use concurrent.futures.ThreadPoolExecutor:

from concurrent.futures import ThreadPoolExecutor
import time

def fetch_url(url: str) -> int:
    time.sleep(0.5)
    return len(url)

urls = [f"https://example.com/page/{i}" for i in range(10)]

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(fetch_url, urls))
elapsed = time.perf_counter() - start

print(f"ThreadPool: {elapsed:.2f}s, results: {results[:3]}...")

ThreadPoolExecutor manages thread lifecycle, limits the pool size, and returns results in order.

When Threading Falls Short

Try threading with the CPU-bound task:

from concurrent.futures import ThreadPoolExecutor
import time, math

def compute_primes(limit: int) -> int:
    count = 0
    for num in range(2, limit):
        if all(num % i != 0 for i in range(2, int(math.sqrt(num)) + 1)):
            count += 1
    return count

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(compute_primes, [50_000] * 4))
elapsed = time.perf_counter() - start
print(f"Threading CPU-bound: {elapsed:.2f}s")
# Often SLOWER than sequential due to GIL contention

The GIL forces threads to take turns executing bytecodes. Context-switching overhead makes this slower than running sequentially.

asyncio: Cooperative Concurrency in One Thread

asyncio uses an event loop and coroutines. When a coroutine hits an await, it yields control so another coroutine can run. There is no thread switching overhead and no GIL contention.

import asyncio
import time

async def fetch_url(url: str) -> int:
    await asyncio.sleep(0.5)  # non-blocking sleep
    return len(url)

async def main() -> list[int]:
    urls = [f"https://example.com/page/{i}" for i in range(10)]
    tasks = [asyncio.create_task(fetch_url(url)) for url in urls]
    return await asyncio.gather(*tasks)

start = time.perf_counter()
results = asyncio.run(main())
elapsed = time.perf_counter() - start
print(f"asyncio: {elapsed:.2f}s for {len(results)} requests")
# asyncio: ~0.50s for 10 requests

The performance is similar to threading for this I/O-bound case, but asyncio uses far fewer system resources since it runs in a single thread.

Real HTTP with aiohttp

In practice, you pair asyncio with an async HTTP library:

import asyncio
import aiohttp
import time

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as response:
        text = await response.text()
        return len(text)

async def main() -> list[int]:
    urls = [f"https://httpbin.org/delay/1" for _ in range(10)]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

start = time.perf_counter()
results = asyncio.run(main())
elapsed = time.perf_counter() - start
print(f"aiohttp: {elapsed:.2f}s for {len(results)} requests")

asyncio Pitfalls

Blocking calls inside a coroutine freeze the entire event loop:

import asyncio
import time

async def bad_coroutine() -> None:
    # This blocks the event loop! Other coroutines cannot run.
    time.sleep(2)

async def good_coroutine() -> None:
    # Run blocking code in a thread pool instead.
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, time.sleep, 2)

Always use await with async-compatible libraries or offload blocking calls with run_in_executor.

Multiprocessing: True Parallelism

multiprocessing spawns separate Python processes, each with its own GIL. This is the only stdlib option that achieves true CPU parallelism.

from concurrent.futures import ProcessPoolExecutor
import time, math

def compute_primes(limit: int) -> int:
    count = 0
    for num in range(2, limit):
        if all(num % i != 0 for i in range(2, int(math.sqrt(num)) + 1)):
            count += 1
    return count

if __name__ == "__main__":
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(compute_primes, [50_000] * 4))
    elapsed = time.perf_counter() - start
    print(f"Multiprocessing: {elapsed:.2f}s, primes: {results}")
    # Nearly 4x faster than sequential on a 4-core machine

The Cost of Multiprocessing

Processes do not share memory. Data must be serialized (pickled) to pass between them. This creates overhead:

from concurrent.futures import ProcessPoolExecutor
import time

def process_chunk(data: list[int]) -> int:
    return sum(x * x for x in data)

if __name__ == "__main__":
    # Large data must be pickled and sent to each process
    big_list = list(range(1_000_000))
    chunks = [big_list[i::4] for i in range(4)]

    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(process_chunk, chunks))
    elapsed = time.perf_counter() - start

    print(f"Total: {sum(results)}, Time: {elapsed:.2f}s")

If the data is large relative to the computation, serialization overhead can eat your speedup. Keep payloads small or use shared memory via multiprocessing.shared_memory.

Head-to-Head Benchmark

Let’s run all three models against both workload types:

import asyncio
import time
import math
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def io_task(n: int) -> int:
    time.sleep(0.5)
    return n

async def async_io_task(n: int) -> int:
    await asyncio.sleep(0.5)
    return n

def cpu_task(limit: int) -> int:
    return sum(1 for num in range(2, limit)
               if all(num % i != 0 for i in range(2, int(math.sqrt(num)) + 1)))

def benchmark_io(count: int = 20) -> dict[str, float]:
    results = {}

    # Sequential
    start = time.perf_counter()
    [io_task(i) for i in range(count)]
    results["sequential"] = time.perf_counter() - start

    # Threading
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=count) as pool:
        list(pool.map(io_task, range(count)))
    results["threading"] = time.perf_counter() - start

    # asyncio
    async def run_async():
        tasks = [async_io_task(i) for i in range(count)]
        await asyncio.gather(*tasks)

    start = time.perf_counter()
    asyncio.run(run_async())
    results["asyncio"] = time.perf_counter() - start

    return results

def benchmark_cpu(count: int = 4, limit: int = 50_000) -> dict[str, float]:
    results = {}

    # Sequential
    start = time.perf_counter()
    [cpu_task(limit) for _ in range(count)]
    results["sequential"] = time.perf_counter() - start

    # Threading
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=count) as pool:
        list(pool.map(cpu_task, [limit] * count))
    results["threading"] = time.perf_counter() - start

    # Multiprocessing
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=count) as pool:
        list(pool.map(cpu_task, [limit] * count))
    results["multiprocessing"] = time.perf_counter() - start

    return results

if __name__ == "__main__":
    print("I/O-bound benchmark:")
    for model, elapsed in benchmark_io().items():
        print(f"  {model:20s}: {elapsed:.2f}s")

    print("\nCPU-bound benchmark:")
    for model, elapsed in benchmark_cpu().items():
        print(f"  {model:20s}: {elapsed:.2f}s")

Typical results on a 4-core machine:

Model	I/O-bound (20 tasks)	CPU-bound (4 tasks)
Sequential	10.0s	8.0s
Threading	0.5s	9.5s (worse!)
asyncio	0.5s	N/A
Multiprocessing	0.6s	2.2s

Decision Framework

Use this table to pick the right model:

Workload	Best Choice	Why
Many HTTP requests	asyncio	Lowest overhead, scales to thousands
File I/O with legacy libs	threading	Works with blocking libraries
Image/video processing	multiprocessing	True CPU parallelism
Data crunching (numpy)	multiprocessing or threading	NumPy releases the GIL for C operations
Mixed I/O + CPU	asyncio + ProcessPoolExecutor	Combine both models

Combining Models

You can mix asyncio with multiprocessing for workloads that need both:

import asyncio
from concurrent.futures import ProcessPoolExecutor
import math

def cpu_heavy(limit: int) -> int:
    return sum(1 for n in range(2, limit)
               if all(n % i != 0 for i in range(2, int(math.sqrt(n)) + 1)))

async def main() -> None:
    loop = asyncio.get_running_loop()
    # Offload CPU work to a process pool while keeping the event loop free
    with ProcessPoolExecutor(max_workers=4) as pool:
        futures = [loop.run_in_executor(pool, cpu_heavy, 50_000) for _ in range(4)]
        results = await asyncio.gather(*futures)
    print(f"Primes found: {results}")

asyncio.run(main())

This pattern keeps your event loop responsive for I/O while offloading heavy computation to separate processes.

Common Mistakes to Avoid

Creating too many threads. Each OS thread consumes memory (typically 8 MB stack). A thousand threads uses 8 GB just for stacks. Use thread pools with sensible limits.

Sharing mutable state without locks. Threads share memory, so concurrent writes cause race conditions:

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100_000):
        with lock:  # Always protect shared state
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # 400000, correct with lock

Forgetting if __name__ == "__main__" with multiprocessing. On macOS and Windows, multiprocessing uses spawn by default. Without the guard, child processes re-import the module and spawn more children infinitely.

Blocking the asyncio event loop. Any synchronous call longer than a few milliseconds should be offloaded to a thread or process executor.

Wrapping Up

Python’s concurrency landscape boils down to three clear choices. Use asyncio when you have many I/O operations and can use async-compatible libraries. Use threading when you need I/O concurrency but are stuck with blocking libraries. Use multiprocessing when you need true CPU parallelism. For complex workloads, combine asyncio’s event loop with ProcessPoolExecutor to get the best of both worlds. The key is understanding the GIL: it makes threading useless for CPU work, but leaves I/O-bound tasks wide open for concurrent execution. Profile first, then pick the model that matches your bottleneck.