Concurrency & performance
asyncio, threading, multiprocessing, and profiling-friendly performance patterns.
Benchmark list.append vs deque.append in Python
Measures and compares the performance of appending to a Python list versus a collections.deque using timeit.repeat, showing best and average timings.
"""Benchmark list.append vs collections.deque.append."""
import timeit
def bench(stmt, setup, repeat=5, number=1_000_000):
times = timeit.repeat(stmt, setup=setup, repeat=repeat, number=number)
return min(times), sum(times) / len(times)
if __name__ == "__main__":
number = 1_000_000
list_best, list_a…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()
…Graceful Shutdown Executor Context Manager in Python
A context manager that starts a background thread and ensures it stops gracefully on exit, handling timeouts and exceptions.
import signal
import threading
import time
from contextlib import contextmanager
@contextmanager
def graceful_shutdown_executor(timeout=5.0):
"""Context manager that runs a task and gracefully stops it on timeout or exception."""
stop_event = threading.Event()
def task():
print("Task started")
…How to Demonstrate the GIL with Python Threads vs Processes
Measure and compare wall-clock time for CPU-bound work using Python threads (limited by the GIL) versus multiprocessing (which bypasses the GIL).
import threading
import multiprocessing
import time
import os
def cpu_heavy(n):
return sum(i * i for i in range(n))
def run_threads(n):
threads = [threading.Thread(target=cpu_heavy, args=(n,)) for _ in range(2)]
start = time.perf_counter()
for t in threads:
t.start()
for t in threads:
…How to Implement a Batch Requests Flush Interval in Python
A simple async batcher that accumulates items and flushes them either when a max batch size is reached or after a time-based flush interval.
import asyncio
from collections import deque
class Batcher:
def __init__(self, flush_interval=0.5, max_batch=5):
self.flush_interval = flush_interval
self.max_batch = max_batch
self.queue = deque()
self.lock = asyncio.Lock()
async def add(self, item):
async with self.l…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 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…Merge K Sorted Lists in Python with heapq
Merge k sorted lists into one sorted list in O(N log k) time using a min-heap of current elements.
import heapq
def merge_k_sorted_lists(lists):
heap = []
for i, lst in enumerate(lists):
if lst: # only push non-empty lists
heapq.heappush(heap, (lst[0], i, 0))
result = []
while heap:
val, list_idx, elem_idx = heapq.heappop(heap)
result.append(val)
if elem…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 …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.