5.3 故障排查与优化

5.3 故障排查与优化 — 国产GPU适配指南运维实战

本节导读：通过系统化的故障排查方法和优化策略，解决国产GPU在实际应用中的各种问题，提升系统稳定性和性能表现，保障AI业务连续运行。

学习目标

• 掌握国产GPU常见故障的诊断方法
• 学会建立完整的故障排查流程
• 理解性能瓶颈的识别和优化策略
• 能够快速定位和解决兼容性问题
• 实现系统的长期稳定运行

核心概念

故障分类体系

国产GPU故障按照影响范围可以分为多个层次：

故障排查方法论

系统性排查流程：

1. 现象观察：记录故障现象、发生时间、影响范围
2. 层次分析：从应用层到硬件层逐层排查
3. 日志分析：检查系统日志、应用日志、GPU日志
4. 测试验证：通过最小化复现验证问题
5. 根因定位：找到根本原因而非表面现象
6. 解决方案：实施修复措施并验证效果

环境准备 / 前置知识

故障排查工具集

# 系统监控工具
psutil>=5.8.0                    # 系统资源监控
nvidia-ml-py>=11.0.0             # NVIDIA GPU监控
py-cpuinfo>=8.0.0                # CPU信息获取

# 日志分析工具
loguru>=0.6.0                    # 日志处理
pandas>=1.3.0                    # 日志数据分析

# 性能分析工具
cProfile                         # Python性能分析
line_profiler                    # 代码行级分析
memory_profiler                  # 内存使用分析

# GPU诊断工具
torch_npu.utils                 # 华为NPU工具
torch_mlu.utils                 # 寒武纪MLU工具
torch.cuda                      # CUDA诊断
​

系统监控配置

import psutil
import time
import json
from datetime import datetime
from typing import Dict, List, Any
import threading

