Производительность Python

Маршрут: Работа в эксплуатации · тема 6 из 6

Навыки: M8, S5, S7

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

Результат: решение, подтверждённое измерением на уровне алгоритма, данных и модели конкурентности.

Основа: сложность и выбор структуры данных. Работа в эксплуатации: измерение, кэширование и выбор конкурентности для CPU/I/O. Углубление: GIL, сборка без GIL, расширения и альтернативные среды выполнения.

Подтверждение: отчёт «до/после» без заранее обещанного ускорения.

Оптимизация начинается с измеримого ограничения: задержки, пропускной способности, памяти или стоимости. После каждого изменения повторяйте тот же опыт на сопоставимой нагрузке.

#1. Цель, метрики и нагрузка

Производительность — не синоним «код выполняется быстрее». Для сервиса обычно важны latency (p50, p95, p99), throughput или RPS, доля ошибок, CPU и память. Выберите целевую метрику и SLO до изменения кода, иначе улучшение одной величины может незаметно ухудшить другую.

Нагрузочный тест должен воспроизводить форму реального потока: набор операций, соотношение чтения и записи, размеры данных, конкурентность и время разогрева. Нагрузку повышают до точки насыщения, наблюдая задержки, ошибки и загрузку ресурсов. Так находят bottleneck, а не просто получают максимальное число RPS.

По закону Амдала ускорение всей программы ограничено долей, которую вы не улучшили. Если 10% времени остаются строго последовательными, теоретический предел ускорения при бесконечном числе работников равен 1 / 0.1 = 10. Накладные расходы делают реальный предел ниже.

Рабочий цикл один: измерить базовую линию, найти узкое место, изменить одну вещь и повторить измерение.

Типовые рычаги — алгоритм и структура данных, кэширование, batching и модель конкурентности. Batching объединяет несколько мелких операций в одну и снижает накладные расходы вызова или round trip, но может увеличить задержку отдельного элемента. Поэтому размер пакета выбирают по latency и throughput вместе.

#2. CPU-bound vs I/O-bound

Прежде чем оптимизировать — определите тип узкого места:

Тип	Характеристика	Примеры	Лучшая стратегия
CPU-bound	Процессор загружен на 100%	Математика, шифрование, обработка изображений, ML	`multiprocessing`, `numba`, `Cython`
I/O-bound	Процессор ждёт (диск, сеть, БД)	HTTP-запросы, файлы, запросы к БД	`asyncio`, `aiohttp`, `asyncpg`
Memory-bound	Узкое место — пропускная способность памяти	Обработка больших массивов	`numpy`, `array`, `__slots__`

import time

# CPU-bound: процессор не простаивает
def cpu_task(n):
    return sum(i * i for i in range(n))

# I/O-bound: большую часть времени ждём
def io_task(url):
    import urllib.request
    return urllib.request.urlopen(url).read()

#3. GIL и модели параллелизма

В стандартной GIL-enabled сборке CPython GIL позволяет только одному потоку за раз выполнять Python-байткод. Отдельная free-threaded сборка имеет другую модель.

Аналогия: GIL — это один повар на кухне. Он переключается между заказами (потоками), пока одни ждут (I/O: варка, жарка), но не может одновременно резать и жарить. multiprocessing — это несколько поваров со своими кухнями. asyncio — один повар, который умело переключается между блюдами, пока ждёт.

Подход	Когда использовать	Обходит GIL?
`threading`	I/O-bound задачи	Частично (GIL снимается при I/O)
`multiprocessing`	CPU-bound задачи	Да (отдельные процессы)
`asyncio`	Много I/O-ожиданий	Нет (один поток, но эффективно переключается)
`concurrent.futures`	Удобный API над threading/multiprocessing	Зависит от executor
`InterpreterPoolExecutor` (3.14+)	Изолированный CPU-bound Python	Да (отдельные интерпретаторы)

from multiprocessing import Pool
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import asyncio

# CPU-bound: multiprocessing
def heavy_computation(n):
    return sum(i * i for i in range(n))

