Python Concurrency — A Comprehensive Guide

Table of Contents

1. Introduction: Concurrency vs Parallelism
2. The Global Interpreter Lock (GIL)
3. The threading Module
4. The multiprocessing Module
5. The asyncio Module
6. The concurrent.futures Module
7. Synchronization Primitives
8. Queues and Inter-Task Communication
9. Choosing the Right Tool
10. Common Pitfalls and Best Practices
11. Advanced Topics
12. Summary Cheat Sheet

---

1. Introduction: Concurrency vs Parallelism

Concurrency is about dealing with multiple tasks at once — structuring a program so multiple tasks can make progress without necessarily running at the exact same instant. Parallelism is about doing multiple tasks at once — actually executing them simultaneously on multiple CPU cores.

Concept	Definition	Example
Concurrency	Multiple tasks in progress, interleaved	Single core switching between tasks
Parallelism	Multiple tasks executing simultaneously	Multiple cores running tasks at the same time
Sequential	One task at a time, in order	No overlap at all

A useful mental model: concurrency is a structure (how you organize the work), parallelism is an execution property (how the hardware actually runs it). You can have concurrency without parallelism (e.g., asyncio on a single core), and you can have parallelism without much concurrency in the “many small tasks” sense (e.g., a single big matrix multiplication split across cores).

Python offers three main concurrency models:

• Threading — good for I/O-bound tasks, limited by the GIL for CPU-bound work.
• Multiprocessing — good for CPU-bound tasks, uses separate processes to bypass the GIL.
• Asyncio — good for I/O-bound tasks with very high concurrency (thousands of connections), single-threaded cooperative multitasking.

---

2. The Global Interpreter Lock (GIL)

The GIL is a mutex that protects access to Python objects, preventing multiple native threads from executing Python bytecode at once in the same process. This exists because CPython’s memory management (reference counting) is not thread-safe by default.

Why the GIL exists

• Reference counting for garbage collection needs to be atomic; without the GIL, race conditions could corrupt memory.
• It simplifies the implementation of CPython and makes single-threaded code fast.
• It makes C extensions easier to write since most don’t need to worry about thread safety.

Practical implications

• CPU-bound multi-threaded code does NOT get faster with more threads in standard CPython, because only one thread executes Python bytecode at a time.
• I/O-bound code benefits from threads because the GIL is released during blocking I/O operations (file reads, network calls, time.sleep, etc.).
• The GIL is released periodically (every ~5ms via sys.setswitchinterval(), configurable) to allow other threads to run even during CPU-bound work, allowing some interleaving but not true parallel computation.

import sys
print(sys.getswitchinterval())  # Default: 0.005 seconds
sys.setswitchinterval(0.001)     # Switch more frequently

The Future: Free-Threaded Python (PEP 703)

As of Python 3.13, an experimental free-threaded build (python3.13t) exists that removes the GIL entirely. This is a major ongoing effort (originally proposed by Sam Gross, “nogil”). It comes with tradeoffs:

• Single-threaded performance may be slightly slower due to per-object locking overhead and biased reference counting.
• Many C extensions are not yet compatible.
• It is not yet the default build; it must be explicitly compiled or installed.

Until free-threaded Python becomes mainstream and universally supported by the ecosystem, the practical guidance below (threads for I/O, processes for CPU) remains the standard approach.

Demonstrating the GIL’s effect

import threading
import time

def cpu_bound_task(n):
    count = 0
    for i in range(n):
        count += i * i
    return count

def run_single_threaded():
    start = time.perf_counter()
    cpu_bound_task(50_000_000)
    cpu_bound_task(50_000_000)
    print(f"Single-threaded: {time.perf_counter() - start:.2f}s")

def run_multi_threaded():
    start = time.perf_counter()
    t1 = threading.Thread(target=cpu_bound_task, args=(50_000_000,))
    t2 = threading.Thread(target=cpu_bound_task, args=(50_000_000,))
    t1.start(); t2.start()
    t1.join(); t2.join()
    print(f"Multi-threaded: {time.perf_counter() - start:.2f}s")

run_single_threaded()
run_multi_threaded()
# On standard CPython, multi-threaded is NOT meaningfully faster for CPU-bound work

---

3. The threading Module

Threads are lightweight and share the same memory space within a process. Best suited for I/O-bound tasks: network requests, file I/O, database queries, waiting on external resources.

Basic thread creation

import threading
import time

def worker(name, delay):
    print(f"Thread {name} starting")
    time.sleep(delay)
    print(f"Thread {name} finished")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(f"Worker-{i}", 1))
    threads.append(t)
    t.start()

