Управление памятью

Управление памятью в Python

Маршрут: Устройство среды выполнения · тема 1 из 3

Навыки: S2, S3, S5

До урока: basic_types, oop, iterators_generators, profiling.

Результат: объяснение времени жизни и поведения памяти с указанием границы языка, реализации и версии.

Основа: ссылки, изменяемость и время жизни объекта. Работа в эксплуатации: поиск утечек, слабые ссылки и tracemalloc. Углубление: PyObject, циклический сборщик, распределители и CPython без GIL.

Подтверждение: опыт с tracemalloc/GC и техническая записка с учётом версии.

Python управляет временем жизни объектов автоматически. На практике важно различать гарантии языка и детали конкретной сборки CPython: счётчики ссылок, структуры объектов и настройки сборщика могут меняться между версиями.

#1. Модель памяти Python

В CPython каждый объект Python — это структура в C:

// Упрощённо:
typedef struct {
    Py_ssize_t ob_refcnt;     // счётчик ссылок
    PyTypeObject *ob_type;    // тип объекта
    // ... данные объекта
} PyObject;

Объекты переменного размера используют заголовок PyVarObject, который добавляет поле ob_size. Это описание относится к CPython; через стабильный C API детали всё чаще скрываются за функциями. На Python-уровне наблюдайте счётчик через sys.getrefcount(), а не читайте память по id(obj) с помощью ctypes: такая раскладка не является гарантией языка.

Heap — вся динамически выделяемая память. Python имеет свой аллокатор поверх системного malloc.

import sys

x = [1, 2, 3]
print(id(x))             # identity на время жизни объекта
print(sys.getsizeof(x))  # поверхностный размер; число зависит от runtime/платформы

# getsizeof — поверхностный размер:
import sys
a = [1, 2, 3]
print(sys.getsizeof(a))   # размер list-объекта без рекурсивного размера элементов

# Для полного размера используйте рекурсивную функцию:
def deep_getsizeof(obj, seen=None):
    if seen is None:
        seen = set()
    obj_id = id(obj)
    if obj_id in seen:
        return 0
    seen.add(obj_id)
    size = sys.getsizeof(obj)
    if isinstance(obj, (list, tuple, set)):
        size += sum(deep_getsizeof(item, seen) for item in obj)
    elif isinstance(obj, dict):
        size += sum(deep_getsizeof(k, seen) + deep_getsizeof(v, seen)
                    for k, v in obj.items())
    return size

В CPython id() обычно соответствует адресу объекта, но это не гарантия языка. Значения getsizeof() нельзя переносить между версиями, сборками и платформами.

#Представления без копирования: `memoryview`

Срез bytes создаёт новый объект и копирует данные. memoryview предоставляет представление объекта с buffer protocol и позволяет читать срез без копии. Исходный буфер должен оставаться жив, пока используется представление.

payload = bytearray(b"header:body")
body = memoryview(payload)[7:]
body[0] = ord("B")

assert payload == bytearray(b"header:Body")
body.release()

memoryview полезен для больших бинарных буферов и сетевых протоколов. Для обычной короткой строки он только усложнит код.

#2. Подсчёт ссылок (Reference Counting)

Основной механизм управления памятью в CPython.

import sys

a = [1, 2, 3]
print(sys.getrefcount(a))  # 2: одна для 'a', одна для аргумента getrefcount

b = a          # refcount = 3
c = [a, b]     # refcount = 5 (a, b, c[0], c[1] + аргумент)

del b          # refcount уменьшается
del c          # refcount уменьшается ещё на 2
# Когда refcount == 0, объект удаляется немедленно

Как работает:

При создании: ob_refcnt = 1
При присваивании: ob_refcnt += 1
При удалении ссылки: ob_refcnt -= 1
При ob_refcnt == 0: деструктор + освобождение памяти

Плюсы:

Детерминированное освобождение (объект удаляется сразу, не позже)
Большая часть объектов не ждёт отдельного цикла GC

Минусы:

Не обрабатывает циклические ссылки
Небольшой overhead на каждую операцию с ссылками
В free-threaded build управление ссылками использует другую синхронизацию; детали зависят от версии CPython

#Фреймы выполнения

