5.3-混合并行策略与性能优化(2)

5.3 混合并行策略与性能优化(2)

本节导读：深入学习混合并行策略的负载均衡和故障容错技术，掌握系统稳定性和可靠性优化方法。

学习目标

• 掌握动态负载均衡技术
• 理解故障容错机制实现
• 学会系统稳定性评估方法
• 具备高可用系统设计能力

负载均衡优化

动态负载分配

import torch
import torch.distributed as dist
from typing import List, Dict, Tuple, Optional
import math
import time

class DynamicLoadBalancer:
    """动态负载均衡器"""
    
    def __init__(self, config: HybridParallelConfig):
        self.config = config
        self.device_loads = {}
        self.load_history = []
        self.adaptation_strategy = 'adaptive'
    
    def update_device_load(self, device_id: int, load_info: Dict):
        """更新设备负载信息"""
        self.device_loads[device_id] = load_info
        
        # 记录负载历史
        self.load_history.append({
            'timestamp': time.time(),
            'device_id': device_id,
            'load': load_info
        })
        
        # 保持历史记录在合理范围内
        if len(self.load_history) > 1000:
            self.load_history = self.load_history[-500:]
    
    def calculate_load_scores(self) -> Dict[int, float]:
        """计算设备负载评分"""
        load_scores = {}
        
        for device_id, load_info in self.device_loads.items():
            # 综合考虑多个因素计算负载评分
            memory_score = self._calculate_memory_score(load_info)
            compute_score = self._calculate_compute_score(load_info)
            network_score = self._calculate_network_score(load_info)
            
            # 加权综合评分
            load_scores[device_id] = (
                0.4 * memory_score +
                0.4 * compute_score +
                0.2 * network_score
            )
        
        return load_scores
    
    def _calculate_memory_score(self, load_info: Dict) -> float:
        """计算内存负载评分"""
        memory_usage = load_info.get('memory_usage', 0)
        memory_total = load_info.get('memory_total', 1)
        
        # 归一化到0-1范围，越高表示负载越重
        return min(memory_usage / memory_total, 1.0)
    
    def _calculate_compute_score(self, load_info: Dict) -> float:
        """计算计算负载评分"""
        compute_util = load_info.get('compute_utilization', 0)
        return min(compute_util / 100.0, 1.0)
    
    def _calculate_network_score(self, load_info: Dict) -> float:
        """计算网络负载评分"""
        network_bw = load_info.get('network_bandwidth', 0)
        network_total = load_info.get('network_total_bandwidth', 1)
        
        return min(network_bw / network_total, 1.0)
    
    def distribute_workload(self, total_workload: int) -> Dict[int, int]:
        """分配工作负载"""
        load_scores = self.calculate_load_scores()
        
        # 根据负载评分分配工作负载
        distribution = {}
        total_score = sum(load_scores.values())
        
        if total_score == 0:
            # 平均分配
            devices = list(self.device_loads.keys())
            workload_per_device = total_workload // len(devices)
            for device_id in devices:
                distribution[device_id] = workload_per_device
        else:
            # 按负载比例反比分配（负载轻的设备分配更多工作）
            for device_id, score in load_scores.items():
                # 负载评分越低，分配的工作越多
                allocation_ratio = (1.0 - score) / (len(load_scores) - total_score)
                distribution[device_id] = int(total_workload * allocation_ratio)
        
        return distribution
    
    def adaptive_load_balancing(self, model: HybridParallelModel, input_data: torch.Tensor) -> torch.Tensor:
        """自适应负载均衡"""
        # 获取当前设备负载
        current_loads = self._get_current_device_loads(model)
        
        # 更新负载信息
        for device_id, load in current_loads.items():
            self.update_device_load(device_id, load)
        
        # 分配工作负载
        workload_distribution = self.distribute_workload(input_data.size(0))
        
        # 根据分配调整数据流
        balanced_input = self._redistribute_data(input_data, workload_distribution)
        
        return balanced_input
    
    def _get_current_device_loads(self, model: HybridParallelModel) -> Dict[int, Dict]:
        """获取当前设备负载"""
        loads = {}
        
        for device_id in range(self.config.tensor_parallel_size * self.config.pipeline_parallel_size):
            device = torch.device(f'cuda:{device_id}')
            
            # 获取设备信息
            load_info = {
                'memory_usage': torch.cuda.memory_allocated(device),
                'memory_total': torch.cuda.get_device_properties(device).total_memory,
                'compute_utilization': self._get_compute_utilization(device),
                'network_bandwidth': self._get_network_bandwidth(device),
                'temperature': self._get_device_temperature(device),
                'power_usage': self._get_device_power(device)
            }
            
            loads[device_id] = load_info
        
        return loads
    
    def _get_compute_utilization(self, device: torch.device) -> float:
        """获取设备利用率"""
        # 简化处理，实际应该从监控接口获取
        return 50.0 + torch.rand(1).item() * 30
    
    def _get_network_bandwidth(self, device: torch.device) -> float:
        """获取网络带宽使用"""
        # 简化处理
        return 1000 + torch.rand(1).item() * 500
    
    def _get_device_temperature(self, device: torch.device) -> float:
        """获取设备温度"""
        # 简化处理
        return 60 + torch.rand(1).item() * 20
    
    def _get_device_power(self, device: torch.device) -> float:
        """获取设备功耗"""
        # 简化处理
        return 200 + torch.rand(1).item() * 100
    
    def _redistribute_data(self, input_data: torch.Tensor, distribution: Dict[int, int]) -> torch.Tensor:
        """重新分配数据"""
        # 根据负载分配重新组织数据
        # 这里简化处理，实际需要根据具体的分配策略实现
        return input_data
