Как понять, в чем проблема, когда приложение потребляет много памяти?

Диагностика утечек памяти и избыточного потребления

Проблемы с памятью в Python приложениях — сложная и часто встречаемая проблема. За мою практику я отладил десятки таких случаев.

Шаг 1: Понять текущее потребление памяти

import psutil
import os

def check_memory_usage():
    process = psutil.Process(os.getpid())
    
    mem_info = process.memory_info()
    print(f"RSS: {mem_info.rss / 1024 / 1024:.2f} MB")
    print(f"VMS: {mem_info.vms / 1024 / 1024:.2f} MB")
    
    mem_percent = process.memory_percent()
    print(f"Memory: {mem_percent:.2f}%")
    
    with open('/proc/self/status') as f:
        for line in f:
            if line.startswith('VmPeak') or line.startswith('VmHWM'):
                print(line.strip())

check_memory_usage()

Шаг 2: Определить где утекает память (tracemalloc)

import tracemalloc

def find_memory_leaks():
    tracemalloc.start()
    
    data = []
    for i in range(1000000):
        data.append({'id': i, 'value': str(i) * 100})
    
    current, peak = tracemalloc.get_traced_memory()
    print(f"Current: {current / 1024 / 1024:.2f} MB")
    print(f"Peak: {peak / 1024 / 1024:.2f} MB")
    
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')
    
    print("[ Top 10 ]")
    for stat in top_stats[:10]:
        print(stat)
    
    tracemalloc.stop()

find_memory_leaks()

Шаг 3: Типичные причины утечек в Python

Проблема 1: Кеши без лимита

# Плохо
_cache = {}

def expensive_function(key):
    if key not in _cache:
        _cache[key] = compute(key)  # Утечка!
    return _cache[key]

# Хорошо
from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_function(key):
    return compute(key)

# Или ручное ограничение
from collections import OrderedDict

class LimitedCache:
    def __init__(self, maxsize=1000):
        self.cache = OrderedDict()
        self.maxsize = maxsize
    
    def get(self, key):
        if key in self.cache:
            self.cache.move_to_end(key)
            return self.cache[key]
        return None
    
    def set(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.maxsize:
            self.cache.popitem(last=False)

Проблема 2: Ссылки на большие объекты в глобальных переменных

# Плохо
class DataProcessor:
    def __init__(self):
        self.large_data = []
    
    def process(self, file_path):
        with open(file_path) as f:
            self.large_data.extend(f.readlines())  # Растёт!

# Хорошо — обработай и забудь
def process_file_streaming(file_path):
    with open(file_path) as f:
        for line in f:
            yield process_line(line)

# Или явно удали большие объекты
def process_large_file(file_path):
    with open(file_path) as f:
        data = f.read()
    
    result = process(data)
    del data  # Явное удаление
    return result

Проблема 3: Циклические ссылки с del

# Плохо
class Node:
    def __init__(self, value):
        self.value = value
        self.next = None
        self.prev = None  # Циклическая ссылка!

# Хорошо — используй weakref
import weakref

class Node:
    def __init__(self, value):
        self.value = value
        self.next = None
        self._prev = None
    
    @property
    def prev(self):
        return self._prev() if self._prev else None
    
    @prev.setter
    def prev(self, node):
        self._prev = weakref.ref(node) if node else None

Проблема 4: Слушатели событий не удаляются

# Плохо
class EventEmitter:
    def __init__(self):
        self.listeners = {}
    
    def on(self, event, callback):
        if event not in self.listeners:
            self.listeners[event] = []
        self.listeners[event].append(callback)

emitter = EventEmitter()
obj = SomeObject()
emitter.on('update', obj.on_update)
# obj удалён, но callback остался!

# Хорошо
import weakref

class EventEmitter:
    def __init__(self):
        self.listeners = {}
    
    def on(self, event, callback):
        if event not in self.listeners:
            self.listeners[event] = []
        self.listeners[event].append(weakref.WeakMethod(callback))
    
    def emit(self, event, *args):
        if event in self.listeners:
            self.listeners[event] = [
                cb for cb in self.listeners[event] if cb() is not None
            ]
            for cb in self.listeners[event]:
                cb()(*args)

Шаг 4: Мониторинг в production

import tracemalloc
import logging

logger = logging.getLogger(__name__)

class MemoryMonitor:
    def __init__(self, threshold_mb=200):
        self.threshold_mb = threshold_mb
        tracemalloc.start()
        self.baseline, _ = tracemalloc.get_traced_memory()
    
    def check(self):
        current, peak = tracemalloc.get_traced_memory()
        growth = (current - self.baseline) / 1024 / 1024
        
        if growth > self.threshold_mb:
            logger.warning(f"Memory growth: {growth:.2f} MB (peak: {peak/1024/1024:.2f} MB)")
            
            snapshot = tracemalloc.take_snapshot()
            for stat in snapshot.statistics('lineno')[:5]:
                logger.warning(stat)
    
    def reset_baseline(self):
        self.baseline, _ = tracemalloc.get_traced_memory()

monitor = MemoryMonitor(threshold_mb=100)

for item in items:
    process(item)
    monitor.check()

Шаг 5: Анализ с memory_profiler

from memory_profiler import profile

@profile
def my_function():
    data = []
    for i in range(1000000):
        data.append(i ** 2)
    return sum(data)

my_function()

# Запуск: python -m memory_profiler script.py
# Результат:
# Line # | Mem usage | Increment | Codecontext
# 5      | 50.5 MiB  | 50.5 MiB  | data = []
# 6      | 100.2 MiB | 49.7 MiB  | for i in range(1000000):

Шаг 6: Профилирование памяти с Pympler

from pympler import tracker

tr = tracker.SummaryTracker()

process_large_dataset()

tr.print_diff()

Практический пример: Django ORM утечка

# Плохо
def get_all_users_cache():
    if not hasattr(get_all_users_cache, 'cache'):
        get_all_users_cache.cache = list(User.objects.all())
    return get_all_users_cache.cache

# Хорошо — используй iterator
def get_all_users_streaming():
    return User.objects.all().iterator(chunk_size=2000)

# Или используй only/defer
def get_user_names():
    return User.objects.only('id', 'name')

Чеклист для диагностики

• Используй psutil для проверки текущего потребления
• Используй tracemalloc для поиска горячих точек
• Проверь кеши — все ли ограничены по размеру
• Ищи циклические ссылки и используй weakref
• Не сохраняй большие объекты в глобальных переменных
• Используй generators/iterators вместо списков
• Явно удаляй большие объекты через del
• Мониторь память в production с логированием
• Используй memory_profiler для построчного анализа
• Регулярно проверяй самые старые процессы

Инструменты

• psutil: мониторинг системных ресурсов
• tracemalloc: встроённое профилирование
• memory_profiler: построчный анализ
• pympler: детальный анализ объектов
• objgraph: визуализация и утечки
• guppy3: heap анализ