Утечки памяти в Python

import sys
import gc
 
# Reference counting
a = []
sys.getrefcount(a)  # 2 (одна — переменная, одна — аргумент функции)
 
b = a
sys.getrefcount(a)  # 3 (добавилась ссылка через b)
 
del b
sys.getrefcount(a)  # 2 (ссылка удалена)
 
# Циклические ссылки — работа для GC
class Node:
    def __init__(self):
        self.ref = None
 
node1 = Node()
node2 = Node()
node1.ref = node2
node2.ref = node1  # Цикл!
 
del node1, node2  # Reference count не достигнет нуля
gc.collect()      # GC обнаружит и освободит цикл

import gc
 
# Настройки порогов сборки мусора
print(gc.get_threshold())  # (700, 10, 10)
# 700 — порог для gen0
# 10 — соотношение gen0/gen1 для сборки gen1
# 10 — соотношение gen1/gen2 для сборки gen2
 
# Статистика
print(gc.get_count())  # (581, 8, 2) — текущее количество объектов

# ❌ Антипаттерн
class FileHandler:
    def __init__(self, filename):
        self.file = open(filename, 'w')
        self.ref = None  # Потенциальный цикл
 
    def __del__(self):
        self.file.close()
 
# Если создать цикл:
handler = FileHandler('test.txt')
handler.ref = handler  # Цикл!
del handler  # Утечка! GC не может вызвать __del__

# ✅ Правильно
class FileHandler:
    def __init__(self, filename):
        self.file = open(filename, 'w')
 
    def __enter__(self):
        return self
 
    def __exit__(self, *args):
        self.file.close()
 
with FileHandler('test.txt') as handler:
    # Файл гарантированно закроется
    pass

# ❌ Утечка: неограниченный кеш
cache = {}
 
def get_user(user_id):
    if user_id not in cache:
        cache[user_id] = fetch_from_db(user_id)  # Кеш растет вечно!
    return cache[user_id]

# ✅ Правильно: LRU-кеш с ограничением
from functools import lru_cache
 
@lru_cache(maxsize=1000)
def get_user(user_id):
    return fetch_from_db(user_id)
 
# ✅ Или TTL-кеш
from cachetools import TTLCache
 
cache = TTLCache(maxsize=1000, ttl=300)  # 5 минут

# ❌ Утечка через замыкание
def create_handler():
    large_data = [0] * 10_000_000  # 80 MB
    small_value = 42
 
    def handler():
        return small_value  # Захватывает и large_data!
 
    return handler
 
callbacks = [create_handler() for _ in range(100)]  # 8 GB утечка!

# ✅ Правильно
def create_handler():
    large_data = [0] * 10_000_000
    small_value = large_data[0]  # Копируем только нужное
    del large_data  # Явно удаляем
 
    def handler():
        return small_value
 
    return handler

# ❌ Утечка файловых дескрипторов
def process_files():
    for filename in large_file_list:
        f = open(filename)  # Не закрыли!
        data = f.read()
        # f остается в памяти
 
# ✅ Правильно
def process_files():
    for filename in large_file_list:
        with open(filename) as f:
            data = f.read()

# ❌ Утечка через event listeners
class EventEmitter:
    def __init__(self):
        self.listeners = []
 
    def on(self, callback):
        self.listeners.append(callback)  # Сильная ссылка!
 
emitter = EventEmitter()
 
class Handler:
    def handle(self, event):
        pass
 
handler = Handler()
emitter.on(handler.handle)  # handler не может быть удален!
del handler  # Утечка — ссылка в listeners

# ✅ Правильно
import weakref
 
class EventEmitter:
    def __init__(self):
        self.listeners = []
 
    def on(self, callback):
        weak_ref = weakref.WeakMethod(callback, self._cleanup)
        self.listeners.append(weak_ref)
 
    def _cleanup(self, ref):
        self.listeners.remove(ref)
 
    def emit(self, event):
        for weak_callback in self.listeners[:]:
            callback = weak_callback()
            if callback is not None:
                callback(event)

import tracemalloc
 
# Старт трассировки
tracemalloc.start()
 
# Код с потенциальной утечкой
large_list = [i for i in range(1_000_000)]
 
# Снимок памяти
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
 
# Топ-10 источников аллокаций
for stat in top_stats[:10]:
    print(stat)

import tracemalloc
import time
 
tracemalloc.start()
 
# Снимок до
snapshot1 = tracemalloc.take_snapshot()
 
# Потенциально текущий код
leaked_data = []
for i in range(10):
    leaked_data.append([0] * 1_000_000)
    time.sleep(0.1)
 
# Снимок после
snapshot2 = tracemalloc.take_snapshot()
 
