+---------------------------------------------------------------------------------------------------+
|                  Ringi 生产真实事故复盘：千卡并行 Checkpoint 诱发 NCCL 级联雪崩全链路                      |
+---------------------------------------------------------------------------------------------------+
|                                                                                                   |
|  [Step 1000 到达]                                                                                  |
|       │                                                                                           |
|       ▼                                                                                           |
|  1024 张 H100 GPU 同时调用 torch.save()                                                           |
|       │                                                                                           |
|       ▼                                                                                           |
|  405B 参数 + FP32 优化器状态 = 单卡 3.2 GB，集群瞬间产生 3.2 TB 突发写并发流量                           |
|       │                                                                                           |
|       ▼                                                                                           |
|  [传统分布式文件系统 CephFS / NFS 遭遇吞吐与元数据暴击]                                           |
|       ├─ 1024 个 Rank 密集在同一父目录下 open(O_CREAT|O_WRONLY) ──> MDS 元数据锁排队严重串行化       |
|       └─ 突发写入打爆 Client Page Cache ──> 脏页回写占满网络 ──> Ceph OSD 日志盘 IOPS 耗尽，Heartbeat 丢包 |
|       │                                                                                           |
|       ▼                                                                                           |
|  [内核级联崩溃]                                                                                    |
|       ├─ 存储客户端出现不可中断睡眠 (D-State)，VFS 锁等待超时                                       |
|       └─ 部分节点写盘卡死 15 分钟，其他已写完节点进入下一步，在下一个 AllReduce 触发 NCCL Watchdog 超时!|
|       │                                                                                           |
|       ▼                                                                                           |
|  [灾难结果] 全集群 128 台物理机被 Kubelet 标记不可调度，任务失败重拉，当晚有效算力损失超 300,000 元！|
|                                                                                                   |
+---------------------------------------------------------------------------------------------------+

+---------------------------------------------------------------------------------------------------+
|                        大模型系统四大极端 IO 负载模式（Access Pattern）对比全景图                         |
+---------------------------------------------------------------------------------------------------+
|                                                                                                   |
|  1. 数据读取 (Dataset Loader)         2. Checkpoint 突发落盘 (Save Ckpt)                          |
|     模式：海量随机不可预测、高并发只读      模式：极端周期性突发大带宽写入、写后基本不读                  |
|     特征：小文件随机 Seek，元数据压力大      特征：全集群千卡同步突发，吞吐需 > 500GB/s                  |
|     痛点：IO 饥饿导致 GPU 等待数据         痛点：长时间停机阻塞训练循环，MFU 直线下滑                   |
|                                                                                                   |
|  3. 权重快速加载 (Weight Warmup)       4. 推理上下文缓存 (KV Cache)                               |
|     模式：冷启动瞬间广播读，全集群共享      模式：长文本前缀毫秒级点对点查写、多级流动                  |
|     特征：高并发拉取大文件 (safetensors)    特征：跨 HBM/DRAM/NVMe 换入换出，延迟要求 < 5ms              |
|     痛点：存储出向网关瓶颈，拉取耗时数十分钟 痛点：GPU 显存寸土寸金，重复 Prefill 计算资源白白浪费       |
|                                                                                                   |
+---------------------------------------------------------------------------------------------------+

+---------------------------------------------------------------------------------------------------+
|                                  Ringi 工程师五问闭环：存储系统设计基石                                   |
+---------------------------------------------------------------------------------------------------+
| 1. Shape 是什么？    | Checkpoint 张量切片尺寸：[TP, PP, DP, Shard_Size]；KV Cache Block: [16, H, D]    |
| 2. Cost 花在哪里？   | 存储网络传输时延 + 内核上下文切换与内存拷贝 + 元数据锁争抢等待时间                  |
| 3. Machine 怎么跑？  | NVMe SSD 直接读写 -> PCIe 5.0 -> Host RAM (Pin Memory) -> RDMA 网卡 -> 存储节点    |
| 4. Evidence 在哪里？ | AI_BOOK/deepseek_3fs/、AI_BOOK/storage/juicefs/ 生产压测基准与源码实现           |
| 5. Production 怎么选？| 训练必须解耦异步快照；大并发选 RDMA+CRAQ (3FS)；模型分发靠 P2P；推理选多级金字塔   |
+---------------------------------------------------------------------------------------------------+

