monitor['temperature'] = 60 + torch.rand(1).item() * 20 # 模拟温度
monitor['power_usage'] = 200 + torch.rand(1).item() * 100 # 模拟功耗

        except Exception as e:
            print(f"Error monitoring GPU {device}: {e}")

def schedule_microbatches(self, total_batch_size: int) -> Dict[int, List[int]]:
    """为每个GPU调度微批次"""
    microbatch_schedules = {}
    
    for i, (device, monitor) in enumerate(self.gpu_monitors.items()):
        self.update_gpu_status()
        
        # 调度微批次
        schedule = self.schedulers[i].schedule_microbatches(
            total_batch_size, 
            monitor['memory_info']
        )
        
        microbatch_schedules[i] = schedule
    
    return microbatch_schedules

def process_microbatch(self, microbatch_idx: int, microbatch_size: int, 
                      device_idx: int) -> torch.Tensor:
    """处理单个微批次"""
    device = self.devices[device_idx]
    scheduler = self.schedulers[device_idx]
    
    # 模拟微批次处理
    start_time = time.time()
    
    # 将模型和数据移动到指定设备
    model_device = self.model.to(device)
    dummy_input = torch.randn(microbatch_size, 128).to(device)  # 简化输入
    
    # 前向传播
    with torch.no_grad():
        output = model_device(dummy_input)
    
    # 计算性能指标
    latency = time.time() - start_time
    throughput = microbatch_size / latency  # tokens per second
    
    # 更新性能统计
    scheduler.update_performance_metrics(microbatch_size, latency, throughput)
    
    return output

def get_system_statistics(self) -> Dict:
    """获取系统统计信息"""
    stats = {
        'num_devices': len(self.devices),
        'gpu_status': [],
        'scheduler_status': []
    }
    
    # GPU状态
    for device, monitor in self.gpu_monitors.items():
        gpu_stat = {
            'device': str(device),
            'memory_usage': monitor['memory_info']['used_memory'] / 1024**3,  # GB
            'memory_free': monitor['memory_info']['free_memory'] / 1024**3,  # GB
            'utilization': monitor['utilization'],
            'temperature': monitor['temperature'],
            'power_usage': monitor['power_usage']
        }
        stats['gpu_status'].append(gpu_stat)
    
    # 调度器状态
    for i, scheduler in enumerate(self.schedulers):
        scheduler_stat = {
            'device_index': i,
            'device': str(self.devices[i]),
            'statistics': scheduler.get_statistics()
        }
        stats['scheduler_status'].append(scheduler_stat)
    
    return stats
​

演示智能微批次调度

def demonstrate_microbatch_scheduling():
"""演示智能微批次调度"""
print("\n智能微批次调度演示")
print("-" * 50)

# 创建设备
devices = [torch.device(f'cuda:{i}') for i in range(2)]

# 创建测试模型
model = nn.Sequential(
    nn.Linear(128, 256),
    nn.ReLU(),
    nn.Linear(256, 128),
    nn.ReLU(),
    nn.Linear(128, 64)
)

# 创建动态微批次管理器
manager = DynamicMicrobatchManager(
    model=model,
​