Профилирование и оптимизация

Профилирование и оптимизация Python

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

Навыки: M8, S7

До урока: python_testing, test_strategy, stdlib.

Результат: локализация узкого места на воспроизводимой нагрузке до изменения кода.

Основа: timeit, cProfile и чтение профиля. Работа в эксплуатации: исходное измерение, проектирование нагрузки и проверка ухудшения производительности. Углубление: tracemalloc, dis, выборочные профилировщики и статистическая интерпретация.

Подтверждение: профиль до оптимизации и обоснованный план эксперимента.

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

#1. Зачем профилировать?

Типичная ошибка: разработчик «знает», что медленно, и оптимизирует — но не то место. Реальность: 90% времени занимает 10% кода. Только профайлер покажет истину.

Порядок работы:
1. Убедитесь что код корректен (тесты!)
2. Измерьте производительность (профайлер)
3. Найдите узкое место
4. Оптимизируйте только это место
5. Измерьте снова — убедитесь, что лучше
6. Повторите

#2. `cProfile` — стандартный профайлер

cProfile — C-расширение, низкий overhead. Отслеживает каждый вызов функции.

#Запуск из командной строки

# Вывод в терминал
python -m cProfile -s cumulative script.py

# Сохранение в файл
python -m cProfile -o profile.out script.py

#Запуск из кода

import cProfile
import pstats

# Способ 1: напрямую
cProfile.run('my_function()', 'profile.out')

# Способ 2: контекстный менеджер
with cProfile.Profile() as pr:
    result = expensive_computation()

pr.dump_stats('profile.out')

#Анализ результатов через `pstats`

import pstats

p = pstats.Stats('profile.out')

# Сортировка и вывод топ-10
p.sort_stats('cumulative').print_stats(10)
p.sort_stats('tottime').print_stats(10)   # самые тяжёлые функции
p.sort_stats('ncalls').print_stats(10)    # самые частые вызовы
p.sort_stats('cumulative', 'calls').print_stats(10)  # несколько критериев

# Фильтр по модулю
p.print_stats('mymodule')

# Ещё один способ — вывести callers/callees
p.print_callers('slow_function')   # кто вызывает slow_function
p.print_callees('main')            # что вызывает main

#Ключевые колонки cProfile

Колонка	Что значит
`ncalls`	Количество вызовов (рекурсивные: `n/n`)
`tottime`	Время внутри функции (без дочерних вызовов)
`percall`	tottime / ncalls
`cumtime`	Суммарное время (с дочерними вызовами)
`percall`	cumtime / ncalls

Что смотреть: высокий tottime → оптимизировать саму функцию. Высокий cumtime при низком tottime → проблема в вызываемых ею функциях.

#Пример интерпретации

# Код
def slow():
    total = 0
    for i in range(1_000_000):
        total += i
    return total

def main():
    for _ in range(100):
        slow()

# Вывод cProfile (сортировка по cumtime):
#    ncalls  tottime  percall  cumtime  percall filename:lineno
#       100    8.234    0.082    8.234    0.082 script.py:1(slow)
#         1    0.001    0.001    8.235    8.235 script.py:6(main)
# → slow() занимает 8.2с суммарно, 0.082с на вызов

#3. `line_profiler` — профилирование по строкам

cProfile показывает функции. line_profiler — каждую строку.

pip install line_profiler

#Использование

# kernprof запускается вместо python
# kernprof -l -v script.py

from line_profiler import LineProfiler

def slow_function(data):
    result = []
    for item in data:            # эта строка ?
        result.append(item * 2)  # или эта?
    return result

profiler = LineProfiler(slow_function)
profiler.run('slow_function(range(1000000))')
profiler.print_stats()

# Через декоратор @profile (при запуске через kernprof)
kernprof -l -v script.py

