Part 4: Resource and Performance Errors

Introduction

I once deployed a background job that processed user uploads. Within hours, the server ran out of memory and crashed. The culprit? I was loading entire files into memory without closing them, creating thousands of open file handles. This incident taught me that resource management isn't optional—it's critical for building reliable systems.

Resource and performance errors often don't appear during development with small datasets. They emerge in production under real load, making them particularly insidious.

Resource Errors

What Are Resource Errors?

Resource errors occur when a program exhausts or mismanages system resources like memory, file handles, network connections, or CPU time. These errors can cause crashes, slowdowns, or system instability.

Real-World Examples from My Projects

Memory Leaks from Unclosed Files

From a log processing service:

# Incorrect - Files not properly closed
def process_all_logs(log_directory):
    """Process all log files in directory"""
    log_files = []
    
    for filename in os.listdir(log_directory):
        if filename.endswith('.log'):
            # Bug: Opening files but never closing them
            f = open(os.path.join(log_directory, filename), 'r')
            log_files.append(f)
    
    # Process files
    for log_file in log_files:
        process_log_content(log_file.read())
    
    # Files never closed - memory leak and file handle exhaustion

# With 1000s of log files, this exhausts file handles
process_all_logs('/var/logs/app/')  # Eventually: OSError: Too many open files

Correct version:

import os

def process_all_logs(log_directory):
    """Process all log files with proper resource management"""
    for filename in os.listdir(log_directory):
        if filename.endswith('.log'):
            filepath = os.path.join(log_directory, filename)
            
            # Use context manager - automatically closes file
            with open(filepath, 'r') as f:
                content = f.read()
                process_log_content(content)

# Better: Process in chunks for large files
def process_all_logs_chunked(log_directory, chunk_size=8192):
    """Process log files in chunks to manage memory"""
    for filename in os.listdir(log_directory):
        if filename.endswith('.log'):
            filepath = os.path.join(log_directory, filename)
            
            with open(filepath, 'r') as f:
                while True:
                    chunk = f.read(chunk_size)
                    if not chunk:
                        break
                    process_log_content(chunk)

Loading Large Datasets into Memory

From a data analytics script:

# Incorrect - Loading entire dataset into memory
def analyze_user_data(csv_file):
    """Analyze user activity data"""
    # Bug: Loading potentially millions of rows into memory
    with open(csv_file, 'r') as f:
        all_lines = f.readlines()  # Loads entire file into memory!
    
    users = []
    for line in all_lines:
        user_data = parse_csv_line(line)
        users.append(user_data)
    
    return calculate_statistics(users)

# With a 10GB CSV file, this causes MemoryError
analyze_user_data('user_activity.csv')

Correct version:

import csv
from collections import defaultdict

def analyze_user_data(csv_file):
    """Analyze user data with streaming processing"""
    stats = defaultdict(int)
    
    # Process line by line - constant memory usage
    with open(csv_file, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            # Update statistics incrementally
            update_statistics(stats, row)
    
    return stats

# Better: Using pandas with chunking for large files
import pandas as pd

def analyze_user_data_pandas(csv_file, chunksize=10000):
    """Analyze large CSV in chunks"""
    aggregated_stats = None
    
    # Process in chunks
    for chunk in pd.read_csv(csv_file, chunksize=chunksize):
        chunk_stats = calculate_chunk_statistics(chunk)
        
        if aggregated_stats is None:
            aggregated_stats = chunk_stats
        else:
            aggregated_stats = merge_statistics(aggregated_stats, chunk_stats)
    
    return aggregated_stats

Connection Pool Exhaustion

From a microservice I built:

import requests

# Incorrect - Creating new connection for each request
def fetch_user_data(user_ids):
    """Fetch data for multiple users"""
    results = []
    
    for user_id in user_ids:
        # Bug: Creating new HTTP connection each time
        response = requests.get(f'https://api.example.com/users/{user_id}')
        results.append(response.json())
    
    return results

# With 1000s of users, this exhausts connections and is slow
users = fetch_user_data(range(10000))

Correct version:

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def create_session_with_retry():
    """Create requests session with connection pooling and retry"""
    session = requests.Session()
    
    # Configure retry strategy
    retry = Retry(
        total=3,
        backoff_factor=0.3,
        status_forcelist=[500, 502, 503, 504]
    )
    
    # Configure connection pooling
    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=10,
        pool_maxsize=20
    )
    
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    
    return session

