Reference library

Concurrency & performance

asyncio, threading, multiprocessing, and profiling-friendly performance patterns.

8 matches
Concurrency & performance medium

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.

profiling cprofile pstats
Python
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()
  …
44 0 Open
Concurrency & performance easy

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.

asyncio lru_cache memoization
Python
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…
12 0 Open
Concurrency & performance easy

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.

lru-cache memoization functools
Python
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({…
15 0 Open
Concurrency & performance medium

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.

asyncio executor concurrency
Python
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_…
13 0 Open
Concurrency & performance medium

How to Use multiprocessing Pool map and starmap in Python

Parallelize functions over iterables with Pool.map, and unpack multiple arguments via Pool.starmap.

multiprocessing parallelism pool
Python
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…
14 0 Open
Concurrency & performance easy

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.

multiprocessing pool cpu-bound
Python
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…
11 0 Open
Concurrency & performance medium

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.

threading rlock concurrency
Python
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…
14 0 Open
Concurrency & performance easy

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.

concurrency threadpoolexecutor parallel
Python
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…
14 0 Open

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.