⁣Память под контролем: __slots__, pympler и гибридные модели

Про __slots__ пишут часто, но в реальных проектах с сотнями классов можно нарваться. Не всё так радужно.

Оверхед словаря и мифы

По умолчанию каждый экземпляр держит атрибуты в __dict__. Словарь с хеш-таблицей для миллионов объектов - это 50-70% оверхеда. __slots__ фиксирует список атрибутов на уровне класса и убирает __dict__ (если не добавить вручную). Экономия до 40-60%. Но в больших кодовых базах всплывают грабли:
- наследование: у родителя нет __slots__, у наследника есть - __dict__ все равно создается, выигрыша нет
- жесткость: нельзя динамически добавить атрибут - ломаются ORM, сериализаторы, половина библиотек
- дебаг: без __dict__ не увидишь текущие поля экземпляра в отладчике, приходится лезть в класс

Профилирование через pympler

Вместо гаданий - профилирование через pympler. Пример:

from pympler import asizeof

class WithSlots:
    __slots__ = ('x', 'y')
    def __init__(self):
        self.x = 10
        self.y = 20

class WithoutSlots:
    def __init__(self):
        self.x = 10
        self.y = 20

print(asizeof.asizeof(WithSlots()))  # ~72 байта
print(asizeof.asizeof(WithoutSlots()))  # ~184 байта

Цифры не врут: видно, где оверхед реально критичен, а где можно забить.

Гибридная модель: __slots__ + __dict__

Когда часть объектов динамическая (dataclass с optional полями), добавляем __dict__ в __slots__:

class FlexibleSlots:
    __slots__ = ('id', 'name', '__dict__')
    def __init__(self, id, name, **kwargs):
        self.id = id
        self.name = name
        for k, v in kwargs.items():
            setattr(self, k, v)

obj = FlexibleSlots(1, 'test', extra_field=42)
print(obj.extra_field)  # 42

Жесткие поля - без оверхеда словаря, динамические - через __dict__. Памяти чуть больше, но гибкость остается.

Что я обычно делаю:
- профилирую не на старте, а когда память реально начинает течь (pympler.muppy, pympler.summary)
- для тысяч объектов - ORM-модели, объекты игр - обязательно __slots__
- если dataclass с наследованием - @dataclass(slots=True) (Python 3.10+), но контролирую дерево наследования
- никогда не пихаю __slots__ в классы, которые подмешиваются через MRO или передаются в гибкие фреймворки

Вывод: __slots__ - мощный инструмент для сокращения памяти, но его применение требует профилирования и осознанного проектирования, особенно при наследовании и в гибридных сценариях с __dict__.

⁣Кастомный семафор с приоритетами и TTL для асинхронных in-memory кэшей

Стандартный asyncio.Semaphore не учитывает приоритеты задач и не имеет time‑to‑live. В production‑кэшах это приводит к тому, что чтение с высоким приоритетом блокируется фоновой записью, а зависшая задача может навсегда захватить ресурс. Решение — семафор на PriorityQueue с TTL.

Архитектура на heapq и TTL
Внутри используется heapq — вставка и извлечение за O(log n). Каждая задача помещается в очередь с приоритетом, а TTL отсчитывается от момента постановки. Если время истекло — выбрасывается TimeoutError. Освобождается слот — подхватывается самая приоритетная задача. Context manager гарантирует вызов release.

Типичные грабли
Если TTL истёк, будущее (Future) нужно явно вычищать из очереди, иначе оно остаётся мёртвым грузом и накапливает память. Также приоритеты инвертированы: heapq считает минимальное значение наивысшим приоритетом. Поэтому в acquire передаётся отрицательное число, если нужно обратное поведение.

Пример из production
В in‑memory кэше расставлены приоритеты: запись от пользователя — 0, фоновое обновление — 5, чтение — 10. Для чтения установлен короткий TTL — при зависании можно отдать stale данные вместо полной блокировки. Это trade‑off между свежестью и доступностью.