# Вывод:
# Line #      Hits         Time  Per Hit   % Time  Line Contents
# =========================================================
#      3     1000K    356000.0      0.4     52.3      for item in data:
#      4     1000K    325000.0      0.3     47.7          result.append(...)

#4. `timeit` — точное измерение маленьких фрагментов

Для сравнения двух реализаций. timeit запускает код многократно, минимизируя влияние системных шумов.

#Из командной строки

# Самый простой способ
python -m timeit "sum(range(1000))"
python -m timeit "[x**2 for x in range(1000)]"

# Сравнение
python -m timeit -n 10000 "[x*x for x in range(100)]"
python -m timeit -n 10000 "list(map(lambda x: x*x, range(100)))"

#Из кода

import timeit

# Одна строка
t = timeit.timeit('sum(range(1000))', number=10000)
print(f"sum: {t:.4f}s")

# Lambda (работает с локальными переменными)
data = list(range(1000))
t = timeit.timeit(lambda: sum(data), number=10000)

# Многострочный код
setup = """
import json
data = {'key': 'value', 'num': 42}
"""
code = "json.dumps(data)"
t = timeit.timeit(code, setup=setup, number=100000)

# repeat — несколько замеров для надёжности
import timeit
results = timeit.repeat(
    lambda: sorted(range(1000), reverse=True),
    repeat=5,
    number=1000
)
print(f"Min: {min(results):.4f}s, Max: {max(results):.4f}s")

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

time.perf_counter() измеряет прошедшее wall-clock время, включая ожидание;
time.process_time() считает CPU time процесса и не включает сон или I/O;
time.time() возвращает календарное время и может измениться при коррекции системных часов, поэтому для длительности не подходит.

В Jupyter %timeit expression измеряет одну строку, а %%timeit в первой строке ячейки — весь многострочный блок. Сравнивайте не только лучший запуск, но и разброс результатов.

#Практический пример: list concatenation vs join

import timeit

# Проблема: Медленно: O(n²)
def concat_bad(words):
    result = ""
    for w in words:
        result += w
    return result

# Вариант: Быстро: O(n)
def concat_good(words):
    return "".join(words)

words = ["word"] * 10000

t1 = timeit.timeit(lambda: concat_bad(words), number=100)
t2 = timeit.timeit(lambda: concat_good(words), number=100)
print(f"concat_bad: {t1:.3f}s, concat_good: {t2:.3f}s")
# concat_bad: 0.242s, concat_good: 0.003s → 80x разница!

#5. `dis` — декомпиляция в байткод

Показывает инструкции текущей версии CPython. Их имена и устройство не являются контрактом языка и могут измениться даже между минорными версиями, поэтому сравнивайте вывод в том же интерпретаторе, в котором запускается приложение.

import dis

def list_append(n):
    result = []
    for i in range(n):
        result.append(i)
    return result

def list_comprehension(n):
    return [i for i in range(n)]

dis.dis(list_append)

dis.dis(list_comprehension)

# Не заучивайте конкретные смещения и opcodes. Сопоставьте число инструкций,
# места вызовов и результат timeit на своей версии Python.

#Что показывает `dis`

instructions = list(dis.Bytecode(lambda x, y: x + y))
print([instruction.opname for instruction in instructions])
# В актуальном CPython сложение обычно представлено BINARY_OP. В старых
# версиях встречался BINARY_ADD; CALL_FUNCTION также заменён семейством CALL.

LOAD_FAST, LOAD_GLOBAL и LOAD_DEREF показывают разные источники имени, но из одного названия инструкции нельзя честно вывести ускорение программы: адаптивная специализация и окружающий код тоже влияют на результат. Проверяйте гипотезу профилировщиком и timeit.

#Полезные факты из dis

# Локальная ссылка быстрее глобальной
import math

def slow_math(n):
    result = 0
    for i in range(n):
        result += math.sqrt(i)  # каждый раз ищет math, потом sqrt
    return result