+---------------------------------------------------------------------------------------------------+
|                        传统 FUSE 用户态文件系统在 AI 场景下的“致命性能开销”链路                          |
+---------------------------------------------------------------------------------------------------+
|                                                                                                   |
|  [用户空间应用: PyTorch DataLoader]                                                                |
|         │                                                                                         |
|         │ 1. read() 系统调用 (触发 User -> Kernel 切换)                                           |
|         ▼                                                                                         |
|  [内核空间: Linux VFS -> /dev/fuse 内核驱动]                                                       |
|         │                                                                                         |
|         │ 2. 数据拷贝到内核缓冲区，挂入 fuse_dev 等待队列，唤醒用户态 Daemon                        |
|         ▼                                                                                         |
|  [用户空间: FUSE Client Daemon (如传统 JuiceFS/s3fs)]                                             |
|         │                                                                                         |
|         │ 3. 再次发生 Kernel -> User 上下文切换，解析请求，查找本地缓存或发起 RPC 访问远端存储       |
|         │ 4. 从网络或磁盘获取数据后，调用 write() 写回 /dev/fuse                                   |
|         ▼                                                                                         |
|  [内核空间: fuse_dev 处理应答]                                                                    |
|         │                                                                                         |
|         │ 5. 再次数据拷贝：从 Daemon 拷贝到内核 Page Cache，唤醒 PyTorch 线程                       |
|         ▼                                                                                         |
|  [用户空间: PyTorch 获得数据]                                                                      |
|                                                                                                   |
|  【总结惨痛代价】：一次原本纯净的 IO，经历了 4 次上下文切换 + 3 次内核/用户态内存拷贝 + 全局自旋锁竞争！|
|  在每秒百万级小样本读取时，CPU 40% 以上时间全部锁死在内核态上下文切换中，GPU 严重饥饿等待！       |
|                                                                                                   |
+---------------------------------------------------------------------------------------------------+

+---------------------------------------------------------------------------------------------------+
|                        DeepSeek 3FS (Fire-Flyer File System) 全景系统架构图                        |
+---------------------------------------------------------------------------------------------------+
|                                                                                                   |
|  [客户端计算节点: 千卡 GPU 计算集群 (Train / Inference Worker)]                                      |
|  ┌─────────────────────────────────────────────────────────────────────────────────────────────┐  |
|  │  PyTorch / SGLang / vLLM                                                                    │  |
|  │      │                                                                                      │  |
|  │  3FS Client Native Lib (零拷贝共享内存环形队列 Ior / Iov, 绕过传统 FUSE)                     │  |
|  └──────┬───────────────────────────────────────────────┬──────────────────────────────────────┘  |
|         │                                               │                                         |
|         │ RDMA (RoCEv2 / InfiniBand, 400Gbps)           │ RDMA Direct NVMe Access                 |
|         ▼                                               ▼                                         |
|  ┌────────────────────────────────────────┐   ┌────────────────────────────────────────────────┐  |
|  │ 3FS Metadata Service (Stateless)       │   │ 3FS Storage Service (Direct NVMe Engine)       │  |
|  │  - 纯无状态设计，多实例水平扩容        │   │  - 基于 SPDK / Kernel Bypass 直访本地 NVMe SSD │  |
|  │  - 移除动态属性跟踪 (stat/mtime)       │   │  - 链式复制读写协议 (CRAQ, Chain Replication)  │  |
|  └──────────────────┬─────────────────────┘   └──────────────────────┬─────────────────────────┘  |
|                     │                                                │                            |
|                     ▼ 强一致分布式事务                               │ 节点间 RDMA 数据流水线      |
|  ┌────────────────────────────────────────┐                          │                            |
|  │ FoundationDB (分片事务型 KV 底座)       │                          ▼                            |
|  │  - 存储全量文件目录树与元数据          │             Node 1 (Head) ──> Node 2 ──> Node 3 (Tail)     |
|  │  - ACID 事务保证，无单点目录锁瓶颈     │             (写从 Head 进)             (读在全链路并发分流)|
|  └────────────────────────────────────────┘                                                       |
|                                                                                                   |
+---------------------------------------------------------------------------------------------------+

+---------------------------------------------------------------------------------------------------+
|                        Checkpoint 停机开销公式五步穿透（No Naked Formula 2.0）                       |
+---------------------------------------------------------------------------------------------------+
| 1. 为什么算？      | 评估定期全量 Checkpoint 阻塞训练导致的千卡集群有效算力利用率 (MFU) 衰减幅度。          |
| 2. 物理直觉        | 训练是流水线。只要 GPU 被同步阻塞挂起去等待数据刷盘，这段时间内数千万机时费就在空转燃烧。 |
| 3. 极简数字手算    | 假设每 500 步保存一次，单步耗时 1.5s (有效训练 750s)。落盘耗时 15 分钟 (900s)。         |
|                   | 总时间 = 750 + 900 = 1650s。有效利用率仅 750 / 1650 = 45.45%！算力腰斩！          |
| 4. 正式物理公式    | 见下文数学推导                                                                     |
| 5. 数量级校验      | 采用 Host RAM 异步快照将停顿降低到 2.5s 时，利用率恢复到 750 / 752.5 = 99.67%！       |
+---------------------------------------------------------------------------------------------------+

