Concurrency & performance
asyncio, threading, multiprocessing, and profiling-friendly performance patterns.
How to Convert Data in Parallel with ThreadPoolExecutor in Python
This example demonstrates converting a list of items in parallel using ThreadPoolExecutor, showing performance gains over serial processing.
import time
from concurrent.futures import ThreadPoolExecutor
def convert_data(item):
"""Simulate a CPU/IO-bound conversion task."""
time.sleep(0.05) # simulate work
return item.upper()
if __name__ == "__main__":
items = [f"item_{i}" for i in range(20)]
start = time.perf_counter()
serial_…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 Run an Async Main with asyncio.run in Python
Show the canonical entry point for an asyncio program: define an async main, then launch it with asyncio.run.
import asyncio
async def main():
print("Hello from async main")
await asyncio.sleep(0.1)
print("Done")
if __name__ == "__main__":
asyncio.run(main())How to Send and Receive Messages Between Processes with multiprocessing.Pipe in Python
Use multiprocessing.Pipe to create a two-way connection between two processes, send a message from parent to child, and receive a reply back.
import multiprocessing
def child_process(conn):
"""Receive from parent and send back a response."""
message = conn.recv()
print(f"Child received: {message}")
conn.send("Hello from child!")
if __name__ == "__main__":
parent_conn, child_conn = multiprocessing.Pipe()
process = multiprocessing…How to Signal asyncio Workers to Stop with an Event in Python
Use an asyncio.Event to coordinate graceful shutdown of multiple concurrent worker tasks in Python.
import asyncio
import random
async def worker(name, stop_event):
while not stop_event.is_set():
await asyncio.sleep(random.uniform(0.1, 0.5))
print(f"Worker {name} processing...")
print(f"Worker {name} stopped.")
async def main():
stop_event = asyncio.Event()
workers = [asyncio.create…How to Test HTTPX Async Client Pool Reuse with Mocks in Python
Mock an httpx.AsyncClient to verify connection pool reuse by asserting GET calls share a single client instance across concurrent async requests.
import asyncio
import httpx
from unittest.mock import AsyncMock, patch, Mock
async def fetch_with_pool(client, url, n_reuses=3):
results = []
for i in range(n_reuses):
resp = await client.get(url)
results.append(resp.status_code)
await asyncio.sleep(0) # yield to loop to mimic real us…How to Time Code Performance with timeit in Python
Benchmark two implementations of the same logic using Python's timeit module and compare their execution speeds.
import timeit
# Implementation 1: Using a list comprehension
def list_comprehension_squares(n):
return [i ** 2 for i in range(n)]
# Implementation 2: Using a for loop with append
def loop_squares(n):
result = []
for i in range(n):
result.append(i ** 2)
return result
if __name__ == "__main__"…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 ThreadPoolExecutor.submit() in Python
Exécute des fonctions en parallèle avec ThreadPoolExecutor.submit(), récupère les résultats avec future.result(), et traite plusieurs tâches simultanément en Python standard.
from concurrent.futures import ThreadPoolExecutor
import time
def square(n):
time.sleep(0.1) # Simulate work
return n * n
if __name__ == "__main__":
with ThreadPoolExecutor(max_workers=3) as executor:
future = executor.submit(square, 5)
result = future.result()
print(f"Result: {r…How to Use bisect.insort in Python to Maintain a Sorted List
Insert items into an already sorted list using Python's bisect.insort to keep it sorted efficiently in O(n) time.
import bisect
def maintain_sorted_list():
data = [3, 1, 4, 1, 5, 9, 2, 6]
sorted_list = []
for num in data:
bisect.insort(sorted_list, num)
print("Original data:", data)
print("Sorted list maintained with insort:", sorted_list)
# Insert new values to maintain sorted orde…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 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.Lock to Synchronize a Counter in Python
Safely increment a shared counter across multiple threads using threading.Lock as a mutex to prevent race conditions.
import threading
counter = 0
lock = threading.Lock()
def increment():
global counter
for _ in range(100000):
with lock:
counter += 1
threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"Final counter valu…How to Use uvloop Faster Event Loop
Install uvloop at startup to replace asyncio's default event loop with a faster libuv-based one, with a graceful fallback when it's unavailable.
import asyncio
try:
import uvloop
uvloop.install()
USING_UVLOOP = True
except ImportError:
USING_UVLOOP = False
async def fetch_data(index):
await asyncio.sleep(0.01)
return f"data-{index}"
async def main():
tasks = [fetch_data(i) for i in range(10)]
results = await asyncio.gather(*…How to Validate Data with ThreadPoolExecutor in Python
This code shows how to validate a list of numbers concurrently using ThreadPoolExecutor, dramatically speeding up slow validation tasks by running them in parallel threads.
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
@dataclass
class Result:
is_valid: bool
value: int
def validate(value: int) -> Result:
time.sleep(0.1) # simulate slow validation (API call, DB check)
return Result(is_valid=0 < value < 100, value=value…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 spawn multiple worker processes in Python with multiprocessing.Process
Spawns three separate worker processes using multiprocessing.Process, runs them concurrently, and waits for all to finish before printing a completion message.
import multiprocessing
import time
def worker(name):
print(f"Worker {name} started")
time.sleep(1)
print(f"Worker {name} finished")
return name
if __name__ == "__main__":
processes = []
for i in range(3):
p = multiprocessing.Process(target=worker, args=(i,))
processes.append(p…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…Run Background Tasks with asyncio.create_task in Python
Create background tasks in an asyncio event loop with asyncio.create_task and run them concurrently using asyncio.gather.
import asyncio
import time
async def background_worker(name, duration):
"""Simulates a long-running background task."""
print(f"{name} started at t={time.monotonic():.1f}")
await asyncio.sleep(duration)
print(f"{name} finished at t={time.monotonic():.1f}")
async def main():
print(f"Main starting …Synchronize Threads with a Barrier in Python
Demonstrates using threading.Barrier to synchronize multiple threads at phase boundaries, ensuring all workers wait for each other before proceeding.
import threading
import time
from random import randint
def worker(barrier, worker_id):
for phase in range(3):
time.sleep(randint(1, 3))
print(f"Worker {worker_id} finished phase {phase} at {time.time():.2f}")
barrier.wait()
print(f"Worker {worker_id}: all phases complete")
if __name_…Thread-Safe Producer Consumer Queue in Python
A producer-consumer pattern using thread-safe queue.Queue with two threads, demonstrating safe communication and synchronized task completion.
import queue
import threading
import time
import random
def producer(q, item_count):
for i in range(item_count):
item = random.randint(1, 100)
q.put(item)
print(f"Producer added: {item}")
time.sleep(0.1)
def consumer(q):
while True:
try:
item = q.get(time…Using a Python Generator Instead of a List to Save Memory
Compare a list approach with a generator to stream values lazily, avoiding memory-heavy storage of large sequences.
def fibonacci_generator(limit):
a, b = 0, 1
count = 0
while count < limit:
yield a
a, b = b, a + b
count += 1
def sum_first_n(generator, n):
total = 0
for i, value in enumerate(generator):
if i >= n:
break
total += value
return total
if __…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.