@asynccontextmanager
async def acquire(self, priority: int, ttl: float):
    future = self._loop.create_future()
    heapq.heappush(self._queue, (priority, future))
    try:
        result = await asyncio.wait_for(future, timeout=ttl)
        yield result
    except asyncio.TimeoutError:
        self._cleanup_expired()
        raise
    finally: ...  # release and pop next

Недостаток и решение
Если очередь большая и приоритеты постоянно высокие, низкоприоритетные задачи могут умирать по TTL повторно. В таких сценариев стоит внедрить dynamic priority aging — со временем повышать приоритет ожидающей задачи. Но для большинства кэшей описанного подхода достаточно.

Вывод: Кастомный семафор с приоритетами и TTL решает конкретные проблемы конкурентного доступа в асинхронных кэшах, но требует аккуратной очистки и осознания trade‑offs при голодании низкого приоритета.

⁣Сборка Python-приложения в единый бинарник: избавляемся от зависимостей от системных библиотек с PyInstaller и staticx

Каждый раз, когда нужно задеплоить Python-приложение не в контейнер, а просто скопировать один файл на сервер, всплывает проблема зависимостей от системных библиотек. PyInstaller собирает бинарник, но он всё равно тащит libc, libssl и другие .so-файлы. На старой Ubuntu или в Docker на scratch это приводит к ошибкам при запуске. Опытные разработчики часто забывают, что статическая сборка решает эту проблему радикально.

Почему одного PyInstaller недостаточно

PyInstaller с флагом --onefile упаковывает Python-интерпретатор и код в один файл, но он динамически линкует системные библиотеки. Проверка через ldd покажет кучу зависимостей. На машине с другой версией glibc или отсутствующей библиотекой (например, libcrypto) приложение упадёт. Это типичная ошибка: разработчик уверен, что бинарник самодостаточен, но на проде получает segfault.

Решение: статическая линковка через staticx

staticx переупаковывает полученный PyInstaller-бинарник, заменяя динамические зависимости на статические. Он встраивает musl libc и делает бинарник полностью независимым от системы. Для этого:

pyinstaller --onefile --name myapp main.py
pip install staticx
staticx dist/myapp dist/myapp_static

На выходе — бинарник, который запускается на любом Linux с ядром >= 2.6.28. Без Python, без pip, без системных библиотек.

Типичные ошибки и trade-offs

* Размер бинарника: 5-30 МБ вместо 5 МБ. Для большинства серверных задач это приемлемо, но для embedded может быть критично.
* Проблемы с glibc-расширениями: staticx использует musl libc, поэтому старые C-расширения (например, NumPy, некоторые ORM) могут сломаться. Решение — собирать всё в Docker на alpine:latest с musl-dev.
* Динамические ctypes: если в коде используются ctypes для загрузки .so, их нужно вручную добавить через --add-binary, иначе приложение упадёт с тишиной.
* Дебаг: ошибки сложнее диагностировать. Помогает флаг --debug в PyInstaller и strace на скомпилированном бинарнике.

Production-пример: CI/CD деплой в корпоративной сети

Представьте: вы разворачиваете сервис на серверах, где IT-отдел одобряет установку только через внутренний репозиторий, а на копирование одного файла — без ограничений. Вы запускаете в CI:

pyinstaller --onefile --name service main.py
staticx dist/service dist/service_static

Копируете service_static на сервер — и он работает сразу, без Python, зависимостей и permission battles. Это особенно ценно для edge-устройств и минималистичных Docker-образов на scratch.

Вывод: staticx превращает PyInstaller-бинарник в truly portable решение, но требует учёта C-расширений и динамических загрузок — без этого сборка будет работать нестабильно на проде.

⁣Асинхронная сериализация через __getstate__ / __setstate__ и Protocol 5 для zero-copy передачи данных между воркерами

Когда данные летят между воркерами через multiprocessing или распределенные очереди, каждый байт копирования бьет по памяти. Стандартный pickle копирует объекты целиком, но переопределение __getstate__ / __setstate__ с protocol 5 позволяет избежать лишних копий.