def fetch_user_data(user_ids):
    """Fetch data for multiple users with connection pooling"""
    results = []
    
    # Reuse session for all requests
    with create_session_with_retry() as session:
        for user_id in user_ids:
            response = session.get(f'https://api.example.com/users/{user_id}')
            response.raise_for_status()
            results.append(response.json())
    
    return results

# Better: Use concurrent requests with limits
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_user_data_concurrent(user_ids, max_workers=10):
    """Fetch user data concurrently with controlled parallelism"""
    results = []
    
    def fetch_single_user(user_id, session):
        response = session.get(f'https://api.example.com/users/{user_id}')
        response.raise_for_status()
        return response.json()
    
    with create_session_with_retry() as session:
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all tasks
            future_to_id = {
                executor.submit(fetch_single_user, uid, session): uid 
                for uid in user_ids
            }
            
            # Collect results as they complete
            for future in as_completed(future_to_id):
                try:
                    result = future.result()
                    results.append(result)
                except Exception as e:
                    user_id = future_to_id[future]
                    logger.error(f"Failed to fetch user {user_id}: {e}")
    
    return results

Database Connection Leaks

From a web application:

import psycopg2

# Incorrect - Not closing database connections
def get_all_users():
    """Fetch all users from database"""
    # Bug: Connection opened but never explicitly closed
    conn = psycopg2.connect(
        host="localhost",
        database="myapp",
        user="admin",
        password="secret"
    )
    
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    users = cursor.fetchall()
    
    # Missing: cursor.close() and conn.close()
    return users

# After many calls, database reaches max connections
for _ in range(1000):
    users = get_all_users()  # Eventually: OperationalError: too many connections

Correct version:

import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """Context manager for database connections"""
    conn = psycopg2.connect(
        host="localhost",
        database="myapp",
        user="admin",
        password="secret"
    )
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

def get_all_users():
    """Fetch all users with proper connection management"""
    with get_db_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            users = cursor.fetchall()
    
    return users

# Better: Use connection pooling
from psycopg2 import pool

class DatabasePool:
    def __init__(self, min_conn=1, max_conn=10):
        self.pool = psycopg2.pool.SimpleConnectionPool(
            min_conn,
            max_conn,
            host="localhost",
            database="myapp",
            user="admin",
            password="secret"
        )
    
    @contextmanager
    def get_connection(self):
        conn = self.pool.getconn()
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            self.pool.putconn(conn)
    
    def close_all(self):
        self.pool.closeall()

# Global pool instance
db_pool = DatabasePool(min_conn=2, max_conn=20)

def get_all_users():
    """Fetch users using connection pool"""
    with db_pool.get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            users = cursor.fetchall()
    
    return users

How I Prevent Resource Errors

1. Always Use Context Managers

# For files
with open('data.txt', 'r') as f:
    data = f.read()

# For database connections
with get_db_connection() as conn:
    # Use connection
    pass

# For locks
from threading import Lock

lock = Lock()
with lock:
    # Critical section
    pass

# Custom context manager
from contextlib import contextmanager

@contextmanager
def temporary_file(filename):
    """Create temporary file and clean up after"""
    f = open(filename, 'w')
    try:
        yield f
    finally:
        f.close()
        os.remove(filename)

with temporary_file('temp.txt') as f:
    f.write('temporary data')
# File automatically closed and deleted

2. Implement Resource Limits

import resource

def set_memory_limit(max_memory_mb=512):
    """Limit memory usage to prevent runaway processes"""
    max_memory = max_memory_mb * 1024 * 1024  # Convert to bytes
    resource.setrlimit(resource.RLIMIT_AS, (max_memory, max_memory))

def set_cpu_time_limit(max_seconds=300):
    """Limit CPU time to prevent infinite loops"""
    resource.setrlimit(resource.RLIMIT_CPU, (max_seconds, max_seconds))

# Use at application startup
set_memory_limit(1024)  # 1GB limit
set_cpu_time_limit(600)  # 10 minute limit

3. Monitor Resource Usage

import psutil
import logging

logger = logging.getLogger(__name__)

def log_resource_usage():
    """Log current resource usage"""
    process = psutil.Process()
    
    # Memory info
    memory_info = process.memory_info()
    memory_mb = memory_info.rss / (1024 * 1024)
    
    # CPU usage
    cpu_percent = process.cpu_percent(interval=1)
    
    # File descriptors
    num_fds = process.num_fds() if hasattr(process, 'num_fds') else 0
    
    logger.info(
        f"Resource usage - Memory: {memory_mb:.2f}MB, "
        f"CPU: {cpu_percent}%, FDs: {num_fds}"
    )

