Concurrency & performance
asyncio, threading, multiprocessing, and profiling-friendly performance patterns.
How to Use asyncio Lock to Protect a Shared Counter in Python
This code demonstrates how to use an asyncio.Lock to safely increment a shared counter from multiple concurrent coroutines.
import asyncio
async def increment(counter, lock, increments):
for _ in range(increments):
async with lock:
counter[0] += 1
async def main():
counter = [0]
lock = asyncio.Lock()
tasks = [
increment(counter, lock, 1000)
for _ in range(5)
]
await asyncio.gath…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 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 Use threading.local for Per-Thread Data in Python
Use threading.local to keep thread-specific data — each thread gets its own copy of the attribute, so values don't leak between threads.
import threading
import time
local_storage = threading.local()
def worker(name):
local_storage.name = name
time.sleep(0.1)
print(f"Thread {threading.current_thread().name}: {local_storage.name}")
if __name__ == "__main__":
threads = []
for i in range(3):
t = threading.Thread(target=worke…Limit Concurrency with asyncio.Semaphore in Python
Use asyncio.Semaphore to cap how many async tasks run at once, throttling a batch of coroutines to a set concurrency limit.
import asyncio
import random
async def fetch_data(i: int, semaphore: asyncio.Semaphore) -> str:
async with semaphore:
print(f"Task {i} starts")
await asyncio.sleep(random.uniform(0.1, 0.5))
print(f"Task {i} finishes")
return f"Result {i}"
async def main() -> None:
semaphore …Mocking Trio's open_nursery and spawn with asyncio.TaskGroup
Show how to mock Trio's nursery pattern using Python's asyncio.TaskGroup to simulate task spawning and completion.
import asyncio
class MockSpawner:
async def spawn(self, nursery):
print("Spawning mock task...")
await asyncio.sleep(1)
print("Mock task completed")
async def open_nursery():
async with asyncio.TaskGroup() as nursery:
mock = MockSpawner()
nursery.create_task(mock.spawn…Thread Pool Map for IO Bound Tasks in Python
Run IO-bound mock tasks concurrently with ThreadPoolExecutor.map and measure total elapsed time in Python.
import concurrent.futures
import time
from pathlib import Path
def mock_io_task(filename):
"""Simulate an IO-bound task by creating a small file and measuring its latency."""
path = Path(filename)
path.write_text("data")
time.sleep(0.1) # Simulate slow disk/network
return f"{filename} written in …asyncio Condition wait notify pattern in Python
Coordinate coroutines with asyncio.Condition: workers wait for notifications and the main task notifies one or all of them.
import asyncio
async def worker(condition, name):
async with condition:
print(f"{name} waiting...")
await condition.wait()
print(f"{name} notified!")
async def main():
condition = asyncio.Condition()
tasks = [asyncio.create_task(worker(condition, f"worker-{i}")) for i in range(3…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.