class SystemHealthMonitor:
    """系统健康监控器"""
    
    def __init__(self, platform: str = 'npu'):
        self.platform = platform
        self.monitoring = False
        self.monitor_thread = None
        self.health_data = []
        
    def start_monitoring(self, interval: float = 5.0):
        """开始系统监控"""
        self.monitoring = True
        self.monitor_thread = threading.Thread(target=self._monitor_loop, args=(interval,))
        self.monitor_thread.start()
        print(f"开始监控 {self.platform} 系统健康状态")
    
    def stop_monitoring(self):
        """停止系统监控"""
        self.monitoring = False
        if self.monitor_thread:
            self.monitor_thread.join()
        print(f"停止监控 {self.platform} 系统健康状态")
    
    def _monitor_loop(self, interval: float):
        """监控循环"""
        while self.monitoring:
            health_data = self.collect_health_data()
            self.health_data.append(health_data)
            self._check_health_status(health_data)
            time.sleep(interval)
    
    def collect_health_data(self) -> Dict[str, Any]:
        """收集系统健康数据"""
        data = {
            'timestamp': datetime.now().isoformat(),
            'platform': self.platform,
            'cpu_usage': psutil.cpu_percent(),
            'memory_usage': psutil.virtual_memory().percent,
            'disk_usage': psutil.disk_usage('/').percent,
            'cpu_temp': self._get_cpu_temperature(),
            'gpu_health': self._get_gpu_health(),
            'system_load': psutil.getloadavg(),
        }
        
        return data
    
    def _get_cpu_temperature(self) -> float:
        """获取CPU温度"""
        try:
            # 读取CPU温度文件
            with open('/sys/class/thermal/thermal_zone0/temp', 'r') as f:
                temp = float(f.read()) / 1000.0
                return temp
        except:
            return 0.0
    
    def _get_gpu_health(self) -> Dict[str, Any]:
        """获取GPU健康状态"""
        gpu_data = {
            'temperature': 0.0,
            'memory_usage': 0.0,
            'utilization': 0.0,
            'power_usage': 0.0,
            'errors': 0
        }
        
        try:
            if self.platform == 'cuda':
                import torch
                if torch.cuda.is_available():
                    gpu_data['memory_usage'] = torch.cuda.memory_allocated() / torch.cuda.max_memory_allocated()
                    gpu_data['temperature'] = self._get_gpu_temperature_cuda()
                    gpu_data['utilization'] = self._get_gpu_utilization_cuda()
            elif self.platform == 'npu':
                # 华为NPU状态检查
                import torch_npu
                if torch_npu.is_available():
                    gpu_data['memory_usage'] = self._get_npu_memory_usage()
                    gpu_data['temperature'] = self._get_npu_temperature()
            elif self.platform == 'mlu':
                # 寒武纪MLU状态检查
                import torch_mlu
                if torch_mlu.is_available():
                    gpu_data['memory_usage'] = self._get_mlu_memory_usage()
                    gpu_data['temperature'] = self._get_mlu_temperature()
        except Exception as e:
            gpu_data['errors'] = str(e)
        
        return gpu_data
    
    def _check_health_status(self, data: Dict[str, Any]):
        """检查健康状态并报警"""
        warnings = []
        
        # CPU使用率警告
        if data['cpu_usage'] > 90:
            warnings.append(f"CPU使用率过高: {data['cpu_usage']}%")
        
        # 内存使用率警告
        if data['memory_usage'] > 85:
            warnings.append(f"内存使用率过高: {data['memory_usage']}%")
        
        # GPU温度警告
        gpu_temp = data['gpu_health']['temperature']
        if gpu_temp > 85:
            warnings.append(f"GPU温度过高: {gpu_temp}°C")
        
        # GPU内存警告
        gpu_memory = data['gpu_health']['memory_usage']
        if gpu_memory > 0.9:
            warnings.append(f"GPU内存使用率过高: {gpu_memory*100}%")
        
        if warnings:
            print(f"⚠️  健康警告:")
            for warning in warnings:
                print(f"    {warning}")
    
    def get_health_report(self, hours: int = 24) -> Dict[str, Any]:
        """生成健康报告"""
        from datetime import datetime, timedelta
        
        cutoff_time = datetime.now() - timedelta(hours=hours)
        recent_data = [d for d in self.health_data if datetime.fromisoformat(d['timestamp']) > cutoff_time]
        
        if not recent_data:
            return {'error': '无健康数据'}
        
        report = {
            'period_hours': hours,
            'total_samples': len(recent_data),
            'cpu_avg': sum(d['cpu_usage'] for d in recent_data) / len(recent_data),
            'memory_avg': sum(d['memory_usage'] for d in recent_data) / len(recent_data),
            'gpu_temp_avg': sum(d['gpu_health']['temperature'] for d in recent_data) / len(recent_data),
            'gpu_mem_avg': sum(d['gpu_health']['memory_usage'] for d in recent_data) / len(recent_data),
            'peak_cpu': max(d['cpu_usage'] for d in recent_data),
            'peak_memory': max(d['memory_usage'] for d in recent_data),
            'peak_gpu_temp': max(d['gpu_health']['temperature'] for d in recent_data),
            'warnings_count': sum(1 for d in recent_data if d['gpu_health']['errors'] != 0)
        }
        
        return report
​

分步实战

步骤1：常见故障排查流程

硬件层故障排查

import subprocess
import re
from typing import List, Dict, Any

