print(f"Failed to initialize NVML: {e}")

    # 初始化CUDA
    try:
        import cuda
        self.cuda_initialized = True
        print("CUDA interface initialized")
    except ImportError:
        print("CUDA interface not available")

def get_nvme_devices(self) -> List[Dict]:
    """获取NVMe设备信息"""
    if not self.nvml_init:
        return []
    
    devices = []
    try:
        handle_count = pynvml.nvmlDeviceGetCount()
        
        for i in range(handle_count):
            handle = pynvml.nvmlDeviceGetHandleByIndex(i)
            
            device_info = {
                'index': i,
                'name': pynvml.nvmlDeviceGetName(handle).decode('utf-8'),
                'serial': pynvml.nvmlDeviceGetSerial(handle).decode('utf-8'),
                'model': pynvml.nvmlDeviceGetModel(handle).decode('utf-8'),
                'firmware': pynvml.nvmlDeviceGetFirmwareVersion(handle).decode('utf-8'),
                'pci_info': {
                    'bus_id': pynvml.nvmlDeviceGetPcieUtilization(handle),
                    'max_bandwidth': pynvml.nvmlDeviceGetPcieThroughput(handle, pynvml.NVML_PCIE_UTILIZATION_READ)
                }
            }
            
            # 获取NVMe信息（需要专门的NVMe工具）
            nvme_info = self._get_nvme_info(device_info['serial'])
            if nvme_info:
                device_info['nvme'] = nvme_info
            
            devices.append(device_info)
    
    except Exception as e:
        print(f"Error getting NVMe devices: {e}")
    
    return devices

def _get_nvme_info(self, serial: str) -> Dict:
    """获取NVMe详细信息"""
    try:
        # 使用nvme-cli工具获取信息
        result = subprocess.run(
            ['sudo', 'nvme', 'list', '-o', 'json'],
            capture_output=True,
            text=True,
            check=True
        )
        
        import json
        nvme_list = json.loads(result.stdout)
        
        for device in nvme_list['Devices']:
            if serial in device['DevicePath']:
                return {
                    'device_path': device['DevicePath'],
                    'firmware': device['Firmware'],
                    'size': device['Size'],
                    'physical_size': device['PhysicalSize'],
                    'used_bytes': device['UsedBytes'],
                    'available_bytes': device['AvailableBytes']
                }
    
    except subprocess.CalledProcessError as e:
        print(f"Error getting NVMe info: {e}")
    
    return None

def get_gpu_memory_info(self) -> Dict:
    """获取GPU内存信息"""
    if not self.cuda_initialized:
        return {}
    
    try:
        import cuda
        
        memory_info = {}
        for i in range(cuda.device_count()):
            memory_info[i] = {
                'total_memory': cuda.mem_get_info(i)[1],
                'free_memory': cuda.mem_get_info(i)[0],
                'used_memory': cuda.mem_get_info(i)[1] - cuda.mem_get_info(i)[0]
            }
        
        return memory_info
    
    except Exception as e:
        print(f"Error getting GPU memory info: {e}")
        return {}

def optimize_pcie_throughput(self, gpu_id: int, nvme_id: int):
    """优化PCIe吞吐量"""
    try:
        # 设置GPU的PCIe带宽限制
        handle = pynvml.nvmlDeviceGetHandleByIndex(gpu_id)
        
        # 获取当前PCIe配置
        current_config = pynvml.nvmlDeviceGetPcieThroughput(handle, pynvml.NVML_PCIE_UTILIZATION_READ)
        print(f"Current PCIe throughput for GPU {gpu_id}: {current_config} MB/s")
        
        # 设置最大PCIe带宽
        max_bandwidth = 32000  # PCIe 4.0 x16的理论最大带宽
        pynvml.nvmlDeviceSetPersistenceMode(handle, pynvml.NVML_FEATURE_ENABLED)
        pynvml.nvmlDeviceSetDriverModel(handle, pynvml.NVML_DRIVER_WDDM)
        
        # 设置PCIe配置（需要管理员权限）
        subprocess.run([
            'sudo', 'nvml-device', f'-d', str(gpu_id),
            '--set-pcie-threshold', str(max_bandwidth)
        ], check=True)
        
        print(f"Optimized PCIe throughput for GPU {gpu_id} to {max_bandwidth} MB/s")
        
    except Exception as e:
        print(f"Error optimizing PCIe throughput: {e}")

