line_profiler and memory_profiler - Line-Level Analysis

Study this function and predict which line is the bottleneck:

def process_log_file(filepath):
    with open(filepath) as f:
        lines = f.readlines()                    # Line A

    entries = []
    for line in lines:
        parts = line.strip().split(',')          # Line B
        if len(parts) >= 4:
            entry = {
                'timestamp': parts[0],
                'level': parts[1],
                'source': parts[2],
                'message': ','.join(parts[3:]),  # Line C
            }
            entries.append(entry)                # Line D

    # Deduplicate by message
    seen = set()
    unique = []
    for entry in entries:
        key = entry['message']
        if key not in seen:                      # Line E
            seen.add(key)
            unique.append(entry)

    return sorted(unique, key=lambda e: e['timestamp'])  # Line F

Most engineers guess Line A (file I/O) or Line F (sorting). On a 2 million line log file with 50% duplicate messages, line_profiler reveals the truth:

Line #  % Time  Line Contents
     3    4.2%      lines = f.readlines()
     7   28.1%      parts = line.strip().split(',')
     9    3.8%          'timestamp': parts[0],
    12   11.3%          'message': ','.join(parts[3:]),
    13   18.7%          entries.append(entry)
    19    2.1%      if key not in seen:
    23   26.4%  return sorted(unique, key=lambda e: e['timestamp'])

Line B (split) and Line D (append) together consume 46.8% of execution time. Not I/O. Not sorting. String splitting and list appending on 2 million iterations. cProfile would have told you "process_log_file is slow." line_profiler tells you exactly which lines to fix.

What You Will Learn

• How to use line_profiler with @profile and kernprof for line-by-line timing
• How to read and interpret line_profiler output
• How to use memory_profiler to track memory consumption per line
• How to use tracemalloc for memory allocation snapshots and diff analysis
• The difference between sys.getsizeof and pympler.asizeof for object sizing
• How to detect and diagnose memory leaks in long-running services
• Real-world patterns for finding memory leaks in production

Prerequisites

• Completed Lesson 2 (cProfile and pstats)
• Understanding of Python's memory model (reference counting, garbage collection from Intermediate course)
• Familiarity with generators and iterators

Part 1 - line_profiler: Time Per Line

line_profiler is a third-party tool that instruments individual lines within functions. Where cProfile tells you "this function took 3.2 seconds," line_profiler tells you "line 7 took 1.8 seconds of that."

Installation

pip install line_profiler

Usage with kernprof

The standard workflow uses kernprof, a command-line wrapper:

# save as example.py

@profile  # This decorator is injected by kernprof - do NOT import it
def compute_statistics(data):
    n = len(data)                              # Line 4
    mean = sum(data) / n                       # Line 5
    squared_diffs = [(x - mean) ** 2 for x in data]  # Line 6
    variance = sum(squared_diffs) / n          # Line 7
    std_dev = variance ** 0.5                  # Line 8

    sorted_data = sorted(data)                 # Line 10
    median_idx = n // 2                        # Line 11
    if n % 2 == 0:
        median = (sorted_data[median_idx - 1] + sorted_data[median_idx]) / 2
    else:
        median = sorted_data[median_idx]       # Line 14

    return {'mean': mean, 'std': std_dev, 'median': median}

import random
data = [random.gauss(0, 1) for _ in range(1_000_000)]
compute_statistics(data)

# Run with kernprof
kernprof -l -v example.py

Output:

Wrote profile results to example.py.lprof
Timer unit: 1e-06 s

Total time: 0.892 s
Function: compute_statistics at line 3

Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
     4         1          1.0      1.0      0.0      n = len(data)
     5         1      35420.0  35420.0      4.0      mean = sum(data) / n
     6         1     380123.0 380123.0     42.6      squared_diffs = [(x - mean) ** 2 for x in data]
     7         1      42310.0  42310.0      4.7      variance = sum(squared_diffs) / n
     8         1          2.0      2.0      0.0      std_dev = variance ** 0.5
    10         1     412890.0 412890.0     46.3      sorted_data = sorted(data)
    11         1          1.0      1.0      0.0      median_idx = n // 2
    14         1          2.0      2.0      0.0          median = sorted_data[median_idx]
    16         1          5.0      5.0      0.0      return {'mean': mean, 'std': std_dev, 'median': median}