def fast_math(n):
    sqrt = math.sqrt  # кэшируем в локальную переменную
    result = 0
    for i in range(n):
        result += sqrt(i)      # LOAD_FAST вместо LOAD_GLOBAL + LOAD_ATTR
    return result

# fast_math примерно в 1.5-2x быстрее на больших n

С CPython 3.11 интерпретатор может специализировать часто исполняемые инструкции по наблюдаемым типам. Эта adaptive specialization происходит во время выполнения, а набор и названия инструкций меняются между версиями. Поэтому dis помогает объяснить наблюдение на конкретном runtime, но не служит вечной гарантией производительности.

#Публичные хуки профилирования и трассировки

sys.setprofile(callback) сообщает о вызовах и возвратах Python-функций, а также о событиях вызова C-функций. sys.settrace(callback) дополнительно получает построчные события и потому обычно создаёт больший overhead. Оба API передают frame-объект; читать внутреннюю память PyFrameObject не нужно.

import sys


def profiler(frame, event, arg):
    if event in {"call", "return"}:
        print(event, frame.f_code.co_name)


sys.setprofile(profiler)
try:
    run_workload()
finally:
    sys.setprofile(None)

#6. `tracemalloc` — профилирование памяти

import tracemalloc

# Запуск (число = глубина стека вызовов в трассировке)
tracemalloc.start(25)

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

# Снимок состояния памяти
snapshot = tracemalloc.take_snapshot()

# Топ по выделениям
top_stats = snapshot.statistics('lineno')
print("Топ 5 потребителей памяти:")
for stat in top_stats[:5]:
    print(stat)

# Текущее и пиковое потребление
current, peak = tracemalloc.get_traced_memory()
print(f"Текущее: {current / 1024:.1f} KB")
print(f"Пиковое: {peak / 1024:.1f} KB")

tracemalloc.stop()

#Сравнение двух снимков — поиск утечек

import tracemalloc
import gc

tracemalloc.start()

# Базовая линия
gc.collect()
snapshot1 = tracemalloc.take_snapshot()

# Подозрительный код
for _ in range(1000):
    process_something()

gc.collect()
snapshot2 = tracemalloc.take_snapshot()

# Разница — что выросло?
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
print("Выросло больше всего:")
for stat in top_stats[:10]:
    print(stat)
# Показывает: файл:строка +N KB (+M аллокаций)

#7. `memory_profiler` — профиль памяти по строкам

pip install memory_profiler

from memory_profiler import profile

@profile
def create_large_structure():
    data = [0] * 10_000_000         # список из 10M нулей
    text = " ".join(str(i) for i in range(100_000))
    del data                         # явное освобождение
    return text

create_large_structure()

# Вывод:
# Line #    Mem usage    Increment   Line Contents
# =================================================
#      4   50.3 MiB     50.3 MiB    @profile
#      5   50.3 MiB      0.0 MiB    def create_large_structure():
#      6  126.7 MiB     76.4 MiB        data = [0] * 10_000_000
#      7  135.1 MiB      8.4 MiB        text = " ".join(...)
#      8   57.3 MiB    -77.8 MiB        del data

# Или запуск через CLI:
python -m memory_profiler script.py

# Построить граф потребления памяти:
mprof run script.py
mprof plot

tracemalloc учитывает выделения памяти, которые отслеживает Python allocator, и хорошо связывает их со строками Python-кода. memory_profiler и системные инструменты смотрят на память процесса. RSS — страницы, находящиеся в физической памяти, VMS — всё виртуальное адресное пространство процесса; ни одна из этих метрик сама по себе не равна сумме размеров живых Python-объектов.

#8. `py-spy` — sampling profiler

Не требует изменений в коде! Подключается к работающему процессу.

pip install py-spy

# Profiling скрипта
py-spy record -o profile.svg -- python script.py

# Attach к работающему процессу (требует sudo)
py-spy record -o profile.svg --pid 12345

# Live top (как htop для Python)
py-spy top -- python script.py

