DEV Community

Deep Fix
Deep Fix

Posted on

Fixing Python Memory Leaks in Production – Step‑by‑Step Guide for Developers

#python #memory #debug #devops

Fixing Python Memory Leaks in Production – A Step‑by‑Step Guide for Developers

TL;DR: Detect, profile, and eliminate memory leaks in live Python services using built‑in tools, third‑party profilers, and automated CI checks. The article walks you through concrete code examples and provides a ready‑to‑use patch script.

1. Why Memory Leaks Matter in Production

Even though Python has automatic garbage collection, objects can stay alive longer than intended because of reference cycles, global caches, or native extensions. In a long‑running service, a few megabytes per hour can quickly become gigabytes, leading to OOM crashes, degraded latency, and costly restarts.

2. Common Sources of Leaks

Category Typical Culprit Example
Reference cycles Objects that reference each other without a __del__ guard node = Node(); node.next = node
Global caches Unbounded lru_cache or custom dicts @lru_cache(maxsize=None)
Native extensions C libraries that allocate memory but don’t expose a free function numpy views, ctypes buffers
File/network handles Forgetting to close files or sockets open('log.txt') without close()

3. Detecting Leaks Early

3.1 Using tracemalloc 

import tracemalloc, time
tracemalloc.start()
# ... run workload ...
time.sleep(5)
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
print("[Top 10 memory blocks]")
for stat in top_stats[:10]:
    print(stat)
Enter fullscreen mode Exit fullscreen mode

tracemalloc shows where allocations originate. Compare snapshots before and after a workload to spot growth.

3.2 memory_profiler 

pip install memory_profiler
Enter fullscreen mode Exit fullscreen mode 
from memory_profiler import profile

@profile
def process_batch(batch):
    # heavy processing
    pass
Enter fullscreen mode Exit fullscreen mode

Run with python -m memory_profiler your_script.py to get line‑by‑line memory usage.

3.3 Visualizing Object Graphs with objgraph 

import objgraph
objgraph.show_refs([your_root_object], filename='refs.png')
Enter fullscreen mode Exit fullscreen mode

A quick visual cue for unexpected reference cycles.

4. Profiling in Production

Instrument production code with low‑overhead metrics:

import psutil, time
from prometheus_client import Gauge, start_http_server

mem_gauge = Gauge('process_resident_memory_bytes', 'Resident memory size')

def update_metrics():
    process = psutil.Process()
    mem_gauge.set(process.memory_info().rss)

if __name__ == "__main__":
    start_http_server(8000)  # Prometheus scrapes here
    while True:
        update_metrics()
        time.sleep(10)
Enter fullscreen mode Exit fullscreen mode

Alert on a steady upward trend to catch leaks before they explode.

5. Fixing the Leak – Step‑by‑Step

  1. Reproduce the leak locally with a realistic workload.
  2. Take a baseline snapshot (tracemalloc.take_snapshot()).
  3. Run the workload and take a second snapshot.
  4. Compare snapshots: 
   snapshot1 = tracemalloc.take_snapshot()
   # workload …
   snapshot2 = tracemalloc.take_snapshot()
   top = snapshot2.compare_to(snapshot1, 'lineno')
   for stat in top[:5]:
       print(stat)
Enter fullscreen mode Exit fullscreen mode
  1. Identify the culprit – look for the file/line that shows the biggest increase.
  2. Apply the fix (break cycles, limit cache size, add finally blocks, use weak references, etc.).
  3. Validate by rerunning the comparison; the delta should be near zero.

6. Real‑World Example: Unbounded LRU Cache

Problem 

from functools import lru_cache

@lru_cache(maxsize=None)  # <-- unlimited cache
def compute(key):
    # expensive calculation
    return heavy_algo(key)
Enter fullscreen mode Exit fullscreen mode

The cache grows forever under heavy traffic.

Fix 

@lru_cache(maxsize=1024)  # limit to 1k entries
def compute(key):
    return heavy_algo(key)
Enter fullscreen mode Exit fullscreen mode

Or manually clear rarely used entries:

compute.cache_clear()
Enter fullscreen mode Exit fullscreen mode

7. Automating Leak Checks in CI

Add a lightweight test that fails when memory growth exceeds a threshold:

# .github/workflows/memory-leak.yml
name: Memory Leak Detection
on: [push, pull_request]
jobs:
  leak-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - run: pip install -r requirements.txt
      - name: Run leak test
        run: |
          python - <<'PY'
          import tracemalloc, subprocess, sys
          tracemalloc.start()
          subprocess.run([sys.executable, '-m', 'pytest', 'tests/'], check=True)
          snapshot = tracemalloc.take_snapshot()
          top = snapshot.statistics('filename')
          total = sum(stat.size for stat in top)
          # Fail if > 5 MB allocated during tests
          if total > 5 * 1024 * 1024:
              print('Memory leak detected!')
              sys.exit(1)
          print('Memory usage within limits')
          PY
Enter fullscreen mode Exit fullscreen mode

The pipeline blocks merges that introduce regressions.

8. Ready‑to‑Use Patch Script

We prepared a pre‑configured script that automates snapshot comparison, alerts on growth, and can be dropped into any service.

Just copy the file, adjust the WORKLOAD_CMD, and run it in a staging environment.

9. Takeaways

  • Python’s GC is powerful but not magical – you still need to watch reference patterns.
  • tracemalloc, memory_profiler, and objgraph give you visibility; Prometheus + psutil give you production‑grade alerts.
  • Limit caches, close resources, and use weak references where appropriate.
  • Codify leak detection in CI to prevent regressions.
  • The provided script (linked above) jump‑starts your monitoring.

Happy debugging!

Top comments (0)