Активный вызов функции представлен frame-объектом. На Python-уровне доступны документированные атрибуты f_code, f_locals, f_globals, f_builtins и f_lasti. Внутреннее устройство PyFrameObject не является стабильным C API: после изменений CPython 3.11 нельзя полагаться на прежние поля вроде f_valuestack или f_localsplus. Для профилирования используйте публичные хуки sys.setprofile() и sys.settrace().

#3. Циклические ссылки и GC

Reference counting не может удалить циклы:

import gc

class Node:
    def __init__(self, value):
        self.value = value
        self.ref = None

a = Node(1)
b = Node(2)
a.ref = b    # a ссылается на b
b.ref = a    # b ссылается на a — цикл!

del a, b
# refcount(a) = 1 (от b.ref), refcount(b) = 1 (от a.ref)
# Никогда не достигнут 0 без GC!

gc.collect()  # явный вызов GC — находит и удаляет цикл
print(gc.collect())  # число удалённых объектов

#Поколения GC: обязательно указывайте версию

В CPython 3.12/3.13 использовались поколения 0, 1 и 2. В CPython 3.14.0 generation 1 удалили, а в 3.14.5 вернули для совместимости поведения. Даже patch-релиз может изменить точную механику, поэтому проверяйте документацию установленного runtime.

import gc

# Не зашивайте ожидаемые значения и семантику порогов:
print(gc.get_threshold())
print(gc.get_stats())

# Менять пороги следует только после измерений на конкретной версии:
current_thresholds = gc.get_threshold()

# Изменять только по результатам воспроизводимого измерения:
gc.set_threshold(*current_thresholds)

# Для производительности можно отключить GC (только если нет циклов):
gc.disable()
gc.enable()

# Принудительная сборка:
gc.collect()        # полная сборка; возвращает collected + uncollectable
gc.collect(0)       # семантика аргумента зависит от версии CPython

# Найти объекты в GC:
for obj in gc.get_objects():
    if isinstance(obj, MyClass):
        print(id(obj), obj)

Более высокие пороги обычно реже запускают циклический GC и могут уменьшить его накладные расходы ценой более долгого удержания циклов. Более низкие пороги запускают проверки чаще. Точная семантика элементов get_threshold() зависит от версии, поэтому сравнивайте паузы и память на реальной нагрузке, а не выбирайте числа по памяти.

Для диагностики циклов можно временно включить gc.set_debug(). Флаг gc.DEBUG_SAVEALL сохраняет найденные недостижимые объекты в gc.garbage, поэтому после эксперимента режим нужно выключить и список очистить. Это диагностический инструмент, а не настройка рабочего процесса по умолчанию.

#4. GIL (Global Interpreter Lock)

GIL — мьютекс, который позволяет только одному потоку выполнять Python-байткод в любой момент.

import threading
import time

counter = 0

def increment():
    global counter
    for _ in range(1_000_000):
        with counter_lock:
            counter += 1

counter_lock = threading.Lock()

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start(); t2.start()
t1.join(); t2.join()
print(counter)  # 2_000_000 — корректно благодаря явной синхронизации

GIL защищает внутреннее состояние интерпретатора, но не является контрактом атомарности составных операций приложения. Для разделяемого изменяемого состояния используйте Lock, очереди, message passing или изоляцию — это важно и для GIL-enabled, и для free-threaded сборки.

Почему GIL не мешает I/O:

# GIL освобождается при системных вызовах:
# - socket.recv(), socket.send()
# - file.read(), file.write()
# - time.sleep()
# - C extensions (если они явно освобождают GIL)

# Поэтому threading работает для I/O-bound задач:
import requests
from concurrent.futures import ThreadPoolExecutor

def fetch(url):
    return requests.get(url).text   # освобождает GIL во время HTTP-запроса

with ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(fetch, urls))   # параллельно!

Free-threaded CPython:

Python 3.13: experimental free-threaded build по PEP 703.
Python 3.14: официально поддерживаемая отдельная сборка по PEP 779.
Это не удаление GIL из обычной сборки и не автоматическая thread-safety пользовательского кода.
Совместимость C-расширений и производительность проверяются на конкретном окружении.

#5. Слабые ссылки (`weakref`)

Слабая ссылка не увеличивает ob_refcnt — объект может быть удалён, даже если слабые ссылки на него существуют.

import weakref

class Cache:
    pass

obj = Cache()
weak = weakref.ref(obj)   # слабая ссылка

print(weak())    # <Cache object> — объект ещё жив
del obj
print(weak())    # None — объект удалён, weak ссылка стала пустой

