performance-optimizer

name	performance-optimizer
description	Identifies and resolves performance bottlenecks in applications, APIs, and systems. Use when profiling slow code, optimizing algorithms, improving response times, reducing memory usage, implementing caching strategies, or analyzing system performance metrics.

Performance Optimizer Skill

This skill helps identify and resolve performance issues across the application stack. Use this whenever you need to profile code, optimize algorithms, implement caching, or improve system throughput.

Performance Principles

1. Optimization Rules

1. Don't optimize prematurely - Measure first, optimize second
2. Profile before guessing - Data beats intuition
3. Optimize the critical path - Focus on what matters most
4. Consider trade-offs - Speed vs memory vs complexity

2. Performance Metrics

Metric	Description	Target
Latency	Time to complete one operation	< 100ms (API)
Throughput	Operations per unit time	Depends on load
p50/p95/p99	Percentile latencies	p99 < 2x p50
TTFB	Time to first byte	< 200ms
Memory usage	RAM consumption	Within limits
CPU usage	Processor utilization	< 70% sustained

3. Big O Quick Reference

Complexity	Name	Example
O(1)	Constant	Hash lookup, array access
O(log n)	Logarithmic	Binary search
O(n)	Linear	Array iteration
O(n log n)	Linearithmic	Efficient sorting
O(n²)	Quadratic	Nested loops
O(2ⁿ)	Exponential	Recursive fibonacci

Profiling Techniques

Python Profiling

# cProfile - Function-level profiling
import cProfile
import pstats

def profile_code():
    profiler = cProfile.Profile()
    profiler.enable()
    
    # Code to profile
    result = expensive_function()
    
    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    stats.print_stats(20)  # Top 20 functions

# Line profiler (install: pip install line-profiler)
# Add @profile decorator to functions
# Run: kernprof -l -v script.py

from line_profiler import profile

@profile
def slow_function():
    result = []
    for i in range(10000):
        result.append(i ** 2)  # Line-by-line timing
    return result

# Memory profiler (install: pip install memory-profiler)
from memory_profiler import profile

@profile
def memory_hungry():
    large_list = [i for i in range(1000000)]
    return sum(large_list)

# Timing decorator
import time
from functools import wraps