Reading the Output

Column	Meaning
Line #	Source line number
Hits	Number of times this line was executed
Time	Total time spent on this line (microseconds)
Per Hit	Time per execution (Time / Hits)
% Time	Percentage of total function time
Line Contents	The source code

In our example, two lines dominate:

• Line 6 (list comprehension for squared diffs): 42.6%
• Line 10 (sorting): 46.3%

Together they consume 88.9% of the function's time. Everything else is noise.

:::tip Focus on % Time, Not Absolute Time Absolute times vary between machines and runs. The % Time column tells you the relative cost of each line, which is stable and actionable. :::

Programmatic Usage (Without kernprof)

You can use line_profiler programmatically, which is useful in notebooks or test suites:

from line_profiler import LineProfiler

def target_function(data):
    result = []
    for item in data:
        processed = item ** 2
        if processed > 100:
            result.append(processed)
    return result

# Create profiler and add function to profile
lp = LineProfiler()
lp.add_function(target_function)

# Run the function through the profiler
data = list(range(10_000))
lp_wrapper = lp(target_function)
lp_wrapper(data)

# Print results
lp.print_stats()

Profiling Multiple Functions

from line_profiler import LineProfiler

def parse(raw):
    tokens = raw.split()
    return [t.lower() for t in tokens]

def transform(tokens):
    return [t[::-1] for t in tokens if len(t) > 3]

def pipeline(raw_data):
    parsed = parse(raw_data)
    result = transform(parsed)
    return result

# Profile all three functions
lp = LineProfiler()
lp.add_function(parse)
lp.add_function(transform)
lp.add_function(pipeline)

lp_wrapper = lp(pipeline)
lp_wrapper("the quick brown fox jumps over the lazy dog " * 100_000)
lp.print_stats()

Part 2 - memory_profiler: Memory Per Line

Where line_profiler shows time per line, memory_profiler shows memory consumption per line. It tracks the process's resident set size (RSS) before and after each line.

Installation

pip install memory_profiler

Basic Usage

# save as mem_example.py
from memory_profiler import profile

@profile
def memory_demo():
    # Baseline memory
    small_list = list(range(1_000))           # Line 5: ~negligible
    big_list = list(range(1_000_000))          # Line 6: ~8 MB (64-bit ints)
    matrix = [list(range(1000)) for _ in range(1000)]  # Line 7: ~8 MB

    # Memory release
    del big_list                               # Line 10: frees ~8 MB

    # String concatenation - surprisingly expensive
    result = ""
    for i in range(100_000):
        result += str(i)                       # Line 14: quadratic memory churn

    return len(result)

memory_demo()

python -m memory_profiler mem_example.py

Output:

Filename: mem_example.py

Line #    Mem usage    Increment  Occurrences   Line Contents
=============================================================
     3     45.2 MiB     45.2 MiB           1   @profile
     4                                         def memory_demo():
     5     45.3 MiB      0.1 MiB           1       small_list = list(range(1_000))
     6     53.4 MiB      8.1 MiB           1       big_list = list(range(1_000_000))
     7     61.6 MiB      8.2 MiB           1       matrix = [list(range(1000)) for _ in range(1000)]
    10     53.5 MiB     -8.1 MiB           1       del big_list
    14     78.9 MiB     25.4 MiB      100001       result += str(i)
    16     78.9 MiB      0.0 MiB           1   return len(result)

Reading the Output

Column	Meaning
Line #	Source line number
Mem usage	Total process memory after this line
Increment	Change in memory from previous line
Occurrences	Number of times this line executed

:::danger String Concatenation is a Memory Disaster Line 14 (result += str(i)) consumes 25.4 MiB because each += creates a new string object. Python must allocate a new string, copy the old content, append the new content, and then the old string becomes garbage. For 100,000 iterations, this creates ~100,000 temporary string objects. Use ''.join() instead. :::

The Fix

from memory_profiler import profile

@profile
def memory_demo_fixed():
    small_list = list(range(1_000))
    big_list = list(range(1_000_000))
    matrix = [list(range(1000)) for _ in range(1000)]

    del big_list

    # Fixed: use join instead of += concatenation
    parts = [str(i) for i in range(100_000)]   # One allocation
    result = ''.join(parts)                     # One final string

    return len(result)

