Reference library

Concurrency & performance

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

12 matches
Concurrency & performance easy

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.

concurrency threadpoolexecutor parallelism
Python
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_…
15 0 Open
Concurrency & performance medium

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).

gil threading multiprocessing
Python
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:
 …
11 0 Open
Concurrency & performance medium

How to Mock anyio.run Backends (asyncio vs trio) in Python

Demonstrates how to mock anyio.run to verify backend selection (asyncio or trio) without actually running the event loop.

anyio async testing
Python
import anyio
from unittest.mock import Mock, patch


async def fetch_data():
    await anyio.sleep(0.1)
    return {"data": 42}


def run_with_backend(backend: str):
    async def main():
        result = await fetch_data()
        print(f"[{backend}] Result: {result}")

    anyio.run(main, backend=backend)


if __nam…
13 0 Open
Concurrency & performance medium

How to Reduce Instance Memory with __slots__ in Python

Demonstrates that classes with __slots__ use less memory per instance than regular classes because they skip the instance __dict__.

__slots__ memory performance
Python
class SlottedPoint:
    __slots__ = ('x', 'y', 'z')

    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z


class RegularPoint:
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z


if __name__ == "__main__":
    regular = RegularPoint(1, 2, 3)…
11 0 Open
Concurrency & performance medium

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.

multiprocessing manager shared-state
Python
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()

       …
13 0 Open
Concurrency & performance easy

How to Use Array Typecodes for Compact Numeric Storage in Python

This code demonstrates how to use the `array` module with typecodes to store integers, floats, and bytes in a memory-efficient way compared to standard Python lists.

array memory performance
Python
from array import array

def demonstrate_array_types():
    # Compact integer arrays
    small_ints = array('i', [1, 2, 3, 4, 5])
    unsigned_ints = array('I', [10, 20, 30])
    
    # Floating point arrays
    floats = array('f', [1.5, 2.5, 3.5])
    doubles = array('d', [1.123456789, 2.987654321])
    
    # Charac…
15 0 Open
Concurrency & performance medium

How to Use a Weakref Cache to Avoid Memory Leaks in Python

This code demonstrates building a value cache with weakref.WeakValueDictionary so objects can be garbage collected when no longer referenced, preventing memory leaks.

weakref caching memory
Python
import weakref
import gc


class ExpensiveObject:
    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return f"ExpensiveObject('{self.name}')"


class ObjectCache:
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()

    def get_or_create(self, name):
       …
13 0 Open
Concurrency & performance medium

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.

asyncio lock concurrency
Python
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…
16 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

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.

threading synchronization barrier
Python
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_…
14 0 Open
Concurrency & performance easy

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.

queue threading producer-consumer
Python
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…
12 0 Open
Concurrency & performance easy

asyncio sleep cooperative scheduling demo in Python

This demo shows how asyncio.sleep yields control between concurrent tasks, letting multiple workers interleave their ticks.

asyncio concurrency scheduling
Python
import asyncio

async def worker(name, delay):
    for i in range(3):
        print(f"{name}: tick {i}")
        await asyncio.sleep(delay)
    return f"{name} done"

async def main():
    tasks = [
        asyncio.create_task(worker("A", 0.1)),
        asyncio.create_task(worker("B", 0.2)),
        asyncio.create_tas…
12 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.