# Используем все ядра процессора
with ProcessPoolExecutor() as executor:
    results = list(executor.map(heavy_computation, [10**6] * 4))

# I/O-bound: threading или asyncio
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(fetch_url, url) for url in urls]
    results = [f.result() for f in futures]

# Async для максимального числа одновременных I/O
async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

ProcessPoolExecutor — высокоуровневый API с объектами Future, одинаковый по форме с ThreadPoolExecutor. multiprocessing.Pool предоставляет старый, более специализированный API (map, apply, imap). Оба подхода выполняют работу в процессах и обычно сериализуют аргументы через pickle.

Для больших числовых буферов multiprocessing.shared_memory позволяет создать область памяти, к которой подключаются несколько процессов без копирования всего массива при каждой задаче. Синхронизацию доступа и жизненный цикл close()/unlink() при этом нужно организовать явно.

asyncio хорошо масштабируется для множества одновременно ожидающих I/O- операций: корутины делят один event loop и не требуют отдельного системного стека на соединение. Потоки тоже подходят для блокирующего I/O, но каждый поток добавляет стек и работу планировщика ОС. Выбор подтверждают нагрузочным тестом конкретного клиента и протокола.

#Free-threaded CPython 3.13–3.14

Python 3.13 представил experimental free-threaded build по PEP 703. В Python 3.14 эта отдельная сборка получила officially supported status по PEP 779. Это не удаление GIL из обычного CPython и не автоматическая thread-safety приложения: проверяйте зависимости, синхронизацию и benchmark на нужной сборке.

#4. Алгоритмическая сложность — главная оптимизация

Изменение алгоритма часто даёт больший эффект, чем микрооптимизация, но коэффициент зависит от данных и реализации.

Алгоритм	Сложность	n=1000	n=1_000_000
Hash lookup	O(1) в среднем	зависит от ключа/runtime	зависит от ключа/runtime
Binary search	O(log n)	~10 шагов	~20 шагов
Linear search	O(n)	1000 шагов	1M шагов
Bubble sort	O(n²)	1M шагов	10¹² шагов

# Сравнение O(n) list membership и O(1) average set membership
lst = list(range(100_000))
s = set(lst)

import timeit
print(timeit.timeit('99999 in lst', globals=globals(), number=1000))
print(timeit.timeit('99999 in s', globals=globals(), number=1000))

# Используйте set/dict для частых проверок наличия
seen = set()
for item in data:
    if item not in seen:
        seen.add(item)
        process(item)

# bisect для отсортированных данных
import bisect
sorted_data = sorted(range(100_000))
idx = bisect.bisect_left(sorted_data, 42)  # O(log n) вместо O(n)

#5. Микрооптимизации Python

#Локальные переменные быстрее глобальных

import timeit

# Медленно: каждое обращение к math.sqrt — поиск в глобальном словаре
import math
def slow():
    for x in range(1000):
        math.sqrt(x)

# Быстро: локальная ссылка — LOAD_FAST вместо LOAD_GLOBAL
def fast():
    sqrt = math.sqrt   # локальная ссылка
    for x in range(1000):
        sqrt(x)

# Измерьте обе версии: результат зависит от версии CPython и тела цикла.

#`str.join()` vs конкатенация

words = ['Hello', 'World'] * 1000

# CPython может оптимизировать некоторые случаи +=, но это не переносимый контракт
slow = ''
for w in words:
    slow += w + ' '

# O(n) — создаётся один раз
fast = ' '.join(words)

#List comprehension и цикл с append

# Явный цикл удобнее, если логика многошаговая
result = []
for x in range(10000):
    if x % 2 == 0:
        result.append(x * x)

# Comprehension часто компактнее и может быть быстрее; измеряйте hotspot
result = [x * x for x in range(10000) if x % 2 == 0]

#`in` для `set` и `dict` — O(1)

item = 9_999
large_list = list(range(10_000))
large_set = set(large_list)
large_dict = dict.fromkeys(large_list)

