在Python中高效判断特定函数是否在栈上

5 投票
2 回答
5185 浏览
提问于 2025-04-15 14:12

在调试的时候,知道某个特定的函数在调用栈中处于什么位置是很有用的。比如,我们通常只想在某个特定的函数调用了我们的时候,才运行调试代码。

一种解决办法是查看所有在调用栈上方的记录,但如果这个函数在调用栈中很深,而且被反复调用,这样做会导致性能下降,浪费很多资源。问题是我们需要找到一种方法,能够有效地判断某个特定的函数是否在调用栈的上方。

相关内容

2 个回答

1

我其实不太喜欢这种做法,不过这是你之前做的一个改进版本:

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核心的情况下这是否可行。)

14

除非你想要的函数有一些特别的方式来标记“我在栈上活跃的实例”,也就是说,如果这个函数是完全独立的,无法意识到你这个特殊的需求,那么就没有其他办法可以做到这一点,只能一帧一帧地向上检查栈,直到你找到要找的函数,或者到达栈顶(而这个函数不在那)。正如一些评论所说,这种优化是否值得是非常值得怀疑的。不过,假设为了讨论的目的,这样做是有价值的……:

编辑:原来的回答(由提问者提供)有很多缺陷,但有些已经修复,所以我在编辑以反映当前的情况,以及为什么某些方面很重要。

首先,在装饰器中使用try/exceptwith是非常重要的,这样可以确保监控的函数的任何退出都能被正确记录,而不仅仅是正常的退出(原版提问者的回答就是这样)。

其次,每个装饰器都应该确保保留被装饰函数的__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

我没有测试这段代码,所以可能包含一些拼写错误或类似问题,但我提供它是因为我希望它涵盖了我上面解释的所有重要技术点。

这一切值得吗?可能不值得,正如之前所解释的。不过,我认为“如果值得做,那就值得做好”;-)。

撰写回答