# Callback при удалении объекта:
def on_destroy(ref):
    print(f"Объект удалён! Ref: {ref}")

obj = Cache()
weak = weakref.ref(obj, on_destroy)
del obj   # выведет: "Объект удалён! Ref: <weakref ...>"

#`weakref.WeakValueDictionary` — кэш без удержания объектов

import weakref

class ExpensiveObject:
    def __init__(self, name):
        self.name = name
        print(f"Создан: {name}")

    def __del__(self):
        print(f"Удалён: {self.name}")

# Обычный dict удерживает объекты:
cache = {}
cache['key'] = ExpensiveObject("obj1")
# При удалении из cache объект остаётся если есть другие ссылки

# WeakValueDictionary: объекты удаляются когда нет других ссылок
cache = weakref.WeakValueDictionary()

obj1 = ExpensiveObject("obj1")
cache['key'] = obj1
print(cache['key'].name)   # obj1 — объект есть

del obj1   # удаляем последнюю сильную ссылку
# Теперь cache['key'] автоматически удаляется из словаря
print('key' in cache)   # False

#`weakref.WeakSet` и `weakref.WeakKeyDictionary`

# WeakSet — набор слабых ссылок на объекты
listeners = weakref.WeakSet()

class EventListener:
    def handle(self, event): ...

listener = EventListener()
listeners.add(listener)
# Когда listener удаляется — автоматически пропадает из listeners

# WeakKeyDictionary — слабые ключи
properties = weakref.WeakKeyDictionary()
obj = SomeObject()
properties[obj] = {"color": "red", "size": 42}
# При удалении obj запись из properties автоматически удалится

#6. `slots` — экономия памяти

import sys

class WithDict:
    def __init__(self, x, y):
        self.x = x
        self.y = y

class WithSlots:
    __slots__ = ('x', 'y', '__weakref__')

    def __init__(self, x, y):
        self.x = x
        self.y = y

d = WithDict(1, 2)
s = WithSlots(1, 2)

regular_size = sys.getsizeof(d) + sys.getsizeof(d.__dict__)
slotted_size = sys.getsizeof(s)
print(regular_size, slotted_size)
# Сравнивайте на целевой версии/платформе и учитывайте наследование.

Ограничения __slots__:

s = WithSlots(1, 2)
s.z = 3   # AttributeError: 'WithSlots' object has no attribute 'z'

# Без '__weakref__' в slots экземпляр обычно нельзя передать в weakref.ref().
# '__dict__' в slots возвращает словарь для произвольных атрибутов и уменьшает
# ожидаемую экономию памяти.

# Наследование: если базовый класс без __slots__, подкласс имеет __dict__
class Base:
    __slots__ = ('a',)

class Child(Base):
    __slots__ = ('b',)   # правильно: добавляем только новые слоты

# Для pickle совместимости нужен __getstate__/__setstate__
class Slotted:
    __slots__ = ('x', 'y')
    def __getstate__(self): return {'x': self.x, 'y': self.y}
    def __setstate__(self, state): self.x = state['x']; self.y = state['y']

#7. Профилирование памяти

#`tracemalloc` — встроенное отслеживание

import tracemalloc

tracemalloc.start(10)   # 10 фреймов в стеке трассировки

# ... ваш код ...
result = [list(range(1000)) for _ in range(100)]

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

print("Топ 3 аллокаторов:")
for stat in top_stats[:3]:
    print(stat)

# Сравнение двух снимков:
snapshot1 = tracemalloc.take_snapshot()
# ... больше кода ...
snapshot2 = tracemalloc.take_snapshot()
top_stats = snapshot2.compare_to(snapshot1, 'lineno')

#`memory_profiler` — построчное профилирование

from memory_profiler import profile

@profile
def memory_heavy():
    a = [0] * 1_000_000         # большой список
    b = {i: i for i in range(100_000)}  # словарь
    del a                        # освобождение
    return b

memory_heavy()
# Выводит потребление памяти построчно

#`objgraph` — граф объектов

import objgraph

# Найти самые частые типы объектов
objgraph.show_most_common_types(limit=10)

# Найти утечки (новые объекты с последнего вызова)
objgraph.show_growth(limit=5)

# Визуализировать ссылки
objgraph.show_backrefs(my_object, max_depth=3)

#8. Memory Pool в CPython

CPython использует собственный аллокатор (pymalloc) поверх системного malloc:

