"""Core execution tracking for ComfyUI's internal methods""" import logging import time import json import os import threading from typing import Dict, List, Optional from collections import defaultdict import functools logger = logging.getLogger('ComfyUI-ExecutionTracker') logger.setLevel(logging.ERROR) class ExecutionTracker: _instance = None _lock = threading.Lock() ENABLED = False #this is a global variable that controls whether the execution tracker is enabled or not def __init__(self): self.data_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "data") os.makedirs(self.data_dir, exist_ok=True) self.trace_file = os.path.join(self.data_dir, "method_traces.json") # Load existing traces if any self.traces = self._load_traces() # Track current execution self.call_stack = [] self.current_execution = None def _load_traces(self) -> Dict: """Load existing traces from file""" if os.path.exists(self.trace_file): try: with open(self.trace_file, 'r') as f: return json.load(f) except Exception as e: logger.error(f"Failed to load traces: {e}") return { "executions": [], "method_stats": defaultdict(lambda: { "total_calls": 0, "total_time": 0, "min_time": float('inf'), "max_time": 0, "avg_time": 0 }) } def _save_traces(self): """Save traces to file""" try: with open(self.trace_file, 'w') as f: # Convert defaultdict to regular dict for JSON serialization traces_copy = { "executions": self.traces["executions"], "method_stats": { k: dict(v) for k, v in self.traces["method_stats"].items() } } json.dump(traces_copy, f, indent=2) except Exception as e: logger.error(f"Failed to save traces: {e}") @classmethod def get_instance(cls) -> 'ExecutionTracker': if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = cls() return cls._instance @classmethod def enable(cls): """Enable execution tracking""" cls.ENABLED = True logger.info("ComfyUI method call tracking enabled") @classmethod def disable(cls): """Disable execution tracking""" cls.ENABLED = False logger.info("ComfyUI method call tracking disabled") def start_execution(self, prompt_id: str): """Start tracking a new execution""" if not self.ENABLED: return with self._lock: self.current_execution = { "prompt_id": prompt_id, "start_time": time.time() * 1000, "method_calls": [], "total_time": 0 } def end_execution(self): """End tracking current execution""" if not self.ENABLED or not self.current_execution: return with self._lock: self.current_execution["end_time"] = time.time() * 1000 self.current_execution["total_time"] = ( self.current_execution["end_time"] - self.current_execution["start_time"] ) self.traces["executions"].append(self.current_execution) self._save_traces() self.current_execution = None def track_method_call(self, method_name: str, class_name: str = None): """Decorator to track method execution time""" def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): if not self.ENABLED: return func(*args, **kwargs) full_name = f"{class_name}.{method_name}" if class_name else method_name start_time = time.time() * 1000 try: with self._lock: self.call_stack.append(full_name) result = func(*args, **kwargs) return result finally: end_time = time.time() * 1000 duration = end_time - start_time with self._lock: # Pop from call stack if self.call_stack: self.call_stack.pop() # Update method stats stats = self.traces["method_stats"][full_name] stats["total_calls"] += 1 stats["total_time"] += duration stats["min_time"] = min(stats.get("min_time", float('inf')), duration) stats["max_time"] = max(stats.get("max_time", 0), duration) stats["avg_time"] = stats["total_time"] / stats["total_calls"] # Record call in current execution with enhanced context if self.current_execution: # Get queue size if available queue_size = None try: import execution if hasattr(execution, 'PromptServer') and hasattr(execution.PromptServer.instance, 'prompt_queue'): queue = execution.PromptServer.instance.prompt_queue.queue queue_size = len(queue) if queue else 0 except: pass # Determine if operation was cached is_cache_hit = False if 'caches' in kwargs and 'current_item' in kwargs: try: is_cache_hit = kwargs['caches'].outputs.get(kwargs['current_item']) is not None except: pass call_info = { "method": full_name, "start_time": start_time, "duration": duration, "stack_depth": len(self.call_stack) + 1, # +1 since we already popped "parent": self.call_stack[-1] if self.call_stack else None, "queue_size": queue_size, "is_cache_hit": is_cache_hit } self.current_execution["method_calls"].append(call_info) return wrapper return decorator def get_method_stats(self) -> Dict: """Get statistics for all tracked methods""" return dict(self.traces["method_stats"]) if self.ENABLED else {}