# Call periodically in long-running processes
def background_task():
    while True:
        log_resource_usage()
        do_work()
        time.sleep(60)

Time Limit Exceeded Errors

What Are Time Limit Errors?

Time limit errors occur when operations take longer than expected or allowed, often due to inefficient algorithms, blocking operations, or external service delays.

Real-World Examples

Inefficient Algorithm

From a data deduplication script:

# Incorrect - O(n²) algorithm, too slow for large datasets
def find_duplicates(items):
    """Find duplicate items in list"""
    duplicates = []
    
    # Bug: Nested loops cause exponential growth
    for i in range(len(items)):
        for j in range(i + 1, len(items)):
            if items[i] == items[j] and items[i] not in duplicates:
                duplicates.append(items[i])
    
    return duplicates

# With 100,000 items, this takes minutes or hours
large_list = list(range(100000))
dupes = find_duplicates(large_list)  # Takes forever!

Correct version:

from collections import Counter

def find_duplicates(items):
    """Find duplicates using hash map - O(n) complexity"""
    counts = Counter(items)
    return [item for item, count in counts.items() if count > 1]

# Much faster - completes in milliseconds
large_list = list(range(100000))
dupes = find_duplicates(large_list)

# Alternative using set
def find_duplicates_set(items):
    """Find duplicates using set - O(n) complexity"""
    seen = set()
    duplicates = set()
    
    for item in items:
        if item in seen:
            duplicates.add(item)
        seen.add(item)
    
    return list(duplicates)

Blocking I/O Operations

From an API service:

import requests

# Incorrect - Blocking calls with no timeout
def fetch_external_data():
    """Fetch data from external API"""
    # Bug: No timeout - could hang forever
    response = requests.get('https://slow-api.example.com/data')
    return response.json()

# If external API is slow or down, this blocks indefinitely
data = fetch_external_data()  # Hangs...

Correct version:

import requests
from requests.exceptions import Timeout, RequestException

def fetch_external_data(timeout=5):
    """
    Fetch data from external API with timeout.
    
    Args:
        timeout: Maximum seconds to wait for response
    
    Raises:
        Timeout: If request exceeds timeout
        RequestException: For other request errors
    """
    try:
        response = requests.get(
            'https://slow-api.example.com/data',
            timeout=timeout  # Timeout in seconds
        )
        response.raise_for_status()
        return response.json()
    
    except Timeout:
        logger.error(f"Request timed out after {timeout}s")
        raise
    except RequestException as e:
        logger.error(f"Request failed: {e}")
        raise

# Better: With retry and exponential backoff
import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def fetch_external_data_with_retry(timeout=5):
    """Fetch with automatic retry on failure"""
    response = requests.get(
        'https://slow-api.example.com/data',
        timeout=timeout
    )
    response.raise_for_status()
    return response.json()

Unbounded Recursive Calls

From a file system traversal tool:

import os

# Incorrect - No depth limit on recursion
def count_files(directory):
    """Count all files in directory recursively"""
    count = 0
    
    for item in os.listdir(directory):
        path = os.path.join(directory, item)
        
        if os.path.isfile(path):
            count += 1
        elif os.path.isdir(path):
            # Bug: Unbounded recursion, could hit recursion limit
            # or take too long on deep directory trees
            count += count_files(path)
    
    return count

# On very deep directory structure: RecursionError
total = count_files('/very/deep/directory/')

Correct version:

import os
from pathlib import Path

def count_files(directory, max_depth=10):
    """
    Count files with depth limit.
    
    Args:
        directory: Root directory to search
        max_depth: Maximum directory depth to traverse
    """
    def _count_recursive(path, current_depth):
        if current_depth > max_depth:
            logger.warning(f"Max depth {max_depth} reached at {path}")
            return 0
        
        count = 0
        try:
            for item in os.listdir(path):
                item_path = os.path.join(path, item)
                
                if os.path.isfile(item_path):
                    count += 1
                elif os.path.isdir(item_path):
                    count += _count_recursive(item_path, current_depth + 1)
        except PermissionError:
            logger.warning(f"Permission denied: {path}")
        
        return count
    
    return _count_recursive(directory, 0)

# Better: Using iterative approach with os.walk
def count_files_iterative(directory, max_depth=10):
    """Count files iteratively - no recursion limit"""
    count = 0
    
    for root, dirs, files in os.walk(directory):
        # Calculate current depth
        depth = root[len(directory):].count(os.sep)
        
        if depth >= max_depth:
            # Clear dirs to prevent deeper traversal
            dirs[:] = []
            continue
        
        count += len(files)
    
    return count

