Concurrency & performance
asyncio, threading, multiprocessing, and profiling-friendly performance patterns.
Build a Python Performance Profiler That Generates Readable Reports
Use cProfile and pstats to profile Python functions and print a sorted performance report showing the top time-consuming calls.
import cProfile
import pstats
import io
from pathlib import Path
def slow_function():
total = 0
for i in range(500_000):
total += i ** 2
return total
def fast_function():
total = sum(i * i for i in range(500_000))
return total
def profile_functions():
profiler = cProfile.Profile()
…How to Memoize Async Functions with lru_cache in Python
Cache async function results with functools.lru_cache to avoid repeated expensive awaits, cutting total execution from ~0.4s to ~0.2s in this example.
from functools import lru_cache
import asyncio
@lru_cache(maxsize=128)
async def fetch_data(user_id: int) -> str:
# Simulate expensive async operation
await asyncio.sleep(0.1)
return f"Data for user {user_id}"
async def main():
start = asyncio.get_event_loop().time()
# First calls (miss cach…How to Memoize Pure Functions with functools.lru_cache in Python
Use functools.lru_cache to memoize a pure Fibonacci function and avoid recomputing repeated values.
from functools import lru_cache
@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
"""Return the nth Fibonacci number (0-indexed) using memoization."""
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
if __name__ == "__main__":
for i in range(10):
print(f"fibonacci({…How to Profile CPU Hot Path in Python with cProfile and sort_stats cumtime
Profile a Python function's CPU usage by running cProfile, sorting stats by cumulative time, and printing a readable report to stdout.
import cProfile
import pstats
import io
def slow_function():
total = 0
for i in range(100_000):
total += i * i
return total
def fast_function():
return sum(i for i in range(100))
def main():
slow_function()
fast_function()
if __name__ == "__main__":
profiler = cProfile.Profi…How to Run Blocking Code in an Executor with asyncio in Python
This code runs blocking functions concurrently without stalling the event loop by offloading them to thread pool executors via asyncio.
import asyncio
import time
def blocking_task(name: str, duration: float) -> str:
"""Simulate a blocking operation."""
time.sleep(duration)
return f"Finished {name} after {duration}s"
async def main() -> None:
loop = asyncio.get_running_loop()
results = await asyncio.gather(
loop.run_in_…How to Use ProcessPoolExecutor for CPU Parallel Map in Python
Run a function over a sequence of inputs in parallel across multiple CPU cores with ProcessPoolExecutor.map.
from concurrent.futures import ProcessPoolExecutor
import math
def compute_square(num):
return num * num
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 = rang…How to Use ThreadPoolExecutor in Python for Parallel Processing
Use ThreadPoolExecutor with executor.map to run a function over many inputs concurrently and collect ordered results.
def worker(item):
return item * item
if __name__ == "__main__":
from concurrent.futures import ThreadPoolExecutor
numbers = list(range(1, 11))
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(worker, numbers))
print("Input: ", numbers)
print("Results:", …How to Use functools.cache for Unbounded Memoization in Python
Speed up repeated recursive calls by memoizing function results with Python's built-in functools.cache decorator.
```python
import functools
import time
@functools.cache
def fib(n):
if n < 2:
return n
return fib(n - 1) + fib(n - 2)
if __name__ == "__main__":
start = time.perf_counter()
result = fib(30)
elapsed = time.perf_counter() - start
print(f"fib(30) = {result}")
print(f"computed in {…How to Use multiprocessing Pool map and starmap in Python
Parallelize functions over iterables with Pool.map, and unpack multiple arguments via Pool.starmap.
from multiprocessing import Pool
def square(x):
return x * x
def add_and_multiply(a, b, c):
return (a + b) * c
if __name__ == "__main__":
numbers = [1, 2, 3, 4, 5]
with Pool(processes=2) as pool:
squares = pool.map(square, numbers)
print(f"squares: {squares}")
starmap_arg…How to Use pool.map for CPU-Bound Tasks in Python
Distribute CPU-intensive functions across processes with multiprocessing.Pool.map and measure the performance gain.
from multiprocessing import Pool
import time
def cpu_bound_task(n):
"""Mock CPU-bound work: compute sum of squares."""
total = 0
for i in range(n):
total += i * i
return total
if __name__ == "__main__":
numbers = [10_000_000, 12_000_000, 8_000_000, 15_000_000]
start = time.perf_count…How to Use threading.RLock in Python
Demonstrates threading.RLock, a reentrant lock that allows the same thread to acquire it multiple times without deadlocking — essential for recursive functions sharing state across threads.
import threading
import time
lock = threading.RLock()
shared_counter = 0
def recursive_increment(value, depth):
global shared_counter
with lock:
shared_counter += 1
print(f"Depth {depth}: counter = {shared_counter}")
if depth > 1:
recursive_increment(value, depth - 1)
def…How to Vectorize a Function with a Pure Python Fallback
Create a decorator that calls a scalar function directly for a single value and routes list inputs to a pure-Python fallback for vectorized processing without NumPy.
import math
def fallback_vectorize(func, fallback=None):
"""Vectorize a scalar function with a pure-Python fallback for lists."""
if fallback is None:
fallback = lambda x: [func(i) for i in x]
def wrapped(*args):
if len(args) == 1 and isinstance(args[0], (list, tuple)):
retur…How to set a timeout with asyncio.wait_for in Python
Use asyncio.wait_for to bound an async function with a timeout, catching TimeoutError when it exceeds the limit.
import asyncio
async def slow_task():
await asyncio.sleep(3)
return "finished"
async def main():
try:
result = await asyncio.wait_for(slow_task(), timeout=1)
print(result)
except asyncio.TimeoutError:
print("Task timed out")
if __name__ == "__main__":
asyncio.run(main())How to use ThreadPoolExecutor for concurrent tasks in Python
Run blocking functions in parallel with ThreadPoolExecutor and as_completed, cutting total runtime from 5 sequential sleeps to about 1 second.
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def fetch_data(item):
"""Simulate a slow operation with a fixed delay."""
time.sleep(0.2)
return item * 2
def main():
items = [1, 2, 3, 4, 5]
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=3) as ex…Browse by section
Each section groups closely related Python snippets.
Concurrency & performance — Python code examples
What you will find here
This page collects concurrency & performance snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.