item in large_list  # O(n)
item in large_set   # O(1) в среднем
item in large_dict  # O(1) в среднем

#6. Кэширование

#`functools.lru_cache` / `functools.cache`

from functools import lru_cache, cache

# lru_cache — ограниченный кэш (LRU eviction)
@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

fibonacci(100)    # мгновенно после прогрева
fibonacci.cache_info()   # CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)
fibonacci.cache_clear()  # очистить кэш

# cache (Python 3.9+) — без ограничения размера
@cache
def factorial(n):
    return 1 if n <= 1 else n * factorial(n - 1)

# Важно: lru_cache работает только с hashable аргументами!
@lru_cache(maxsize=None)
def compute(x, y):    # x, y должны быть hashable
    return x * y

Подробнее о lru_cache, cached_property и других приёмах кэширования — в profiling.

#`functools.cached_property` — ленивое вычисление

from functools import cached_property

class Circle:
    def __init__(self, radius):
        self.radius = radius

    @cached_property
    def area(self):
        import math
        return math.pi * self.radius ** 2  # вычисляется один раз при первом обращении

c = Circle(5)
c.area   # вычисляется
c.area   # берётся из кэша

#Redis для распределённого кэша

import redis
import json

r = redis.Redis(host='localhost', port=6379)

def get_user(user_id: int) -> dict:
    cache_key = f"user:{user_id}"

    # Проверяем кэш
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # Загружаем из БД
    user = db.query(User).filter_by(id=user_id).first()
    data = user.to_dict()

    # Кэшируем на 1 час
    r.setex(cache_key, 3600, json.dumps(data))
    return data

#7. Экономия памяти

#`slots` — экономия памяти для классов

class RegularPoint:
    def __init__(self, x, y):
        self.x = x
        self.y = y
    # каждый экземпляр обычно имеет __dict__

class SlottedPoint:
    __slots__ = ['x', 'y']
    def __init__(self, x, y):
        self.x = x
        self.y = y
    # при согласованной иерархии экземпляр может быть без __dict__

import sys
r = RegularPoint(1, 2)
s = SlottedPoint(1, 2)
regular_size = sys.getsizeof(r) + sys.getsizeof(r.__dict__)
slotted_size = sys.getsizeof(s)
print(regular_size, slotted_size)

Универсального процента экономии нет. Измеряйте целевой класс, учитывая наследование, __dict__, __weakref__ и вложенные значения.

#`array` vs `list` для числовых данных

import array
import sys

# list хранит ссылки на отдельные Python-объекты; точный размер зависит от runtime
lst = list(range(1_000_000))
print(sys.getsizeof(lst))   # только контейнер; int-объекты считаются отдельно

# array — компактное представление C-типов
arr = array.array('i', range(1_000_000))   # 'i' = signed int
print(sys.getsizeof(arr))   # ~4 MB

# NumPy — ещё эффективнее
import numpy as np
np_arr = np.arange(1_000_000, dtype=np.int32)
print(np_arr.nbytes)  # 4 MB + незначительный overhead

#Генераторы вместо списков

# 100 MB данных в памяти сразу
def process_v1(filename):
    lines = open(filename).readlines()   # вся файла в памяти
    return [line.strip() for line in lines]

# O(1) памяти — обрабатываем построчно
def process_v2(filename):
    with open(filename) as f:
        for line in f:           # итератор строк
            yield line.strip()   # обрабатываем по одной

Генераторные выражения и функции из itertools образуют ленивый конвейер: элемент проходит через него по мере запроса, а не после загрузки всего набора в RAM. Это не означает буквально O(1) для любого pipeline: буферизацию могут добавить сортировка, tee, группировка или сторонняя библиотека.

#Буферизованная запись

io.BufferedWriter накапливает небольшие записи и передаёт их операционной системе более крупными блоками. Это уменьшает число системных вызовов, но требует flush() или закрытия файла перед чтением результата другим процессом.

