"""FlorenceForge内存管理工具模块 提供内存监控、优化和管理功能 """ import gc import math import psutil import torch import threading import time import warnings from dataclasses import dataclass from contextlib import contextmanager from collections import deque from typing import List, Dict, Any, Union, Optional, Tuple from pathlib import Path @dataclass class MemoryInfo: """内存信息数据类""" total: float # 总内存 (GB) available: float # 可用内存 (GB) used: float # 已用内存 (GB) percent: float # 使用百分比 def __str__(self) -> str: """返回内存信息的字符串表示 Returns: str: 格式化的内存使用情况字符串,显示已用/总内存和使用百分比 """ return f"Memory: {self.used:.1f}GB/{self.total:.1f}GB ({self.percent:.1f}%)" @dataclass class GPUMemoryInfo: """GPU内存信息数据类""" device_id: int name: str total: float # 总显存 (GB) allocated: float # 已分配显存 (GB) reserved: float # 已保留显存 (GB) free: float # 空闲显存 (GB) def __str__(self) -> str: """返回GPU内存信息的字符串表示 Returns: str: 格式化的GPU内存使用情况字符串,显示设备ID、名称和已分配/总显存 """ return f"GPU {self.device_id} ({self.name}): {self.allocated:.1f}GB/{self.total:.1f}GB" def get_memory_usage() -> MemoryInfo: """获取系统内存使用情况 Returns: 内存信息 """ memory = psutil.virtual_memory() return MemoryInfo( total=memory.total / (1024**3), available=memory.available / (1024**3), used=memory.used / (1024**3), percent=memory.percent ) def get_gpu_memory_usage() -> List[GPUMemoryInfo]: """获取GPU内存使用情况 Returns: GPU内存信息列表 """ gpu_info = [] if torch.cuda.is_available(): for i in range(torch.cuda.device_count()): props = torch.cuda.get_device_properties(i) allocated = torch.cuda.memory_allocated(i) / (1024**3) reserved = torch.cuda.memory_reserved(i) / (1024**3) total = props.total_memory / (1024**3) free = total - reserved gpu_info.append(GPUMemoryInfo( device_id=i, name=props.name, total=total, allocated=allocated, reserved=reserved, free=free )) return gpu_info def clear_cache() -> None: """清理缓存""" # 清理Python垃圾回收 gc.collect() # 清理PyTorch缓存 if torch.cuda.is_available(): torch.cuda.empty_cache() torch.cuda.synchronize() def clear_gpu_cache() -> None: """清理GPU缓存 专门用于清理GPU内存缓存的函数 """ if torch.cuda.is_available(): torch.cuda.empty_cache() torch.cuda.synchronize() # 强制垃圾回收 gc.collect() def optimize_memory( aggressive: bool = False, clear_gradients: bool = True, model: Optional[torch.nn.Module] = None, ) -> Dict[str, float]: """优化内存使用 Args: aggressive: 是否进行激进优化 clear_gradients: 是否清理梯度 model: 可选模型;提供时仅遍历其参数释放梯度,避免全局对象扫描 Returns: 优化前后的内存使用情况 """ # 记录优化前的内存使用 before_memory = get_memory_usage() before_gpu = get_gpu_memory_usage() # 基础优化 if clear_gradients: if model is not None: for param in model.parameters(): if param.grad is not None: param.grad = None if torch.cuda.is_available(): torch.cuda.empty_cache() # 清理缓存 clear_cache() # 激进优化 if aggressive: # 强制垃圾回收多次 for _ in range(3): gc.collect() # 清理未使用的张量 if torch.cuda.is_available(): torch.cuda.empty_cache() torch.cuda.ipc_collect() # 记录优化后的内存使用 after_memory = get_memory_usage() after_gpu = get_gpu_memory_usage() result = { 'system_memory_freed': before_memory.used - after_memory.used, 'system_memory_before': before_memory.used, 'system_memory_after': after_memory.used } if before_gpu and after_gpu: result.update({ 'gpu_memory_freed': before_gpu[0].allocated - after_gpu[0].allocated, 'gpu_memory_before': before_gpu[0].allocated, 'gpu_memory_after': after_gpu[0].allocated }) return result @contextmanager def memory_monitor(name: str = "Operation", log_func=None): """内存监控上下文管理器 Args: name: 操作名称 log_func: 日志函数 """ if log_func is None: log_func = print # 记录开始时的内存 start_memory = get_memory_usage() start_gpu = get_gpu_memory_usage() log_func(f"[{name}] 开始 - {start_memory}") if start_gpu: log_func(f"[{name}] 开始 - {start_gpu[0]}") try: yield finally: # 记录结束时的内存 end_memory = get_memory_usage() end_gpu = get_gpu_memory_usage() memory_diff = end_memory.used - start_memory.used log_func(f"[{name}] 结束 - {end_memory} (变化: {memory_diff:+.1f}GB)") if start_gpu and end_gpu: gpu_diff = end_gpu[0].allocated - start_gpu[0].allocated log_func(f"[{name}] 结束 - {end_gpu[0]} (变化: {gpu_diff:+.1f}GB)") class MemoryTracker: """内存跟踪器 持续监控内存使用情况 """ def __init__( self, interval: float = 1.0, max_history: int = 1000 ): """初始化内存跟踪器 Args: interval: 监控间隔(秒) max_history: 最大历史记录数 """ self.interval = interval self.max_history = max(0, max_history) self.history = deque(maxlen=self.max_history) self.is_tracking = False self.thread: Optional[threading.Thread] = None def start(self) -> None: """开始跟踪""" if self.is_tracking: return self.is_tracking = True self.thread = threading.Thread(target=self._track_loop, daemon=True) self.thread.start() def stop(self) -> None: """停止跟踪""" self.is_tracking = False if self.thread: self.thread.join() def _track_loop(self) -> None: """跟踪循环""" while self.is_tracking: timestamp = time.time() memory_info = get_memory_usage() gpu_info = get_gpu_memory_usage() record = { 'timestamp': timestamp, 'system_memory': { 'total': memory_info.total, 'used': memory_info.used, 'percent': memory_info.percent } } if gpu_info: record['gpu_memory'] = [ { 'device_id': gpu.device_id, 'allocated': gpu.allocated, 'reserved': gpu.reserved, 'total': gpu.total } for gpu in gpu_info ] self.history.append(record) time.sleep(self.interval) def get_peak_usage(self) -> Dict[str, float]: """获取峰值使用情况 Returns: 峰值使用情况 """ if not self.history: return {} peak_system = max(record['system_memory']['used'] for record in self.history) result = {'peak_system_memory': peak_system} # GPU峰值 if self.history[0].get('gpu_memory'): for i in range(len(self.history[0]['gpu_memory'])): peak_gpu = max( record['gpu_memory'][i]['allocated'] for record in self.history if 'gpu_memory' in record and i < len(record['gpu_memory']) ) result[f'peak_gpu_{i}_memory'] = peak_gpu return result def get_average_usage(self) -> Dict[str, float]: """获取平均使用情况 Returns: 平均使用情况 """ if not self.history: return {} avg_system = sum(record['system_memory']['used'] for record in self.history) / len(self.history) result = {'avg_system_memory': avg_system} # GPU平均 if self.history[0].get('gpu_memory'): for i in range(len(self.history[0]['gpu_memory'])): gpu_values = [ record['gpu_memory'][i]['allocated'] for record in self.history if 'gpu_memory' in record and i < len(record['gpu_memory']) ] if gpu_values: result[f'avg_gpu_{i}_memory'] = sum(gpu_values) / len(gpu_values) return result def export_history(self, file_path: Union[str, Path]) -> None: """导出历史记录 Args: file_path: 文件路径 """ import json file_path = Path(file_path) file_path.parent.mkdir(parents=True, exist_ok=True) with open(file_path, 'w') as f: json.dump(list(self.history), f, indent=2) class MemoryPool: """内存池 管理可重用的张量内存 """ def __init__(self, device: Optional[torch.device] = None): """初始化内存池 Args: device: 设备 """ self.device = device or torch.device('cpu') self.pools: Dict[Tuple[torch.Size, torch.dtype], List[torch.Tensor]] = {} self.max_pool_size = 10 def get_tensor( self, shape: Union[torch.Size, Tuple[int, ...]], dtype: torch.dtype = torch.float32 ) -> torch.Tensor: """获取张量 Args: shape: 张量形状 dtype: 数据类型 Returns: 张量 """ if isinstance(shape, tuple): shape = torch.Size(shape) key = (shape, dtype) if key in self.pools and self.pools[key]: tensor = self.pools[key].pop() tensor.zero_() # 清零 return tensor else: return torch.zeros(shape, dtype=dtype, device=self.device) def return_tensor(self, tensor: torch.Tensor) -> None: """归还张量 Args: tensor: 要归还的张量 """ if tensor.device != self.device: return key = (tensor.shape, tensor.dtype) if key not in self.pools: self.pools[key] = [] if len(self.pools[key]) < self.max_pool_size: # 分离梯度并移动到CPU(如果需要) tensor = tensor.detach() self.pools[key].append(tensor) def clear(self) -> None: """清空内存池""" self.pools.clear() clear_cache() def get_stats(self) -> Dict[str, Any]: """获取内存池统计信息 Returns: 统计信息 """ total_tensors = sum(len(pool) for pool in self.pools.values()) total_memory = 0 for (shape, dtype), pool in self.pools.items(): tensor_size = math.prod(shape) * torch.tensor([], dtype=dtype).element_size() total_memory += len(pool) * tensor_size return { 'total_pools': len(self.pools), 'total_tensors': total_tensors, 'total_memory_mb': total_memory / (1024**2), 'pools': { f"{shape}_{dtype}": len(pool) for (shape, dtype), pool in self.pools.items() } } def check_memory_requirements( model_size_gb: float, batch_size: int, sequence_length: int, safety_factor: float = 1.5 ) -> Dict[str, Any]: """检查内存需求 Args: model_size_gb: 模型大小(GB) batch_size: 批次大小 sequence_length: 序列长度 safety_factor: 安全系数 Returns: 内存需求分析 """ # 估算内存需求 # 模型参数 + 梯度 + 优化器状态 + 激活值 estimated_memory = model_size_gb * 4 # 参数 + 梯度 + Adam状态 # 激活值内存(粗略估算) activation_memory = batch_size * sequence_length * 1024 * 4 / (1024**3) # 假设1024维隐藏层 total_estimated = (estimated_memory + activation_memory) * safety_factor # 获取当前可用内存 current_memory = get_memory_usage() gpu_memory = get_gpu_memory_usage() result = { 'estimated_memory_gb': total_estimated, 'model_memory_gb': model_size_gb, 'activation_memory_gb': activation_memory, 'safety_factor': safety_factor, 'system_available_gb': current_memory.available, 'system_sufficient': current_memory.available > total_estimated } if gpu_memory: gpu = gpu_memory[0] result.update({ 'gpu_available_gb': gpu.free, 'gpu_sufficient': gpu.free > total_estimated }) return result def suggest_batch_size( model_size_gb: float, available_memory_gb: float, sequence_length: int = 512, safety_factor: float = 0.8 ) -> int: """建议批次大小 Args: model_size_gb: 模型大小(GB) available_memory_gb: 可用内存(GB) sequence_length: 序列长度 safety_factor: 安全系数 Returns: 建议的批次大小 """ # 保留内存用于模型和其他开销 usable_memory = (available_memory_gb - model_size_gb * 3) * safety_factor if usable_memory <= 0: return 1 # 估算每个样本的内存需求 memory_per_sample = sequence_length * 1024 * 4 / (1024**3) # 粗略估算 suggested_batch_size = max(1, int(usable_memory / memory_per_sample)) return suggested_batch_size @contextmanager def low_memory_mode(): """低内存模式上下文管理器""" # 保存原始设置 original_benchmark = torch.backends.cudnn.benchmark original_deterministic = torch.backends.cudnn.deterministic try: # 设置为低内存模式 torch.backends.cudnn.benchmark = False torch.backends.cudnn.deterministic = True # 清理缓存 clear_cache() yield finally: # 恢复原始设置 torch.backends.cudnn.benchmark = original_benchmark torch.backends.cudnn.deterministic = original_deterministic def monitor_memory_leaks( threshold_mb: float = 100.0, check_interval: float = 60.0 ) -> None: """监控内存泄漏 Args: threshold_mb: 内存增长阈值(MB) check_interval: 检查间隔(秒) """ initial_memory = get_memory_usage().used * 1024 # 转换为MB def check_leak(): """检查内存泄漏的内部函数 检查当前内存使用量与初始内存的差值,如果超过阈值则记录警告日志。 """ current_memory = get_memory_usage().used * 1024 growth = current_memory - initial_memory if growth > threshold_mb: warnings.warn( f"检测到可能的内存泄漏: 内存增长 {growth:.1f}MB", UserWarning ) # 启动监控线程 def monitor_loop(): """内存监控循环函数 在后台线程中持续运行,按指定间隔检查内存泄漏情况。 """ while True: time.sleep(check_interval) check_leak() monitor_thread = threading.Thread(target=monitor_loop, daemon=True) monitor_thread.start()