# Mem usage for the join approach: ~2 MiB instead of ~25 MiB

Part 3 - tracemalloc: Allocation-Level Memory Analysis

tracemalloc is a built-in module (since Python 3.4) that tracks memory allocations at the Python level. Unlike memory_profiler which tracks RSS, tracemalloc tracks individual malloc calls, giving you precise attribution.

Basic Snapshot

import tracemalloc

tracemalloc.start()

# Your code here
data = [dict(id=i, name=f"user_{i}", scores=[j*i for j in range(100)])
        for i in range(10_000)]

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("Top 10 memory allocations:")
for stat in top_stats[:10]:
    print(f"  {stat}")

# Output:
# <ipython>:4: size=76.3 MiB, count=1010001, average=79 B
# <ipython>:4: size=8.4 MiB, count=10000, average=880 B
# ...

Comparing Snapshots (Leak Detection)

The killer feature of tracemalloc is snapshot comparison. Take a snapshot before and after some operation, and see exactly what memory was allocated and not freed:

import tracemalloc
import gc

tracemalloc.start()

# Snapshot before
gc.collect()
snapshot1 = tracemalloc.take_snapshot()

# Run the suspicious operation
leaked_data = []
for i in range(1000):
    # Simulate a leak: accumulating data that is never cleaned up
    leaked_data.append({
        'id': i,
        'payload': b'\x00' * 10_000,  # 10KB per entry
        'metadata': {'processed': False, 'retries': 0},
    })

# Snapshot after
gc.collect()
snapshot2 = tracemalloc.take_snapshot()

# Compare
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
print("Memory changes (top 10):")
for stat in top_stats[:10]:
    print(f"  {stat}")

# Output:
#   example.py:12: size=9.77 MiB (+9.77 MiB), count=1000 (+1000)
#   example.py:13: size=78.2 KiB (+78.2 KiB), count=1000 (+1000)
#   ...

Grouping by Traceback

For deeper analysis, group by the full call stack, not just the line:

import tracemalloc

tracemalloc.start(25)  # Store up to 25 frames in traceback

# ... your code ...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('traceback')

# Show the full traceback for the top allocation
print("Largest allocation traceback:")
for line in top_stats[0].traceback.format():
    print(f"  {line}")

# Output:
#   File "models.py", line 45
#     return cls(**row)
#   File "database.py", line 120
#     results = [Model.from_row(row) for row in cursor.fetchall()]
#   File "views.py", line 30
#     courses = get_all_courses()

Monitoring Memory Over Time

import tracemalloc
import time

def monitor_memory(func, interval_seconds=1.0, duration_seconds=30.0):
    """
    Monitor memory allocations over time during a long-running operation.
    Useful for detecting gradual leaks.
    """
    tracemalloc.start()
    snapshots = []

    start_time = time.time()
    baseline = tracemalloc.take_snapshot()

    # Start the function in a thread or run it in chunks
    # Here we simulate periodic monitoring
    import threading

    result_holder = [None]
    def run():
        result_holder[0] = func()

    thread = threading.Thread(target=run)
    thread.start()

    while thread.is_alive() and (time.time() - start_time) < duration_seconds:
        time.sleep(interval_seconds)
        snapshot = tracemalloc.take_snapshot()
        current, peak = tracemalloc.get_traced_memory()
        elapsed = time.time() - start_time
        snapshots.append({
            'time': elapsed,
            'current_mb': current / 1024 / 1024,
            'peak_mb': peak / 1024 / 1024,
        })
        print(f"  [{elapsed:.1f}s] Current: {current/1024/1024:.1f} MiB, "
              f"Peak: {peak/1024/1024:.1f} MiB")

    thread.join()

    # Final comparison
    final = tracemalloc.take_snapshot()
    diff = final.compare_to(baseline, 'lineno')
    print("\nMemory changes from start to finish:")
    for stat in diff[:10]:
        print(f"  {stat}")

    tracemalloc.stop()
    return result_holder[0], snapshots

Part 4 - Object Sizing: sys.getsizeof vs pympler

Understanding how much memory an object actually consumes is surprisingly tricky in Python.

