Performance

import cProfile
import pstats
import io

pr = cProfile.Profile()
pr.enable()
result = my_slow_function()
pr.disable()

s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(20)  # top 20 slowest functions
print(s.getvalue())

# Profile a whole script from the command line:
# python -m cProfile -o output.prof script.py
# snakeviz output.prof  # opens interactive flame graph in browser

# pip install memory-profiler
from memory_profiler import profile

@profile
def my_function():
    # annotated line-by-line memory usage
    data = [x for x in range(10_000_000)]
    return data

# CPU flame graph
npx 0x -- node dist/server.js
# Opens a generated .html flame graph in the browser

# Heap snapshot + event loop lag
npx clinic doctor -- node dist/server.js

# CPU flame graph via clinic
npx clinic flame -- node dist/server.js

# Async waterfall / I/O bottlenecks
npx clinic bubbleprof -- node dist/server.js

import (
    "net/http"
    _ "net/http/pprof" // side-effect import registers /debug/pprof handlers
)

// In main(), run alongside your app server:
go func() {
    http.ListenAndServe("localhost:6060", nil)
}()

# CPU profile (30-second sample)
go tool pprof http://localhost:6060/debug/pprof/profile?seconds=30

# Memory (heap) profile
go tool pprof http://localhost:6060/debug/pprof/heap

# In the pprof interactive prompt:
# (pprof) top10          — top 10 functions by CPU or memory
# (pprof) web            — open flame graph in browser (requires graphviz)
# (pprof) list FuncName  — annotated source with per-line costs

def get_user(user_id: str) -> User:
    # 1. Check cache first
    cached = redis.get(f"user:{user_id}")
    if cached:
        return User.from_json(cached)

    # 2. Cache miss — fetch from DB
    user = db.query(User).filter(User.id == user_id).first()

    # 3. Populate cache with TTL
    redis.setex(f"user:{user_id}", 300, user.to_json())  # 5 min TTL
    return user

# Python
from functools import lru_cache

@lru_cache(maxsize=1000)
def get_config(key: str) -> str:
    return db.get_config(key)

// TypeScript
import LRU from 'lru-cache';

const cache = new LRU<string, string>({ max: 1000, ttl: 1000 * 60 * 5 });

function getConfig(key: string): string {
  if (cache.has(key)) return cache.get(key)!;
  const value = db.getConfig(key);
  cache.set(key, value);
  return value;
}

// Go
import "github.com/hashicorp/golang-lru/v2"

cache, _ := lru.New[string, string](1000)

func getConfig(key string) string {
    if val, ok := cache.Get(key); ok {
        return val
    }
    val := db.GetConfig(key)
    cache.Add(key, val)
    return val
}

# First request
GET /api/products/123
→ 200 OK
   ETag: "v2-abc123"
   Cache-Control: max-age=60

# After TTL expires — client sends ETag back
GET /api/products/123
If-None-Match: "v2-abc123"
→ 304 Not Modified   (no response body — saves bandwidth)
# or, if product changed:
→ 200 OK
   ETag: "v3-def456"

# BAD: N+1 — 1 query for orders + 1 query per order for its user
orders = db.query(Order).all()      # 1 query
for order in orders:
    print(order.user.name)          # N queries (lazy load per order)

from sqlalchemy.orm import selectinload, joinedload

# selectinload: 2 queries total — 1 for orders, 1 IN query for all related users
orders = db.query(Order).options(selectinload(Order.user)).all()

# joinedload: 1 query with a JOIN (better for single related object)
orders = db.query(Order).options(joinedload(Order.user)).all()

// BAD
const orders = await prisma.order.findMany();
for (const order of orders) {
  const user = await prisma.user.findUnique({ where: { id: order.userId } });
}

// GOOD — Prisma batches the related fetches automatically
const orders = await prisma.order.findMany({
  include: { user: true },
});

var orders []Order

// BAD — N separate queries inside the loop
db.Find(&orders)
for i := range orders {
    db.First(&orders[i].User, orders[i].UserID)
}

// GOOD — Preload issues a single IN query for all users
db.Preload("User").Find(&orders)

import asyncio
import aiohttp

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)  # concurrent, not parallel

async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(data: list) -> int:
    return sum(x ** 2 for x in data)

async def process_many(chunks: list[list]) -> list[int]:
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        results = await asyncio.gather(
            *[loop.run_in_executor(pool, cpu_heavy, chunk) for chunk in chunks]
        )
    return results

import fs from 'fs';
import { Worker, isMainThread, workerData, parentPort } from 'worker_threads';

// BAD: sync read blocks the event loop for all requests
const data = fs.readFileSync('large-file.json', 'utf8');

// GOOD: async I/O — yields control back to event loop
const data = await fs.promises.readFile('large-file.json', 'utf8');

// BAD: CPU-heavy work in the main thread stalls all requests
const result = heavyComputation(data);

// GOOD: offload CPU work to a worker thread
function runInWorker(payload: unknown): Promise<unknown> {
  return new Promise((resolve, reject) => {
    const worker = new Worker(__filename, { workerData: payload });
    worker.on('message', resolve);
    worker.on('error', reject);
  });
}

// Fan-out: fire N goroutines, collect with WaitGroup + channel
func fetchAll(urls []string) []Result {
    results := make(chan Result, len(urls))
    var wg sync.WaitGroup

    for _, url := range urls {
        wg.Add(1)
        go func(u string) {
            defer wg.Done()
            resp, err := http.Get(u)
            results <- Result{URL: u, Err: err, Body: readBody(resp)}
        }(url)
    }

    wg.Wait()
    close(results)

    var out []Result
    for r := range results {
        out = append(out, r)
    }
    return out
}