from io import BufferedWriter

with open("events.bin", "wb", buffering=0) as raw:
    with BufferedWriter(raw, buffer_size=64 * 1024) as output:
        for event in events:
            output.write(encode_event(event))

#8. NumPy — векторизация числовых вычислений

import numpy as np
import timeit

# Python цикл — медленно
def python_sum(n):
    return sum(i * i for i in range(n))

# NumPy — векторизованная операция
def numpy_sum(n):
    arr = np.arange(n)
    return (arr * arr).sum()

# Сравните на своей версии NumPy, размере массива и CPU
n = 1_000_000
print(timeit.timeit(lambda: python_sum(n), number=3))
print(timeit.timeit(lambda: numpy_sum(n), number=3))

# Broadcasting — операции без циклов
a = np.array([1, 2, 3, 4, 5])
b = np.array([10, 20, 30, 40, 50])
result = a * b + a    # векторизованно: [11, 42, 93, 164, 255]

Broadcasting совмещает совместимые формы массивов без материализации повторённых входных данных. Однако сама операция всё равно может создать выходной или промежуточный массив; проверяйте пики памяти отдельно.

Pandas строится поверх массивов и часто переносит фильтрацию, агрегацию и арифметику столбцов из Python-циклов в NumPy или другой нативный код. Это полезно для табличных данных, но DataFrame.apply(axis=1) с Python-функцией может вернуть накладные расходы обратно. Сравнивайте векторное выражение с реальной исходной реализацией и учитывайте стоимость создания DataFrame.

#9. C-расширения и JIT-компиляция

#Numba — JIT для числового кода

from numba import jit, prange
import numpy as np

@jit(nopython=True)    # компилирует в машинный код при первом вызове
def matrix_multiply(a, b):
    n = len(a)
    result = np.zeros((n, n))
    for i in range(n):
        for j in range(n):
            for k in range(n):
                result[i, j] += a[i, k] * b[k, j]
    return result

# Первый вызов: медленно (компиляция)
# Последующие вызовы не платят полную цену компиляции; выигрыш измеряется.

#Cython — Python → C

# math_utils.pyx
def fast_sum(int n):    # аннотации типов = C-код
    cdef int i
    cdef long total = 0
    for i in range(n):
        total += i * i
    return total

# После компиляции: скорость C-кода с Python-интерфейсом

#ctypes — вызов C библиотек

import ctypes
import os

# Загружаем системную библиотеку
libc = ctypes.CDLL("libc.so.6")

# Вызываем C функцию
result = libc.strlen(b"hello")  # 5

#10. Профилирование — основа оптимизации

# cProfile — где тратится время
import cProfile
cProfile.run('my_function()', sort='cumulative')

# timeit — точные измерения
import timeit
t = timeit.timeit(
    'sum(x**2 for x in range(1000))',
    number=10000
)
print(f"{t:.3f} sec для 10000 итераций")

# memory_profiler — потребление памяти по строкам
from memory_profiler import profile

@profile
def memory_heavy():
    data = [0] * 1_000_000
    return sum(data)

# tracemalloc — встроенное отслеживание
import tracemalloc
tracemalloc.start()
# ... ваш код ...
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
for stat in top_stats[:3]:
    print(stat)

#11. Практические рекомендации

Проблема	Решение
Поиск в большом списке	`set` или `dict`
Строковая конкатенация в цикле	`str.join()`
Повторные вычисления	`@lru_cache` / `@cache`
CPU-bound задачи	`multiprocessing` / `ProcessPoolExecutor`
I/O-bound задачи	`asyncio` / `ThreadPoolExecutor`
Много экземпляров класса	`__slots__`
Числовые вычисления	`numpy` / `numba`
Большие файлы	Генераторы (построчно)
Медленный алгоритм	Сначала улучши алгоритм!

#Антипаттерны

# 1. Оптимизация без измерений (premature optimization)
# Сначала: cProfile → находим узкое место → оптимизируем