sys.getsizeof - Shallow Size

sys.getsizeof returns the shallow size of an object - the memory consumed by the object itself, not including objects it references.

import sys

# Integers
print(sys.getsizeof(0))        # 28 bytes (CPython 3.11+)
print(sys.getsizeof(1))        # 28 bytes
print(sys.getsizeof(2**30))    # 32 bytes (larger int needs more space)
print(sys.getsizeof(2**100))   # 44 bytes

# Strings
print(sys.getsizeof(''))       # 49 bytes (empty string has overhead)
print(sys.getsizeof('hello'))  # 54 bytes (49 + 5 chars, Latin-1)
print(sys.getsizeof('hello' * 100))  # 549 bytes

# Collections - SHALLOW size only!
data = [1, 2, 3, 4, 5]
print(sys.getsizeof(data))     # 120 bytes - size of the list OBJECT
# But this does NOT include the integers it contains!
# Total = 120 (list) + 5 * 28 (ints) = 260 bytes

# Dict
d = {'a': 1, 'b': 2}
print(sys.getsizeof(d))        # 232 bytes - size of the dict OBJECT
# Does NOT include keys or values

:::danger sys.getsizeof Lies About Container Size sys.getsizeof([1, 2, 3]) returns ~88 bytes, but the list actually consumes 88 + 3 * 28 = 172 bytes when you include the integer objects. For nested structures, the undercount is dramatic. Never use sys.getsizeof alone for containers. :::

pympler.asizeof - Deep (Recursive) Size

pympler traverses the entire object graph and sums up the total memory:

from pympler import asizeof

# Compare shallow vs deep sizing
data = [list(range(100)) for _ in range(100)]

shallow = sys.getsizeof(data)                    # Just the outer list
deep = asizeof.asizeof(data)                     # Outer list + inner lists + ints

print(f"sys.getsizeof: {shallow:,} bytes")       # ~856 bytes
print(f"pympler.asizeof: {deep:,} bytes")        # ~326,456 bytes
print(f"Ratio: {deep/shallow:.0f}x")             # ~381x undercount!

# Deep sizing a dict of objects
class User:
    def __init__(self, name, email, scores):
        self.name = name
        self.email = email
        self.scores = scores

users = [User(f"user_{i}", f"user_{i}@test.com", list(range(50)))
         for i in range(1000)]

print(f"1000 Users: {asizeof.asizeof(users) / 1024 / 1024:.1f} MiB")

When to Use Each

Tool	Use case
sys.getsizeof	Quick check on primitive types, understanding per-object overhead
pympler.asizeof	Accurate total memory of complex structures
tracemalloc	Finding which code allocated the memory
memory_profiler	Line-by-line process RSS changes

Part 5 - Finding Memory Leaks in Long-Running Services

Memory leaks in Python are not true leaks (unreachable memory) - they are almost always unintentional retention: objects that remain reachable through some reference chain, preventing garbage collection.

Common Leak Patterns

Pattern 1: Growing Collections

# LEAK: global list that grows forever
_request_log = []

async def handle_request(request):
    result = await process(request)
    _request_log.append({  # This list grows without bound
        'timestamp': time.time(),
        'path': request.path,
        'response_size': len(result),
    })
    return result

# FIX: Use a bounded collection
from collections import deque
_request_log = deque(maxlen=10_000)  # Automatically evicts oldest entries

Pattern 2: Callback Leaks

# LEAK: registered callbacks hold strong references to large objects
class DataProcessor:
    def __init__(self, data):
        self.data = data  # 100 MB of data
        event_bus.subscribe('on_update', self.handle_update)

    def handle_update(self, event):
        # process event using self.data
        pass

# Even after all external references to DataProcessor are dropped,
# the event_bus subscription keeps self alive, retaining the 100 MB.

# FIX: Use weakref callbacks (covered in Lesson 5) or explicit unsubscribe
class DataProcessor:
    def __init__(self, data):
        self.data = data
        self._subscription = event_bus.subscribe('on_update', self.handle_update)

    def close(self):
        event_bus.unsubscribe(self._subscription)

    def __del__(self):
        self.close()

Pattern 3: Traceback Reference Retention

# LEAK: exception handling retains local variables via traceback
import sys

