Reference library

Concurrency & performance

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

4 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 Implement a Token Bucket Rate Limiter with asyncio in Python

This code implements a thread-safe token bucket rate limiter for asyncio, allowing you to limit the rate of async tasks or API calls.

asyncio rate-limiting token-bucket

Python

import asyncio
import time


class TokenBucket:
    def __init__(self, rate_per_second, capacity):
        self.rate = rate_per_second
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.monotonic()
        self.lock = asyncio.Lock()

    async def acquire(self):
        asy…

14 0 Open

Concurrency & performance easy

How to set a timeout with asyncio.wait_for in Python

Use asyncio.wait_for to bound an async function with a timeout, catching TimeoutError when it exceeds the limit.

asyncio timeout concurrency

Python

import asyncio

async def slow_task():
    await asyncio.sleep(3)
    return "finished"

async def main():
    try:
        result = await asyncio.wait_for(slow_task(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("Task timed out")

if __name__ == "__main__":
    asyncio.run(main())

13 0 Open

Concurrency & performance medium

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.

asyncio concurrency semaphore

Python

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 …

13 0 Open

Browse by section

Each section groups closely related Python snippets.