​

智能调度算法

class IntelligentScheduler:
    """智能调度器"""
    
    def __init__(self, config: HybridParallelConfig):
        self.config = config
        self.scheduling_history = []
        self.scheduling_strategies = {
            'load_balanced': self._load_balanced_scheduling,
            'performance_optimized': self._performance_optimized_scheduling,
            'memory_aware': self._memory_aware_scheduling,
            'hybrid': self._hybrid_scheduling
        }
        self.current_strategy = 'hybrid'
    
    def schedule_workload(self, model: HybridParallelModel, requests: List[Dict]) -> Dict[int, List[Dict]]:
        """调度工作负载"""
        # 根据系统状态选择调度策略
        strategy = self._select_scheduling_strategy(model)
        self.current_strategy = strategy
        
        # 执行调度
        schedule = self.scheduling_strategies[strategy](model, requests)
        
        # 记录调度历史
        self._record_scheduling(schedule, strategy)
        
        return schedule
    
    def _select_scheduling_strategy(self, model: HybridParallelModel) -> str:
        """选择调度策略"""
        # 分析系统状态
        system_state = self._analyze_system_state(model)
        
        # 根据系统状态选择最佳策略
        if system_state['memory_pressure'] > 0.8:
            return 'memory_aware'
        elif system_state['performance_pressure'] > 0.8:
            return 'performance_optimized'
        elif system_state['load_imbalance'] > 0.5:
            return 'load_balanced'
        else:
            return 'hybrid'
    
    def _analyze_system_state(self, model: HybridParallelModel) -> Dict:
        """分析系统状态"""
        system_state = {
            'memory_pressure': 0.0,
            'performance_pressure': 0.0,
            'load_imbalance': 0.0
        }
        
        # 分析内存压力
        total_memory = 0
        used_memory = 0
        for device_id in range(self.config.tensor_parallel_size * self.config.pipeline_parallel_size):
            device = torch.device(f'cuda:{device_id}')
            total_memory += torch.cuda.get_device_properties(device).total_memory
            used_memory += torch.cuda.memory_allocated(device)
        
        system_state['memory_pressure'] = used_memory / total_memory if total_memory > 0 else 0
        
        # 分析性能压力
        system_state['performance_pressure'] = 0.6 + torch.rand(1).item() * 0.3
        
        # 分析负载不均衡
        load_balancer = DynamicLoadBalancer(self.config)
        current_loads = load_balancer._get_current_device_loads(model)
        load_scores = load_balancer.calculate_load_scores()
        
        if load_scores:
            max_score = max(load_scores.values())
            min_score = min(load_scores.values())
            system_state['load_imbalance'] = (max_score - min_score) / (max_score + min_score + 1e-6)
        
        return system_state
    
    def _load_balanced_scheduling(self, model: HybridParallelModel, requests: List[Dict]) -> Dict[int, List[Dict]]:
        """负载均衡调度"""
        load_balancer = DynamicLoadBalancer(self.config)
        
        # 计算设备负载
        current_loads = load_balancer._get_current_device_loads(model)
        load_scores = load_balancer.calculate_load_scores()
        
        # 按负载分配请求
        schedule = {}
        devices = list(load_scores.keys())
        
        for device_id in devices:
            schedule[device_id] = []
        
        # 负载评分越低的设备分配更多请求
        sorted_devices = sorted(devices, key=lambda x: load_scores[x])
        
        for i, request in enumerate(requests):
            device_id = sorted_devices[i % len(sorted_devices)]
            schedule[device_id].append(request)
        
        return schedule
    
    def _performance_optimized_scheduling(self, model: HybridParallelModel, requests: List[Dict]) -> Dict[int, List[Dict]]:
        """性能优化调度"""
        # 根据请求类型和性能特征进行调度
        schedule = {}
        
        for device_id in range(self.config.tensor_parallel_size * self.config.pipeline_parallel_size):
            schedule[device_id] = []
        
        # 分类请求
        computation_intensive = []
        memory_intensive = []
        io_intensive = []
        
        for request in requests:
            if request.get('type') == 'computation':
                computation_intensive.append(request)
            elif request.get('type') == 'memory':
                memory_intensive.append(request)
            else:
                io_intensive.append(request)
        
        # 将计算密集型请求分配到性能好的GPU
        devices = list(range(self.config.tensor_parallel_size * self.config.pipeline_parallel_size))
        devices.sort(key=lambda x: self._get_device_performance_score(x))
        
        # 分配计算密集型请求
        for i, request in enumerate(computation_intensive):
            device_id = devices[i % len(devices)]
            schedule[device_id].append(request)
        
        # 分配内存密集型请求
        for i, request in enumerate(memory_intensive):
            device_id = devices[len(devices) - 1 - (i % len(devices))]  # 反向分配
            schedule[device_id].append(request)
        
        # 分配IO密集型请求
        for i, request in enumerate(io_intensive):
            device_id = devices[i % len(devices)]
            schedule[device_id].append(request)
        
        return schedule
    
    def _memory_aware_scheduling(self, model: HybridParallelModel, requests: List[Dict]) -> Dict[int, List[Dict]]:
        """内存感知调度"""
        load_balancer = DynamicLoadBalancer(self.config)
        current_loads = load_balancer._get_current_device_loads(model)
        
        schedule = {}
        
        for device_id in range(self.config.tensor_parallel_size * self.config.pipeline_parallel_size):
            schedule[device_id] = []
        
        # 根据内存压力分配请求
        sorted_devices = sorted(
            range(self.config.tensor_parallel_size * self.config.pipeline_parallel_size),
            key=lambda x: current_loads[x]['memory_usage'] / current_loads[x]['memory_total']
        )
        
        for i, request in enumerate(requests):
            device_id = sorted_devices[i % len(sorted_devices)]
            schedule[device_id].append(request)
        
        return schedule
    
    def _hybrid_scheduling(self, model: HybridParallelModel, requests: List[Dict]) -> Dict[int, List[Dict]]:
        """混合调度策略"""
        # 结合多种策略的优势
        load_balanced_schedule = self._load_balanced_scheduling(model, requests)
        performance_schedule = self._performance_optimized_scheduling(model, requests)
        
        # 混合调度结果
        hybrid_schedule = {}
        
        for device_id in range(self.config.tensor_parallel_size * self.config.pipeline_parallel_size):
            hybrid_schedule[device_id] = []
            
            # 从两种调度策略中按比例选择
            lb_ratio = 0.6
            perf_ratio = 0.4
            
            # 从负载均衡调度中取60%的请求
            lb_requests = load_balanced_schedule[device_id]
            hybrid_requests = lb_requests[:int(len(lb_requests) * lb_ratio)]
            
            # 从性能优化调度中取40%的请求
            perf_requests = performance_schedule[device_id]
            hybrid_requests.extend(perf_requests[:int(len(perf_requests) * perf_ratio)])
            
            hybrid_schedule[device_id] = hybrid_requests
        
        return hybrid_schedule
    
    def _get_device_performance_score(self, device_id: int) -> float:
        """获取设备性能评分"""
        device = torch.device(f'cuda:{device_id}')
        
        # 综合考虑性能因素
        compute_score = self._get_device_compute_score(device)
        memory_score = self._get_device_memory_score(device)
        network_score = self._get_device_network_score(device)
        
        return 0.5 * compute_score + 0.3 * memory_score + 0.2 * network_score
    
    def _get_device_compute_score(self, device: torch.device) -> float:
        """获取设备计算性能评分"""
        # 简化处理，实际应该根据设备具体性能指标
        return 0.7 + torch.rand(1).item() * 0.3
    
    def _get_device_memory_score(self, device: torch.device) -> float:
        """获取设备内存性能评分"""
        try:
            props = torch.cuda.get_device_properties(device)
            memory_gb = props.total_memory / (1024**3)
            return min(memory_gb / 80.0, 1.0)  # 80GB为满分
        except:
            return 0.5
    
    def _get_device_network_score(self, device: torch.device) -> float:
        """获取设备网络性能评分"""
        # 简化处理
        return 0.6 + torch.rand(1).item() * 0.4
    
    def _record_scheduling(self, schedule: Dict[int, List[Dict]], strategy: str):
        """记录调度历史"""
        self.scheduling_history.append({
            'timestamp': time.time(),
            'strategy': strategy,
            'schedule': schedule,
            'total_requests': sum(len(requests) for requests in schedule.values())
        })
        
        # 保持历史记录在合理范围内
        if len(self.scheduling_history) > 100:
            self.scheduling_history = self.scheduling_history[-50:]
    
    def get_scheduling_statistics(self) -> Dict:
        """获取调度统计信息"""
        if not self.scheduling_history:
            return {'error': 'no_scheduling_history'}
        
        # 计算统计信息
        total_schedules = len(self.scheduling_history)
        strategy_usage = {}
        
        for history in self.scheduling_history:
            strategy = history['strategy']
            strategy_usage[strategy] = strategy_usage.get(strategy, 0) + 1
        
        avg_requests_per_schedule = sum(
            h['total_requests'] for h in self.scheduling_history
        ) / total_schedules
        
        return {
            'total_schedules': total_schedules,
            'strategy_distribution': strategy_usage,
            'avg_requests_per_schedule': avg_requests_per_schedule,
            'recent_strategies': [h['strategy'] for h in self.scheduling_history[-10:]]
        }
​