for t in threads:
    t.join()  # Wait for all threads to complete
print("All threads done")

Subclassing Thread

class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(f"Running {self.name}")

t = MyThread("custom-thread")
t.start()
t.join()

Daemon threads

Daemon threads are killed automatically when the main program exits, unlike normal threads which block program exit until they finish.

t = threading.Thread(target=worker, args=("daemon", 10), daemon=True)
t.start()
# Program can exit even if this thread is still running

Thread-local data

threading.local() gives each thread its own isolated storage — useful for per-thread state like database connections.

local_data = threading.local()

def process():
    local_data.value = threading.current_thread().name
    print(local_data.value)

for i in range(3):
    threading.Thread(target=process).start()

ThreadPoolExecutor (preview — covered fully in Section 6)

from concurrent.futures import ThreadPoolExecutor

def fetch(url):
    # simulate network call
    time.sleep(0.5)
    return f"data from {url}"

urls = ["http://a.com", "http://b.com", "http://c.com"]
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(fetch, urls))
print(results)

Race conditions and why locks matter

counter = 0
lock = threading.Lock()

def increment_unsafe():
    global counter
    for _ in range(100_000):
        counter += 1  # NOT atomic! read-modify-write race condition

def increment_safe():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1  # Atomic thanks to the lock

threads = [threading.Thread(target=increment_unsafe) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)  # Likely NOT 400,000 due to race conditions

---

4. The multiprocessing Module

Processes have separate memory spaces, bypassing the GIL entirely. Ideal for CPU-bound work: numerical computation, image/video processing, data transformation, simulations.

Basic process creation

import multiprocessing as mp

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

if __name__ == "__main__":  # REQUIRED on Windows/macOS (spawn method)
    p = mp.Process(target=cpu_task, args=(10_000_000,))
    p.start()
    p.join()

Pool for parallel mapping

import multiprocessing as mp

def square(x):
    return x * x

if __name__ == "__main__":
    with mp.Pool(processes=4) as pool:
        results = pool.map(square, range(20))
        print(results)

        # Other Pool methods:
        result_async = pool.apply_async(square, (10,))
        print(result_async.get())

        # imap: lazy, preserves order
        for r in pool.imap(square, range(5)):
            print(r)

        # imap_unordered: lazy, order not guaranteed, often faster
        for r in pool.imap_unordered(square, range(5)):
            print(r)

Sharing data between processes

Because processes don’t share memory, you need explicit mechanisms:

from multiprocessing import Process, Value, Array, Manager

def increment(shared_val, shared_arr):
    with shared_val.get_lock():
        shared_val.value += 1
    shared_arr[0] += 10

if __name__ == "__main__":
    val = Value('i', 0)          # shared integer
    arr = Array('i', [0, 0, 0])  # shared array

    processes = [Process(target=increment, args=(val, arr)) for _ in range(4)]
    for p in processes: p.start()
    for p in processes: p.join()
    print(val.value, arr[:])

    # Manager for more complex shared objects (dict, list)
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()
        shared_dict["key"] = "value"
        shared_list.append(1)

Inter-process communication: Pipes and Queues

from multiprocessing import Process, Pipe, Queue

def pipe_worker(conn):
    conn.send("hello from child")
    conn.close()

def queue_worker(q):
    q.put("hello from child via queue")

if __name__ == "__main__":
    # Pipe: two-way, for two processes
    parent_conn, child_conn = Pipe()
    p = Process(target=pipe_worker, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    p.join()

    # Queue: many-to-many, process-safe
    q = Queue()
    p2 = Process(target=queue_worker, args=(q,))
    p2.start()
    print(q.get())
    p2.join()

Process start methods

Method	Description	Default OS
fork	Child copies parent’s memory (fast, but can be unsafe with threads)	Linux, macOS (pre-3.8)
spawn	Fresh Python interpreter started, imports module cleanly	Windows, macOS (3.8+)
forkserver	Server process forks new workers, avoiding some fork pitfalls	Available on Unix

import multiprocessing as mp
mp.set_start_method('spawn')  # must be called once, before creating processes

Costs of multiprocessing

• Serialization overhead: Data passed between processes must be pickled/unpickled.
• Memory overhead: Each process has its own Python interpreter and memory copy.
• Startup latency: Creating a process is slower than creating a thread.
• Best used for coarse-grained parallelism (large chunks of work) rather than many tiny tasks.

---

5. The asyncio Module

asyncio provides single-threaded, single-process cooperative multitasking using an event loop. It excels at I/O-bound workloads with very high concurrency (e.g., thousands of simultaneous network connections) with much lower overhead than threads.

Core concepts

• Coroutine: a function defined with async def that can be paused and resumed.
• Event loop: the engine that schedules and runs coroutines, callbacks, and I/O.
• Task: a wrapper around a coroutine that schedules it to run on the event loop.
• Future: a low-level awaitable representing an eventual result.
• await: pauses the coroutine until the awaited operation completes, yielding control back to the event loop.

Basic example

import asyncio

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)

