multi_view/DiffSynth-Studio-main/diffsynth/trainers/timer.py

import time
import torch

# 模块级的私有字典，用于存储所有计时器实例
_TIMERS = {}

class Timer:
    """
    一个用于测量代码块执行时间的计时器。
    它能自动选择使用 CUDA Events（用于精确 GPU 计时）或标准 time 模块（用于 CPU）。
    """
    def __init__(self, name: str, use_gpu: bool = True):
        self.name = name
        self._use_gpu = use_gpu and torch.cuda.is_available()
        self.reset()

        if self._use_gpu:
            self.start_event = torch.cuda.Event(enable_timing=True)
            self.end_event = torch.cuda.Event(enable_timing=True)

    def reset(self):
        """重置计时器状态。"""
        self.total_time_ms = 0.0
        self.count = 0

    def start(self):
        """标记计时的开始。"""
        if self._use_gpu:
            # 在 CUDA 流中记录一个事件
            self.start_event.record()
        else:
            self.start_cpu_time = time.perf_counter()

    def stop(self) -> float:
        """
        标记计时的结束，并返回本次计时的耗时（毫秒）。
        这个操作会强制同步 CUDA 流以获得准确时间。
        """
        if self._use_gpu:
            self.end_event.record()
            # 等待直到 end_event 之前的所有 CUDA 核心任务完成
            torch.cuda.synchronize()
            # 计算两个事件之间的毫秒数
            elapsed_ms = self.start_event.elapsed_time(self.end_event)
        else:
            end_cpu_time = time.perf_counter()
            elapsed_ms = (end_cpu_time - self.start_cpu_time) * 1000.0
        
        self.total_time_ms += elapsed_ms
        self.count += 1
        return elapsed_ms

    def average_ms(self) -> float:
        """返回平均耗时（毫秒）。"""
        return self.total_time_ms / self.count if self.count > 0 else 0.0

class _Timers:
    """
    一个管理所有 Timer 实例的单例访问类。
    通过 `__getattr__` 魔法方法，可以在首次访问时动态创建计时器。
    例如 `timers.step_time` 会自动创建或返回名为 'step_time' 的计时器。
    """
    def __init__(self):
        self._use_gpu = torch.cuda.is_available()

    def __getattr__(self, name: str) -> Timer:
        if name not in _TIMERS:
            _TIMERS[name] = Timer(name, use_gpu=self._use_gpu)
        return _TIMERS[name]
    
    def __call__(self, name: str) -> Timer:
        """允许使用函数调用语法获取计时器，如 timers('step_time')。"""
        return self.__getattr__(name)

    def reset(self):
        """重置所有已创建的计时器。"""
        for timer in _TIMERS.values():
            timer.reset()

# 创建一个全局单例
_global_timers_instance = _Timers()

def get_timers() -> _Timers:
    """
    获取全局计时器管理器实例。
    这是从任何文件访问计时器的推荐方式。
    """
    return _global_timers_instance