def process_large_data():
    huge_array = bytearray(100_000_000)  # 100 MB
    try:
        result = risky_operation(huge_array)
    except Exception:
        # sys.exc_info()[2] holds the traceback, which references
        # the local variable 'huge_array' through frame objects
        exc_type, exc_value, exc_tb = sys.exc_info()
        log_error(exc_type, exc_value, exc_tb)
        # huge_array is retained as long as exc_tb exists!

    # Even after the except block, the 100 MB may not be freed
    # until the next GC cycle or until exc_tb is explicitly deleted

# FIX: Delete traceback references explicitly
def process_large_data():
    huge_array = bytearray(100_000_000)
    try:
        result = risky_operation(huge_array)
    except Exception:
        exc_type, exc_value, exc_tb = sys.exc_info()
        try:
            log_error(exc_type, exc_value, exc_tb)
        finally:
            del exc_tb  # Break the reference chain

Leak Detection Framework

import tracemalloc
import gc
import time
from typing import Callable

class LeakDetector:
    """
    Detect memory leaks by running an operation repeatedly
    and checking if memory grows monotonically.
    """

    def __init__(self, operation: Callable, warmup_rounds: int = 5):
        self.operation = operation
        self.warmup_rounds = warmup_rounds

    def run(self, rounds: int = 20, gc_between: bool = True) -> dict:
        tracemalloc.start(10)

        # Warmup - let caches fill, JIT warm up, etc.
        for _ in range(self.warmup_rounds):
            self.operation()
        if gc_between:
            gc.collect()

        # Baseline
        baseline_snapshot = tracemalloc.take_snapshot()
        memory_readings = []

        for i in range(rounds):
            self.operation()
            if gc_between:
                gc.collect()

            current, peak = tracemalloc.get_traced_memory()
            memory_readings.append(current)

        # Final snapshot
        final_snapshot = tracemalloc.take_snapshot()

        # Analyze growth
        if len(memory_readings) >= 4:
            first_quarter = sum(memory_readings[:len(memory_readings)//4]) / (len(memory_readings)//4)
            last_quarter = sum(memory_readings[-len(memory_readings)//4:]) / (len(memory_readings)//4)
            growth_ratio = last_quarter / first_quarter if first_quarter > 0 else 0
        else:
            growth_ratio = 0

        # Diff snapshots
        diff = final_snapshot.compare_to(baseline_snapshot, 'traceback')

        result = {
            'leak_detected': growth_ratio > 1.10,  # >10% growth = likely leak
            'growth_ratio': growth_ratio,
            'memory_start_mb': memory_readings[0] / 1024 / 1024,
            'memory_end_mb': memory_readings[-1] / 1024 / 1024,
            'top_allocations': [],
        }

        for stat in diff[:5]:
            if stat.size_diff > 0:
                result['top_allocations'].append({
                    'size_diff_kb': stat.size_diff / 1024,
                    'count_diff': stat.count_diff,
                    'traceback': [str(line) for line in stat.traceback.format()],
                })

        tracemalloc.stop()
        return result


# Usage
def suspicious_operation():
    """This operation might leak memory."""
    global _cache  # Simulated unbounded cache
    if not hasattr(suspicious_operation, '_cache'):
        suspicious_operation._cache = {}
    suspicious_operation._cache[time.time()] = list(range(1000))

detector = LeakDetector(suspicious_operation, warmup_rounds=3)
result = detector.run(rounds=50)

if result['leak_detected']:
    print(f"LEAK DETECTED! Growth ratio: {result['growth_ratio']:.2f}x")
    print(f"Memory: {result['memory_start_mb']:.1f} MiB -> {result['memory_end_mb']:.1f} MiB")
    print("Top growing allocations:")
    for alloc in result['top_allocations']:
        print(f"  +{alloc['size_diff_kb']:.1f} KB ({alloc['count_diff']} objects)")
        for line in alloc['traceback'][:3]:
            print(f"    {line}")
else:
    print("No leak detected.")

Part 6 - Real-World: Profiling a Long-Running Service

Here is a complete example of diagnosing a memory leak in a background task worker:

import tracemalloc
import gc
import asyncio
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class MemoryMonitor:
    """
    Periodic memory monitoring for long-running async services.
    Integrates with tracemalloc for allocation tracking.
    """

    def __init__(self, interval_seconds: float = 60.0,
                 alert_growth_mb: float = 50.0):
        self.interval = interval_seconds
        self.alert_threshold = alert_growth_mb
        self.readings = []
        self._task = None

    async def start(self):
        """Start periodic monitoring."""
        tracemalloc.start(5)
        self._baseline = tracemalloc.take_snapshot()
        self._task = asyncio.create_task(self._monitor_loop())
        logger.info("Memory monitoring started")

    async def stop(self):
        """Stop monitoring and produce final report."""
        if self._task:
            self._task.cancel()
        self._produce_report()
        tracemalloc.stop()

    async def _monitor_loop(self):
        while True:
            await asyncio.sleep(self.interval)
            gc.collect()

            current, peak = tracemalloc.get_traced_memory()
            current_mb = current / 1024 / 1024
            peak_mb = peak / 1024 / 1024

            self.readings.append({
                'timestamp': datetime.utcnow().isoformat(),
                'current_mb': current_mb,
                'peak_mb': peak_mb,
            })

            logger.info(f"Memory: current={current_mb:.1f}MB, peak={peak_mb:.1f}MB")

            # Check for concerning growth
            if len(self.readings) >= 10:
                recent = [r['current_mb'] for r in self.readings[-10:]]
                growth = recent[-1] - recent[0]
                if growth > self.alert_threshold:
                    logger.warning(
                        f"Memory grew {growth:.1f}MB in last "
                        f"{10 * self.interval:.0f}s - possible leak!"
                    )
                    self._dump_leak_info()

    def _dump_leak_info(self):
        """Dump detailed allocation info when a leak is suspected."""
        snapshot = tracemalloc.take_snapshot()
        diff = snapshot.compare_to(self._baseline, 'traceback')

        logger.warning("Top memory growth sources:")
        for stat in diff[:10]:
            if stat.size_diff > 1024 * 1024:  # Only show >1MB growth
                logger.warning(f"  +{stat.size_diff/1024/1024:.1f}MB "
                             f"({stat.count_diff} allocations)")
                for line in stat.traceback.format()[:5]:
                    logger.warning(f"    {line}")

    def _produce_report(self):
        """Final summary report."""
        if not self.readings:
            return

        print("\n" + "=" * 60)
        print("MEMORY MONITORING REPORT")
        print("=" * 60)
        print(f"Duration: {self.readings[0]['timestamp']} to "
              f"{self.readings[-1]['timestamp']}")
        print(f"Readings: {len(self.readings)}")
        print(f"Start:    {self.readings[0]['current_mb']:.1f} MB")
        print(f"End:      {self.readings[-1]['current_mb']:.1f} MB")
        print(f"Peak:     {max(r['peak_mb'] for r in self.readings):.1f} MB")

        total_growth = self.readings[-1]['current_mb'] - self.readings[0]['current_mb']
        print(f"Growth:   {total_growth:+.1f} MB")

        if total_growth > self.alert_threshold:
            print("STATUS:   LEAK SUSPECTED")
        else:
            print("STATUS:   OK")


# Integration with a FastAPI app
# from fastapi import FastAPI
# app = FastAPI()
# monitor = MemoryMonitor(interval_seconds=30, alert_growth_mb=100)
#
# @app.on_event("startup")
# async def startup():
#     await monitor.start()
#
# @app.on_event("shutdown")
# async def shutdown():
#     await monitor.stop()

Combining line_profiler and memory_profiler

For the deepest analysis, use both tools on the same function:

# Step 1: Use line_profiler to find the slow lines
# kernprof -l -v my_module.py

# Step 2: Use memory_profiler on the same function to see memory impact
# python -m memory_profiler my_module.py

# Step 3: Cross-reference - lines that are both slow AND memory-heavy
# are your highest-priority optimization targets

# Example: a function that is both slow and memory-hungry
from memory_profiler import profile as mem_profile

@mem_profile
def process_csv(filepath):
    import csv

    # Time: 5%,  Memory: +0.1 MB - not a concern
    with open(filepath) as f:
        reader = csv.DictReader(f)
        rows = list(reader)          # Time: 15%, Memory: +200 MB - PROBLEM

    # Time: 60%, Memory: +400 MB - BIGGEST PROBLEM
    enriched = []
    for row in rows:
        enriched.append({
            **row,
            'full_name': f"{row['first']} {row['last']}",
            'score_avg': sum(float(row[f's{i}']) for i in range(1, 11)) / 10,
        })

    # Time: 20%, Memory: +0 MB - slow but not a memory issue
    sorted_data = sorted(enriched, key=lambda r: r['score_avg'], reverse=True)

    return sorted_data[:100]

# The fix: use generators, process in chunks, don't materialize everything
def process_csv_fixed(filepath):
    import csv
    import heapq

    def generate_enriched(filepath):
        with open(filepath) as f:
            for row in csv.DictReader(f):
                yield {
                    **row,
                    'full_name': f"{row['first']} {row['last']}",
                    'score_avg': sum(float(row[f's{i}']) for i in range(1, 11)) / 10,
                }

    # heapq.nlargest consumes the generator without materializing the full list
    return heapq.nlargest(100, generate_enriched(filepath),
                          key=lambda r: r['score_avg'])

Key Takeaways

• line_profiler shows time per line: use kernprof -l -v or the programmatic API to find exactly which lines within a function are the bottleneck.
• memory_profiler shows RSS per line: the Increment column reveals which lines allocate (or free) significant memory.
• tracemalloc tracks allocations at the Python level: snapshot comparison (compare_to) is the most effective tool for finding memory leaks.
• sys.getsizeof is shallow: it returns the size of the container object itself, not its contents. Use pympler.asizeof for deep/recursive sizing.
• Python memory leaks are retention problems: objects remain reachable through growing collections, callback registrations, or traceback references.
• Monitor memory in production: periodic tracemalloc snapshots catch leaks before they cause OOM crashes.
• Combine tools: line_profiler for time, memory_profiler for memory, tracemalloc for allocation tracking - used together, they give you the complete picture.

Graded Practice Challenges

Level 1 - Predict the Output

Question 1: What does sys.getsizeof([1, 2, 3]) return, and why is it misleading?

Answer

On CPython 3.11 (64-bit), it returns approximately 88 bytes. This is misleading because it only measures the list object's internal storage (pointer array + overhead), not the integer objects it references. The actual total memory is approximately 88 + 3 * 28 = 172 bytes. For deeply nested structures, sys.getsizeof can undercount by 100x or more.

Question 2: In a memory_profiler report, you see Increment: -15.2 MiB on a line that says del large_list. Does this mean 15.2 MiB was freed?

Answer

Not necessarily. It means the process's RSS (Resident Set Size) decreased by 15.2 MiB. The memory was released by CPython's allocator, and the OS may or may not have reclaimed the pages. In practice, CPython often retains memory arenas for reuse (pymalloc arena behavior), so the OS-level RSS drop may be smaller or delayed. However, for practical purposes, if you see a negative increment after del, the Python objects were deallocated.

Question 3: You use tracemalloc's compare_to between two snapshots and see the line models.py:45 size=50.2 MiB (+50.2 MiB), count=100000 (+100000). What does this tell you?

Answer

Between the two snapshots, line 45 of models.py allocated 100,000 objects totaling 50.2 MiB that were not freed. This is a strong indicator of a memory leak or an unbounded accumulation. The average object size is ~527 bytes. You should inspect line 45 to see what objects are being created and trace why they are not being garbage collected (likely stored in a growing collection).

Level 2 - Debug Challenge

This memory monitoring code is supposed to detect leaks but fails to catch a real leak. Find the bugs:

import tracemalloc
import gc

def check_for_leaks(func, rounds=10):
    tracemalloc.start()

    snapshot_before = tracemalloc.take_snapshot()

    for _ in range(rounds):
        func()

    snapshot_after = tracemalloc.take_snapshot()

    diff = snapshot_after.compare_to(snapshot_before, 'lineno')
    leaks = [s for s in diff if s.size_diff > 0]

    if leaks:
        print("Leaks detected!")
    else:
        print("No leaks.")

    tracemalloc.stop()

# Test with a known leak
_cache = {}
counter = [0]

def leaky_function():
    counter[0] += 1
    _cache[counter[0]] = [0] * 1000

check_for_leaks(leaky_function)  # Prints "No leaks." - wrong!

Answer

Two bugs:

1. No gc.collect() before snapshots: Without forcing garbage collection, some objects may be in gc's pending list, skewing the snapshots. Add gc.collect() before each tracemalloc.take_snapshot().

2. The main bug: tracemalloc.start() is called after the global _cache = {} and counter = [0] already exist. tracemalloc only tracks allocations that occur after start() is called. The _cache dict was created before tracing began. When items are added to it, tracemalloc does track those allocations. However, the real issue is more subtle: on some Python versions, the _cache[counter[0]] = [0] * 1000 allocation might be attributed to the list creation line in leaky_function, not to the dict. The diff filter s.size_diff > 0 should catch this.

Actually, the code should work for detecting the leak. Let me re-examine. The real bug is that the function does produce output - the issue is that leaky_function creates [0] * 1000 which allocates lists, and those allocations ARE tracked. So the likely issue is the test setup: run it and verify. The more likely real-world bug is:

Missing warmup: The first few calls may trigger module imports or one-time initializations that look like leaks. Add warmup rounds before the baseline snapshot:

def check_for_leaks(func, rounds=10, warmup=3):
    tracemalloc.start()

    for _ in range(warmup):
        func()
    gc.collect()

    snapshot_before = tracemalloc.take_snapshot()

    for _ in range(rounds):
        func()
    gc.collect()

    snapshot_after = tracemalloc.take_snapshot()

    diff = snapshot_after.compare_to(snapshot_before, 'lineno')
    leaks = [s for s in diff if s.size_diff > 1024]  # Ignore tiny allocations

    if leaks:
        print("Leaks detected!")
        for s in leaks[:5]:
            print(f"  {s}")
    else:
        print("No leaks.")

    tracemalloc.stop()

Level 3 - Design Challenge

You operate a Python microservice that processes 10,000 events per minute. After 24 hours of uptime, memory usage grows from 200 MB to 1.8 GB, at which point the OOM killer terminates the process. Design a comprehensive strategy to:

1. Identify which objects are leaking
2. Find the code path responsible
3. Fix the leak without significant downtime
4. Prevent future leaks

Include specific tools, code snippets, and a monitoring architecture.

Solution Sketch

Phase 1: Characterize the leak (2 hours)

Start tracemalloc in production with low overhead:

# Add to service startup
import tracemalloc
tracemalloc.start(3)  # 3 frames - low overhead

# Add a diagnostic endpoint (authenticated!)
@app.get("/debug/memory")
async def memory_debug(auth: AdminAuth = Depends()):
    snapshot = tracemalloc.take_snapshot()
    stats = snapshot.statistics('traceback')
    return {
        "current_mb": tracemalloc.get_traced_memory()[0] / 1024 / 1024,
        "peak_mb": tracemalloc.get_traced_memory()[1] / 1024 / 1024,
        "top_allocations": [
            {"size_mb": s.size / 1024 / 1024,
             "count": s.count,
             "traceback": s.traceback.format()[:3]}
            for s in stats[:20]
        ],
    }

Poll this endpoint every 5 minutes and plot memory over time.

Phase 2: Reproduce locally (4 hours)

Use the LeakDetector class from this lesson. Feed it with production-like event data:

detector = LeakDetector(
    lambda: process_event(sample_event),
    warmup_rounds=100,
)
result = detector.run(rounds=10_000)

Phase 3: Fix

Common fixes based on root cause:

• Growing dict/list: Use collections.deque(maxlen=N) or add periodic cleanup
• Callback leak: Use weakref.WeakMethod for registered callbacks
• Traceback retention: Explicitly del traceback references in exception handlers
• Thread-local accumulation: Clear thread-locals after each request

Phase 4: Prevention

• Add a memory budget assertion to CI tests
• Add the MemoryMonitor class to production with alerting
• Add objgraph snapshots to staging environment for periodic analysis
• Code review checklist: "Does this function accumulate state across calls?"

What's Next

Now that you can find exactly where time and memory are spent, the next lesson covers the most impactful optimization technique: Caching Strategies. You will learn how to trade memory for speed using functools.lru_cache, TTL caches, and external caching with Redis.