Генерирует flame graph (SVG) — интерактивная визуализация:

Ширина блока = время в этой функции
Высота = глубина стека вызовов
Нажмите на блок — zoom in

py-spy периодически снимает стеки процесса извне, не добавляя вызовы в исходный код приложения. Нагрузка остаётся, но обычно ниже, чем у детерминированного профилировщика каждого вызова.

#9. `snakeviz` — визуализация cProfile в браузере

pip install snakeviz

# Генерируем профиль
python -m cProfile -o profile.out script.py

# Открываем в браузере
snakeviz profile.out

Icicle chart — интерактивная диаграмма. Клик → zoom. Намного удобнее текстового вывода pstats.

#10. Стратегии оптимизации

#Алгоритмическая сложность (самый большой выигрыш)

# O(n²) — поиск дубликатов через вложенный цикл
def find_duplicates_slow(lst):
    result = []
    for i in range(len(lst)):
        for j in range(i + 1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in result:
                result.append(lst[i])
    return result

# O(n) — через множество
def find_duplicates_fast(lst):
    seen = set()
    duplicates = set()
    for x in lst:
        if x in seen:
            duplicates.add(x)
        seen.add(x)
    return list(duplicates)

# На n=10000: slow ~2.5s, fast ~0.001s → 2500x разница

#Правильные структуры данных

# Поиск элемента: list O(n) vs set O(1)
items_list = list(range(100_000))
items_set = set(range(100_000))

# timeit: 99999 in items_list → 1.8ms
# timeit: 99999 in items_set  → 0.04ms → 45x быстрее

# Очередь: list.pop(0) O(n) vs deque.popleft() O(1)
from collections import deque
queue = deque(range(10000))
queue.popleft()   # O(1), не O(n) как list.pop(0)

# Counter вместо ручного подсчёта
from collections import Counter
counter = Counter(words)   # один проход, C-оптимизированный

#Кэширование результатов

from functools import lru_cache, cache

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

# cache (Python 3.9+) — без ограничений (быстрее lru_cache)
@cache
def expensive_computation(x, y):
    return x ** y

# 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

#Векторизация через NumPy

import numpy as np
import time

n = 1_000_000

# Чистый Python: ~200ms
data = list(range(n))
start = time.perf_counter()
result = [x ** 2 for x in data]
print(f"Python: {time.perf_counter() - start:.3f}s")

# NumPy: ~2ms (100x быстрее!)
arr = np.arange(n)
start = time.perf_counter()
result = arr ** 2
print(f"NumPy: {time.perf_counter() - start:.3f}s")

# Ключевые функции вместо Python-циклов:
# np.sum, np.mean, np.std, np.where, np.vectorize
# np.dot (матричное умножение)

NumPy хранит однородные значения плотно и выполняет векторные циклы в нативном коде. Обычный список хранит ссылки на отдельные Python-объекты. Векторизация убирает большую часть накладных расходов интерпретатора, но выигрыш зависит от размера данных, dtype и числа временных массивов.

#Память экземпляров через `slots`

class Point:
    __slots__ = ("x", "y")

    def __init__(self, x: float, y: float) -> None:
        self.x = x
        self.y = y

Slots могут убрать __dict__ экземпляра и снизить расход памяти, если таких объектов много. Эффект зависит от наследования, поддержки слабых ссылок и сборки Python, поэтому сравнивайте полный размер объектов на целевой среде. Подробнее о __slots__ в контексте производительности — в python_performance.

#Локальные переменные vs глобальные

import math

# Медленно: каждый раз LOAD_GLOBAL + LOAD_ATTR
def slow_sqrt(n):
    result = 0.0
    for i in range(n):
        result += math.sqrt(i)
    return result

# Быстро: локальная ссылка
def fast_sqrt(n):
    local_sqrt = math.sqrt   # LOAD_FAST вместо двух операций
    result = 0.0
    for i in range(n):
        result += local_sqrt(i)
    return result

# fast_sqrt в 1.3-1.5x быстрее для n=1_000_000

#Строки: join вместо конкатенации

# Проблема: O(n²): каждая конкатенация создаёт новый объект
def slow_concat(words):
    result = ""
    for w in words:
        result += w   # каждый раз новая строка!
    return result

# Вариант: O(n): одна аллокация
def fast_concat(words):
    return "".join(words)

# Для форматирования с разными типами:
parts = [str(x) for x in data]
result = ", ".join(parts)

#Генераторы для больших данных

# Загружает всё в память: 800 MB для 10M чисел
def squares_list(n):
    return [x**2 for x in range(n)]

# Ленивое вычисление: ~0 памяти
def squares_gen(n):
    return (x**2 for x in range(n))

# Читаем большой файл построчно
def count_lines(filename):
    with open(filename) as f:
        return sum(1 for _ in f)  # не загружает весь файл

#JIT компиляция через Numba

from numba import njit
import numpy as np

# Обычная Python функция: ~500ms для n=1M
def python_sum(arr):
    total = 0.0
    for x in arr:
        total += x
    return total

# JIT-компилированная версия: ~2ms (250x быстрее!)
@njit
def numba_sum(arr):
    total = 0.0
    for x in arr:
        total += x
    return total

arr = np.random.rand(1_000_000)
numba_sum(arr)  # первый вызов — компиляция (~200ms)
numba_sum(arr)  # последующие — быстро

#Cython и вызов C-библиотек

Cython переводит .pyx-модуль в C extension. Простая компиляция Python-кода не гарантирует ускорения: выигрыш появляется, когда статические типы (cdef и typed memoryviews) убирают операции с Python-объектами в горячем цикле. Обычный workflow — описать модуль, собрать его через cythonize() и сравнить результат с исходной реализацией одним benchmark.

ctypes входит в стандартную библиотеку и вызывает функции по ABI уже собранной динамической библиотеки. cffi — внешняя зависимость с более декларативным описанием C API и режимом предварительной компиляции. Выбор между ними определяется интерфейсом библиотеки и стоимостью поддержки, а не предполагаемой скоростью без измерения.

#Параллелизм

import multiprocessing

# CPU-bound: используйте ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor

def cpu_task(data):
    return sum(x**2 for x in data)

chunks = [range(i*1000, (i+1)*1000) for i in range(16)]

# Без параллелизма: 16 * T
results = [cpu_task(chunk) for chunk in chunks]

# С параллелизмом: ~4 * T (на 4-ядерном CPU)
with ProcessPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(cpu_task, chunks))