# Сравнение
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
 
for stat in top_stats[:5]:
    print(stat)

pip install memory-profiler

from memory_profiler import profile
 
@profile
def my_function():
    a = [1] * (10 ** 6)
    b = [2] * (2 * 10 ** 7)
    del b
    return a
 
if __name__ == '__main__':
    my_function()

python -m memory_profiler script.py

pip install pympler

from pympler import tracker
 
tr = tracker.SummaryTracker()
 
# Код
for i in range(3):
    leaked = [0] * 1_000_000
    tr.print_diff()  # Показывает изменения между итерациями

from pympler import asizeof
 
my_list = [1, 2, 3, [4, 5, 6]]
print(asizeof.asizeof(my_list))  # Полный размер с учетом вложенных объектов

from pympler import muppy, summary
 
# Снимок всех объектов в памяти
all_objects = muppy.get_objects()
 
# Суммарная статистика
sum1 = summary.summarize(all_objects)
summary.print_(sum1)

pip install objgraph

import objgraph
 
# Создание утечки
class Node:
    def __init__(self):
        self.ref = None
 
a = Node()
b = Node()
a.ref = b
b.ref = a
 
# Поиск циклов
objgraph.show_backrefs([a], filename='refs.png')  # Граф ссылок
 
# Топ-10 типов объектов в памяти
objgraph.show_most_common_types(limit=10)

import objgraph
 
objgraph.show_growth(limit=5)
 
# Код с утечкой
leaked = []
for i in range(1000):
    leaked.append([0] * 1000)
 
objgraph.show_growth(limit=5)  # Покажет прирост

pip install guppy3

from guppy import hpy
 
h = hpy()
print(h.heap())

from prometheus_client import Gauge, start_http_server
import psutil
import os
 
# Метрики
memory_usage = Gauge('process_memory_bytes', 'Memory usage in bytes')
gc_collections = Gauge('gc_collections_total', 'GC collections', ['generation'])
 
def collect_metrics():
    process = psutil.Process(os.getpid())
    memory_usage.set(process.memory_info().rss)
 
    import gc
    stats = gc.get_stats()
    for i, stat in enumerate(stats):
        gc_collections.labels(generation=i).set(stat['collections'])
 
# Запуск метрик на порту 8000
start_http_server(8000)
 
# Периодический сбор
import threading
def monitor():
    while True:
        collect_metrics()
        threading.Event().wait(10)
 
threading.Thread(target=monitor, daemon=True).start()

# Рост памяти более 100 MB за 5 минут
rate(process_memory_bytes[5m]) > 100_000_000

import gc
import threading
import time
 
def periodic_gc():
    while True:
        time.sleep(300)  # Каждые 5 минут
        collected = gc.collect()
        print(f"GC collected {collected} objects")
 
# Запуск в фоне
gc_thread = threading.Thread(target=periodic_gc, daemon=True)
gc_thread.start()

import os
import tracemalloc
 
if os.environ.get('MEMORY_PROFILING') == '1':
    tracemalloc.start()
 
    # Периодический дамп
    def dump_snapshot():
        snapshot = tracemalloc.take_snapshot()
        with open(f'/tmp/memory_{time.time()}.dump', 'w') as f:
            for stat in snapshot.statistics('lineno')[:50]:
                f.write(f"{stat}\n")
 
    # Вызов по расписанию или по сигналу
    signal.signal(signal.SIGUSR1, lambda sig, frame: dump_snapshot())

# Включить профилирование
export MEMORY_PROFILING=1
systemctl restart myapp
 
# Получить снимок
kill -USR1 <pid>
 
# Анализ
cat /tmp/memory_*.dump

import psutil
import os
import sys
 
MAX_MEMORY_MB = 1024  # 1 GB
 
def check_memory():
    process = psutil.Process(os.getpid())
    memory_mb = process.memory_info().rss / 1024 / 1024
 
    if memory_mb > MAX_MEMORY_MB:
        print(f"Memory limit exceeded: {memory_mb:.2f} MB. Restarting...")
        # Graceful shutdown
        cleanup()
        os.execv(sys.executable, [sys.executable] + sys.argv)
 
# Проверка каждую минуту
import threading
def monitor():
    while True:
        check_memory()
        threading.Event().wait(60)
 
threading.Thread(target=monitor, daemon=True).start()

# Установка gdb
apt-get install gdb
 
# Дамп процесса
gcore <pid>
 
# Анализ с помощью pyrasite (для живого процесса)
pip install pyrasite
pyrasite-shell <pid>

import gc
import objgraph
 
objgraph.show_most_common_types(limit=20)
gc.collect()