Concurrency Basics

9 examples to get you started with Concurrency - 6 basic and 3 intermediate.

Prerequisites

• Python 3.14.0 installed.
• Read Choosing a Concurrency Model for I/O vs CPU guidance.

Basic Examples

1. Concurrency vs Parallelism

Concurrency interleaves tasks; parallelism runs tasks simultaneously on multiple cores.

# Concurrent: one thread switches while another waits on I/O
# Parallel: two processes on two CPU cores compute at once

• I/O-bound work benefits from concurrency (threads/async).
• CPU-bound work needs parallelism (processes) due to the GIL.
• Async is concurrent but usually single-threaded.

Related: The GIL & Free-Threaded Python - GIL details

2. Thread for I/O Wait

Threads overlap blocking I/O.

import time
from threading import Thread
 
def fetch(_id: int) -> None:
    time.sleep(0.1)  # stand-in for network
    print("done", _id)
 
threads = [Thread(target=fetch, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

• sleep simulates blocking I/O release of the GIL.
• Join ensures all work completes before exit.
• Threads share memory - protect mutable state with locks.

Related: threading - locks and safety

3. Process for CPU Work

Processes bypass the GIL with separate interpreters.

from multiprocessing import Pool
 
def square(n: int) -> int:
    return n * n
 
if __name__ == "__main__":
    with Pool(2) as pool:
        print(pool.map(square, range(4)))

• if __name__ == "__main__" guard required on spawn platforms.
• Processes do not share memory by default.
• Startup cost higher than threads - batch work.

Related: multiprocessing - pools and IPC

4. concurrent.futures ThreadPool

High-level pool API for threads and processes.

from concurrent.futures import ThreadPoolExecutor, as_completed
 
def work(n: int) -> int:
    return n * 2
 
with ThreadPoolExecutor(max_workers=4) as ex:
    futures = [ex.submit(work, i) for i in range(6)]
    for fut in as_completed(futures):
        print(fut.result())

• submit returns Future; map preserves order.
• as_completed yields as each finishes.
• Context manager shuts down workers cleanly.

Related: concurrent.futures - full API

5. queue.Queue Coordination

Producer/consumer decouples stages.

from queue import Queue
from threading import Thread
 
q: Queue[int] = Queue()
 
def producer() -> None:
    for i in range(3):
        q.put(i)
    q.put(-1)  # sentinel
 
def consumer() -> None:
    while True:
        item = q.get()
        if item == -1:
            break
        print("got", item)
 
Thread(target=producer).start()
Thread(target=consumer).start()

• Queue is thread-safe blocking.
• Sentinel ends consumer loop cleanly.
• Prefer join() on queue for production pipelines.

Related: Queues & Producer/Consumer - pipelines

6. Shared State Risk

Unprotected increments race.

import threading
 
counter = 0
 
def inc() -> None:
    global counter
    for _ in range(100_000):
        counter += 1
 
threads = [threading.Thread(target=inc) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # often < 200_000

• counter += 1 is read-modify-write - not atomic in Python bytecode.
• Use threading.Lock or atomic types for shared counters.
• Prefer message passing over shared mutable state.

Related: Concurrency Best Practices - avoid races

Intermediate Examples

7. asyncio Concurrent I/O

Single-threaded event loop with many awaiting tasks.

import asyncio
 
async def fetch(n: int) -> int:
    await asyncio.sleep(0.1)
    return n
 
async def main() -> None:
    results = await asyncio.gather(*(fetch(i) for i in range(3)))
    print(results)
 
asyncio.run(main())

• asyncio suits many concurrent I/O waits.
• Does not parallelize CPU-bound Python on one thread.
• See asyncio section for production patterns.

Related: Asyncio Basics - event loop

8. Subinterpreters Preview

Isolated interpreters reduce GIL contention (3.12+ experimental APIs).

# Conceptual - use interpreters module when enabled for your workload
# Each subinterpreter may own a GIL in free-threaded builds
print("evaluate subinterpreters for CPU isolation in 3.13+")

• Subinterpreters target CPU parallelism without full process overhead.
• APIs evolve - read dedicated page before production use.
• Not a drop-in replacement for all multiprocessing cases.

Related: Subinterpreters - when to adopt

9. Pick Model Checklist

Quick decision before importing a library.

def model_hint(bound: str) -> str:
    if bound == "io":
        return "asyncio or threads"
    if bound == "cpu":
        return "multiprocessing or subprocess"
    return "profile first"

• Profile before switching models - assumptions lie.
• Mixed workloads split I/O async layer + process pool for CPU.
• Document choice in ADR for team alignment.

Related: Choosing a Concurrency Model - full checklist

Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.