# I/O-bound: используйте asyncio или ThreadPoolExecutor

Асинхронный workload профилируют теми же инструментами, но интерпретируют по задачам: asyncio.create_task() планирует независимую работу, а asyncio.gather() ждёт набор результатов. Они не являются профилировщиками; время CPU по-прежнему измеряют cProfile/py-spy, а ожидания — отдельными метриками операций.

#11. Сводная таблица инструментов

Инструмент	Что измеряет	Когда использовать
`cProfile`	Время по функциям	Первый шаг профилирования
`line_profiler`	Время по строкам	После cProfile нашёл функцию
`timeit`	Время одного фрагмента	Сравнение двух реализаций
`dis`	Байткод инструкции	Понимание микрооптимизаций
`tracemalloc`	Аллокации памяти	Поиск утечек памяти
`memory_profiler`	Память по строкам	После tracemalloc нашёл функцию
`py-spy`	Выборочное (без изменений кода)	Профилирование в рабочей среде
`snakeviz`	Визуализация cProfile	Анализ больших профилей

#12. Как читать flame graph

Flame graph — самый эффективный способ понять профиль:

                  ┌──────────────────────────────┐
           3      │          main()              │
                  ├─────────────┬────────────────┤
           2      │  read_db()  │   process()    │
                  ├──────┬──────┤    ──────────  │
           1      │ sql()│parse()│              │
                  └──────┴──────┴────────────────┘
                    Время →

