Concurrency & performance
asyncio, threading, multiprocessing, and profiling-friendly performance patterns.
How to Build a Producer-Consumer Pattern with asyncio.Queue in Python
This code implements a classic producer-consumer pattern using asyncio.Queue to coordinate one producer task that generates items and two consumer tasks that process them concurrently, with a sentinel value to signal completion.
import asyncio
import random
async def producer(queue, item_count):
for i in range(item_count):
item = random.randint(1, 100)
await queue.put(item)
print(f"Produced: {item}")
await asyncio.sleep(0.1)
await queue.put(None) # Sentinel to signal end
async def consumer(queue, n…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 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 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 Share Memory Between Processes in Python with multiprocessing.Value and Array
Share a numeric value and a list-like array across multiple Python processes using multiprocessing.Value and multiprocessing.Array, with each process modifying the same memory.
import multiprocessing
def worker(shared_value, shared_array, index):
shared_value.value += 10
shared_array[index] = shared_array[index] * 2
if __name__ == "__main__":
shared_value = multiprocessing.Value("i", 5)
shared_array = multiprocessing.Array("i", [1, 2, 3, 4, 5])
processes = []
for i…How to Share a Dict and List Between Processes with multiprocessing Manager in Python
This code demonstrates how to share a dictionary and a list between multiple processes using multiprocessing.Manager, enabling safe concurrent updates.
import multiprocessing as mp
def worker(shared_dict, shared_list, name):
shared_dict[name] = name.upper()
shared_list.append(name)
print(f"{name} added to shared structures")
def main():
with mp.Manager() as manager:
shared_dict = manager.dict()
shared_list = manager.list()
…How to Share a Queue Between Processes in Python
Use multiprocessing.Queue to pass work from a producer process to multiple consumer processes, coordinating with a sentinel stop message.
import multiprocessing
import time
def producer(queue, items):
for item in items:
queue.put(item)
time.sleep(0.1)
queue.put("STOP")
def consumer(queue, name):
while True:
item = queue.get()
if item == "STOP":
break
print(f"{name} processed: {item}")
…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 and ProcessPoolExecutor in Python
Compares ThreadPoolExecutor and ProcessPoolExecutor by running CPU-bound and I/O-tolerant tasks over a large list, printing elapsed times and first results.
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import math
numbers = list(range(1, 1000001))
def compute_square(n):
return n * n
def compute_sqrt(n):
return math.sqrt(n)
def run_executor(executor, func, data):
start = time.perf_counter()
results = list(executo…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 as_completed to Process Futures in Order of Completion
Submit multiple tasks to a ThreadPoolExecutor and process each result as soon as it finishes using as_completed.
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def fetch_data(item_id):
time.sleep(1)
return f"item-{item_id}"
def main():
with ThreadPoolExecutor(max_workers=3) as executor:
future_map = {executor.submit(fetch_data, i): i for i in range(1, 6)}
for future 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 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 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…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_…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.