Арены
  └── Пулы
       └── Блоки до 512 байт

Размеры arena/pool — детали конкретной версии и платформы CPython. Кроме того, free-threaded build Python 3.14 по умолчанию использует mimalloc, а GIL-enabled build — pymalloc, поэтому универсальная схема для всех сборок неверна.

Важное следствие:

Объекты размером > 512 байт: сразу в системный malloc
Небольшие объекты: из пула (быстрее + меньше фрагментации)
Освобождённая память возвращается в пул, а не сразу ОС

# Кэш целых чисел: -5 до 256 создаются при старте интерпретатора
a = 42
b = 42
print(a is b)    # True — один и тот же объект!

a = 1000
b = 1000
print(a == b)    # True — сравнение значений
print(a is b)    # не полагайтесь на результат: компилятор может переиспользовать константу

# Интернирование строк:
import sys

a = "hello"
b = "hello"
print(a is b)    # True — Python кэширует короткие "идентификаторные" строки

a = "hello world!"
b = "hello world!"
print(a is b)    # Зависит от реализации — не полагайтесь на это!

# Явное интернирование возвращает канонический объект для равной строки:
a = sys.intern("generated value")
b = sys.intern("generated " + "value")
assert a is b

Автоматический кэш малых целых и строк — деталь CPython. В распространённых сборках малые целые обычно лежат в диапазоне от -5 до 256, но программа не должна зависеть от этого. sys.intern() даёт явный контракт канонизации строк; значения всё равно сравнивайте через ==.

#9. Практические советы

# 1. Используйте del для явного удаления больших объектов
large_data = load_gigabyte_file()
process(large_data)
del large_data   # намекаем GC что объект можно удалить
gc.collect()     # принудительно если нужно немедленно

# 2. Генераторы вместо списков для больших данных
# Плохо:
all_data = [parse(line) for line in open('huge_file.csv')]
# Хорошо:
all_data = (parse(line) for line in open('huge_file.csv'))

# 3. __slots__ для классов с миллионами экземпляров
class Point:
    __slots__ = ('x', 'y')

# 4. WeakRef для кэшей
import functools
@functools.cached_property
def expensive_prop(self):
    return compute()   # вычисляется один раз, хранится в self.__dict__

# 5. array.array для однотипных данных
import array
floats = array.array('d', range(1_000_000))   # вместо list float

# 6. numpy для матриц и массивов
import numpy as np
matrix = np.zeros((1000, 1000), dtype=np.float32)  # 4 MB вместо ~400 MB

#10. Поиск и устранение утечек памяти

#Типичные причины утечек

# 1. Глобальные списки/словари накапливают объекты
_event_handlers = []  # никогда не очищается!

def register(handler):
    _event_handlers.append(handler)  # объект удерживается вечно

# Решение: weakref
import weakref
_event_handlers = weakref.WeakSet()

# 2. Замыкания захватывают большие объекты
def make_handler(large_data):
    def handler(event):
        # large_data захвачен в замыкании!
        return process(event, large_data)
    return handler

# Решение: захватывать только нужное
def make_handler(large_data):
    summary = extract_summary(large_data)  # только что нужно
    def handler(event):
        return process(event, summary)
    return handler

# 3. Циклические ссылки с __del__
class Node:
    def __init__(self):
        self.child = None
    def __del__(self):   # __del__ + цикл = GC не может удалить!
        print("deleted")

# GC ≥3.4 умеет удалять циклы с __del__, но finalize порядок не гарантирован

#Отладка утечек через `tracemalloc`

import tracemalloc
import linecache

def display_top(snapshot, key_type='lineno', limit=10):
    stats = snapshot.statistics(key_type)
    print(f"Top {limit} lines")
    for index, stat in enumerate(stats[:limit], 1):
        frame = stat.traceback[0]
        print(f"#{index}: {frame.filename}:{frame.lineno}: "
              f"{stat.size/1024:.1f} KiB")
        line = linecache.getline(frame.filename, frame.lineno).strip()
        if line:
            print(f"    {line}")
    other = stats[limit:]
    if other:
        size = sum(stat.size for stat in other)
        print(f"{len(other)} other: {size/1024:.1f} KiB")
    total = sum(stat.size for stat in stats)
    print(f"Total allocated size: {total/1024:.1f} KiB")

# Использование:
tracemalloc.start(25)

# ... ваш код с подозрением на утечку ...
run_application()