def setup_direct_memory_access(self, gpu_id: int, nvme_device_path: str):
    """设置GPU与NVMe的直接内存访问"""
    try:
        # 创建CUDA内存池
        import cuda
        cuda.set_device(gpu_id)
        
        # 创建NVMe内存映射
        mempool = cuda.Mempool()
        mempool.register_memory(nvme_device_path)
        
        # 设置内存属性
        mempool.set_access_flags(cuda.AccessFlags.GPU_READ | cuda.AccessFlags.GPU_WRITE)
        mempool.set_location(cuda.MemoryLocation.GPU)
        
        print(f"Set up direct memory access from GPU {gpu_id} to NVMe {nvme_device_path}")
        
        return mempool
    
    except Exception as e:
        print(f"Error setting up direct memory access: {e}")
        return None

def benchmark_io_performance(self, gpu_id: int, nvme_id: int, test_size_gb: float = 1.0):
    """基准测试I/O性能"""
    try:
        # 测试数据准备
        test_size = int(test_size_gb * 1024**3)  # GB to bytes
        test_data = torch.randn(test_size // 4).cuda()  # float32 = 4 bytes
        
        # 测试GPU到NVMe的写入
        start_time = time.time()
        
        # 实际实现应该使用专门的NVMe写入接口
        # 这里简化为模拟
        torch.save(test_data, f'/tmp/test_gpu_to_nvme_{gpu_id}.pt')
        
        write_time = time.time() - start_time
        write_throughput = test_size / write_time / (1024**3)  # GB/s
        
        # 测试NVMe到GPU的读取
        start_time = time.time()
        loaded_data = torch.load(f'/tmp/test_gpu_to_nvme_{gpu_id}.pt')
        read_time = time.time() - start_time
        read_throughput = test_size / read_time / (1024**3)  # GB/s
        
        # 清理测试文件
        os.remove(f'/tmp/test_gpu_to_nvme_{gpu_id}.pt')
        
        result = {
            'gpu_id': gpu_id,
            'nvme_id': nvme_id,
            'test_size_gb': test_size_gb,
            'write_throughput_gb_s': write_throughput,
            'read_throughput_gb_s': read_throughput,
            'write_time_s': write_time,
            'read_time_s': read_time
        }
        
        return result
    
    except Exception as e:
        print(f"Error benchmarking I/O performance: {e}")
        return None
​

使用示例

def demo_nvme_gpu_interface():
"""演示NVMe与GPU接口功能"""
interface = NVMeGPUInterface()

# 获取设备信息
nvme_devices = interface.get_nvme_devices()
gpu_memory_info = interface.get_gpu_memory_info()

print("NVMe Devices:")
for device in nvme_devices:
    print(f"  {device['name']} (Serial: {device['serial']})")

print("\nGPU Memory Info:")
for gpu_id, info in gpu_memory_info.items():
    print(f"  GPU {gpu_id}: {info['used_memory']/1024**3:.1f}GB / {info['total_memory']/1024**3:.1f}GB used")

# 优化PCIe吞吐量
if nvme_devices and gpu_memory_info:
    interface.optimize_pcie_throughput(0, 0)
    
    # 设置直接内存访问
    nvme_path = nvme_devices[0]['device_path']
    mempool = interface.setup_direct_memory_access(0, nvme_path)
    
    # 性能测试
    benchmark_result = interface.benchmark_io_performance(0, 0, 1.0)
    if benchmark_result:
        print("\nI/O Performance Benchmark:")
        print(f"  Write throughput: {benchmark_result['write_throughput_gb_s']:.2f} GB/s")
        print(f"  Read throughput: {benchmark_result['read_throughput_gb_s']:.2f} GB/s")
​

demo_nvme_gpu_interface()

## 2. 分级存储实现策略

### 2.1 智能数据调度算法

**基于访问模式的数据迁移**

```python
import heapq
from collections import deque, defaultdict
from typing import List, Dict, Set, Tuple
import numpy as np
from datetime import datetime, timedelta

class AccessPatternAnalyzer:
    """访问模式分析器"""
    
    def __init__(self, window_size_minutes: int = 60):
        self.window_size = window_size_minutes * 60  # 转换为秒
        self.access_history = deque(maxlen=10000)  # 访问历史记录
        self.block_access_patterns = defaultdict(list)
        self.access_statistics = defaultdict(lambda: {
            'total_access': 0,
            'last_access': 0,
            'access_intervals': [],
            'access_frequency': 0.0,
            'predicted_next_access': None
        })
    
    def record_access(self, block_id: str, timestamp: float = None):
        """记录访问事件"""
        if timestamp is None:
            timestamp = time.time()
        
        self.access_history.append((block_id, timestamp))
        self.block_access_patterns[block_id].append(timestamp)
        
        # 更新访问统计
        stats = self.access_statistics[block_id]
        stats['total_access'] += 1
        stats['last_access'] = timestamp
        
        # 计算访问间隔
        if len(self.block_access_patterns[block_id]) > 1:
            last_time = self.block_access_patterns[block_id][-2]
            interval = timestamp - last_time
            stats['access_intervals'].append(interval)
            
            # 保持最近100个间隔
            if len(stats['access_intervals']) > 100:
                stats['access_intervals'] = stats['access_intervals'][-100:]
        
        # 计算访问频率
        time_span = timestamp - self._get_window_start_time(timestamp)
        if time_span > 0:
            stats['access_frequency'] = stats['total_access'] / (time_span / 3600)  # 每小时访问次数
        
        # 预测下次访问时间
        self._predict_next_access(block_id, timestamp)
    
    def _get_window_start_time(self, timestamp: float) -> float:
        """获取时间窗口的开始时间"""
        return timestamp - self.window_size
    
    def _predict_next_access(self, block_id: str, timestamp: float):
        """预测下次访问时间"""
        stats = self.access_statistics[block_id]
        
        if len(stats['access_intervals']) >= 3:
            # 使用最近的访问间隔进行简单预测
            recent_intervals = stats['access_intervals'][-3:]
            avg_interval = sum(recent_intervals) / len(recent_intervals)
            
            # 添加一些随机性
            predicted_time = timestamp + avg_interval * np.random.uniform(0.8, 1.2)
            stats['predicted_next_access'] = predicted_time
    
    def get_access_pattern(self, block_id: str) -> Dict:
        """获取数据块的访问模式"""
        stats = self.access_statistics[block_id]
        
        if not stats['access_intervals']:
            return {
                'pattern_type': 'cold',
                'access_frequency': 0.0,
                'access_intervals': [],
                'prediction': None
            }
        
        # 分析访问模式
        intervals = np.array(stats['access_intervals'])
        avg_interval = np.mean(intervals)
        std_interval = np.std(intervals)
        
        # 判断访问模式类型
        pattern_type = self._classify_pattern(intervals, avg_interval, std_interval)
        
        return {
            'pattern_type': pattern_type,
            'access_frequency': stats['access_frequency'],
            'avg_interval_seconds': avg_interval,
            'std_interval_seconds': std_interval,
            'coefficient_of_variation': std_interval / (avg_interval + 1e-6),
            'prediction': stats['predicted_next_access'],
            'last_access': stats['last_access'],
            'total_access': stats['total_access']
        }
    
    def _classify_pattern(self, intervals: np.ndarray, avg_interval: float, std_interval: float) -> str:
        """分类访问模式"""
        cv = std_interval / (avg_interval + 1e-6)  # 变异系数
        
        if avg_interval < 10:  # 10秒内访问
            return 'hot'
        elif avg_interval < 60:  # 1分钟内访问
            return 'warm'
        elif avg_interval < 300:  # 5分钟内访问
            return 'lukewarm'
        elif cv > 2.0:  # 高变异系数，不规律访问
            return 'irregular'
        elif cv < 0.5:  # 低变异系数，规律访问
            return 'periodic'
        else:
            return 'normal'

class TieredStorageScheduler:
    """分级存储调度器"""
    
    def __init__(self, 
                 storage_manager: NVMeOffloadManager,
                 access_analyzer: AccessPatternAnalyzer):
        self.storage_manager = storage_manager
        self.access_analyzer = access_analyzer
        self.scheduling_queue = []
        self.migration_thresholds = {
            'hot_to_warm': 300,    # 5分钟
            'warm_to_lukewarm': 1800,  # 30分钟
            'lukewarm_to_cold': 7200,  # 2小时
            'cold_to_nvme': 28800  # 8小时
        }
    
    def add_scheduling_request(self, block_id: str, priority: int = 0):
        """添加调度请求"""
        heapq.heappush(self.scheduling_queue, (priority, time.time(), block_id))
    
    def process_scheduling_queue(self):
        """处理调度队列"""
        if not self.scheduling_queue:
            return
        
        # 按优先级处理请求
        priority, timestamp, block_id = heapq.heappop(self.scheduling_queue)
        
        # 检查数据块是否存在
        if block_id not in self.storage_manager.memory_blocks:
            return
        
        # 分析访问模式
        pattern = self.access_analyzer.get_access_pattern(block_id)
        current_tier = self.storage_manager.memory_blocks[block_id].storage_tier
        
        # 确定目标层次
        target_tier = self._determine_target_tier(current_tier, pattern)
        
        # 如果需要迁移
        if target_tier != current_tier:
            asyncio.create_task(
                self.storage_manager.move_to_tier(block_id, target_tier)
            )
            print(f"Scheduled migration of {block_id} from {current_tier} to {target_tier}")
    
    def _determine_target_tier(self, current_tier: StorageTier, pattern: Dict) -> StorageTier:
        """根据访问模式确定目标存储层次"""
        pattern_type = pattern['pattern_type']
        last_access = pattern['last_access']
        time_now = time.time()
​