Méthode efficace pour déterminer si une fonction particulière est sur la pile en Python

Question

Pour le débogage, il est souvent utile d'indiquer si une fonction particulière est située plus haut dans la pile d'appels. Par exemple, nous ne voulons souvent exécuter du code de débogage que lorsqu'une fonction particulière nous appelle.

Une solution consiste à examiner toutes les entrées de la pile plus haut, mais dans une fonction située au fond de la pile et appelée à plusieurs reprises, cela entraîne une surcharge excessive. La question est de trouver une méthode qui nous permette de déterminer si une fonction particulière est située plus haut dans la pile d’appels d’une manière raisonnablement efficace.

similaire

• Obtention de références aux objets de fonction sur la pile d’exécution à partir de l’objet frame? - Cette question porte sur l’obtention des objets de fonction, plutôt que sur la détermination de l’affectation d’une fonction particulière. Bien que les mêmes techniques puissent être appliquées, elles risquent de devenir extrêmement inefficaces.

Answer 1

Sauf si la fonction que vous visez fait quelque chose de très spécial pour marquer "une instance de moi est active sur la pile". (IOW: si la fonction est vierge et intouchable et qu'il est impossible de la mettre au courant de ce besoin particulier), il n'y a pas d'autre solution envisageable que de marcher image par image dans la pile jusqu'à ce que vous atteigniez le sommet (et la fonction est pas là) ou un cadre de pile pour votre fonction d'intérêt. Comme plusieurs commentaires à la question l'indiquent, il est extrêmement douteux que cela soit intéressant d'optimiser cela. Mais supposons, pour l’argumentation, que cela en valait la peine ...:

Modifier : la réponse d'origine (par le PO) comportait de nombreux défauts, mais certains ont été corrigés depuis. Je modifie donc pour refléter la situation actuelle et expliquer pourquoi certains aspects sont importants.

Tout d’abord, il est essentiel d’utiliser essayer / sauf ou avec dans le décorateur, de sorte que TOUTE sortie d’une fonction être surveillé est correctement comptabilisé, pas seulement les normaux (comme le faisait la version originale de la propre réponse du PO).

Deuxièmement, chaque décorateur doit s’assurer que le __ nom __ et le __ doc __ de la fonction décorée sont intacts - c’est le but de functools.wraps . Il existe d'autres moyens, mais encapsule le simplifie).

Troisièmement, tout aussi crucial que le premier point, un set , qui était la structure de données initialement choisie par le PO, est un mauvais choix: une fonction peut être sur la pile plusieurs fois (directement). ou récursion indirecte). Nous avons clairement besoin d'un " multi-set " (également connu sous le nom de "sac"), une structure en forme de jeu qui garde la trace de "combien de fois" chaque article est présent. En Python, l'implémentation naturelle d'un multiset se présente sous la forme d'une clé de mappage dict permettant de compter, qui est elle-même plus facilement implémentée sous la forme collections.defaultdict (int) .

Quatrièmement, une approche générale devrait être threadsafe (quand cela peut être accompli facilement, au moins ;-). Heureusement, threading.local le rend trivial, le cas échéant - et ici, il devrait sûrement l'être (chaque pile ayant son propre thread d'appel distinct).

Cinquièmement, une question intéressante qui a été abordée dans certains commentaires (notant à quel point les décorateurs proposés jouent mal dans certaines réponses avec d’autres décorateurs: le décorateur de surveillance semble être le DERNIER (le plus extérieur), sinon les pauses de vérification. Cela vient du choix naturel mais malheureux d’utiliser l’objet fonction lui-même comme clé de dictée de surveillance.

Je propose de résoudre ce problème par un choix de clé différent: faire en sorte que le décorateur prenne un argument (chaîne, disons) identifier qui doit être unique (dans chaque fil donné) et utiliser l'identifiant comme clé dans le dict de surveillance. Le code vérifiant la pile doit bien sûr connaître l'identifiant et l'utiliser également.

Au moment de la décoration, le décorateur peut vérifier la propriété d’unicité (en utilisant un jeu séparé). L'identifiant peut être laissé avec le nom de fonction par défaut (il est donc explicitement nécessaire de conserver la flexibilité de surveillance des fonctions homonymes dans le même espace de noms); il est possible de renoncer explicitement à la propriété d’unicité lorsque plusieurs fonctions surveillées doivent être considérées comme "identiques". à des fins de contrôle (cela peut être le cas si une instruction def donnée doit être exécutée plusieurs fois dans des contextes légèrement différents pour créer plusieurs objets fonction que les programmeurs souhaitent considérer comme "la même fonction" surveillance). Enfin, il devrait être possible de revenir facultativement à "l'objet fonction en tant qu'identificateur". pour les rares cas dans lesquels il est connu que la décoration est encore impossible, ce peut être la façon la plus commode de garantir l'unicité).

Donc, en rassemblant ces nombreuses considérations, nous pourrions avoir (y compris une fonction utilitaire threadlocal_var qui sera probablement déjà dans un module de boîte à outils bien sûr ;-) quelque chose comme ce qui suit ...:

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

Je n'ai pas testé ce code, il peut donc contenir des fautes de frappe ou autres, mais je le propose car j'espère qu'il couvre tous les points techniques importants que j'ai expliqués ci-dessus.

Cela en vaut-il la peine? Probablement pas, comme expliqué précédemment. Cependant, je pense que "si cela en vaut la peine, alors il vaut la peine de le faire correctement"; -).

Answer 2

Je n'aime pas vraiment cette approche, mais voici une version corrigée de ce que vous faisiez:

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

Ceci gère la récursivité, les threads et les exceptions.

Je n'aime pas cette approche pour deux raisons:

• Cela ne fonctionne pas si la fonction est décorée davantage: ce doit être le décorateur final.
• Si vous l'utilisez pour le débogage, cela signifie que vous devez éditer le code à deux endroits pour l'utiliser. un pour ajouter le décorateur et un pour l'utiliser. Il est beaucoup plus pratique de simplement examiner la pile pour ne modifier que le code dans le code en cours de débogage.

Une meilleure approche consisterait à examiner directement la pile (éventuellement en tant qu’extension native pour la vitesse) et, si possible, à trouver un moyen de mettre en cache les résultats pour la durée de vie du cadre de la pile. (Je ne sais pas si cela est possible sans modifier le noyau Python, cependant.)