+---------------------------------------------------------------------------------------------------+
|                           工业级异步非阻塞 Checkpoint 三级流水线架构设计                              |
+---------------------------------------------------------------------------------------------------+
|                                                                                                   |
|  [主训练流水线 (GPU 计算线程)]                                                                     |
|       │                                                                                           |
|       │ Step N 结束，触发快照                                                                     |
|       ├─ 1. 在独立 CUDA Stream 下发 D2H (Device to Host) 异步内存拷贝                             |
|       │    将当前卡负责的参数与优化器分片写入 Host Pinned Memory Buffer (专用环形预分配池)          |
|       ├─ 2. cudaStreamSynchronize(stream_d2h) 仅耗时几百毫秒完成 CPU 内存快照                     |
|       ▼                                                                                           |
|  [主训练线程立即释放！毫不拖延，直接进入 Step N+1 前向计算]                                          |
|                                                                                                   |
|  [后台异步工作线程 (CPU Background Worker Thread)]                                                |
|       │                                                                                           |
|       │ 接管刚刚快照的 Host Pinned Memory                                                         |
|       ├─ 3. 并发写入本地极致快速的 NVMe SSD Scratch 目录 (耗时 5~10 秒)                           |
|       │    (保证若此时节点突发断电，本地 SSD 仍留存最新完好副本)                                    |
|       ├─ 4. 通过后台 IO 队列通过 RDMA/网络异步上传至分布式共享存储 (3FS / S3 / JuiceFS)              |
|       ├─ 5. 分布式元数据原子更名 (Rename / Manifest Commit)                                        |
|       └─ 6. 释放 Host Pinned Memory 缓冲区，供下一次迭代循环复用                                  |
|                                                                                                   |
+---------------------------------------------------------------------------------------------------+