class HardwareDiagnostics:
    """硬件诊断工具"""
    
    def __init__(self, platform: str):
        self.platform = platform
    
    def check_gpu_health(self) -> Dict[str, Any]:
        """检查GPU硬件健康状态"""
        print(f"检查 {self.platform} GPU硬件状态...")
        
        health_report = {
            'platform': self.platform,
            'hardware_status': 'unknown',
            'temperature': 0.0,
            'memory_errors': 0,
            'fan_status': 'unknown',
            'power_status': 'unknown',
            'warnings': []
        }
        
        try:
            if self.platform == 'npu':
                health_report = self._check_npu_hardware()
            elif self.platform == 'mlu':
                health_report = self._check_mlu_hardware()
            elif self.platform == 'cuda':
                health_report = self._check_cuda_hardware()
            else:
                health_report['warnings'].append(f"不支持的平台: {self.platform}")
        
        except Exception as e:
            health_report['warnings'].append(f"硬件检查失败: {str(e)}")
        
        return health_report
    
    def _check_npu_hardware(self) -> Dict[str, Any]:
        """检查华为NPU硬件状态"""
        try:
            import torch_npu
            
            # 获取设备信息
            devices = torch_npu.list_devices()
            health_report = {
                'platform': 'npu',
                'hardware_status': 'healthy' if devices else 'no_devices',
                'device_count': len(devices),
                'device_info': devices,
                'warnings': []
            }
            
            # 检查内存状态
            if devices:
                for i, device in enumerate(devices):
                    try:
                        # 分配测试内存
                        test_tensor = torch.randn(1000, 1000, device=device)
                        del test_tensor
                        torch.npu.empty_cache()
                    except Exception as e:
                        health_report['warnings'].append(f"设备{i}内存测试失败: {str(e)}")
            
            return health_report
            
        except ImportError:
            return {
                'platform': 'npu',
                'hardware_status': 'driver_not_installed',
                'warnings': ['华为NPU驱动未安装']
            }
    
    def check_system_resources(self) -> Dict[str, Any]:
        """检查系统资源状态"""
        import psutil
        import datetime
        
        resource_report = {
            'timestamp': datetime.datetime.now().isoformat(),
            'cpu': {
                'usage_percent': psutil.cpu_percent(),
                'core_count': psutil.cpu_count(),
                'load_avg': psutil.getloadavg(),
                'temperature': self._get_cpu_temperature()
            },
            'memory': {
                'total': psutil.virtual_memory().total,
                'used': psutil.virtual_memory().used,
                'available': psutil.virtual_memory().available,
                'percent': psutil.virtual_memory().percent
            },
            'disk': {
                'total': psutil.disk_usage('/').total,
                'used': psutil.disk_usage('/').used,
                'free': psutil.disk_usage('/').free,
                'percent': psutil.disk_usage('/').percent
            },
            'network': {
                'bytes_sent': psutil.net_io_counters().bytes_sent,
                'bytes_recv': psutil.net_io_counters().bytes_recv
            }
        }
        
        return resource_report
    
    def _get_cpu_temperature(self) -> float:
        """获取CPU温度"""
        try:
            # Linux系统CPU温度读取
            temps = psutil.sensors_temperatures()
            if 'coretemp' in temps:
                return temps['coretemp'][0].current
            elif 'acpitz' in temps:
                return temps['acpitz'][0].current
        except:
            pass
        return 0.0

# 使用示例
def run_hardware_diagnostics(platform: str = 'npu'):
    """运行硬件诊断"""
    print(f"开始 {platform} 硬件诊断...")
    
    diagnostic = HardwareDiagnostics(platform)
    
    # 检查GPU硬件状态
    gpu_health = diagnostic.check_gpu_health()
    print("GPU健康状态:")
    print(json.dumps(gpu_health, indent=2, ensure_ascii=False))
    
    # 检查系统资源
    system_resources = diagnostic.check_system_resources()
    print("\n系统资源状态:")
    print(json.dumps(system_resources, indent=2, ensure_ascii=False))
    
    return gpu_health, system_resources
​

驱动层故障排查