# 2. Строковая конкатенация в цикле
result = ""
for item in huge_list:
    result += str(item)   # O(n^2)! Используй: "".join(str(i) for i in huge_list)

# 3. Поиск в списке вместо множества
allowed_list = list(range(1_001))
allowed_set = set(allowed_list)
x in allowed_list  # O(n)
x in allowed_set   # O(1) в среднем

# 4. Повторные вычисления без кэширования
for i in range(n):
    if expensive_check(i):  # вызывается n раз!
        ...
# Решение: кэшировать результат

#12. `concurrent.futures` — унифицированный API

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import urllib.request

# ThreadPoolExecutor — для I/O-bound задач
def download(url):
    with urllib.request.urlopen(url) as r:
        return r.read()

urls = ["https://example.com"] * 20

with ThreadPoolExecutor(max_workers=10) as executor:
    # map — сохраняет порядок, блокирует до завершения
    results = list(executor.map(download, urls))

    # submit + as_completed — обрабатываем по мере завершения
    futures = {executor.submit(download, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f"{url}: {len(data)} bytes")
        except Exception as e:
            print(f"{url} failed: {e}")

# ProcessPoolExecutor — для CPU-bound задач
def compute(n):
    return sum(i*i for i in range(n))

with ProcessPoolExecutor() as executor:
    results = list(executor.map(compute, [10**6] * 4))

#Паттерн: смешанный pipeline

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import asyncio

async def pipeline(items):
    """I/O в потоках, CPU в процессах, координация через asyncio."""
    loop = asyncio.get_running_loop()

    # Шаг 1: загрузка (I/O-bound) — в потоках
    with ThreadPoolExecutor(max_workers=20) as io_pool:
        raw_data = await asyncio.gather(*[
            loop.run_in_executor(io_pool, fetch_item, item)
            for item in items
        ])

    # Шаг 2: обработка (CPU-bound) — в процессах
    with ProcessPoolExecutor() as cpu_pool:
        processed = await asyncio.gather(*[
            loop.run_in_executor(cpu_pool, heavy_process, data)
            for data in raw_data
        ])

    return processed

#13. Оптимизация структур данных

from collections import deque, defaultdict, Counter
import heapq

# deque — O(1) добавление/удаление с обоих концов
# list — O(n) для appendleft/popleft
dq = deque(maxlen=100)   # FIFO с автоматическим выталкиванием старых
dq.append(1)             # O(1)
dq.appendleft(0)         # O(1)

# defaultdict — избегает KeyError проверок
graph = defaultdict(list)
graph['a'].append('b')   # не нужен 'if a not in graph'

# Counter — частотный анализ
words = ["the", "quick", "the", "fox", "the"]
freq = Counter(words)
freq.most_common(2)     # [('the', 3), ('quick', 1)]
freq['the']             # 3

# heapq — приоритетная очередь (мин-куча)
heap = []
heapq.heappush(heap, (priority, item))
priority, item = heapq.heappop(heap)    # O(log n)

# Для N наименьших/наибольших:
top5 = heapq.nlargest(5, data, key=lambda x: x.score)

#14. PyPy — JIT-компилятор для Python

# PyPy — альтернативная реализация Python с JIT
# Установка: pypy3 вместо python3
# Совместим с большинством Python кода

# Когда PyPy быстрее:
# - Длинные циклы
# - Числовые вычисления без numpy
# - Интерпретируемый код

# Когда PyPy хуже:
# - C-расширения (numpy, scipy) — могут быть медленнее
# - Короткие скрипты (JIT не прогревается)
# - Память: PyPy обычно потребляет больше

# Прогоните один и тот же representative benchmark после warmup.
# Не переносите коэффициенты из чужого окружения.

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

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

Сначала улучшайте алгоритм и структуру данных, затем измеряйте реализацию. Результат основы — сравнение с одинаковой семантикой, фиксированными данными и несколькими повторами.

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

from statistics import median
from timeit import repeat

values = list(range(5_000))
lookup = set(values)
target = 4_999

list_times = repeat("target in values", globals=globals(), number=1_000, repeat=5)
set_times = repeat("target in lookup", globals=globals(), number=1_000, repeat=5)

assert (target in values) == (target in lookup)
print({"list_median": median(list_times), "set_median": median(set_times)})

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

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

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

Для одной CPU-нагрузки сравните потоки, процессы и доступные варианты интерпретаторов. Отдельно укажите сборку CPython, влияние GIL, сериализации и нативного кода; решение должно зависеть от нагрузки, а не от лозунга.

#Упражнения

Профилируйте функцию с cProfile и найдите узкое место.
Сравните list.index() и dict lookup для 10^6 элементов.
Реализуйте LRU-кэш вручную используя collections.OrderedDict.
Измерьте разницу памяти между классом с __slots__ и без.
Перепишите цикл суммирования через numpy и измерьте разницу.
Используйте ProcessPoolExecutor для параллельной обработки списка файлов — измерьте speedup на разном числе воркеров.
Профилируйте утечку памяти в коде с циклическими ссылками через tracemalloc.

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

Сначала сохраните профиль и эталонный результат, затем меняйте только измеренное узкое место.
Для lookup включите стоимость построения dict отдельным измерением: она окупается только при достаточном числе поисков.
Ручной LRU проверяется на чтении существующего ключа, вытеснении, обновлении и capacity 0; сравните поведение с functools.lru_cache.
В сравнении __slots__ учитывайте __dict__ обычного объекта и измеряйте множество экземпляров.
NumPy и цикл получают одинаковые dtype и результат; отдельно покажите накладные расходы на маленьком входе.
ProcessPool прогоняется с 1, 2 и доступным числом CPU, включая сериализацию данных; отсутствие ускорения — допустимый результат.
Рост памяти считается утечкой только если объекты остаются достижимыми или повторные снимки показывают устойчивое удержание после gc.collect().

#Практическая лаборатория

Закрепите тему в лаборатории «Пакетный поиск с точкой окупаемости индекса». Вы отделите стоимость построения индекса от запросов, сохраните сырые измерения и сформулируете вывод, ограниченный конкретной средой и нагрузкой.

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 python_performance/tests

Тип

Характеристика

Примеры

Лучшая стратегия

CPU-bound

Процессор загружен на 100%

Математика, шифрование, обработка изображений, ML

multiprocessing, numba, Cython

I/O-bound

Процессор ждёт (диск, сеть, БД)

HTTP-запросы, файлы, запросы к БД

asyncio, aiohttp, asyncpg

Memory-bound

Узкое место — пропускная способность памяти

Обработка больших массивов

numpy, array, __slots__

Подход

Когда использовать

Обходит GIL?

threading

I/O-bound задачи

Частично (GIL снимается при I/O)

multiprocessing

CPU-bound задачи

Да (отдельные процессы)

asyncio

Много I/O-ожиданий

Нет (один поток, но эффективно переключается)

concurrent.futures

Удобный API над threading/multiprocessing

Зависит от executor

InterpreterPoolExecutor (3.14+)

Изолированный CPU-bound Python

Да (отдельные интерпретаторы)

Алгоритм

Сложность

n=1000

n=1_000_000

Hash lookup

O(1) в среднем

зависит от ключа/runtime

Binary search

O(log n)

~10 шагов

~20 шагов

Linear search

O(n)

1000 шагов

1M шагов

Bubble sort

O(n²)

1M шагов

10¹² шагов

Проблема

Решение

Поиск в большом списке

set или dict

Строковая конкатенация в цикле

str.join()

Повторные вычисления

@lru_cache / @cache

CPU-bound задачи

multiprocessing / ProcessPoolExecutor

I/O-bound задачи

asyncio / ThreadPoolExecutor

Много экземпляров класса

__slots__

Числовые вычисления

numpy / numba

Большие файлы

Генераторы (построчно)

Медленный алгоритм

Сначала улучши алгоритм!

Производительность Python