+---------------------------------------------------------------------------------------------------+
|                        推理系统四级存储金字塔（Multi-Tier Storage Pyramid）                       |
+---------------------------------------------------------------------------------------------------+
|                                                                                                   |
|               ▲                    [Tier 0: GPU HBM 显存]                                         |
|              / \                   容量：80 ~ 144 GB | 带宽：2.0 ~ 3.3 TB/s | 时延：< 10 ns        |
|             /   \                  用途：当前活跃 Step 运行计算区 (PagedAttention Block)            |
|            /     \                                                                                |
|           /       \                [Tier 1: Host RAM 宿主机内存]                                   |
|          /         \               容量：1 ~ 2 TB | 带宽：32 ~ 64 GB/s (PCIe 5.0) | 时延：< 100 ns |
|         /           \              用途：本地热点会话 KV Cache 换出暂存区                          |
|        /             \                                                                            |
|       /               \            [Tier 2: 本地 NVMe SSD 闪存阵列]                               |
|      /                 \           容量：8 ~ 30 TB | 带宽：7 ~ 14 GB/s | 时延：< 10 us             |
|     /                   \          用途：单节点长文本 Prefix 缓存池                                 |
|    /                     \                                                                        |
|   /                       \        [Tier 3: 远端分布式存储 (3FS / JuiceFS / S3)]                   |
|  /─────────────────────────\       容量：PB 级无上限 | 带宽：10 ~ 50 GB/s (RDMA) | 时延：1 ~ 5 ms   |
|                                    用途：全局共享跨实例 Prompt 库、跨节点迁移与知识库持久化        |
|                                                                                                   |
+---------------------------------------------------------------------------------------------------+

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File: async_checkpoint_engine.py
Description: 工业级大模型训练异步非阻塞 Checkpoint 保存模拟器
Author: Ringi (AI Infra Architect)
Standards: Full-Output Enforcement, Zero-Omission, Production-Ready
"""

import os
import time
import queue
import threading
import torch
import shutil

class AsyncCheckpointManager:
    """
    异步无感 Checkpoint 管理器
    架构设计：
      1. GPU 显存 -> Host Pinned Memory (主线程极速异步 D2H 拷贝并同步流)
      2. Host Pinned Memory -> Disk/Distributed Storage (后台独立 IO 线程落盘)
      3. 双缓冲轮转机制，防止内存分配抖动与并发读写冲突
    """
    def __init__(self, checkpoint_dir: str = "./checkpoints_scratch", num_buffers: int = 2):
        self.checkpoint_dir = checkpoint_dir
        os.makedirs(self.checkpoint_dir, exist_ok=True)
        
        self.io_queue = queue.Queue(maxsize=num_buffers)
        self.shutdown_event = threading.Event()
        
        # 启动后台 IO 落盘工作线程
        self.worker_thread = threading.Thread(target=self._io_worker, daemon=True)
        self.worker_thread.start()
        
        print(f"[Init] AsyncCheckpointManager 初始化完成，后台写线程已就绪，落地目录: {self.checkpoint_dir}")

    def _io_worker(self):
        """后台线程：负责将 CPU 内存数据安全持久化到磁盘"""
        while not self.shutdown_event.is_set():
            try:
                task = self.io_queue.get(timeout=0.5)
            except queue.Empty:
                continue

            step, cpu_state_dict, start_time = task
            save_path = os.path.join(self.checkpoint_dir, f"model_step_{step}.pt")
            temp_path = save_path + ".tmp"
            
            disk_start = time.perf_counter()
            # 模拟落盘写入（先写临时文件，再原子更名）
            torch.save(cpu_state_dict, temp_path)
            os.replace(temp_path, save_path)
            disk_cost = time.perf_counter() - disk_start
            total_cost = time.perf_counter() - start_time
            
            print(f"[Async-IO] Step {step} 异步落盘完成! 磁盘写入耗时: {disk_cost*1000:.2f} ms, 全链路总耗时: {total_cost*1000:.2f} ms")
            self.io_queue.task_done()

    def save_checkpoint_sync(self, step: int, model: torch.nn.Module) -> float:
        """基线对照：传统同步保存模式（主线程全量阻塞等待磁盘写完）"""
        start = time.perf_counter()
        state_dict = model.state_dict()
        save_path = os.path.join(self.checkpoint_dir, f"sync_model_step_{step}.pt")
        torch.save(state_dict, save_path)
        cost = time.perf_counter() - start
        return cost

    def save_checkpoint_async(self, step: int, model: torch.nn.Module) -> float:
        """工业级优化：异步快照保存模式（主线程仅等待内存拷贝，立即返回）"""
        start = time.perf_counter()
        
        # 1. 在 CPU 侧创建 Pinned Memory 存储副本（模拟快速显存到主机内存拷贝）
        # 在真实多卡 GPU 场景中，此处采用异步 CUDA Stream 下发: tensor.to('cpu', non_blocking=True)
        cpu_state_dict = {}
        for k, v in model.state_dict().items():
            # clone 保证后续迭代修改权重时不会破坏当前快照数据（Copy-on-Write 逻辑）
            cpu_state_dict[k] = v.detach().cpu().clone().pin_memory()
            
        staging_cost = time.perf_counter() - start
        
        # 2. 将快照推送至后台队列，主线程全身而退
        self.io_queue.put((step, cpu_state_dict, start))
        return staging_cost

    def close(self):
        """等待所有积压的 Checkpoint 写盘完成并平稳退出"""
        print("[Shutdown] 等待未完成的异步写入任务提交...")
        self.io_queue.join()
        self.shutdown_event.set()
        self.worker_thread.join()
        print("[Shutdown] 异步写入管理系统安全退出。")


def run_test():
    print("=" * 70)
    print(">> 实战 1：同步 vs 异步 Checkpoint 性能基准实测")
    print("=" * 70)
    
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f">> 当前计算运行设备: {device}")
    
    # 构建一个中等尺寸的模拟模型（包含多层 Linear，产生数千万参数）
    layers = []
    for _ in range(5):
        layers.append(torch.nn.Linear(2048, 2048))
    model = torch.nn.Sequential(*layers).to(device)
    
    total_params = sum(p.numel() for p in model.parameters())
    print(f">> 模型总参数量: {total_params / 1e6:.2f} M Float32 数据")
    
    manager = AsyncCheckpointManager(checkpoint_dir="./test_ckpt_output")
    
    # 1. 测试同步阻塞保存
    sync_costs = []
    for step in range(1, 4):
        cost = manager.save_checkpoint_sync(step, model)
        sync_costs.append(cost)
        print(f"[Sync-Test] Step {step} 主线程完全阻塞停滞耗时: {cost*1000:.2f} ms")
        
    avg_sync = sum(sync_costs) / len(sync_costs)
    
    # 2. 测试异步快照保存
    async_costs = []
    for step in range(4, 7):
        cost = manager.save_checkpoint_async(step, model)
        async_costs.append(cost)
        print(f"[Async-Test] Step {step} 主线程感知停滞耗时 (仅内存快照): {cost*1000:.2f} ms")
        time.sleep(0.1) # 模拟训练下一步计算
        
    avg_async = sum(async_costs) / len(async_costs)
    
    manager.close()
    
    # 清理测试生成的临时文件
    shutil.rmtree("./test_ckpt_output", ignore_errors=True)
    
    print("-" * 70)
    print(f"【性能结果穿透分析】:")
    print(f"  - 传统同步阻塞平均耗时: {avg_sync*1000:.2f} ms")
    print(f"  - 异步非阻塞快照平均耗时: {avg_async*1000:.2f} ms")
    print(f"  - 训练主流程停顿降低幅度: {(1 - avg_async/avg_sync)*100:.2f}%！")
    print("=" * 70)

if __name__ == "__main__":
    run_test()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File: p2p_weight_distributor.py
Description: 大模型千卡集群冷启动 P2P 权重分块广播与完整性哈希校验系统
Author: Ringi (AI Infra Architect)
Standards: Full-Output Enforcement, Zero-Omission, Production-Ready
"""

