GPU 监控与调试

#!/usr/bin/env python3
"""
gpu_monitor.py - GPU 监控和告警脚本
用于生产环境的 GPU 监控，支持阈值告警
"""

import subprocess
import json
import time
import sys
from datetime import datetime
from typing import Dict, List, Optional
import argparse

class GPUMonitor:
    """GPU 监控类"""

    # 默认告警阈值
    DEFAULT_THRESHOLDS = {
        'temperature': {'warning': 75, 'critical': 85},
        'memory_percent': {'warning': 85, 'critical': 95},
        'power_percent': {'warning': 90, 'critical': 100},
        'ecc_errors': {'warning': 1, 'critical': 10},
    }

    def __init__(self, thresholds: Optional[Dict] = None):
        self.thresholds = thresholds or self.DEFAULT_THRESHOLDS

    def query_gpus(self) -> List[Dict]:
        """查询所有 GPU 状态"""
        query_fields = [
            'index',
            'name',
            'uuid',
            'driver_version',
            'memory.total',
            'memory.used',
            'memory.free',
            'temperature.gpu',
            'power.draw',
            'power.limit',
            'utilization.gpu',
            'utilization.memory',
            'pstate',
            'ecc.errors.uncorrected.volatile.total',
            'ecc.errors.corrected.volatile.total',
            'pcie.link.gen.current',
            'pcie.link.width.current',
        ]

        cmd = [
            'nvidia-smi',
            f'--query-gpu={",".join(query_fields)}',
            '--format=csv,noheader,nounits'
        ]

        try:
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        except subprocess.CalledProcessError as e:
            print(f"Error running nvidia-smi: {e}", file=sys.stderr)
            return []
        except FileNotFoundError:
            print("nvidia-smi not found", file=sys.stderr)
            return []

        gpus = []
        for line in result.stdout.strip().split('\n'):
            if not line:
                continue
            values = [v.strip() for v in line.split(',')]

            # 解析数值，处理 [N/A] 情况
            def parse_value(v, default=0):
                if v in ['[N/A]', '[Not Supported]', '']:
                    return default
                try:
                    return float(v)
                except ValueError:
                    return v

            gpu = {
                'index': int(values[0]),
                'name': values[1],
                'uuid': values[2],
                'driver_version': values[3],
                'memory_total': parse_value(values[4]),
                'memory_used': parse_value(values[5]),
                'memory_free': parse_value(values[6]),
                'temperature': parse_value(values[7]),
                'power_draw': parse_value(values[8]),
                'power_limit': parse_value(values[9]),
                'utilization_gpu': parse_value(values[10]),
                'utilization_memory': parse_value(values[11]),
                'pstate': values[12],
                'ecc_uncorrected': parse_value(values[13]),
                'ecc_corrected': parse_value(values[14]),
                'pcie_gen': parse_value(values[15]),
                'pcie_width': parse_value(values[16]),
            }

            # 计算百分比
            if gpu['memory_total'] > 0:
                gpu['memory_percent'] = (gpu['memory_used'] / gpu['memory_total']) * 100
            else:
                gpu['memory_percent'] = 0

            if gpu['power_limit'] > 0:
                gpu['power_percent'] = (gpu['power_draw'] / gpu['power_limit']) * 100
            else:
                gpu['power_percent'] = 0

            gpus.append(gpu)

        return gpus

    def check_alerts(self, gpu: Dict) -> List[Dict]:
        """检查告警条件"""
        alerts = []

        # 温度告警
        temp = gpu['temperature']
        if temp >= self.thresholds['temperature']['critical']:
            alerts.append({
                'level': 'CRITICAL',
                'metric': 'temperature',
                'message': f"GPU {gpu['index']} temperature critical: {temp}°C",
                'value': temp
            })
        elif temp >= self.thresholds['temperature']['warning']:
            alerts.append({
                'level': 'WARNING',
                'metric': 'temperature',
                'message': f"GPU {gpu['index']} temperature high: {temp}°C",
                'value': temp
            })

        # 显存告警
        mem_pct = gpu['memory_percent']
        if mem_pct >= self.thresholds['memory_percent']['critical']:
            alerts.append({
                'level': 'CRITICAL',
                'metric': 'memory',
                'message': f"GPU {gpu['index']} memory critical: {mem_pct:.1f}%",
                'value': mem_pct
            })
        elif mem_pct >= self.thresholds['memory_percent']['warning']:
            alerts.append({
                'level': 'WARNING',
                'metric': 'memory',
                'message': f"GPU {gpu['index']} memory high: {mem_pct:.1f}%",
                'value': mem_pct
            })

        # ECC 错误告警
        ecc = gpu['ecc_uncorrected']
        if ecc >= self.thresholds['ecc_errors']['critical']:
            alerts.append({
                'level': 'CRITICAL',
                'metric': 'ecc_errors',
                'message': f"GPU {gpu['index']} ECC uncorrected errors critical: {ecc}",
                'value': ecc
            })
        elif ecc >= self.thresholds['ecc_errors']['warning']:
            alerts.append({
                'level': 'WARNING',
                'metric': 'ecc_errors',
                'message': f"GPU {gpu['index']} ECC uncorrected errors: {ecc}",
                'value': ecc
            })

        return alerts

    def format_output(self, gpus: List[Dict], format: str = 'text') -> str:
        """格式化输出"""
        if format == 'json':
            output = {
                'timestamp': datetime.now().isoformat(),
                'gpus': gpus
            }
            return json.dumps(output, indent=2)

        # 文本格式
        lines = []
        lines.append(f"GPU Monitor - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        lines.append("=" * 80)

        for gpu in gpus:
            lines.append(f"\nGPU {gpu['index']}: {gpu['name']}")
            lines.append(f"  UUID: {gpu['uuid']}")
            lines.append(f"  Memory: {gpu['memory_used']:.0f} / {gpu['memory_total']:.0f} MiB ({gpu['memory_percent']:.1f}%)")
            lines.append(f"  Temperature: {gpu['temperature']:.0f}°C")
            lines.append(f"  Power: {gpu['power_draw']:.1f} / {gpu['power_limit']:.1f} W ({gpu['power_percent']:.1f}%)")
            lines.append(f"  GPU Util: {gpu['utilization_gpu']:.0f}%")
            lines.append(f"  Memory BW Util: {gpu['utilization_memory']:.0f}%")
            lines.append(f"  P-State: {gpu['pstate']}")
            lines.append(f"  PCIe: Gen{gpu['pcie_gen']:.0f} x{gpu['pcie_width']:.0f}")
            lines.append(f"  ECC Errors: {gpu['ecc_uncorrected']:.0f} uncorrected, {gpu['ecc_corrected']:.0f} corrected")

            # 检查告警
            alerts = self.check_alerts(gpu)
            if alerts:
                lines.append("  Alerts:")
                for alert in alerts:
                    lines.append(f"    [{alert['level']}] {alert['message']}")

        return '\n'.join(lines)

    def run_continuous(self, interval: int = 5, format: str = 'text'):
        """持续监控"""
        try:
            while True:
                gpus = self.query_gpus()
                output = self.format_output(gpus, format)

                # 清屏并输出
                print('\033[2J\033[H', end='')
                print(output)

                time.sleep(interval)
        except KeyboardInterrupt:
            print("\nMonitoring stopped.")


def main():
    parser = argparse.ArgumentParser(description='GPU Monitoring Tool')
    parser.add_argument('-f', '--format', choices=['text', 'json'], default='text',
                        help='Output format')
    parser.add_argument('-c', '--continuous', action='store_true',
                        help='Continuous monitoring mode')
    parser.add_argument('-i', '--interval', type=int, default=5,
                        help='Monitoring interval in seconds (for continuous mode)')
    parser.add_argument('--json', action='store_true',
                        help='Output in JSON format (shortcut for -f json)')

    args = parser.parse_args()

    format = 'json' if args.json else args.format
    monitor = GPUMonitor()

    if args.continuous:
        monitor.run_continuous(interval=args.interval, format=format)
    else:
        gpus = monitor.query_gpus()
        print(monitor.format_output(gpus, format))


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
dcgm_monitor.py - 使用 DCGM Python API 进行 GPU 监控
"""

import pydcgm
import dcgm_fields
import dcgm_structs
import time
from typing import List, Dict, Optional
import json

class DCGMMonitor:
    """DCGM 监控封装类"""

    def __init__(self, embedded: bool = True):
        """
        初始化 DCGM 连接

        Args:
            embedded: True 使用嵌入模式, False 连接到 nv-hostengine
        """
        if embedded:
            self.handle = pydcgm.DcgmHandle(
                opMode=dcgm_structs.DCGM_OPERATION_MODE_AUTO
            )
        else:
            self.handle = pydcgm.DcgmHandle(
                ipAddress='127.0.0.1',
                opMode=dcgm_structs.DCGM_OPERATION_MODE_AUTO
            )

        self.system = self.handle.GetSystem()
        self.group = None

    def discover_gpus(self) -> List[int]:
        """发现所有 GPU"""
        gpu_ids = self.system.discovery.GetAllSupportedGpuIds()
        return list(gpu_ids)

    def create_group(self, name: str, gpu_ids: Optional[List[int]] = None) -> pydcgm.DcgmGroup:
        """创建 GPU 组"""
        if gpu_ids is None:
            gpu_ids = self.discover_gpus()

        self.group = self.system.GetGroupWithGpuIds(name, gpu_ids)
        return self.group

    def get_gpu_info(self, gpu_id: int) -> Dict:
        """获取 GPU 基本信息"""
        attributes = self.system.discovery.GetGpuAttributes(gpu_id)
        return {
            'gpu_id': gpu_id,
            'name': attributes.identifiers.deviceName,
            'uuid': attributes.identifiers.uuid,
            'pci_bus_id': attributes.identifiers.pciBusId,
            'serial': attributes.identifiers.serial,
            'driver_version': attributes.identifiers.driverVersion,
            'memory_total': attributes.memoryInfo.memorySizeGb,
        }

    def get_gpu_metrics(self, gpu_id: int) -> Dict:
        """获取 GPU 实时指标"""
        field_ids = [
            dcgm_fields.DCGM_FI_DEV_GPU_TEMP,
            dcgm_fields.DCGM_FI_DEV_POWER_USAGE,
            dcgm_fields.DCGM_FI_DEV_GPU_UTIL,
            dcgm_fields.DCGM_FI_DEV_MEM_COPY_UTIL,
            dcgm_fields.DCGM_FI_DEV_FB_FREE,
            dcgm_fields.DCGM_FI_DEV_FB_USED,
            dcgm_fields.DCGM_FI_DEV_SM_CLOCK,
            dcgm_fields.DCGM_FI_DEV_MEM_CLOCK,
            dcgm_fields.DCGM_FI_DEV_ECC_SBE_VOL_TOTAL,
            dcgm_fields.DCGM_FI_DEV_ECC_DBE_VOL_TOTAL,
        ]

        field_group = pydcgm.DcgmFieldGroup(
            self.handle,
            name="metrics_group",
            fieldIds=field_ids
        )

        # 创建临时组用于查询
        temp_group = self.system.GetGroupWithGpuIds("temp", [gpu_id])

        # 启用监控
        temp_group.samples.WatchFields(
            field_group,
            updateFreq=1000000,  # 1秒
            maxKeepAge=10.0,
            maxKeepSamples=10
        )

        time.sleep(1)  # 等待数据收集

        # 获取数据
        data = temp_group.samples.GetLatest(field_group)

        metrics = {}
        for gpu_data in data.values:
            for field_data in gpu_data:
                field_id = field_data.fieldId
                value = field_data.value

                field_name = dcgm_fields.DcgmFieldGetById(field_id).tag
                metrics[field_name] = value

        # 清理
        field_group.Delete()
        temp_group.Delete()

        return metrics

    def run_health_check(self, gpu_ids: Optional[List[int]] = None) -> Dict:
        """运行健康检查"""
        if self.group is None:
            self.create_group("health_check_group", gpu_ids)

        health = self.group.health

        # 设置健康监控
        health.Set(dcgm_structs.DCGM_HEALTH_WATCH_ALL)

        # 获取健康状态
        result = health.Check()

        return {
            'overall_health': result.overallHealth,
            'incidents': [
                {
                    'gpu_id': incident.entityInfo.entityId,
                    'health': incident.health,
                    'error_string': incident.error.msg,
                }
                for incident in result.incidents
            ]
        }

    def run_diagnostic(self, level: int = 1, gpu_ids: Optional[List[int]] = None) -> Dict:
        """
        运行诊断测试

        Args:
            level: 诊断级别 1-3
        """
        if self.group is None:
            self.create_group("diag_group", gpu_ids)

        diag = self.group.action.RunDiagnostic(
            dcgm_structs.DCGM_DIAG_LVL_SHORT if level == 1 else
            dcgm_structs.DCGM_DIAG_LVL_MED if level == 2 else
            dcgm_structs.DCGM_DIAG_LVL_LONG
        )

        results = []
        for test_result in diag.perGpuResponses:
            results.append({
                'gpu_id': test_result.gpuId,
                'status': test_result.status,
                'info': test_result.info,
                'warning': test_result.warning,
            })

        return {
            'level': level,
            'version': diag.version,
            'results': results
        }

    def close(self):
        """关闭 DCGM 连接"""
        if self.group:
            self.group.Delete()
        self.handle.Shutdown()


def main():
    monitor = DCGMMonitor(embedded=True)

    try:
        # 发现 GPU
        gpu_ids = monitor.discover_gpus()
        print(f"Found {len(gpu_ids)} GPU(s)")

        # 获取每个 GPU 信息
        for gpu_id in gpu_ids:
            info = monitor.get_gpu_info(gpu_id)
            print(f"\nGPU {gpu_id}:")
            print(json.dumps(info, indent=2))

            metrics = monitor.get_gpu_metrics(gpu_id)
            print("Metrics:")
            print(json.dumps(metrics, indent=2))

        # 运行健康检查
        print("\n=== Health Check ===")
        health = monitor.run_health_check()
        print(json.dumps(health, indent=2))

        # 运行快速诊断
        print("\n=== Diagnostic (Level 1) ===")
        diag = monitor.run_diagnostic(level=1)
        print(json.dumps(diag, indent=2))

    finally:
        monitor.close()


if __name__ == '__main__':
    main()

# 使用 NVTX 标注代码区域
import torch
import nvtx

class TrainingLoop:
    def __init__(self, model, optimizer):
        self.model = model
        self.optimizer = optimizer

    def train_epoch(self, dataloader):
        for batch_idx, (data, target) in enumerate(dataloader):
            # 标注数据传输
            with nvtx.annotate("Data to GPU", color="blue"):
                data = data.cuda()
                target = target.cuda()

            # 标注前向传播
            with nvtx.annotate("Forward", color="green"):
                output = self.model(data)
                loss = F.cross_entropy(output, target)

            # 标注反向传播
            with nvtx.annotate("Backward", color="red"):
                self.optimizer.zero_grad()
                loss.backward()

            # 标注优化器更新
            with nvtx.annotate("Optimizer Step", color="yellow"):
                self.optimizer.step()

# PyTorch 内置的 profiler 也可以导出 nsys 格式
with torch.profiler.profile(
    activities=[
        torch.profiler.ProfilerActivity.CPU,
        torch.profiler.ProfilerActivity.CUDA,
    ],
    schedule=torch.profiler.schedule(wait=1, warmup=1, active=3),
    on_trace_ready=torch.profiler.tensorboard_trace_handler('./log'),
    record_shapes=True,
    with_stack=True
) as prof:
    for step in range(5):
        train_step()
        prof.step()

#!/usr/bin/env python3
"""
pytorch_profiler.py - PyTorch GPU 性能分析
"""

import torch
import torch.nn as nn
import torch.profiler
from torch.profiler import profile, record_function, ProfilerActivity
import time

def create_model():
    """创建示例模型"""
    return nn.Sequential(
        nn.Linear(1024, 4096),
        nn.ReLU(),
        nn.Linear(4096, 4096),
        nn.ReLU(),
        nn.Linear(4096, 1024),
    ).cuda()

def profile_basic():
    """基础性能分析"""
    model = create_model()
    x = torch.randn(32, 1024).cuda()

    # 预热
    for _ in range(10):
        model(x)
    torch.cuda.synchronize()

    # 性能分析
    with profile(
        activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
        record_shapes=True,
        profile_memory=True,
        with_stack=True,
    ) as prof:
        with record_function("model_inference"):
            output = model(x)

    # 打印结果
    print(prof.key_averages().table(
        sort_by="cuda_time_total",
        row_limit=20
    ))

    # 导出 Chrome trace
    prof.export_chrome_trace("trace.json")

    # 导出 stacks (火焰图)
    prof.export_stacks("stacks.txt", "self_cuda_time_total")

def profile_training_loop():
    """分析训练循环"""
    model = create_model()
    optimizer = torch.optim.Adam(model.parameters())
    criterion = nn.MSELoss()

    x = torch.randn(32, 1024).cuda()
    y = torch.randn(32, 1024).cuda()

    # 使用 schedule 控制采集
    with profile(
        activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
        schedule=torch.profiler.schedule(
            wait=1,      # 等待 1 步
            warmup=1,    # 预热 1 步
            active=3,    # 采集 3 步
            repeat=2     # 重复 2 次
        ),
        on_trace_ready=torch.profiler.tensorboard_trace_handler('./log/profiler'),
        record_shapes=True,
        profile_memory=True,
        with_stack=True,
    ) as prof:
        for step in range(10):
            with record_function("forward"):
                output = model(x)
                loss = criterion(output, y)

            with record_function("backward"):
                optimizer.zero_grad()
                loss.backward()

            with record_function("optimizer"):
                optimizer.step()

            prof.step()  # 标记步结束

def profile_memory():
    """内存分析"""
    model = create_model()

    # 重置内存统计
    torch.cuda.reset_peak_memory_stats()
    torch.cuda.empty_cache()

    print(f"Initial memory: {torch.cuda.memory_allocated() / 1024**2:.2f} MB")

    with profile(
        activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
        profile_memory=True,
        record_shapes=True,
    ) as prof:
        x = torch.randn(64, 1024).cuda()
        print(f"After input: {torch.cuda.memory_allocated() / 1024**2:.2f} MB")

        output = model(x)
        print(f"After forward: {torch.cuda.memory_allocated() / 1024**2:.2f} MB")

        loss = output.sum()
        loss.backward()
        print(f"After backward: {torch.cuda.memory_allocated() / 1024**2:.2f} MB")

    # 打印内存事件
    print("\n=== Memory Events ===")
    print(prof.key_averages().table(
        sort_by="self_cuda_memory_usage",
        row_limit=10
    ))

    print(f"\nPeak memory: {torch.cuda.max_memory_allocated() / 1024**2:.2f} MB")

def profile_custom_kernels():
    """分析自定义操作"""
    model = create_model()
    x = torch.randn(32, 1024).cuda()

    with profile(
        activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
        with_stack=True,
    ) as prof:
        # 标注不同阶段
        with record_function("stage_1_preprocess"):
            x_norm = (x - x.mean()) / x.std()

        with record_function("stage_2_inference"):
            output = model(x_norm)

        with record_function("stage_3_postprocess"):
            result = torch.softmax(output, dim=-1)

    # 打印按阶段分组的结果
    print(prof.key_averages(group_by_input_shape=True).table(
        sort_by="cuda_time_total",
        row_limit=20
    ))

class GPUMemoryTracker:
    """GPU 内存追踪器"""

    def __init__(self, device=0):
        self.device = device
        self.history = []

    def snapshot(self, tag: str = ""):
        """记录当前内存状态"""
        self.history.append({
            'tag': tag,
            'allocated': torch.cuda.memory_allocated(self.device),
            'reserved': torch.cuda.memory_reserved(self.device),
            'max_allocated': torch.cuda.max_memory_allocated(self.device),
            'timestamp': time.time()
        })

    def report(self):
        """生成报告"""
        print("\n=== GPU Memory Report ===")
        print(f"{'Tag':<30} {'Allocated':<15} {'Reserved':<15} {'Max Alloc':<15}")
        print("-" * 75)

        for h in self.history:
            print(f"{h['tag']:<30} "
                  f"{h['allocated']/1024**2:>12.2f} MB "
                  f"{h['reserved']/1024**2:>12.2f} MB "
                  f"{h['max_allocated']/1024**2:>12.2f} MB")

    def reset(self):
        """重置统计"""
        self.history = []
        torch.cuda.reset_peak_memory_stats(self.device)
        torch.cuda.empty_cache()


def analyze_tensor_core_usage():
    """分析 Tensor Core 使用情况"""
    # Tensor Core 需要特定的数据类型和维度
    # FP16 或 BF16, 矩阵维度是 8 的倍数

    # 不使用 Tensor Core
    a = torch.randn(1000, 1000).cuda()
    b = torch.randn(1000, 1000).cuda()

    with profile(activities=[ProfilerActivity.CUDA]) as prof:
        with record_function("FP32_matmul"):
            c = torch.mm(a, b)

    print("FP32 (No Tensor Core):")
    print(prof.key_averages().table(row_limit=5))

    # 使用 Tensor Core (FP16, 维度是 8 的倍数)
    a_fp16 = torch.randn(1024, 1024, dtype=torch.float16).cuda()
    b_fp16 = torch.randn(1024, 1024, dtype=torch.float16).cuda()

    with profile(activities=[ProfilerActivity.CUDA]) as prof:
        with record_function("FP16_matmul"):
            c_fp16 = torch.mm(a_fp16, b_fp16)

    print("\nFP16 (With Tensor Core):")
    print(prof.key_averages().table(row_limit=5))


if __name__ == '__main__':
    print("=== Basic Profiling ===")
    profile_basic()

    print("\n=== Training Loop Profiling ===")
    profile_training_loop()

    print("\n=== Memory Profiling ===")
    profile_memory()

    print("\n=== Custom Kernels Profiling ===")
    profile_custom_kernels()

    print("\n=== Tensor Core Analysis ===")
    analyze_tensor_core_usage()