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

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

4 голосов
/ 10 сентября 2009

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

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

Аналогичные

Ответы [ 2 ]

11 голосов
/ 10 сентября 2009

Если только функция, к которой вы стремитесь, не сделает что-то особенное, чтобы отметить «один экземпляр меня активен в стеке» (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

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

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

1 голос
/ 10 сентября 2009

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

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.)

...