import hashlib
import threading
import time
from typing import Dict, List, Tuple

CHUNK_SIZE = 1024 * 1024  # 1MB 逻辑切片

class ChunkMetadata:
    """分块元数据契约"""
    def __init__(self, chunk_id: int, offset: int, size: int, sha256: str):
        self.chunk_id = chunk_id
        self.offset = offset
        self.size = size
        self.sha256 = sha256

class StorageNode:
    """模拟集群节点（既是数据拉取者，也是向邻居供血的 Seed 种子）"""
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.downloaded_chunks: Dict[int, bytes] = {}
        self.lock = threading.Lock()

    def has_chunk(self, chunk_id: int) -> bool:
        with self.lock:
            return chunk_id in self.downloaded_chunks

    def put_chunk(self, chunk_id: int, data: bytes, expected_sha: str) -> bool:
        # 严格验证数据完整性
        actual_sha = hashlib.sha256(data).hexdigest()
        if actual_sha != expected_sha:
            raise ValueError(f"节点 {self.node_id} 分块 {chunk_id} 数据损坏! 期望 {expected_sha}, 实际 {actual_sha}")
        with self.lock:
            self.downloaded_chunks[chunk_id] = data
        return True

    def get_chunk(self, chunk_id: int) -> bytes:
        with self.lock:
            return self.downloaded_chunks[chunk_id]


class P2PDistributionMesh:
    """BitTorrent 式去中心化广播控制器"""
    def __init__(self, raw_weights: bytes):
        self.raw_weights = raw_weights
        self.total_size = len(raw_weights)
        self.chunks_meta: List[ChunkMetadata] = []
        self._prepare_chunks()

    def _prepare_chunks(self):
        """将原始权重文件切分为定长分块，并预先计算哈希指纹"""
        offset = 0
        chunk_id = 0
        while offset < self.total_size:
            end = min(offset + CHUNK_SIZE, self.total_size)
            chunk_data = self.raw_weights[offset:end]
            sha = hashlib.sha256(chunk_data).hexdigest()
            self.chunks_meta.append(ChunkMetadata(chunk_id, offset, len(chunk_data), sha))
            offset = end
            chunk_id += 1
        print(f"[P2P-Root] 权重文件总大小: {self.total_size} 字节，成功划分为 {len(self.chunks_meta)} 个独立分块。")

    def simulate_broadcast(self, target_nodes: List[StorageNode]):
        """
        模拟分层级联 P2P 广播：
        1. 根节点将不同分块打散先随机发给不同 Worker；
        2. Worker 之间互通有无，并行从邻居拉取缺少的分块。
        """
        start_time = time.perf_counter()
        
        # Step 1: 种子分发 (Seed Phase) - Root 只需向网络发送相当于原始权重 1 倍的总流量
        print("\n[Phase 1] 根源节点向集群打散注入种子分块...")
        for i, meta in enumerate(self.chunks_meta):
            target_node = target_nodes[i % len(target_nodes)]
            chunk_bytes = self.raw_weights[meta.offset : meta.offset + meta.size]
            target_node.put_chunk(meta.chunk_id, chunk_bytes, meta.sha256)
            
        print(f"[Phase 1 完毕] 种子注入完成，耗时: {(time.perf_counter() - start_time)*1000:.2f} ms")

        # Step 2: 邻居互传 (Peer-to-Peer Gossiping)
        print("\n[Phase 2] 节点间高并发并行对等互传 (P2P Mesh Transfer)...")
        p2p_start = time.perf_counter()
        
        rounds = 0
        while True:
            rounds += 1
            all_done = True
            transfers_in_round = 0
            
            for node in target_nodes:
                for meta in self.chunks_meta:
                    if not node.has_chunk(meta.chunk_id):
                        all_done = False
                        # 寻找谁拥有这个分块
                        providers = [n for n in target_nodes if n.has_chunk(meta.chunk_id)]
                        if providers:
                            provider = providers[0] # 选择第一个可用邻居
                            data = provider.get_chunk(meta.chunk_id)
                            node.put_chunk(meta.chunk_id, data, meta.sha256)
                            transfers_in_round += 1
                            
            if all_done:
                break
                
        p2p_cost = time.perf_counter() - p2p_start
        print(f"[Phase 2 完毕] 全集群经过 {rounds} 轮互传达成全量同步，P2P 传输耗时: {p2p_cost*1000:.2f} ms")
        
        # 最终校验各节点组装是否完美无缺
        for node in target_nodes:
            assert len(node.downloaded_chunks) == len(self.chunks_meta)
        print("[Verify] 校验通过：所有计算节点权重 100% 字节级一致！")