def timing(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = f(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{f.__name__} took {elapsed:.4f} seconds")
        return result
    return wrapper

@timing
def my_function():
    # Function code
    pass

JavaScript Profiling

// Console timing
console.time('operation');
expensiveOperation();
console.timeEnd('operation');

// Performance API
const start = performance.now();
expensiveOperation();
const end = performance.now();
console.log(`Execution time: ${end - start}ms`);

// Node.js profiling
// Run: node --prof app.js
// Process: node --prof-process isolate-*.log > profile.txt

// Memory snapshot (Chrome DevTools)
// 1. Open DevTools > Memory tab
// 2. Take heap snapshot
// 3. Compare snapshots to find leaks

// Express.js middleware for timing
const timingMiddleware = (req, res, next) => {
  const start = process.hrtime.bigint();
  
  res.on('finish', () => {
    const end = process.hrtime.bigint();
    const duration = Number(end - start) / 1e6; // Convert to ms
    console.log(`${req.method} ${req.path} - ${duration.toFixed(2)}ms`);
  });
  
  next();
};

Database Query Profiling

-- PostgreSQL
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT * FROM orders 
WHERE customer_id = 123 
ORDER BY created_at DESC 
LIMIT 10;

-- Look for:
-- - Seq Scan (full table scan)
-- - High actual time
-- - Large rows removed by filter
-- - Missing indexes

-- Enable slow query logging
-- postgresql.conf:
-- log_min_duration_statement = 1000  -- Log queries > 1s

-- MySQL
SET profiling = 1;
SELECT * FROM orders WHERE customer_id = 123;
SHOW PROFILE FOR QUERY 1;

-- Enable slow query log
-- SET GLOBAL slow_query_log = 'ON';
-- SET GLOBAL long_query_time = 1;

Common Bottlenecks & Solutions

1. Algorithm Optimization

# ❌ SLOW: O(n²) - Nested loop search
def find_duplicates_slow(items):
    duplicates = []
    for i, item in enumerate(items):
        for j, other in enumerate(items):
            if i != j and item == other and item not in duplicates:
                duplicates.append(item)
    return duplicates

# ✅ FAST: O(n) - Hash set
def find_duplicates_fast(items):
    seen = set()
    duplicates = set()
    for item in items:
        if item in seen:
            duplicates.add(item)
        seen.add(item)
    return list(duplicates)

# ❌ SLOW: O(n) lookup in list
def find_user_slow(users, user_id):
    for user in users:
        if user['id'] == user_id:
            return user
    return None

# ✅ FAST: O(1) lookup in dict
users_by_id = {user['id']: user for user in users}
def find_user_fast(user_id):
    return users_by_id.get(user_id)

2. Loop Optimization

# ❌ SLOW: Creating objects in loop
def process_items_slow(items):
    results = []
    for item in items:
        config = load_config()  # Loaded every iteration!
        result = process(item, config)
        results.append(result)
    return results

# ✅ FAST: Hoist invariants outside loop
def process_items_fast(items):
    config = load_config()  # Load once
    results = []
    for item in items:
        result = process(item, config)
        results.append(result)
    return results

# ❌ SLOW: String concatenation in loop
def build_string_slow(items):
    result = ""
    for item in items:
        result += str(item) + ", "  # Creates new string each time
    return result

# ✅ FAST: Join list of strings
def build_string_fast(items):
    return ", ".join(str(item) for item in items)

# ❌ SLOW: Append to list in comprehension equivalent
def squares_slow(n):
    result = []
    for i in range(n):
        result.append(i ** 2)
    return result

# ✅ FAST: List comprehension
def squares_fast(n):
    return [i ** 2 for i in range(n)]

3. I/O Optimization

# ❌ SLOW: Synchronous I/O
import requests

def fetch_all_slow(urls):
    results = []
    for url in urls:
        response = requests.get(url)  # Blocking!
        results.append(response.json())
    return results

# ✅ FAST: Async I/O
import aiohttp
import asyncio

async def fetch_all_fast(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def fetch_one(session, url):
    async with session.get(url) as response:
        return await response.json()

# ❌ SLOW: Reading file line by line into memory
def read_slow(filename):
    with open(filename) as f:
        lines = f.readlines()  # All in memory
    return [process(line) for line in lines]

# ✅ FAST: Streaming/generator
def read_fast(filename):
    with open(filename) as f:
        for line in f:  # One line at a time
            yield process(line)

4. Memory Optimization

# ❌ MEMORY HOG: Store everything in list
def process_large_file_slow(filename):
    results = []
    with open(filename) as f:
        for line in f:
            results.append(transform(line))
    return results  # All in memory

# ✅ MEMORY EFFICIENT: Generator
def process_large_file_fast(filename):
    with open(filename) as f:
        for line in f:
            yield transform(line)  # One at a time

# ❌ MEMORY HOG: Dict for simple data
class Point:
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z

# ✅ MEMORY EFFICIENT: __slots__
class Point:
    __slots__ = ['x', 'y', 'z']
    
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z

# ✅ EVEN BETTER: namedtuple or dataclass
from dataclasses import dataclass

@dataclass(slots=True)
class Point:
    x: float
    y: float
    z: float

# ❌ SLOW: Creating many small objects
def create_many_dicts():
    return [{'x': i, 'y': i*2} for i in range(1000000)]

# ✅ FAST: NumPy array for numerical data
import numpy as np

def create_array():
    return np.array([[i, i*2] for i in range(1000000)])

Caching Strategies

In-Memory Caching

from functools import lru_cache
from cachetools import TTLCache, LRUCache
import time

# Python built-in LRU cache
@lru_cache(maxsize=1000)
def expensive_computation(n):
    time.sleep(1)  # Simulate expensive work
    return n ** 2

# TTL cache with cachetools
cache = TTLCache(maxsize=100, ttl=300)  # 5 minutes

def get_user(user_id):
    if user_id in cache:
        return cache[user_id]
    
    user = db.fetch_user(user_id)
    cache[user_id] = user
    return user

# Manual cache with expiration
class SimpleCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = ttl_seconds
    
    def get(self, key):
        if key in self.cache:
            value, expiry = self.cache[key]
            if time.time() < expiry:
                return value
            del self.cache[key]
        return None
    
    def set(self, key, value):
        self.cache[key] = (value, time.time() + self.ttl)
    
    def invalidate(self, key):
        self.cache.pop(key, None)

Redis Caching

import redis
import json
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_result(ttl=300):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Build cache key
            key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
            
            # Try cache first
            cached = redis_client.get(key)
            if cached:
                return json.loads(cached)
            
            # Compute and cache
            result = func(*args, **kwargs)
            redis_client.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@cache_result(ttl=600)
def get_product_details(product_id):
    return db.query_product(product_id)

# Cache invalidation patterns
def update_product(product_id, data):
    db.update_product(product_id, data)
    # Invalidate specific key
    redis_client.delete(f"get_product_details:{hash(str((product_id,)))}")
    # Or invalidate pattern
    for key in redis_client.scan_iter(f"product:{product_id}:*"):
        redis_client.delete(key)

# Cache-aside pattern with write-through
class CacheAside:
    def __init__(self, cache, db):
        self.cache = cache
        self.db = db
    
    def read(self, key):
        # Try cache
        value = self.cache.get(key)
        if value:
            return value
        
        # Fallback to DB
        value = self.db.get(key)
        if value:
            self.cache.set(key, value)
        return value
    
    def write(self, key, value):
        self.db.set(key, value)
        self.cache.set(key, value)  # Write-through
        # OR
        self.cache.delete(key)  # Write-invalidate

HTTP Caching

from fastapi import FastAPI, Response
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

@app.get("/products/{product_id}")
async def get_product(product_id: int, response: Response):
    product = await fetch_product(product_id)
    
    # Cache for 1 hour, allow stale for 1 day while revalidating
    response.headers["Cache-Control"] = "public, max-age=3600, stale-while-revalidate=86400"
    response.headers["ETag"] = f'"{product["version"]}"'
    
    return product

@app.get("/user/profile")
async def get_profile(response: Response):
    # Private data - no shared cache
    response.headers["Cache-Control"] = "private, max-age=60"
    return current_user.profile

@app.post("/products")
async def create_product(response: Response):
    # Mutations - no caching
    response.headers["Cache-Control"] = "no-store"
    return await create_new_product()

Database Optimization

Query Optimization

# ❌ N+1 Problem
users = User.query.all()
for user in users:
    orders = Order.query.filter_by(user_id=user.id).all()  # N queries!

# ✅ Eager loading
users = User.query.options(
    joinedload(User.orders)
).all()

# ✅ Batch loading
user_ids = [u.id for u in users]
orders = Order.query.filter(Order.user_id.in_(user_ids)).all()
orders_by_user = {}
for order in orders:
    orders_by_user.setdefault(order.user_id, []).append(order)

# ❌ SLOW: Fetching unused columns
users = db.execute("SELECT * FROM users WHERE status = 'active'")

# ✅ FAST: Select only needed columns
users = db.execute("""
    SELECT id, name, email 
    FROM users 
    WHERE status = 'active'
""")

# ✅ Use database-level pagination
def get_page(page, per_page):
    offset = (page - 1) * per_page
    return db.execute("""
        SELECT * FROM products
        ORDER BY created_at DESC
        LIMIT :limit OFFSET :offset
    """, {"limit": per_page, "offset": offset})

Connection Pooling

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    "postgresql://user:pass@localhost/db",
    poolclass=QueuePool,
    pool_size=10,           # Persistent connections
    max_overflow=20,        # Extra connections when busy
    pool_timeout=30,        # Wait time for connection
    pool_recycle=1800,      # Refresh connections after 30min
    pool_pre_ping=True,     # Verify connection health
)

Frontend Performance

JavaScript Optimization

// ❌ SLOW: DOM manipulation in loop
function updateItems(items) {
  const container = document.getElementById('container');
  items.forEach(item => {
    const div = document.createElement('div');
    div.textContent = item.name;
    container.appendChild(div);  // Reflow on each append!
  });
}

// ✅ FAST: Batch DOM updates
function updateItemsFast(items) {
  const container = document.getElementById('container');
  const fragment = document.createDocumentFragment();
  
  items.forEach(item => {
    const div = document.createElement('div');
    div.textContent = item.name;
    fragment.appendChild(div);
  });
  
  container.appendChild(fragment);  // Single reflow
}

// ❌ SLOW: Layout thrashing
function resizeAll(elements) {
  elements.forEach(el => {
    const width = el.offsetWidth;  // Read
    el.style.height = width + 'px';  // Write - forces reflow!
  });
}

// ✅ FAST: Batch reads, then writes
function resizeAllFast(elements) {
  const widths = elements.map(el => el.offsetWidth);  // All reads
  elements.forEach((el, i) => {
    el.style.height = widths[i] + 'px';  // All writes
  });
}

// Debounce expensive operations
function debounce(fn, delay) {
  let timeoutId;
  return (...args) => {
    clearTimeout(timeoutId);
    timeoutId = setTimeout(() => fn(...args), delay);
  };
}

const handleSearch = debounce((query) => {
  performSearch(query);
}, 300);

// Throttle continuous events
function throttle(fn, limit) {
  let inThrottle;
  return (...args) => {
    if (!inThrottle) {
      fn(...args);
      inThrottle = true;
      setTimeout(() => inThrottle = false, limit);
    }
  };
}

const handleScroll = throttle(() => {
  updatePosition();
}, 16);  // ~60fps

React Performance

// ❌ SLOW: Creating new function on each render
function List({ items, onSelect }) {
  return items.map(item => (
    <Item 
      key={item.id}
      onClick={() => onSelect(item.id)}  // New function each render!
    />
  ));
}

// ✅ FAST: useCallback for stable references
function List({ items, onSelect }) {
  const handleClick = useCallback((id) => {
    onSelect(id);
  }, [onSelect]);
  
  return items.map(item => (
    <Item 
      key={item.id}
      id={item.id}
      onClick={handleClick}
    />
  ));
}

// ❌ SLOW: Computing on every render
function ExpensiveList({ items, filter }) {
  const filtered = items.filter(i => i.name.includes(filter));
  const sorted = filtered.sort((a, b) => a.name.localeCompare(b.name));
  
  return sorted.map(item => <Item key={item.id} {...item} />);
}

// ✅ FAST: Memoize expensive computations
function ExpensiveList({ items, filter }) {
  const processed = useMemo(() => {
    const filtered = items.filter(i => i.name.includes(filter));
    return filtered.sort((a, b) => a.name.localeCompare(b.name));
  }, [items, filter]);
  
  return processed.map(item => <Item key={item.id} {...item} />);
}

// ✅ Use React.memo for pure components
const Item = React.memo(({ id, name, onClick }) => {
  return <div onClick={() => onClick(id)}>{name}</div>;
});

// ✅ Virtualize long lists
import { FixedSizeList } from 'react-window';

function VirtualizedList({ items }) {
  const Row = ({ index, style }) => (
    <div style={style}>{items[index].name}</div>
  );
  
  return (
    <FixedSizeList
      height={400}
      width={300}
      itemCount={items.length}
      itemSize={35}
    >
      {Row}
    </FixedSizeList>
  );
}

Async & Concurrency

Python Async

import asyncio
import aiohttp

# ❌ SLOW: Sequential async calls
async def fetch_sequential(urls):
    results = []
    async with aiohttp.ClientSession() as session:
        for url in urls:
            async with session.get(url) as response:
                results.append(await response.json())
    return results

# ✅ FAST: Concurrent async calls
async def fetch_concurrent(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def fetch_one(session, url):
    async with session.get(url) as response:
        return await response.json()

# Limit concurrency with semaphore
async def fetch_with_limit(urls, max_concurrent=10):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch_limited(session, url):
        async with semaphore:
            async with session.get(url) as response:
                return await response.json()
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_limited(session, url) for url in urls]
        return await asyncio.gather(*tasks)

Thread/Process Pools

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import multiprocessing

# I/O-bound: Use threads
def fetch_urls_threaded(urls):
    def fetch_one(url):
        response = requests.get(url)
        return response.json()
    
    with ThreadPoolExecutor(max_workers=10) as executor:
        return list(executor.map(fetch_one, urls))

# CPU-bound: Use processes
def process_items_parallel(items):
    def process_one(item):
        # CPU-intensive work
        return heavy_computation(item)
    
    with ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
        return list(executor.map(process_one, items))

Performance Checklist

Backend

• Profile before optimizing
• No N+1 queries
• Appropriate indexes
• Connection pooling
• Async for I/O-bound tasks
• Caching for hot data
• Pagination for lists

Frontend

• Bundle size minimized
• Images optimized
• Lazy loading
• Virtual scrolling for long lists
• Debounce/throttle events
• Memoization where appropriate

Infrastructure

• CDN for static assets
• HTTP caching headers
• Compression enabled
• Load balancing
• Auto-scaling configured

Output Format

When optimizing performance, provide:

## Performance Analysis

### Bottleneck Identified
**Type:** [Algorithm / I/O / Memory / Database]
**Location:** `file.py:42`
**Impact:** [High/Medium/Low]

### Current Performance
- Latency: X ms
- Memory: Y MB
- Complexity: O(...)

### Optimized Solution

**Before:**
```python
[Original slow code]

After:

[Optimized code]

Expected Improvement

• Latency: X ms → Y ms (Z% improvement)
• Memory: X MB → Y MB
• Complexity: O(...) → O(...)

Trade-offs

• [Trade-off 1]
• [Trade-off 2]

Additional Recommendations

1. [Recommendation 1]
2. [Recommendation 2]

## Notes

- Always measure before and after
- Profile in production-like conditions
- Consider the 80/20 rule - optimize what matters
- Don't sacrifice readability for micro-optimizations
- Cache invalidation is hard - plan carefully
- Premature optimization is the root of all evil