Zero-copy через out-of-band буферы
В protocol 5 (Python 3.8+) появляется zero-copy передача через out-of-band буферы. Данные не упаковываются в байтовый поток, а передаются отдельно через shared memory или IPC. __getstate__ возвращает только словарь метаданных, а __setstate__ восстанавливает объект из буфера.

Пример с NumPy

import pickle
import numpy as np

class ZeroCopyArray:
    def __init__(self, data):
        self.data = data
    def __getstate__(self):
        return {'shape': self.data.shape, 'dtype': self.data.dtype}
    def __setstate__(self, state):
        self.shape, self.dtype, self.data = state['shape'], state['dtype'], None

При дампе с protocol=5 данные физически не копируются, а None в сериализованном виде — expected.

Где это критично
По умолчанию pickle копирует большие массивы при передаче через Redis, RabbitMQ или multiprocessing.Queue. Два процесса — два полных копирования. Protocol 5 позволяет не копировать, если оба процесса на одной машине.
* data pipelines с NumPy/Pandas, где каждый гигабайт на счету
* дублирование конфигов с тяжелыми данными — сериализуем только параметры
* потоковый парсинг бинарных протоколов

Типичная ошибка и предупреждение
Ошибка: думать, что zero-copy работает везде. Protocol 5 не поддерживается старым Celery с kombu, а out-of-band буферы требуют ручной передачи — они не входят в байтовый поток. На разных машинах zero-copy превращается в обычную копию из-за отсутствия shared memory.

Практический совет
Всегда ставьте protocol=pickle.HIGHEST_PROTOCOL. Для критичных сценариев используйте 5-й протокол с правильными __getstate__ / __setstate__. Изучите PEP 574, документацию pickle.Pickler и код dask с cloudpickle — это реальные production-источники.

Вывод: Protocol 5 с кастомными методами сериализации дает zero-copy передачу данных между воркерами на одной машине, экономя память и время, но требует контроля совместимости и ручной передачи буферов.

⁣Динамический мониторинг утечек памяти: gc.get_objects и tracemalloc

Память нельзя мониторить только статикой. OOM Killer приходит неожиданно, когда забытый lru_cache или циклическая ссылка в asyncio жрут все ресурсы. Решение — фоновый поток, который раз в N секунд снимает срезы gc.get_objects и tracemalloc, сравнивает с порогами и шлет алерт.

Как работает

gc.get_objects возвращает все отслеживаемые GC объекты. Считаешь количество инстансов своих классов через Counter. tracemalloc дает раскладку по строкам кода — сразу видно, где аллоцируется больше всего. Сравниваешь снимки, ловишь аномальный рост.

Пример production-сбора

import gc
import tracemalloc
from collections import Counter

_MEMORY_THRESHOLD = 50 * 1024 * 1024  # 50 MB
_OBJECT_THRESHOLD = 10000

def check_leaks():
    alerts = {}
    type_counts = Counter(type(o).__name__ for o in gc.get_objects())
    for type_name, count in type_counts.items():
        if count > _OBJECT_THRESHOLD:
            alerts[f"obj_{type_name}"] = count

    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')
    total_size = sum(stat.size for stat in top_stats[:10])
    if total_size > _MEMORY_THRESHOLD:
        alerts["memory_total"] = total_size
    return alerts

Подводные камни

- gc.get_objects замораживает GIL. При миллионе объектов пауза ощутима — решай, готов ли к этому в пиковую нагрузку.
- tracemalloc.start() дает оверхед ~20% на аллокации. Отключай в проде, когда не нужен, или включай дозировано.
- tracemalloc не видит объекты, созданные до старта. Запускай сразу после инициализации приложения.
- C-расширения вроде numpy не поддаются мониторингу — учитывай при анализе.

Практический совет

Ставь пороги на 70% от лимита контейнера. Экспортируй метрики в Prometheus через gc.get_count() и tracemalloc.get_traced_memory(). Для глубокой копки используй objgraph на визуализацию ссылок, pympler или memray. Если сработал алерт, бери py-spy или scalene — ищи конкретную причину.

