Эффективный способ определить, находится ли определенная функция в стеке Python

Question

Для отладки часто бывает полезно определить, находится ли конкретная функция выше в стеке вызовов.Например, мы часто хотим запускать код отладки только тогда, когда нас вызывает определенная функция.

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

Похожий

• Получение ссылок на функциональные объекты в стеке выполнения из объекта фрейма? — Этот вопрос направлен на получение объектов функции, а не на определение того, находимся ли мы в конкретной функции.Хотя можно применять те же методы, они могут оказаться крайне неэффективными.

Answer 1

Если только функция, к которой вы стремитесь, не делает что-то особенное, чтобы отметить «один экземпляр меня активен в стеке» (IOW:если функция нетронута и неприкосновенна и не может быть уведомлена об этой вашей особой потребности), нет мыслимой альтернативы прохождению кадр за кадром вверх по стеку, пока вы не достигнете вершины (а функции там нет) или кадр стека для интересующей вас функции.Как показывают несколько комментариев к вопросу, крайне сомнительно, стоит ли стремиться к оптимизации этого.Но если предположить ради аргументации, что это был стоит...:

Редактировать:в исходном ответе (от ОП) было много дефектов, но некоторые из них с тех пор были исправлены, поэтому я редактирую, чтобы отразить текущую ситуацию и то, почему некоторые аспекты важны.

Прежде всего, очень важно использовать try/except, или with, в декораторе, так что ЛЮБОЙ выход из отслеживаемой функции правильно учитывается, а не только обычные (как это было в исходной версии собственного ответа ОП).

Во-вторых, каждый декоратор должен гарантировать, что он сохраняет свойства декорируемой функции. __name__ и __doc__ нетронутый - вот что functools.wraps для (есть и другие способы, но wraps делает это проще всего).

В-третьих, столь же важный, как и первый пункт, set, изначально выбранная ОП, является неправильным выбором:функция может находиться в стеке несколько раз (прямая или косвенная рекурсия).Нам явно нужен «мульти-набор» (также известный как «сумка»), структура, подобная набору, которая отслеживает, «сколько раз» присутствует каждый элемент.В Python естественной реализацией мультимножества является сопоставление ключей со счетчиками, что, в свою очередь, удобнее всего реализовать как collections.defaultdict(int).

В-четвертых, общий подход должен быть потокобезопасным (по крайней мере, если это можно легко сделать ;-).К счастью, threading.local делает это тривиальным, когда это применимо - и здесь это обязательно должно быть (каждый стек имеет свой отдельный поток вызовов).

В-пятых, интересная проблема, затронутая в некоторых комментариях (обратите внимание, насколько плохо предложенные в некоторых ответах декораторы играют с другими декораторами:декоратор мониторинга должен быть ПОСЛЕДНИМ (самым внешним), иначе проверка прерывается.Это происходит из-за естественного, но неудачного выбора использования самого функционального объекта в качестве ключа в словаре мониторинга.

Предлагаю решить эту проблему другим выбором ключа:заставить декоратора взять (скажем, строку) identifier аргумент, который должен быть уникальным (в каждом данном потоке) и использовать идентификатор в качестве ключа в словаре мониторинга.Код, проверяющий стек, конечно же, должен знать об идентификаторе и также использовать его.

Во время декорирования декоратор может проверить свойство уникальности (используя отдельный набор).Идентификатор можно оставить по умолчанию равным имени функции (поэтому он явно требуется только для сохранения гибкости мониторинга одноименных функций в одном и том же пространстве имен);от свойства уникальности можно явно отказаться, если несколько отслеживаемых функций должны считаться «одинаковыми» для целей мониторинга (это может случиться, если заданная def оператор предназначен для выполнения несколько раз в немного разных контекстах, чтобы создать несколько функциональных объектов, которые программисты хотят считать «одной и той же функцией» для целей мониторинга).Наконец, должна быть возможность при необходимости вернуться к «функциональному объекту в качестве идентификатора» для тех редких случаев, когда ЗАВЕДЕНО, что дальнейшее декорирование невозможно (поскольку в этих случаях это может быть самый удобный способ гарантировать уникальность).

Итак, объединив эти многочисленные соображения, мы могли бы (включая threadlocal_var служебная функция, которая, вероятно, уже будет в модуле панели инструментов;-) что-то вроде следующего...:

import collections
import functools
import threading

threadlocal = threading.local()

def threadlocal_var(varname, factory, *a, **k):
  v = getattr(threadlocal, varname, None)
  if v is None:
    v = factory(*a, **k)
    setattr(threadlocal, varname, v)
  return v

def monitoring(identifier=None, unique=True, use_function=False):
  def inner(f):
    assert (not use_function) or (identifier is None)
    if identifier is None:
      if use_function:
        identifier = f
      else:
        identifier = f.__name__
    if unique:
      monitored = threadlocal_var('uniques', set)
      if identifier in monitored:
        raise ValueError('Duplicate monitoring identifier %r' % identifier)
      monitored.add(identifier)
    counts = threadlocal_var('counts', collections.defaultdict, int)
    @functools.wraps(f)
    def wrapper(*a, **k):
      counts[identifier] += 1
      try:
        return f(*a, **k)
      finally:
        counts[identifier] -= 1
    return wrapper
  return inner

Я не тестировал этот код, поэтому он может содержать какую-либо опечатку или что-то в этом роде, но я предлагаю его, потому что надеюсь, что он охватывает все важные технические моменты, которые я объяснил выше.

Стоит ли все это того?Вероятно, нет, как объяснялось ранее.Однако я думаю в духе «если вообще стоит что-то делать, то стоит делать правильно» ;-).

Answer 2

Мне не очень нравится этот подход, но вот исправленная версия того, что вы делали:

from collections import defaultdict
import threading
functions_on_stack = threading.local()

def record_function_on_stack(f):
    def wrapped(*args, **kwargs):
        if not getattr(functions_on_stack, "stacks", None):
            functions_on_stack.stacks = defaultdict(int)
        functions_on_stack.stacks[wrapped] += 1

        try:
            result = f(*args, **kwargs)
        finally:
            functions_on_stack.stacks[wrapped] -= 1
            if functions_on_stack.stacks[wrapped] == 0:
                del functions_on_stack.stacks[wrapped]
        return result

    wrapped.orig_func = f
    return wrapped

def function_is_on_stack(f):
    return f in functions_on_stack.stacks

def nested():
    if function_is_on_stack(test):
        print "nested"

@record_function_on_stack
def test():
    nested()

test()

Это обрабатывает рекурсию, многопоточность и исключения.

Мне не нравится такой подход по двум причинам:

• Это не сработает, если функция будет декорирована дополнительно:это, должно быть, последний декоратор.
• Если вы используете это для отладки, это означает, что вам придется редактировать код в двух местах, чтобы его использовать;один для добавления декоратора и один для его использования.Гораздо удобнее просто изучить стек, так что вам нужно будет редактировать только тот код, который вы отлаживаете.

Лучшим подходом было бы проверить стек напрямую (возможно, как собственное расширение для скорости) и, если возможно, найти способ кэшировать результаты на время жизни кадра стека.(Однако я не уверен, возможно ли это без изменения ядра Python.)