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

Question

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

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

Аналогичный

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

Answer 1

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

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

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

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

В-третьих, столь же важный, как первый пункт, set, которая была структурой данных, первоначально выбранной OP, является неправильным выбором: функция может быть в стеке несколько раз (прямая или косвенная рекурсия). Нам явно нужен "множественный набор" (также известный как "сумка"), подобная множеству структура, которая отслеживает "сколько раз" присутствует каждый элемент. В 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.)