async def main():
    print("started")
    await say_after(1, "hello")
    await say_after(1, "world")
    print("finished")  # Total: ~2 seconds (sequential awaits)

asyncio.run(main())

Running tasks concurrently

import asyncio

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)
    return message

async def main():
    # Sequential: ~2 seconds total
    # await say_after(1, "hello")
    # await say_after(1, "world")

    # Concurrent: ~1 second total
    task1 = asyncio.create_task(say_after(1, "hello"))
    task2 = asyncio.create_task(say_after(1, "world"))
    await task1
    await task2

asyncio.run(main())

asyncio.gather for running many coroutines

import asyncio

async def fetch_data(id):
    await asyncio.sleep(1)
    return f"data-{id}"

async def main():
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3),
    )
    print(results)  # ['data-1', 'data-2', 'data-3'] — all in ~1 second

asyncio.run(main())

asyncio.TaskGroup (Python 3.11+) — structured concurrency

import asyncio

async def fetch_data(id):
    await asyncio.sleep(1)
    if id == 2:
        raise ValueError("failed on id 2")
    return f"data-{id}"

async def main():
    results = []
    try:
        async with asyncio.TaskGroup() as tg:
            for i in range(3):
                tg.create_task(fetch_data(i))
    except* ValueError as eg:
        print(f"Caught exception group: {eg.exceptions}")

asyncio.run(main())

TaskGroup automatically cancels sibling tasks if one fails, and properly propagates exceptions — much safer than manually managing a list of tasks.

Timeouts and cancellation

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2)
    except asyncio.TimeoutError:
        print("Operation timed out")

    # Python 3.11+ timeout context manager
    try:
        async with asyncio.timeout(2):
            await slow_operation()
    except TimeoutError:
        print("Timed out via asyncio.timeout")

asyncio.run(main())

Async context managers and iterators

class AsyncResource:
    async def __aenter__(self):
        print("acquiring resource")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("releasing resource")
        await asyncio.sleep(0.1)

async def main():
    async with AsyncResource() as res:
        print("using resource")

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.i = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.i >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.i += 1
        return self.i

async def iterate():
    async for num in AsyncCounter(3):
        print(num)

asyncio.run(main())
asyncio.run(iterate())

Running blocking / CPU-bound code inside asyncio

Blocking calls (like time.sleep or CPU-heavy loops) will freeze the entire event loop, blocking all other coroutines. Use executors to offload this work.

import asyncio
import time

def blocking_io():
    time.sleep(2)  # simulate blocking I/O — this would freeze the event loop
    return "blocking result"

async def main():
    loop = asyncio.get_running_loop()
    # Offload to a thread pool
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

    # Or use asyncio.to_thread (Python 3.9+, simpler API)
    result2 = await asyncio.to_thread(blocking_io)
    print(result2)

asyncio.run(main())

Async generators and streaming

async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for i in async_range(5):
        print(i)

asyncio.run(main())

asyncio.Queue for producer-consumer patterns

import asyncio
import random

async def producer(queue, n):
    for i in range(n):
        item = f"item-{i}"
        await queue.put(item)
        print(f"Produced {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
    await queue.put(None)  # sentinel to signal completion

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

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

asyncio.run(main())

---

6. The concurrent.futures Module

Provides a high-level, unified interface for both threads and processes, abstracting away much of the manual bookkeeping.

ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch(url):
    time.sleep(1)
    return f"result from {url}"

urls = [f"http://site{i}.com" for i in range(5)]

with ThreadPoolExecutor(max_workers=3) as executor:
    # Method 1: map (preserves order, blocks until all done)
    for result in executor.map(fetch, urls):
        print(result)

    # Method 2: submit + as_completed (process results as they finish)
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f"{url}: {data}")
        except Exception as exc:
            print(f"{url} generated an exception: {exc}")

ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor
import math

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