Database Query Performance

From a reporting system:

# Incorrect - N+1 query problem
def get_users_with_orders():
    """Get all users with their orders"""
    users = db.query("SELECT * FROM users")
    
    result = []
    for user in users:
        # Bug: Separate query for each user - very slow!
        orders = db.query(
            "SELECT * FROM orders WHERE user_id = ?", 
            user['id']
        )
        result.append({
            'user': user,
            'orders': orders
        })
    
    return result

# With 10,000 users: 10,001 database queries!
users = get_users_with_orders()  # Takes many seconds

Correct version:

def get_users_with_orders():
    """Get users with orders using JOIN - single query"""
    query = """
        SELECT 
            u.id as user_id,
            u.name,
            u.email,
            o.id as order_id,
            o.total,
            o.created_at
        FROM users u
        LEFT JOIN orders o ON u.id = o.user_id
        ORDER BY u.id, o.created_at
    """
    
    rows = db.query(query)
    
    # Group results by user
    users_dict = {}
    for row in rows:
        user_id = row['user_id']
        
        if user_id not in users_dict:
            users_dict[user_id] = {
                'id': user_id,
                'name': row['name'],
                'email': row['email'],
                'orders': []
            }
        
        if row['order_id']:
            users_dict[user_id]['orders'].append({
                'id': row['order_id'],
                'total': row['total'],
                'created_at': row['created_at']
            })
    
    return list(users_dict.values())

# Single database query - much faster!

How I Prevent Performance Errors

1. Profile Before Optimizing

import cProfile
import pstats
from io import StringIO

def profile_function(func):
    """Decorator to profile function execution"""
    def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        profiler.enable()
        
        result = func(*args, **kwargs)
        
        profiler.disable()
        
        # Print stats
        s = StringIO()
        ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
        ps.print_stats()
        print(s.getvalue())
        
        return result
    
    return wrapper

@profile_function
def slow_function():
    # Function to profile
    pass

2. Use Appropriate Data Structures

# Bad: Using list for membership testing - O(n)
items = []
for i in range(10000):
    if i not in items:  # Slow!
        items.append(i)

# Good: Using set for membership testing - O(1)
items = set()
for i in range(10000):
    if i not in items:  # Fast!
        items.add(i)

3. Implement Caching

from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_calculation(n):
    """Cache results of expensive calculations"""
    # Expensive operation
    time.sleep(1)  # Simulating slow operation
    return n * n

# First call: slow
result1 = expensive_calculation(10)  # Takes 1 second

# Second call: instant (cached)
result2 = expensive_calculation(10)  # Instant!

4. Add Timeouts to All External Calls

import signal
from contextlib import contextmanager

class TimeoutError(Exception):
    pass

@contextmanager
def timeout(seconds):
    """Context manager for timeouts"""
    def timeout_handler(signum, frame):
        raise TimeoutError(f"Operation timed out after {seconds}s")
    
    # Set alarm
    old_handler = signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(seconds)
    
    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

# Usage
try:
    with timeout(5):
        slow_operation()
except TimeoutError:
    logger.error("Operation took too long")

Tools I Use

Resource Monitoring

# Install monitoring tools
pip install psutil memory_profiler

# Memory profiling
python -m memory_profiler script.py

# Line-by-line profiling
python -m cProfile -o output.prof script.py
python -m pstats output.prof

Performance Testing

import timeit

# Measure execution time
def test_performance():
    # Setup code
    setup = "from __main__ import my_function"
    
    # Code to test
    code = "my_function(1000)"
    
    # Run 1000 times
    time = timeit.timeit(code, setup=setup, number=1000)
    print(f"Average time: {time/1000:.6f}s")

Key Takeaways

1. Use context managers: Always close resources properly

2. Implement connection pooling: Reuse expensive connections

3. Process data in chunks: Don't load everything into memory

4. Add timeouts everywhere: External calls should never block indefinitely

5. Choose right algorithms: O(n) vs O(n²) matters at scale

6. Profile before optimizing: Measure to find real bottlenecks

7. Monitor resource usage: Track memory, CPU, and connections

8. Set resource limits: Prevent runaway processes

Next in Series

In Part 5: Interface and Integration Errors, we'll explore errors that occur when different parts of a system interact—API mismatches, version conflicts, and the challenges of integrating with external services.

---

Lessons from scaling systems from prototype to production, handling millions of requests, and debugging memory leaks at 3 AM.