Reference library

Concurrency & performance

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

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

46 0 Open

Concurrency & performance medium

How to Implement a Batch Requests Flush Interval in Python

A simple async batcher that accumulates items and flushes them either when a max batch size is reached or after a time-based flush interval.

asyncio batching concurrency

Python

import asyncio
from collections import deque

class Batcher:
    def __init__(self, flush_interval=0.5, max_batch=5):
        self.flush_interval = flush_interval
        self.max_batch = max_batch
        self.queue = deque()
        self.lock = asyncio.Lock()

    async def add(self, item):
        async with self.l…

14 0 Open

Concurrency & performance medium

How to Profile CPU Hot Path in Python with cProfile and sort_stats cumtime

Profile a Python function's CPU usage by running cProfile, sorting stats by cumulative time, and printing a readable report to stdout.

cprofile profiling performance

Python

import cProfile
import pstats
import io


def slow_function():
    total = 0
    for i in range(100_000):
        total += i * i
    return total


def fast_function():
    return sum(i for i in range(100))


def main():
    slow_function()
    fast_function()


if __name__ == "__main__":
    profiler = cProfile.Profi…

13 0 Open

Concurrency & performance medium

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.

multiprocessing queue concurrency

Python

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}")

…

14 0 Open

Concurrency & performance medium

Merge K Sorted Lists in Python with heapq

Merge k sorted lists into one sorted list in O(N log k) time using a min-heap of current elements.

heapq merge sorted-lists

Python

import heapq

def merge_k_sorted_lists(lists):
    heap = []
    for i, lst in enumerate(lists):
        if lst:  # only push non-empty lists
            heapq.heappush(heap, (lst[0], i, 0))
    result = []
    while heap:
        val, list_idx, elem_idx = heapq.heappop(heap)
        result.append(val)
        if elem…

13 0 Open

Concurrency & performance medium

Profile Memory Usage with tracemalloc Snapshot Diff in Python

Use tracemalloc to take two memory snapshots, compute a diff, and print the top changes (size and count) by line number.

tracemalloc memory-profile performance

Python

import tracemalloc

def profile_memory():
    tracemalloc.start()
    
    # Allocate some objects to track
    data = [i * 2 for i in range(10000)]
    text = "x" * 5000
    nested = {"key": [1, 2, 3], "value": (4, 5)}
    
    # Take first snapshot
    snapshot1 = tracemalloc.take_snapshot()
    
    # Free some mem…

11 0 Open

Browse by section

Each section groups closely related Python snippets.