if __name__ == "__main__":
    numbers = list(range(100_000, 100_100))
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(is_prime, numbers))
    primes = [n for n, p in zip(numbers, results) if p]
    print(primes)

Handling exceptions in futures

from concurrent.futures import ThreadPoolExecutor

def risky(x):
    if x == 3:
        raise ValueError("bad value")
    return x * 2

with ThreadPoolExecutor() as executor:
    futures = [executor.submit(risky, i) for i in range(5)]
    for f in futures:
        try:
            print(f.result())
        except ValueError as e:
            print(f"Caught: {e}")

Callbacks with add_done_callback

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    time.sleep(1)
    return n * n

def callback(future):
    print(f"Task completed with result: {future.result()}")

with ThreadPoolExecutor() as executor:
    future = executor.submit(task, 5)
    future.add_done_callback(callback)

---

7. Synchronization Primitives

When multiple threads (or processes) access shared state, you need synchronization to prevent race conditions, deadlocks, and data corruption.

Lock (mutex)

Ensures only one thread accesses a critical section at a time.

import threading

lock = threading.Lock()
shared_resource = []

def add_item(item):
    with lock:  # equivalent to lock.acquire() / lock.release()
        shared_resource.append(item)

RLock (reentrant lock)

Allows the same thread to acquire the lock multiple times without deadlocking itself — useful for recursive functions.

import threading

rlock = threading.RLock()

def recursive_function(n):
    with rlock:
        if n > 0:
            print(n)
            recursive_function(n - 1)  # Same thread re-acquires lock, no deadlock

Semaphore

Limits the number of threads that can access a resource simultaneously — useful for connection pools, rate-limiting.

import threading
import time

semaphore = threading.Semaphore(3)  # only 3 threads at once

def limited_resource(name):
    with semaphore:
        print(f"{name} acquired the resource")
        time.sleep(1)
        print(f"{name} released the resource")

for i in range(10):
    threading.Thread(target=limited_resource, args=(f"Thread-{i}",)).start()

BoundedSemaphore

Like Semaphore but raises an error if released more times than acquired — catches programming bugs.

Event

A simple flag threads can wait on, used for one-to-many signaling.

import threading
import time

event = threading.Event()

def waiter():
    print("Waiting for event...")
    event.wait()  # blocks until event.set() is called
    print("Event received, proceeding!")

def setter():
    time.sleep(2)
    print("Setting event")
    event.set()

threading.Thread(target=waiter).start()
threading.Thread(target=setter).start()

Condition

Combines a lock with the ability to wait for a notification — used for producer-consumer patterns.

import threading

condition = threading.Condition()
items = []

def consumer():
    with condition:
        while not items:
            condition.wait()  # releases lock, waits for notify
        item = items.pop(0)
        print(f"Consumed {item}")

def producer():
    with condition:
        items.append("item")
        print("Produced item")
        condition.notify()  # wakes up one waiting thread

threading.Thread(target=consumer).start()
threading.Thread(target=producer).start()

Barrier

Makes a group of threads wait until all of them reach the barrier point.

import threading

barrier = threading.Barrier(3)

def worker(name):
    print(f"{name} waiting at barrier")
    barrier.wait()  # blocks until all 3 threads call wait()
    print(f"{name} passed the barrier")

for i in range(3):
    threading.Thread(target=worker, args=(f"Worker-{i}",)).start()

Deadlocks: how they happen and how to avoid them

A deadlock occurs when two or more threads wait on each other indefinitely.

import threading

lock_a = threading.Lock()
lock_b = threading.Lock()

def thread_1():
    with lock_a:
        # ... does some work
        with lock_b:  # thread_2 might hold lock_b already
            pass

def thread_2():
    with lock_b:
        with lock_a:  # thread_1 might hold lock_a already — DEADLOCK
            pass

Prevention strategies:

• Always acquire locks in the same global order across all threads.
• Use lock.acquire(timeout=...) to avoid waiting forever.
• Minimize the scope/duration of locked sections.
• Prefer higher-level constructs (Queue, concurrent.futures) over manual locking where possible.

asyncio synchronization primitives

asyncio provides analogous primitives designed for coroutines: asyncio.Lock, asyncio.Event, asyncio.Condition, asyncio.Semaphore, asyncio.Barrier (3.11+). These are NOT thread-safe and must only be used within a single event loop.

import asyncio

async def main():
    lock = asyncio.Lock()
    async with lock:
        print("critical section")

asyncio.run(main())

---

8. Queues and Inter-Task Communication

Queues are thread/process-safe data structures for passing data between concurrent units of work — often the safest way to share data instead of relying on locks.