Ось X = время (или % вызовов), ось Y = глубина стека
Широкий блок = функция занимает много времени
Плато наверху = нет дочерних функций — именно здесь тратится время
Ищите широкие плато наверху — это ваши узкие места

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

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

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

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

import cProfile
import io
import pstats


def workload() -> int:
    return sum(value * value for value in range(10_000))


profile = cProfile.Profile()
profile.enable()
result = workload()
profile.disable()

stream = io.StringIO()
pstats.Stats(profile, stream=stream).sort_stats("cumulative").print_stats(5)
assert result > 0
assert "workload" in stream.getvalue()

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

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

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

Сопоставьте детерминированный профиль с выборочным профилировщиком и tracemalloc. Объясните влияние самого инструмента, прогрева, системного шума и почему один запуск не доказывает улучшение.

#Упражнения

Профилируйте функцию сортировки пузырьком через cProfile и line_profiler. Найдите узкое место.
Сравните timeit для трёх способов создания списка от 0 до N: цикл + append, list comprehension, list(range(N)).
Используйте tracemalloc для сравнения потребления памяти между list и generator при обработке 1M элементов.
Напишите декоратор @profile_call который замеряет время через timeit.default_timer и количество аллокаций через tracemalloc.
Оптимизируйте функцию поиска пересечения двух списков: вначале O(n²), затем O(n) через множество. Сравните через timeit.
Изучите flame graph для своего кода через py-spy. Объясните, что означает самый широкий блок.

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

Сохраните версию Python, входные данные, число прогонов и исходный результат: оптимизированная функция обязана возвращать то же значение.
timeit сравнивает одинаковую работу, включает несколько повторов и сообщает распределение, а не один красивый запуск.
Для list/generator отдельно измерьте память контейнера и память результата; материализация генератора устраняет ожидаемое преимущество.
Декоратор профилирования не смешивает аллокации параллельного кода с измеряемым вызовом без оговорки.
Переход к set может менять порядок и удалять дубликаты — зафиксируйте семантику до заявления об O(n).
Ширина блока flame graph означает долю собранных sample, а не автоматически «плохую функцию».

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

Закрепите тему в лаборатории «Защищаемые измерения производительности». Вы сначала проверите эквивалентность реализаций, затем соберёте чередующуюся серию измерений на свежих наборах и объясните границы полученного вывода.

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

Порядок работы: 1. Убедитесь что код корректен (тесты!) 2. Измерьте производительность (профайлер) 3. Найдите узкое место 4. Оптимизируйте только это место 5. Измерьте снова — убедитесь, что лучше 6. Повторите

Колонка

Что значит

ncalls

Количество вызовов (рекурсивные: n/n)

tottime

Время внутри функции (без дочерних вызовов)

percall

tottime / ncalls

cumtime

Суммарное время (с дочерними вызовами)

percall

cumtime / ncalls

Инструмент

Что измеряет

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

cProfile

Время по функциям

Первый шаг профилирования

line_profiler

Время по строкам

После cProfile нашёл функцию

timeit

Время одного фрагмента

Сравнение двух реализаций

dis

Байткод инструкции

Понимание микрооптимизаций

tracemalloc

Аллокации памяти

Поиск утечек памяти

memory_profiler

Память по строкам

После tracemalloc нашёл функцию

py-spy

Выборочное (без изменений кода)

Профилирование в рабочей среде

snakeviz

Визуализация cProfile

Анализ больших профилей

┌──────────────────────────────┐ 3 │ main() │ ├─────────────┬────────────────┤ 2 │ read_db() │ process() │ ├──────┬──────┤ ────────── │ 1 │ sql()│parse()│ │ └──────┴──────┴────────────────┘ Время →