class DriverDiagnostics:
    """驱动诊断工具"""
    
    def __init__(self, platform: str):
        self.platform = platform
    
    def check_driver_status(self) -> Dict[str, Any]:
        """检查驱动状态"""
        print(f"检查 {self.platform} 驱动状态...")
        
        driver_report = {
            'platform': self.platform,
            'driver_version': 'unknown',
            'driver_status': 'unknown',
            'installation_path': 'unknown',
            'compatibility_check': 'unknown',
            'warnings': []
        }
        
        try:
            if self.platform == 'npu':
                driver_report = self._check_npu_driver()
            elif self.platform == 'mlu':
                driver_report = self._check_mlu_driver()
            elif self.platform == 'cuda':
                driver_report = self._check_cuda_driver()
            else:
                driver_report['warnings'].append(f"不支持的平台: {self.platform}")
        
        except Exception as e:
            driver_report['warnings'].append(f"驱动检查失败: {str(e)}")
        
        return driver_report
    
    def _check_npu_driver(self) -> Dict[str, Any]:
        """检查华为NPU驱动"""
        try:
            import torch_npu
            
            # 检查驱动版本
            if hasattr(torch_npu, '__version__'):
                driver_version = torch_npu.__version__
            else:
                driver_version = "unknown"
            
            # 检查设备可用性
            devices = torch_npu.list_devices()
            driver_status = "healthy" if devices else "no_devices"
            
            if not devices:
                driver_report['warnings'].append("未检测到NPU设备")
            
            driver_report = {
                'platform': 'npu',
                'driver_version': driver_version,
                'driver_status': driver_status,
                'device_count': len(devices),
                'devices': devices,
                'warnings': driver_report['warnings']
            }
            
            return driver_report
            
        except ImportError:
            return {
                'platform': 'npu',
                'driver_version': 'not_installed',
                'driver_status': 'driver_not_installed',
                'warnings': ['华为NPU驱动未安装']
            }

# 使用示例
def run_driver_diagnostics(platform: str = 'npu'):
    """运行驱动诊断"""
    print(f"开始 {platform} 驱动诊断...")
    
    diagnostic = DriverDiagnostics(platform)
    
    # 检查驱动状态
    driver_status = diagnostic.check_driver_status()
    print("驱动状态:")
    print(json.dumps(driver_status, indent=2, ensure_ascii=False))
    
    return driver_status
​

步骤2：性能问题诊断与优化

性能瓶颈分析

import cProfile
import pstats
import time
from typing import Dict, List, Any

class PerformanceProfiler:
    """性能分析器"""
    
    def __init__(self, platform: str = 'npu'):
        self.platform = platform
        self.profiler_results = {}
    
    def profile_tensor_operations(self, operation_type: str = 'matrix_multiply') -> Dict[str, Any]:
        """分析张量操作性能"""
        print(f"分析张量操作: {operation_type}")
        
        if operation_type == 'matrix_multiply':
            return self._profile_matrix_multiply()
        else:
            return {'error': f'不支持的操作类型: {operation_type}'}
    
    def _profile_matrix_multiply(self) -> Dict[str, Any]:
        """分析矩阵乘法性能"""
        import torch
        
        # 测试不同矩阵大小
        sizes = [512, 1024, 2048]
        results = {}
        
        for size in sizes:
            print(f"测试矩阵大小: {size}x{size}")
            
            # 准备测试数据
            if self.platform == 'npu':
                import torch_npu
                device = torch.device("npu")
            elif self.platform == 'mlu':
                import torch_mlu
                device = torch.device("mlu")
            elif self.platform == 'cuda':
                device = torch.device("cuda")
            else:
                device = torch.device("cpu")
            
            matrix_a = torch.randn(size, size, dtype=torch.float32).to(device)
            matrix_b = torch.randn(size, size, dtype=torch.float32).to(device)
            
            # 预热
            for _ in range(3):
                _ = torch.mm(matrix_a, matrix_b)
            
            # 同步设备
            if self.platform == 'npu':
                import torch_npu
                torch.npu.synchronize()
            elif self.platform == 'mlu':
                import torch_mlu
                torch.mlu.synchronize()
            else:
                torch.cuda.synchronize()
            
            # 测试
            start_time = time.time()
            result = torch.mm(matrix_a, matrix_b)
            
            # 同步设备
            if self.platform == 'npu':
                import torch_npu
                torch.npu.synchronize()
            elif self.platform == 'mlu':
                import torch_mlu
                torch.mlu.synchronize()
            else:
                torch.cuda.synchronize()
            
            end_time = time.time()
            
            # 计算性能指标
            execution_time = end_time - start_time
            flops = 2 * size ** 3
            tflops = flops / execution_time / 1e12
            
            results[size] = {
                'execution_time': execution_time,
                'tflops': tflops,
                'memory_allocated': 0
            }
        
        return results

