Reference library

Concurrency & performance

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

5 matches

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.

profiling cprofile pstats

Python

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()
  …

44 0 Open

Concurrency & performance medium

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.

asyncio queue concurrency

Python

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…

14 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 easy

How to Use functools.cache for Unbounded Memoization in Python

Speed up repeated recursive calls by memoizing function results with Python's built-in functools.cache decorator.

functools memoization performance

Python

```python
import functools
import time


@functools.cache
def fib(n):
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)


if __name__ == "__main__":
    start = time.perf_counter()
    result = fib(30)
    elapsed = time.perf_counter() - start

    print(f"fib(30) = {result}")
    print(f"computed in {…

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

Browse by section

Each section groups closely related Python snippets.