Parallel and Concurrent Processing in Python: threading, multiprocessing and asyncio

Parallel and Concurrent Processing in Python: threading, multiprocessing and asyncio

Python offers three main approaches for concurrent or parallel execution. The right choice depends on your task type: IO-bound (waiting on disk, network, APIs) or CPU-bound (heavy computation).

1. The GIL and when it matters

The Global Interpreter Lock (GIL) in CPython allows only one thread to execute Python bytecode at a time. This means:

• CPU-bound: the GIL prevents real parallelism between threads → use multiprocessing.
• IO-bound: threads release the GIL while waiting for IO → threading works well.

IO-bound tasks (network, disk)    → threading or asyncio
CPU-bound tasks (computation)     → multiprocessing or ProcessPoolExecutor
Many concurrent IO operations     → asyncio (single thread, very efficient)

2. threading: threads for IO-bound tasks

import threading
import time
import requests

def download(url, results, index):
    r = requests.get(url, timeout=10)
    results[index] = len(r.content)
    print(f"[{index}] {url} → {len(r.content)} bytes")

urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

results = [None] * len(urls)
threads = []

start = time.perf_counter()
for i, url in enumerate(urls):
    t = threading.Thread(target=download, args=(url, results, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Total: {time.perf_counter() - start:.2f} s")  # ~1 s instead of ~3 s

Thread with Lock (shared state)

import threading

counter = 0
lock = threading.Lock()

def increment(n):
    global counter
    for _ in range(n):
        with lock:
            counter += 1

threads = [threading.Thread(target=increment, args=(10000,)) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Final counter: {counter}")  # 50000

3. concurrent.futures.ThreadPoolExecutor

The simplest high-level API for threading:

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch(url):
    r = requests.get(url, timeout=10)
    return url, r.status_code, len(r.content)

urls = [f"https://httpbin.org/anything/{i}" for i in range(8)]

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url, status, size = future.result()
        print(f"{status} | {size:6d} bytes | {url}")

map() for ordered results

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as ex:
    results = list(ex.map(lambda n: n * n, range(10)))

print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

4. multiprocessing: processes for CPU-bound tasks

import multiprocessing
import time

def is_prime(n):
    if n < 2: return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0: return False
    return True

numbers = list(range(100_000, 100_200))

# Sequential
start = time.perf_counter()
seq_results = [is_prime(n) for n in numbers]
print(f"Sequential: {time.perf_counter() - start:.3f} s")

# Parallel Pool
start = time.perf_counter()
with multiprocessing.Pool() as pool:
    par_results = pool.map(is_prime, numbers)
print(f"Parallel  : {time.perf_counter() - start:.3f} s")

primes = [n for n, p in zip(numbers, par_results) if p]
print(f"Primes found: {len(primes)}")

Pool.starmap for multiple arguments

from multiprocessing import Pool

def power(base, exp):
    return base ** exp

pairs = [(2, 10), (3, 8), (5, 6), (7, 5)]

with Pool(processes=4) as pool:
    results = pool.starmap(power, pairs)

print(results)  # [1024, 6561, 15625, 16807]

5. concurrent.futures.ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor, as_completed
import math

def factorial_info(n):
    return n, math.factorial(n)

numbers = [1000, 2000, 3000, 4000, 5000]

with ProcessPoolExecutor() as executor:
    futures = {executor.submit(factorial_info, n): n for n in numbers}
    for future in as_completed(futures):
        n, result = future.result()
        print(f"{n}! has {len(str(result))} digits")

6. asyncio: async concurrency for IO

asyncio uses a single thread and coroutines to handle thousands of concurrent IO operations:

import asyncio
import aiohttp   # pip install aiohttp
import time

async def fetch(session, url):
    async with session.get(url) as r:
        content = await r.read()
        return url, r.status, len(content)

async def main():
    urls = [f"https://httpbin.org/anything/{i}" for i in range(10)]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

    for url, status, size in results:
        print(f"{status} | {size:5d} bytes | {url}")

start = time.perf_counter()
asyncio.run(main())
print(f"Total: {time.perf_counter() - start:.2f} s")

asyncio.gather vs asyncio.wait

import asyncio

async def task(name, seconds):
    await asyncio.sleep(seconds)
    return name

async def demo():
    # gather: wait for all, results in order
    results = await asyncio.gather(
        task("A", 1), task("B", 2), task("C", 1),
    )
    print(f"Results: {results}")

    # wait: fine-grained control
    tasks = [asyncio.create_task(task(f"T{i}", i)) for i in range(1, 4)]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    for t in done:
        print(f"First done: {t.result()}")
    for t in pending:
        t.cancel()

asyncio.run(demo())

7. asyncio with aiofiles

import asyncio
import aiofiles   # pip install aiofiles

async def read_file(path):
    async with aiofiles.open(path, 'r', encoding='utf-8') as f:
        content = await f.read()
    return path, len(content)

async def process_files():
    paths = ["file1.txt", "file2.txt", "file3.txt"]
    results = await asyncio.gather(*[read_file(p) for p in paths], return_exceptions=True)
    for res in results:
        if isinstance(res, Exception):
            print(f"Error: {res}")
        else:
            path, size = res
            print(f"{path}: {size} bytes")

asyncio.run(process_files())

8. Producer-consumer queue

import asyncio

async def producer(queue: asyncio.Queue, items):
    for item in items:
        await queue.put(item)
        print(f"Produced: {item}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # sentinel

async def consumer(queue: asyncio.Queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"  Consumed: {item}")
        await asyncio.sleep(0.2)

async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(producer(queue, range(10)), consumer(queue))

asyncio.run(main())

9. Error handling in parallel tasks

from concurrent.futures import ThreadPoolExecutor, as_completed

def might_fail(n):
    if n == 3:
        raise ValueError(f"Error on {n}")
    return n * 2

with ThreadPoolExecutor(max_workers=4) as ex:
    futures = {ex.submit(might_fail, i): i for i in range(6)}
    for future in as_completed(futures):
        try:
            print(f"OK: {future.result()}")
        except Exception as e:
            print(f"ERROR: {e}")

10. Quick decision guide

11. Run blocking code inside asyncio

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_task(n):
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as executor:
        result = await loop.run_in_executor(executor, cpu_task, 1_000_000)
    print(f"Result: {result}")

asyncio.run(main())

12. Best practices

• Don't share mutable state between processes; use Queue or Manager if needed.
• Limit pool size to os.cpu_count() for CPU-bound; too many workers hurt rather than help.
• asyncio is not multithreaded: await yields control, but synchronous code inside coroutines still blocks.
• Use as_completed when you want to process results as they arrive, not after all finish.
• Cancel pending tasks in asyncio with task.cancel() to avoid resource leaks.
• Profile first: measure where the bottleneck actually is before parallelizing.

Related conversions

Frequent conversions across the catalogue:

• JPG to PNG
• MP4 to MP3
• PDF to DOCX
• PNG to JPG
• MP3 to WAV
• DOCX to PDF