Reference library

Concurrency & performance

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

3 matches

How to Test HTTPX Async Client Pool Reuse with Mocks in Python

Mock an httpx.AsyncClient to verify connection pool reuse by asserting GET calls share a single client instance across concurrent async requests.

httpx async-await mock

Python

import asyncio
import httpx
from unittest.mock import AsyncMock, patch, Mock

async def fetch_with_pool(client, url, n_reuses=3):
    results = []
    for i in range(n_reuses):
        resp = await client.get(url)
        results.append(resp.status_code)
        await asyncio.sleep(0)  # yield to loop to mimic real us…

13 0 Open

Concurrency & performance easy

How to Use pool.map for CPU-Bound Tasks in Python

Distribute CPU-intensive functions across processes with multiprocessing.Pool.map and measure the performance gain.

multiprocessing pool cpu-bound

Python

from multiprocessing import Pool
import time

def cpu_bound_task(n):
    """Mock CPU-bound work: compute sum of squares."""
    total = 0
    for i in range(n):
        total += i * i
    return total

if __name__ == "__main__":
    numbers = [10_000_000, 12_000_000, 8_000_000, 15_000_000]

    start = time.perf_count…

11 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

Browse by section

Each section groups closely related Python snippets.