在 Python 中确定特定函数是否在堆栈上的有效方法

Question

对于调试，判断一个特定函数是否在调用堆栈上更高通常很有用。例如，我们经常只想在某个函数调用我们时才运行调试代码。

一种解决方案是检查所有更高的堆栈条目，但这是在堆栈深处并重复调用的函数中，这会导致过多的开销。问题是要找到一种方法，让我们能够以一种相当有效的方式确定某个特定函数是否在调用堆栈上更高。

类似

• Obtaining references to function objects on the execution stack from the frame object? - 这个问题的重点是获取函数对象，而不是确定我们是否在一个特定的函数中。尽管可以应用相同的技术，但它们最终可能效率极低。

Answer 1

除非您的目标函数做了一些非常特殊的事情来标记“我的一个实例在堆栈上处于活动状态”（IOW：如果该函数是原始且不可触及并且不可能意识到这种特殊需要你的），没有其他替代方法可以逐帧遍历堆栈，直到您到达顶部（并且该函数不存在）或您感兴趣的函数的堆栈帧。正如该问题的几位专家所指出的那样，是否值得努力优化这一点非常值得怀疑。但是，为了论证的目的，假设它是值得的......：

编辑：最初的答案（由 OP 提供）有很多缺陷，但有些已经修复，所以我正在编辑以反映当前情况以及为什么某些方面很重要。

首先，在装饰器中使用try/except 或with 非常重要，以便正确考虑从被监视函数的任何退出，而不仅仅是正常退出（如原始OP自己的答案的版本确实如此）。

其次，每个装饰器都应该确保它保持被装饰函数的__name__ 和__doc__ 完整——这就是functools.wraps 的用途（还有其他方法，但wraps 使它最简单）。

第三，与第一点一样重要，set，这是 OP 最初选择的数据结构，是错误的选择：一个函数可以在堆栈上多次（直接或间接递归）。我们显然需要一个“多集”（也称为“袋子”），一种类似集的结构，可以跟踪每个项目出现的“次数”。在 Python 中，multiset 的自然实现是作为将键映射到计数的 dict，而这反过来最容易实现为 collections.defaultdict(int)。

第四，一般的方法应该是线程安全的（至少当它可以很容易地实现时；-）。幸运的是，threading.local 在适用时使它变得微不足道——在这里，它肯定应该是（每个堆栈都有自己独立的调用线程）。

第五，在一些 cmets 中提出了一个有趣的问题（注意在某些答案中提供的装饰器与其他装饰器的关系有多么糟糕：监控装饰器似乎必须是最后一个（最外层）的装饰器，否则检查会中断。这来自于使用函数对象本身作为监控字典的键的自然但不幸的选择。

我建议通过选择不同的键来解决这个问题：让装饰器采用（例如）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 核心的情况下是否可行。）