snapshot = tracemalloc.take_snapshot()
display_top(snapshot)

#`gc.get_referrers()` — кто держит объект

import gc

class LeakedObject:
    pass

obj = LeakedObject()
ref = obj   # вторая ссылка

referrers = gc.get_referrers(obj)
for r in referrers:
    print(type(r), r)
# <class 'dict'> {'obj': ..., 'ref': ...}  — frame locals

#11. `del` и финализация объектов

class Resource:
    def __init__(self, name):
        self.name = name
        print(f"Открыт: {name}")

    def __del__(self):
        # Важно: НЕ полагайтесь на __del__ для освобождения ресурсов!
        # Может не вызваться при завершении интерпретатора
        print(f"Закрыт: {self.name}")

# Лучше: используйте контекстные менеджеры
class Resource:
    def __enter__(self):
        return self
    def __exit__(self, *args):
        self.close()

# Или weakref.finalize — надёжная альтернатива __del__
import weakref

class Resource:
    def __init__(self, name):
        self.name = name
        self._finalizer = weakref.finalize(self, Resource._cleanup, name)

    @staticmethod
    def _cleanup(name):
        print(f"Cleaned up: {name}")  # вызывается при сборке мусора

r = Resource("conn")
del r   # → "Cleaned up: conn"

#Практика: от пробы к рабочему решению

#Шаг 1. Соберите минимальный пример

Ищите удерживающую ссылку, а не «объект, который Python не удалил». Результат основы — воспроизводимый опыт с контрольной точкой до и после освобождения ссылок.

#Шаг 2. Проверьте поведение

Эта проба исследует циклический сборщик CPython; укажите реализацию и версию в своём отчёте.

import gc
import weakref


class Node:
    pass


node = Node()
node.link = node
reference = weakref.ref(node)
del node

collected = gc.collect()
assert collected >= 1
assert reference() is None

#Шаг 3. Доведите решение до рабочего сценария

Практика с подсказками: воспроизведите рост памяти на серии одинаковых запросов, снимите два tracemalloc snapshot и найдите удерживающий путь. Проверьте исправление на той же нагрузке, не используя gc.collect() как маскировку.

#Шаг 4. Объясните внутренний механизм

Сравните счётчик ссылок, циклический сборщик и распределитель CPython с гарантиями языка. Повторите опыт на другой версии или сборке и зафиксируйте, какая часть вывода перестала быть устойчивой.

#Упражнения

Напишите функцию, которая определяет, есть ли циклическая ссылка в объекте через gc.is_tracked() и gc.get_referents().
Создайте кэш с weakref.WeakValueDictionary — убедитесь что объекты удаляются при отсутствии внешних ссылок.
Сравните потребление памяти класса с __slots__ и без через sys.getsizeof().
Используйте tracemalloc чтобы найти где в вашем коде аллоцируется больше всего памяти.
Объясните почему sys.getrefcount(x) всегда возвращает как минимум 2.
Напишите декоратор @track_memory который логирует потребление памяти до и после вызова функции.
Зарегистрируйте освобождение внешнего ресурса через weakref.finalize и докажите, что callback вызывается после исчезновения последней сильной ссылки. Объясните, почему finalize сам по себе не реализует TTL.

#Самопроверка

gc.is_tracked() говорит об участии объекта в циклическом GC, а не доказывает наличие цикла. Для поиска цикла ведите множество посещённых объектов и путь.
WeakValueDictionary теряет запись только после удаления всех сильных ссылок; не полагайтесь в тесте на немедленный сбор без gc.collect().
sys.getsizeof() измеряет объект неглубоко. Для обычного экземпляра учтите __dict__, а для сравнения множества экземпляров используйте tracemalloc.
Отчёт tracemalloc содержит снимки до/после, выбранный фильтр и строку, ответственную за разницу.
Декоратор памяти не должен объявлять временный пик «утечкой»: повторите вызов и проверьте удерживаемую память.

#Исполняемая лаборатория

Закрепите тему в работе «Поиск удержания подписчиков». Нужно воспроизвести сильное удержание связанного метода, выбрать правильную слабую ссылку и защитить выводы измерением tracemalloc с указанием версии CPython.

git clone --branch v0.6.2 --depth 1 https://gitlab.potapov.me/courses/python-labs.git
cd python-labs
uv sync --group test
uv run --group test pytest memory_gc/tests