# 使用示例
def profile_performance(platform: str = 'npu'):
    """性能分析"""
    print(f"开始 {platform} 性能分析...")
    
    profiler = PerformanceProfiler(platform)
    
    # 分析矩阵乘法性能
    matrix_results = profiler.profile_tensor_operations('matrix_multiply')
    print("矩阵乘法性能分析结果:")
    print(json.dumps(matrix_results, indent=2, ensure_ascii=False))
    
    return matrix_results
​

常见问题 FAQ

Q1：国产GPU常见故障有哪些典型表现？

A：国产GPU常见故障表现为：

• 硬件故障：过热、内存错误、风扇异常、电源不足
• 驱动问题：设备识别失败、版本冲突、配置错误
• 软件兼容性：框架版本不匹配、编译失败、运行时错误
• 性能问题：推理延迟高、内存溢出、计算超时、性能退化

Q2：如何快速定位GPU故障的根本原因？

A：系统化定位流程：

1. 分层排查：从应用层→软件层→驱动层→硬件层逐层检查
2. 日志分析：检查系统日志、应用日志、GPU专用日志
3. 最小化复现：创建最小化测试用例复现问题
4. 对比测试：在不同平台和环境进行对比测试
5. 工具诊断：使用专业诊断工具获取详细状态信息

Q3：国产GPU性能优化有哪些有效策略？

A：有效的性能优化策略：

• 内存优化：合理使用缓存、及时释放内存、使用混合精度
• 计算优化：使用适当的批次大小、优化数据布局、利用并行计算
• 编译优化：启用编译器优化、使用自定义算子、调整编译参数
• 系统优化：调整系统参数、优化调度策略、改善散热条件

Q4：如何确保国产GPU系统的长期稳定运行？

A：确保长期稳定运行的方法：

• 健康监控：建立实时监控系统，及时发现异常
• 定期维护：定期更新驱动、优化系统、清理缓存
• 容错设计：实现故障检测和自动恢复机制
• 负载均衡：合理分配计算负载，避免单点过载
• 备份策略：重要数据和配置定期备份

最佳实践与避坑

故障排查最佳实践

• 建立标准流程：制定标准化的故障排查流程和检查清单
• 详细记录：详细记录故障现象、处理过程和解决结果
• 经验总结：定期总结经验教训，完善知识库
• 团队协作：建立技术协作机制，共享故障处理经验

性能优化建议

• 基准测试：建立性能基准，定期进行对比测试
• 持续优化：持续监控系统性能，不断优化改进
• 资源监控：建立完善的资源监控和告警机制
• 版本管理：合理管理软件版本，避免版本冲突

本节小结

本节详细讲解了国产GPU故障排查与优化的完整体系，涵盖了：

1. 故障分类体系：建立从硬件到应用层的完整故障分类框架
2. 排查方法论：提供系统化的故障排查流程和方法
3. 硬件诊断：硬件层故障检查和系统资源监控工具
4. 驱动诊断：驱动状态检查和兼容性验证工具
5. 性能分析：性能瓶颈分析和优化策略实施
6. 运维策略：确保系统长期稳定运行的实用建议

通过系统化的故障排查和优化策略，可以大幅提升国产GPU的稳定性和性能表现。

延伸阅读

• 官方文档：各GPU厂商故障诊断指南
• 相关章节：本教程 4.3 节性能调优实战
• 技术资料：国产GPU运维管理最佳实践

关键词：国产GPU适配指南,故障排查,性能优化,系统监控,运维实战
难度：进阶
预计阅读：50分钟