queue.Queue (threading)

import queue
import threading

q = queue.Queue(maxsize=10)

def producer():
    for i in range(5):
        q.put(f"item-{i}")

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Got {item}")
        q.task_done()

threading.Thread(target=producer).start()
t = threading.Thread(target=consumer)
t.start()

q.join()  # wait until all items processed
q.put(None)
t.join()

Queue variants: Queue (FIFO), LifoQueue (stack-like), PriorityQueue (ordered by priority).

multiprocessing.Queue

Same API, but safe to share across processes (uses pipes and locks internally, pickles data).

asyncio.Queue

Same conceptual API but designed for coroutines (await queue.put(), await queue.get()).

---

9. Choosing the Right Tool

Workload Type	Recommended Tool	Why
I/O-bound, few connections	threading	Simple, works well, GIL released during I/O
I/O-bound, thousands of connections	asyncio	Extremely low overhead per task vs. threads
CPU-bound (heavy computation)	multiprocessing / ProcessPoolExecutor	Bypasses GIL, uses multiple cores
Mixed I/O + light CPU work	asyncio + run_in_executor/to_thread	Keeps event loop responsive
Simple parallel “map” over data	concurrent.futures (Thread or Process pool)	High-level, less boilerplate
Need shared mutable state across workers	threading (with locks) or multiprocessing.Manager	Processes need explicit sharing mechanisms

Decision flowchart (in words)

1. Is the task I/O-bound (network, disk, waiting on external systems)?
   • Yes, and you control the whole call stack with async libraries available → use asyncio.
   • Yes, but working with blocking libraries (e.g., older DB drivers) → use threading or ThreadPoolExecutor.
2. Is the task CPU-bound (computation-heavy, tight loops, number crunching)?
   • Yes → use multiprocessing or ProcessPoolExecutor.
   • Consider whether NumPy/Cython/native libraries can release the GIL — if so, threading might still help.
3. Do you need extremely high concurrency (10,000+ simultaneous connections)?
   • Yes → asyncio is essentially required; threads become too expensive (memory/context switching).

---

10. Common Pitfalls and Best Practices

Pitfall 1: Believing threads speed up CPU-bound Python code

As discussed, the GIL prevents this in standard CPython. Use multiprocessing instead.

Pitfall 2: Forgetting if __name__ == "__main__": guard with multiprocessing

On Windows and macOS (spawn method), forgetting this guard causes infinite process spawning or crashes, because the child process re-imports the main module.

# WRONG on Windows/macOS
import multiprocessing as mp

def worker():
    print("working")

p = mp.Process(target=worker)
p.start()  # This line re-executed by every spawned child = chaos

# CORRECT
if __name__ == "__main__":
    p = mp.Process(target=worker)
    p.start()
    p.join()

Pitfall 3: Blocking the asyncio event loop

Calling time.sleep() or doing heavy CPU work directly inside a coroutine freezes the whole event loop.

# WRONG
async def bad():
    time.sleep(5)  # Blocks EVERYTHING

# RIGHT
async def good():
    await asyncio.sleep(5)  # Non-blocking
    # or offload CPU-heavy work:
    await asyncio.to_thread(cpu_heavy_function)

Pitfall 4: Race conditions from unprotected shared state

# WRONG - race condition
counter = 0
def increment():
    global counter
    counter += 1  # Not atomic

# RIGHT
lock = threading.Lock()
def increment_safe():
    global counter
    with lock:
        counter += 1

Pitfall 5: Deadlocks from inconsistent lock ordering

Always acquire multiple locks in a consistent global order across the codebase.

Pitfall 6: Forgetting to join/await tasks (orphaned work)

# Threads: forgetting t.join() may cause the program to exit before work finishes,
# or leave zombie threads if not daemonized.

# Asyncio: forgetting to await a task means "fire and forget" —
# exceptions inside it are silently swallowed unless you keep a reference
# and check it, or use TaskGroup which handles this properly.

Pitfall 7: Using mutable default arguments / shared objects across processes

Passing large, complex, or unpicklable objects to multiprocessing.Process/Pool causes errors or serialization overhead. Keep worker function arguments small and simple (numbers, strings, lists of primitives).

Pitfall 8: Not setting a max size for queues under heavy producer load

Without maxsize, a fast producer can cause unbounded memory growth if the consumer can’t keep up.

Best practices summary

