sys.settrace: как устроены дебаггеры, coverage и profilers в Python
Когда запускаешь pytest --cov код выполняется как обычно, но в конце появляется отчёт о покрытии. Как pytest узнаёт, какие строки выполнились? Ответ в sys.settrace, это низкоуровневый хук, который позволяет перехватывать каждый шаг интерпретатора.
На этом механизме построены coverage.py, pdb, PyCharm debugger, hunters, и десятки других инструментов. Разберём, как это работает изнутри и почему трассировка устроена именно так.
Модель выполнения CPython
Прежде чем говорить о трассировке, нужно понять, как CPython выполняет код. Python-код компилируется в байткод — последовательность инструкций для виртуальной машины. Каждая функция имеет объект code, содержащий байткод и метаданные.
Когда функция вызывается, создаётся frame — объект, хранящий состояние выполнения: локальные переменные, текущую позицию в байткоде, ссылку на код. Frames образуют стек вызовов.
Интерпретатор — это цикл, который берёт следующую инструкцию из текущего frame и выполняет её. Трассировка встраивается в этот цикл: перед выполнением определённых инструкций интерпретатор проверяет, установлена ли trace-функция, и если да — вызывает её.
Базовый API: sys.settrace
sys.settrace(func) устанавливает глобальную trace-функцию. Она вызывается при определённых событиях:
import sys
def trace_calls(frame, event, arg):
"""
frame — текущий frame выполнения
event — тип события ('call', 'line', 'return', 'exception')
arg — дополнительные данные (зависит от event)
"""
print(f"{event}: {frame.f_code.co_name}, line {frame.f_lineno}")
return trace_calls # Возвращаем trace-функцию для локальных событий
sys.settrace(trace_calls)
def example():
x = 1
y = 2
return x + y
example()
sys.settrace(None)
Вывод:
call: example, line 1 line: example, line 2 line: example, line 3 line: example, line 4 return: example, line 4
Четыре типа событий:
- call — вызов функции,
arg= None - line — переход на новую строку,
arg= None - return — возврат из функции,
arg= возвращаемое значение - exception — возникло исключение,
arg= (type, value, traceback)
trace-функция должна вернуть функцию для отслеживания локальных событий (line, return, exception). Если вернуть None,то локальные события этого frame игнорируются. Это механизм фильтрации: можно отслеживать вызовы всех функций, но line-события только для интересующих.
Анатомия frame-объекта
Frame — это окно в состояние выполнения. Доступные атрибуты:
def inspect_frame(frame, event, arg):
# Информация о коде
code = frame.f_code
print(f"Function: {code.co_name}")
print(f"Filename: {code.co_filename}")
print(f"First line: {code.co_firstlineno}")
# Текущее состояние
print(f"Current line: {frame.f_lineno}")
print(f"Locals: {frame.f_locals}")
# Стек вызовов
print(f"Caller frame: {frame.f_back}")
return inspect_frame
f_locals — словарь локальных переменных. Но есть нюанс, это снапшот на момент вызова trace-функции. Изменения в f_locals не отражаются на реальных переменных (кроме специальных случаев с ctypes). Для дебаггеров это проблема,нельзя просто так изменить переменную во время отладки.
f_back — ссылка на frame вызывающей функции. Позволяет ходить по стеку вызовов.
Как работает coverage.py
Coverage.py — стандартный инструмент измерения покрытия кода. Его архитектура: trace-функция отмечает выполненные строки, потом сравниваем с исходным кодом.
Упрощённая версия:
class SimpleCoverage:
def __init__(self):
self.executed_lines = {} # filename -> set of line numbers
def start(self):
sys.settrace(self._trace)
def stop(self):
sys.settrace(None)
def _trace(self, frame, event, arg):
if event == 'call':
# Для каждого вызова возвращаем локальный tracer
return self._trace_lines
return None
def _trace_lines(self, frame, event, arg):
if event == 'line':
filename = frame.f_code.co_filename
lineno = frame.f_lineno
if filename not in self.executed_lines:
self.executed_lines[filename] = set()
self.executed_lines[filename].add(lineno)
return self._trace_lines
def report(self):
for filename, lines in self.executed_lines.items():
print(f"{filename}: {sorted(lines)}")
Реальный coverage.py сложнее, он написан на C для производительности, отслеживает арки (transitions между строками) для branch coverage, интегрируется с multiprocessing.
Кстати, тут еще интересно, что coverage должен знать, какие строки вообще являются исполняемыми. Пустые строки, комментарии, строки с только else: не являются. Для этого coverage.py парсит AST исходного кода и определяет «исполняемые» строки.
Дебаггеры
pdb использует тот же механизм, но с другой логикой:
class SimpleDebugger:
def __init__(self):
self.breakpoints = set() # (filename, lineno)
self.step_mode = False
def set_break(self, filename, lineno):
self.breakpoints.add((filename, lineno))
def _trace(self, frame, event, arg):
if event == 'call':
return self._trace_lines
return None
def _trace_lines(self, frame, event, arg):
if event != 'line':
return self._trace_lines
filename = frame.f_code.co_filename
lineno = frame.f_lineno
should_stop = (
self.step_mode or
(filename, lineno) in self.breakpoints
)
if should_stop:
self._interact(frame)
return self._trace_lines
def _interact(self, frame):
"""REPL для отладки"""
while True:
cmd = input(f"(Pdb) {frame.f_code.co_name}:{frame.f_lineno}> ")
if cmd == 'n': # next
self.step_mode = True
break
elif cmd == 'c': # continue
self.step_mode = False
break
elif cmd == 'p': # print locals
print(frame.f_locals)
elif cmd == 'q': # quit
sys.exit(0)
Команда step (войти в функцию) vs next (перешагнуть) реализуется через анализ событий: при step мы реагируем на следующее событие, при next игнорируем вложенные call/return и ждём line в текущем frame.
Условные breakpoints — просто проверка условия при остановке:
# breakpoint с условием x > 10
if eval(condition, frame.f_globals, frame.f_locals):
self._interact(frame)
Почему трассировка медленная
Trace-функция вызывается на каждую строку кода. Все это беда для производительности. Почему так:
- Вызов Python-функции — сам по себе дорогой (создание frame, setup arguments)
- Переход Python ↔ C — trace-функция на Python, интерпретатор на C
- Невозможность оптимизаций — трассировка отключает многие оптимизации CPython
Поэтому coverage.py имеет C-расширение, а профилировщики используют другие механизмы.
sys.setprofile: легковесная альтернатива
sys.setprofile похож на settrace, но события только на границах функций:
- call
- return
- c_call (вызов C-функции)
- c_return
- c_exception
Нет событий line, огромная экономия. cProfile построен на этом:
import sys
call_counts = {}
def profiler(frame, event, arg):
if event == 'call':
name = frame.f_code.co_name
call_counts[name] = call_counts.get(name, 0) + 1
return profiler
sys.setprofile(profiler)
# ... код ...
sys.setprofile(None)
Для профилирования по времени можно использовать time.perf_counter() на call/return и считать разницу.
Per-thread трассировка
sys.settrace устанавливает trace-функцию для текущего потока. Для многопоточных приложений нужно устанавливать в каждом потоке:
import threading
import sys
def install_tracer():
sys.settrace(my_trace_func)
# Устанавливаем в главном потоке
sys.settrace(my_trace_func)
# Для новых потоков
threading.settrace(install_tracer) # deprecated в 3.12
# Или вручную
original_start = threading.Thread.start
def patched_start(self):
original_run = self.run
def traced_run():
sys.settrace(my_trace_func)
original_run()
self.run = traced_run
original_start(self)
threading.Thread.start = patched_start
coverage.py делает подобный monkey-patching для отслеживания всех потоков.
Python 3.12+: sys.monitoring
Python 3.12 добавил sys.monitoring — новый, более эффективный API для инструментации:
import sys
def line_handler(code, line_number):
print(f"Line {line_number} in {code.co_name}")
# Регистрируем callback
sys.monitoring.register_callback(
sys.monitoring.PROFILER_ID,
sys.monitoring.events.LINE,
line_handler
)
# Включаем события
sys.monitoring.set_events(
sys.monitoring.PROFILER_ID,
sys.monitoring.events.LINE
)
Меньше overhead — callbacks вызываются напрямую, без создания frame, плюсом имеем гранулярный контроль, можно включать события для конкретных функций.
API ещё стабилизируется.
Детектор утечек
Пример нетипичного применения — отслеживание незакрытых файлов:
import sys
import weakref
import traceback
open_files = {} # id -> (weakref, traceback)
original_open = open
def tracked_open(*args, **kwargs):
f = original_open(*args, **kwargs)
# Сохраняем traceback места открытия
tb = ''.join(traceback.format_stack())
open_files[id(f)] = (weakref.ref(f), tb)
return f
def trace_closes(frame, event, arg):
if event == 'return':
# Проверяем, не закрылся ли файл
for fid, (ref, tb) in list(open_files.items()):
if ref() is None:
del open_files[fid]
return trace_closes
# Установка
import builtins
builtins.open = tracked_open
sys.settrace(trace_closes)
# В конце программы
def report_leaks():
for fid, (ref, tb) in open_files.items():
f = ref()
if f and not f.closed:
print(f"Unclosed file: {f.name}")
print(f"Opened at:\n{tb}")
В проде использовали бы del или context managers.
sys.settrace — низкоуровневый механизм. Для большинства задач достаточно готовых инструментов.
Хотите разбираться в Python не только на уровне «пишу код», но и понимать, что делает интерпретатор, где теряется время и как устроены инструменты? Специализация разработчик на Python помогает пройти путь от базы до уверенного профессионала: практика, код-ревью, проекты и инженерные привычки.
Источник: https://habr.com/ru/companies/otus/articles/988880/