def run_p2p_demo():
    print("=" * 70)
    print(">> 实战 2：大模型千卡权重加载 P2P 广播分发验证")
    print("=" * 70)
    
    # 模拟生成 10MB 的虚拟模型权重二进制流
    dummy_weight_bytes = os.urandom(10 * 1024 * 1024)
    
    # 构建 8 个模拟计算节点
    nodes = [StorageNode(f"Worker_Rank_{i}") for i in range(8)]
    
    distributor = P2PDistributionMesh(dummy_weight_bytes)
    distributor.simulate_broadcast(nodes)
    print("=" * 70)

if __name__ == "__main__":
    run_p2p_demo()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File: multi_tier_kv_cache.py
Description: 大模型长文本推理多级 KV Cache (HBM -> Host RAM -> NVMe) 分层引擎
Author: Ringi (AI Infra Architect)
Standards: Full-Output Enforcement, Zero-Omission, Production-Ready
"""

import collections
import hashlib
import time
from typing import Optional, Tuple

class CacheBlock:
    """KV Cache 物理分块抽象"""
    def __init__(self, block_hash: str, token_ids: list, size_mb: float):
        self.block_hash = block_hash
        self.token_ids = token_ids
        self.size_mb = size_mb
        self.last_access_time = time.time()

class MultiTierKVCacheManager:
    """
    四级推理缓存金字塔调度器
    - Tier 0: GPU HBM (最快，容量最小，单位 MB)
    - Tier 1: Host RAM (次快，容量中等，单位 MB)
    - Tier 2: Local NVMe (大容量，速度略慢，单位 MB)
    """
    def __init__(self, hbm_limit_mb: float = 128.0, ram_limit_mb: float = 512.0, nvme_limit_mb: float = 2048.0):
        self.limits = {"HBM": hbm_limit_mb, "RAM": ram_limit_mb, "NVME": nvme_limit_mb}
        self.current_usage = {"HBM": 0.0, "RAM": 0.0, "NVME": 0.0}
        
        # 使用 OrderedDict 实现精准 O(1) LRU 淘汰链表
        self.tiers = {
            "HBM": collections.OrderedDict(),
            "RAM": collections.OrderedDict(),
            "NVME": collections.OrderedDict()
        }
        
        print(f"[Init] 多级 KV Cache 体系构建完成: HBM上限={hbm_limit_mb}MB, RAM上限={ram_limit_mb}MB, NVMe上限={nvme_limit_mb}MB")

    def _generate_prefix_hash(self, token_ids: list) -> str:
        """根据 Token 前缀序列生成唯一确定性哈希指纹"""
        token_str = ",".join(map(str, token_ids))
        return hashlib.sha256(token_str.encode('utf-8')).hexdigest()[:16]

    def get(self, token_ids: list) -> Tuple[Optional[str], float]:
        """
        根据 Token 前缀查询 KV Cache 命中层级
        返回值: (命中层级 Tier_Name 或 None, 读取时延微秒)
        """
        block_hash = self._generate_prefix_hash(token_ids)
        
        # 1. 检查 Tier 0: GPU HBM
        if block_hash in self.tiers["HBM"]:
            block = self.tiers["HBM"][block_hash]
            block.last_access_time = time.time()
            self.tiers["HBM"].move_to_end(block_hash)
            return "HBM", 0.05 # 模拟 50ns
            
        # 2. 检查 Tier 1: Host RAM (命中后触发换入 HBM)
        if block_hash in self.tiers["RAM"]:
            block = self.tiers["RAM"][block_hash]
            block.last_access_time = time.time()
            self.tiers["RAM"].move_to_end(block_hash)
            self._promote_to_hbm(block)
            return "RAM", 1.50 # 模拟 1.5us PCIe 传输
            
        # 3. 检查 Tier 2: Local NVMe (命中后触发逐级回填)
        if block_hash in self.tiers["NVME"]:
            block = self.tiers["NVME"][block_hash]
            block.last_access_time = time.time()
            self.tiers["NVME"].move_to_end(block_hash)
            self._promote_to_hbm(block)
            return "NVME", 25.0 # 模拟 25us NVMe 读取
            
        return None, 0.0

    def put(self, token_ids: list, size_mb: float):
        """新 Prefill 阶段生成的 KV Cache 写入体系"""
        block_hash = self._generate_prefix_hash(token_ids)
        block = CacheBlock(block_hash, token_ids, size_mb)
        
        # 优先直接灌入最高层 HBM
        self._ensure_capacity("HBM", size_mb)
        self.tiers["HBM"][block_hash] = block
        self.current_usage["HBM"] += size_mb

    def _promote_to_hbm(self, block: CacheBlock):
        """将深层缓存提升至 HBM，加速后续高频 Token 生成"""
        # 从原本所在层剥离
        for tier in ["RAM", "NVME"]:
            if block.block_hash in self.tiers[tier]:
                del self.tiers[tier][block.block_hash]
                self.current_usage[tier] -= block.size_mb
                break
                
        # 腾出 HBM 空间并写入
        self._ensure_capacity("HBM", block.size_mb)
        self.tiers["HBM"][block.block_hash] = block
        self.current_usage["HBM"] += block.size_mb

    def _ensure_capacity(self, tier: str, required_mb: float):
        """LRU 级联逐出保障水线健康"""
        while self.current_usage[tier] + required_mb > self.limits[tier]:
            if not self.tiers[tier]:
                break
            # 弹出最久未访问的块
            oldest_hash, oldest_block = self.tiers[tier].popitem(last=False)
            self.current_usage[tier] -= oldest_block.size_mb
            
            # 降级到下一层 (Cascade Eviction)
            if tier == "HBM":
                self._ensure_capacity("RAM", oldest_block.size_mb)
                self.tiers["RAM"][oldest_hash] = oldest_block
                self.current_usage["RAM"] += oldest_block.size_mb
            elif tier == "RAM":
                self._ensure_capacity("NVME", oldest_block.size_mb)
                self.tiers["NVME"][oldest_hash] = oldest_block
                self.current_usage["NVME"] += oldest_block.size_mb
            else:
                # NVMe 也满了，彻底丢弃 (Drop Cache)
                pass

    def report_status(self):
        """输出各级存储水位实时体检报告"""
        print("-" * 65)
        for tier in ["HBM", "RAM", "NVME"]:
            usage = self.current_usage[tier]
            limit = self.limits[tier]
            count = len(self.tiers[tier])
            pct = (usage / limit) * 100
            print(f"[{tier:4s}] 水位: {usage:6.2f} / {limit:6.2f} MB ({pct:5.1f}%) | 包含 Block 数: {count}")
        print("-" * 65)


def run_cache_simulation():
    print("=" * 70)
    print(">> 实战 3：大模型多级 KV Cache 换入换出与分级命中全流程演示")
    print("=" * 70)
    
    cache_mgr = MultiTierKVCacheManager(hbm_limit_mb=64.0, ram_limit_mb=128.0, nvme_limit_mb=256.0)
    
    # 模拟灌入 5 个不同请求的超长 Prompt KV Cache（每个 32MB）
    prompts = [
        [101, 2054, 2003, 1037, i] + [99] * 100 for i in range(1, 6)
    ]
    
    print("\n>> 1. 逐步写入 5 个 32MB 的 KV Block，观察 HBM 溢出后向 RAM/NVMe 自动降级...")
    for idx, p in enumerate(prompts):
        print(f"写入 Prompt #{idx+1} (32MB)...")
        cache_mgr.put(p, size_mb=32.0)
        cache_mgr.report_status()
        
    print("\n>> 2. 查询 Prompt #1 (最先写入的数据，当前应该已被挤出 HBM)...")
    hit_tier, latency = cache_mgr.get(prompts[0])
    print(f"Prompt #1 命中层级: {hit_tier}, 感知时延: {latency} us")
    cache_mgr.report_status()
    
    print("\n>> 3. 查询全新未见过的 Prompt #999...")
    hit_tier, latency = cache_mgr.get([999, 888, 777])
    print(f"全新 Prompt 命中结果: {hit_tier} (Miss, 需执行完整 Prefill 运算)")
    print("=" * 70)

if __name__ == "__main__":
    run_cache_simulation()

#!/usr/bin/env bash
# ==============================================================================
# File: mount_ai_storage_production.sh
# Description: 生产级大模型高性能存储客户端挂载与内核网络 IO 极致调优脚本
# Author: Ringi (AI Infra Architect)
# Target: Ubuntu 22.04 LTS / Enterprise Linux, Mellanox ConnectX-6/7 RDMA
# ==============================================================================

set -euo pipefail

echo "========================================================================"
echo ">> [1/4] Linux 操作系统内核网络与虚拟内存极限调优"
echo "========================================================================"

# 1. 调大系统文件句柄上限（防止海量小文件句柄耗尽）
sysctl -w fs.file-max=2097152

# 2. 调优脏页回写参数，防止 Checkpoint 突发落盘打满 Page Cache 诱发 D 状态卡死
# 当脏页占总内存达到 10% 时，后台 pdflush 线程立即静默回写
sysctl -w vm.dirty_background_ratio=5
# 当脏页占总内存达到 20% 时，强制阻塞产生脏页的进程，防止雪崩
sysctl -w vm.dirty_ratio=15

# 3. 调大网络 Socket 读写缓冲区（适配 400Gbps 高带宽延迟积 BDP）
sysctl -w net.core.rmem_max=67108864
sysctl -w net.core.wmem_max=67108864
sysctl -w net.ipv4.tcp_rmem="4096 87380 33554432"
sysctl -w net.ipv4.tcp_wmem="4096 65536 33554432"

echo ">> 内核参数生效完毕。"

echo "========================================================================"
echo ">> [2/4] 本地极速 NVMe Scratch 缓存盘挂载准备"
echo "========================================================================"

LOCAL_CACHE_DIR="/mnt/nvme_cache"
mkdir -p "${LOCAL_CACHE_DIR}"

# 生产规范：缓存盘必须启用 noatime 和 discard (TRIM) 挂载选项，压榨 SSD 随机性能
# mount -o noatime,nodiratime,discard /dev/nvme0n1 "${LOCAL_CACHE_DIR}"
echo ">> 本地 NVMe 缓存目录已就绪: ${LOCAL_CACHE_DIR}"

echo "========================================================================"
echo ">> [3/4] 执行 JuiceFS 生产级挂载命令 (模拟与规范)"
echo "========================================================================"

MOUNT_POINT="/data/model_repo"
mkdir -p "${MOUNT_POINT}"

# 核心调优参数深度剖析：
# --cache-dir: 绑定到本地极速 NVMe SSD，阻断远端网络回源
# --cache-size: 分配 1.5TB 本地只读预热缓存
# --free-space-ratio: 预留 10% 磁盘空间防爆仓
# --max-uploads: 调大并发异步上传线程数至 64，应对 Checkpoint 并发分块写入
# --buffer-size: 读写内存缓冲区放大至 1024MB
# --prefetch: 启用 5 个分块预读，彻底隐藏顺序读取时延
# --writeback: 启用客户端写缓冲（先落本地 NVMe，后台异步推向对象存储，训练必须开启！）

cat << 'EOF' > /tmp/juicefs_mount_example.sh
juicefs mount \
    --background \
    --cache-dir /mnt/nvme_cache \
    --cache-size 1500000 \
    --free-space-ratio 0.1 \
    --max-uploads 64 \
    --buffer-size 1024 \
    --prefetch 5 \
    --writeback \
    --open-cache 300 \
    --attr-cache 300 \
    --entry-cache 300 \
    --dir-entry-cache 300 \
    --no-usage-report \
    redis://10.0.0.10:6379/1 \
    /data/model_repo
EOF

chmod +x /tmp/juicefs_mount_example.sh
echo ">> 生产 JuiceFS 挂载模板已生成至 /tmp/juicefs_mount_example.sh"

echo "========================================================================"
echo ">> [4/4] DeepSeek 3FS 客户端部署黄金规范提示"
echo "========================================================================"
cat << 'EOF'
【DeepSeek 3FS 生产部署三大黄金军规】：
1. 存储网络与计算通信网络物理/子网隔离：
   3FS RDMA 流量切勿与 NCCL 集合通信流量争抢相同的 RoCE PFC 优先级队列！
   建议 NCCL 使用 DSCP 46 (Priority 3)，3FS 使用 DSCP 26 (Priority 4)。
2. FoundationDB 事务引擎高可用：
   元数据底座必须采用跨机架（Double/Triple Replication）部署，
   事务日志盘（Transaction Log）必须独占企业级高 DWPD NVMe SSD。
3. 容器挂载标准：
   利用 Kubernetes HostPath 配合特权容器挂载大页内存（HugePages 2MB/1GB），
   供 3FS 客户端原生无锁 Ring Buffer 独占访问！
EOF
echo "========================================================================"

千卡训练怕落盘，元数据锁最要命；
海量样本打包读，散装文件休进门；
异步快照留内存，主线秒退继算程；
推理长文显存贵，四级缓存金字塔；
全线极速找三埃（3FS），链式读写全打通！

+---------------------------------------------------------------------------------------------------+
|                            大厂 AI 存储与缓存系统 10 条高频白板考察要点                             |
+---------------------------------------------------------------------------------------------------+
| 1. 为什么千卡训练时传统的 POSIX 文件系统 (如 NFS/CephFS) 会出现严重的元数据锁死？                 |
| 2. 对象存储 (S3) 为什么不适合直接作为大模型训练的实时数据读写盘？                                    |
| 3. 详细画出并推演 FUSE 用户态文件系统的内部调用链路，说明其性能损耗的本质。                         |
| 4. DeepSeek 3FS 是如何利用 CRAQ (链式复制) 协议实现线性写吞吐与高并发广播读扩展的？                |
| 5. 3FS 与 FoundationDB 的配合是如何从根本上消除传统 MDS 单点目录锁瓶颈的？                          |
| 6. 异步非阻塞 Checkpoint 的实现原理是什么？如何避免主线程内存拷贝时的显存与主机内存抖动？           |
| 7. 推理长文本场景下，单 Token 的 KV Cache 显存大小如何计算？为什么需要跨节点多级缓存？              |
| 8. 详细阐述 Radix Attention 与前缀缓存 (Prefix Caching) 的命中判定与匹配逻辑。                     |
| 9. 在 RoCEv2 无损以太网环境下，如何避免存储写盘突发流量干扰 NCCL 分布式训练集合通信？              |
| 10. 针对千卡集群冷启动，如何基于 P2P (如 Dragonfly) 技术设计高效的模型权重分发架构？              |
+---------------------------------------------------------------------------------------------------+