1. Prefer high-level APIs (concurrent.futures, asyncio.TaskGroup) over manual thread/process management.
2. Use context managers (with lock:, with ThreadPoolExecutor() as ex:) to guarantee cleanup.
3. Keep critical sections (locked code) as short as possible.
4. Prefer message passing (queues) over shared mutable state where feasible.
5. Always handle exceptions from futures/tasks — they don’t automatically crash the program but can be silently lost.
6. Profile before parallelizing — concurrency adds complexity and sometimes overhead outweighs the benefit for small tasks.
7. Test on the target OS — fork vs spawn differences can cause bugs that only appear on certain platforms.
8. Use timeouts on locks, joins, and network calls to avoid indefinite hangs.

---

11. Advanced Topics

asyncio and threads together: safely calling back into the event loop

When a background thread needs to interact with asyncio, use run_coroutine_threadsafe:

import asyncio
import threading

async def notify(message):
    print(f"Notified: {message}")

def background_worker(loop):
    import time
    time.sleep(1)
    asyncio.run_coroutine_threadsafe(notify("done from thread"), loop)

async def main():
    loop = asyncio.get_running_loop()
    threading.Thread(target=background_worker, args=(loop,)).start()
    await asyncio.sleep(2)

asyncio.run(main())

Structured concurrency philosophy

Modern concurrent programming favors structured concurrency: concurrent tasks should have clear, bounded lifetimes tied to a scope (like TaskGroup or async with), rather than “fire and forget” tasks that can leak or outlive their intended context. This makes error handling and cancellation predictable.

Combining multiprocessing with asyncio

For CPU-bound work inside an async application, offload to a ProcessPoolExecutor:

import asyncio
from concurrent.futures import ProcessPoolExecutor

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

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy, 10_000_000)
        print(result)

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

multiprocessing.shared_memory (Python 3.8+)

Allows true zero-copy shared memory blocks between processes, avoiding pickling overhead for large data like NumPy arrays.

from multiprocessing import shared_memory
import numpy as np

# Create shared memory block
arr = np.array([1, 2, 3, 4, 5])
shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
shared_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
shared_arr[:] = arr[:]

# In another process, attach to the same block by shm.name
existing_shm = shared_memory.SharedMemory(name=shm.name)
attached_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=existing_shm.buf)
print(attached_arr)

shm.close()
shm.unlink()  # free the shared memory

Async comprehensions

import asyncio

async def get_value(x):
    await asyncio.sleep(0.1)
    return x * 2

async def main():
    results = [await get_value(x) for x in range(5)]
    print(results)

    # Async generator comprehension
    async def gen():
        for i in range(5):
            yield i

    results2 = [x async for x in gen()]
    print(results2)

asyncio.run(main())

uvloop — a faster event loop

A drop-in replacement event loop built on libuv, offering significant performance gains for I/O-heavy asyncio applications (not part of the standard library; install via pip).

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

GIL release in C extensions

Libraries like NumPy release the GIL during heavy numeric operations, meaning threading can provide real speedups for workloads dominated by such calls (not pure Python loops).

---

12. Summary Cheat Sheet

Feature	threading	multiprocessing	asyncio
Parallelism (multi-core)	No (GIL-limited)	Yes	No (single-threaded)
Best for	I/O-bound, few tasks	CPU-bound	I/O-bound, many tasks
Memory	Shared	Separate per process	Shared (single process)
Overhead per unit	Medium	High	Very low
Communication	Shared variables + locks	Pipes, Queues, shared memory	Shared variables (single loop) + queues
Failure isolation	Low (shared memory)	High (separate process)	Low (single process)
Debugging difficulty	Medium-High	Medium	Medium (different mental model)
Requires explicit await/callbacks	No	No	Yes

Quick reference imports

import threading                                  # Thread, Lock, Event, Condition, Semaphore
import multiprocessing as mp                      # Process, Pool, Queue, Value, Array, Manager
import asyncio                                    # coroutines, tasks, event loop
from concurrent.futures import (
    ThreadPoolExecutor, ProcessPoolExecutor, as_completed
)
import queue                                      # Queue, LifoQueue, PriorityQueue (thread-safe)

Golden rules

1. I/O-bound + few tasks → threading.
2. I/O-bound + many tasks → asyncio.
3. CPU-bound → multiprocessing.
4. Prefer concurrent.futures for simple parallel maps over manual thread/process handling.
5. Never call blocking code directly inside a coroutine — offload it.
6. Always guard multiprocessing entry points with if __name__ == "__main__":.
7. Prefer message passing (queues) over shared mutable state.
8. Measure before you optimize — concurrency isn’t free.