Reference library

Concurrency & performance

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

6 matches

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 Run Blocking Code in an Executor with asyncio in Python

This code runs blocking functions concurrently without stalling the event loop by offloading them to thread pool executors via asyncio.

asyncio executor concurrency

Python

import asyncio
import time


def blocking_task(name: str, duration: float) -> str:
    """Simulate a blocking operation."""
    time.sleep(duration)
    return f"Finished {name} after {duration}s"


async def main() -> None:
    loop = asyncio.get_running_loop()
    results = await asyncio.gather(
        loop.run_in_…

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 easy

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.

threading lock mutex

Python

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…

14 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

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.

concurrency threadpoolexecutor parallel

Python

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…

14 0 Open

Browse by section

Each section groups closely related Python snippets.