Вывод: Динамический мониторинг с gc.get_objects и tracemalloc — первый эшелон защиты от утечек, который позволяет поймать аномалию до падения прода.

⁣Проектирование кастомного RLock-совместимого диспетчера блокировок с поддержкой deadline и приоритетного наследования в асинхронных воркерах

Стандартный asyncio.Lock не поддерживает дедлайны и приоритеты — в production это приводит к инверсии приоритетов и зависанию воркеров. Без наследования приоритетов низкоприоритетная задача может блокировать критичный поток, а без дедлайна блокировка навсегда замораживает воркер.

Проблема и решение
В high-throughput системе с асинхронными воркерами (например, Celery + asyncio) конкуренция за общий ресурс без управления приоритетами вызывает голодание. Решение — кастомный диспетчер поверх asyncio.Lock с очередью на heapq и механизмом приоритетного наследования: когда высокоприоритетный воркер ждёт блокировку, его приоритет временно передаётся текущему владельцу.

Архитектура диспетчера
Ключевые элементы:
* Очередь_Request с priority и deadline, реализованная через heapq. Приоритет инвертируется (отрицание) для корректной сортировки.
* Рекурсивный счётчик для RLock-совместимости — блокировка может быть захвачена тем же воркером повторно.
* Future для пробуждения: каждый запрос ассоциирован с asyncio.Future, которая вызывается при освобождении.
* При наследовании приоритет владельца повышается до максимального из ожидающих, при освобождении сбрасывается.

Упрощённый код ядра

import asyncio, heapq
from dataclasses import dataclass, field

@dataclass(order=True)
class _LockRequest:
    deadline: float
    priority: int
    future: asyncio.Future = field(compare=False)

class PriorityRLock:
    def __init__(self):
        self._owner = None
        self._current_priority = 0
        self._queue = []
        self._recursion_level = 0
        self._cond = asyncio.Condition()

    async def acquire(self, priority=0, timeout=None):
        loop = asyncio.get_event_loop()
        deadline = loop.time() + timeout if timeout else float('inf')
        future = loop.create_future()
        heapq.heappush(self._queue, _LockRequest(deadline, -priority, future))
        async with self._cond:
            if self._owner and priority > self._current_priority:
                self._current_priority = priority  # наследование
            while True:
                if self._owner is None and self._queue[0].future == future:
                    self._owner = id(asyncio.current_task())
                    self._current_priority = priority
                    self._recursion_level += 1
                    return True
                try:
                    await asyncio.wait_for(future, timeout)
                except asyncio.TimeoutError:
                    self._queue.remove(future)
                    heapq.heapify(self._queue)
                    return False

    async def release(self):
        async with self._cond:
            self._recursion_level -= 1
            if self._recursion_level == 0:
                self._owner = None
                self._current_priority = 0
                if self._queue:
                    top = self._queue[0]
                    loop.call_soon(top.future.set_result, None)

Типичная ошибка и практический совет
Ошибка: забыть про сортировку heap при удалении — use _LockRequest с compare=False для future, иначе сравнение объектов ломает кучу. Практический совет: в production замените asyncio.wait_for на кастомный таймер с асинхронным сном, чтобы избежать оверхеда исключений при частых таймаутах. Для простых кейсов (один пул, без жёстких дедлайнов) используйте Lock с timeout — не усложняйте.

Когда это критично?
* Очереди с приоритетами в Celery/arq для воркеров к общему ресурсу (БД, кэш).
* Системы с жёсткими дедлайнами, где превышение времени блокировки порождает цепную реакцию.
* High-throughput сценарии, где инверсия приоритетов проявляется только под нагрузкой. Минусы: сложность отладки и оверхед на сортировку Heap (O(log n)). Но в продакшене без этого ловятся странные гонки и зависания, не воспроизводимые локально.

Вывод: Проектируйте диспетчер блокировок с приоритетным наследованием и дедлайнами только под реальную нагрузку, иначе ст

андартный Lock с таймаутом будет дешевле и понятнее.