6.3.1 Transfer Engine （传输引擎）

Transfer Engine是 Mooncake 的核心组件，负责高性能的 KV Cache 数据传输。它是Mooncake最早开源的组件，也是整个系统的基础。

6.3.1.1 基于ZeroMQ的通信

Transfer Engine使用ZeroMQ作为底层通信框架，提供高效的异步消息传递能力：

cpp

// Transfer Engine 核心架构
class TransferEngine {
public:
//   初始化Transfer Engine
int initialize(const std::string& local_hostname,
                const std::string& metadata_server,
                const std::string& protocol,
                const std::string& device_name);


//   注册本地内存区域
int batchRegisterMemory(const std::vector<void*>& addrs,
                const std::vector<size_t>& sizes);


//   执行数据传输
int batchTransferSyncWrite(const std::string& remote_session,
                const std::vector<uintptr_t>& src_addrs,
                const std::vector<uintptr_t>& dst_addrs,
                const std::vector<size_t>& sizes);


//   获取RPC端口
int getRpcPort() const;


private:
    std::unique_ptr<Transport> transport_;
    std::shared_ptr<MetadataService> metadata_service_;
};

ZeroMQ 提供了以下优势：

异步I/O：支持高并发连接
多传输协议：支持TCP、IPC、inproc等
灵活的消息模式：支持pub-sub、req-rep、push-pull等

6.3.1.2 RDMA 支持

Transfer Engine原生支持RDMA（Remote Direct Memory Access），实现零拷贝、低延迟的数据传输：

支持的RDMA传输方式： 1. InfiniBand RDMA：传统InfiniBand网络 2. RoCEv2：基于以太网的RDMA 3. GPUDirect RDMA（GDR）：GPU内存直接访问 4. 阿里云eRDMA：阿里云自研弹性RDMA

cpp

// RDMA 传输层实现
class RdmaTransport : public Transport {
public:
    int initialize(const std::string& local_hostname,
                    const std::string& device_name) override;   
    
    //   注册GPU内存用于GDR
    int registerGpuMemory(void* addr, size_t size, int cuda_device);    
    
    //   执行RDMA写操作
    int write(const std::string& remote_session,
            uintptr_t src_addr, uintptr_t dst_addr, size_t size);


private:
    struct ibv_context* ctx_;
    struct ibv_pd* pd_;
    struct ibv_cq* cq_;
    std::vector<struct ibv_qp*> qps_;
};

6.3.1.3序列化和反序列化

Transfer Engine使用高效的序列化方案：

python

# Mooncake Transfer Engine Python API
import mooncake_transfer_engine as mte

# 初始化Transfer Engine
engine = mte.TransferEngine()
engine.initialize(
      local_hostname="192.168.1.10",
      metadata_server="etcd://192.168.1.1:2379",
      protocol="rdma",
      device_name="mlx5_0"
)


# 注册内存区域
import torch
kv_cache = torch.empty((num_blocks, block_size, num_heads, head_dim),
                         dtype=torch.float16, device='cuda')
engine.batch_register_memory([kv_cache.data_ptr()], [kv_cache.nbytes])


# 执行数据传输
engine.batch_transfer_sync_write(
      remote_session="192.168.1.20:50051",
      src_addrs=[src_ptr],
      dst_addrs=[dst_ptr],
      sizes=[transfer_size]
)

6.3.1.4 性能优化

Transfer Engine实现了多项性能优化

多网卡聚合 - 支持最多8×400Gbps网卡带宽聚合 - 拓扑感知路由 - 故障容错和负载均衡
零拷贝传输 - GPU内存直接通过 GDR传输 - 避免 CPU内存拷贝开销 - 减少延迟和 CPU占用
批量传输 - 支持批量内存注册和传输 - 减少系统调用开销 - 提高吞吐量
拓扑感知 - 识别节点间的网络拓扑 - 优化数据传输路径 - 避免网络拥塞

6.3.2 P2P Store （点对点存储）

P2P Store是Mooncake提供的去中心化分布式对象存储系统，特别适用于机器学习场景中的和参数同步。

6.3.2.1 设计原理

P2P Store采用纯客户端架构，不依赖中心化的Master节点：

核心设计特点： 1. 去中心化：无单点故障，所有节点对等** 2. 智能数据分发：新节点可从**多个源并行下载 3. 内存级传输：基于RDMA的直接内存访问 4. 弹性扩展：节点可动态加入和离开

6.3.2.2 节点间数据共享

P2P Store实现了高效的节点间数据共享机制

cpp

// P2P Store 核心API
class P2PStore {
    public:
    //   初始化P2P Store
    int initialize(const std::string& etcd_endpoints);
     
    //   写入对象
    int put(const std::string& key, const void* data, size_t size);
     
    //   读取对象
    int get(const std::string& key, void* buffer, size_t* size);
    
    //   从多个源并行获取
    int getFromPeers(const std::string& key,
                        const std::vector<std::string>& peers,
                        void* buffer, size_t* size);
    
    
private:
    std::shared_ptr<TransferEngine> transfer_engine_;
    std::shared_ptr<MetadataClient> metadata_client_;
};

数据分片策略：

大对象自动分片为固定大小的chunk
每个chunk可从不同peer并行获取
支持断点续传和错误恢复

6.3.2.3 一致性保证

P2P Store采用最终一致性模型

python

# P2P Store  使用示例
from mooncake import P2PStore

#   初始化store
store = P2PStore()
store.initialize("etcd://localhost:2379")

#   写入检查点
checkpoint_data = model.state_dict()
store.put("checkpoint/step_1000", checkpoint_data)

#   广播到多个推理节点
peers = store.discover_peers("inference_nodes")
for peer in peers:
    store.replicate_to("checkpoint/step_1000", peer)

#   推理节点读取检查点
checkpoint = store.get("checkpoint/step_1000")
model.load_state_dict(checkpoint)

6.3.3 高性能内存存储

Mooncake Store是Mooncake的分布式KV Cache存储系统，提供多级缓存能力。

6.3.3.1 内存池管理

Mooncake Store实现了高效的内存池管理：

cpp

// 内存池管理器
class MemoryPool {
public:
    //   创建内存池
    static std::unique_ptr<MemoryPool> create(
            size_t gpu_pool_size,
            size_t cpu_pool_size,
            const std::string& ssd_path,
            size_t ssd_pool_size
    );
        
    //   分配内存块
    Buffer allocate(size_t size, MemoryTier tier);
        
    //   释放内存块
    void deallocate(Buffer buffer);   
    
    //   在不同层级间迁移数据
    int migrate(const Buffer& src, Buffer& dst, MemoryTier dst_tier);

private:
    std::unique_ptr<GPUMemoryPool> gpu_pool_;
    std::unique_ptr<CPUMemoryPool> cpu_pool_;
    std::unique_ptr<SSDStorage> ssd_pool_;
};

内存层级： 1. L1（ GPU HBM）：延迟最低，容量最小 2. L2（ CPU DRAM）：延迟中等，容量较大 3. L3（SSD）：延迟最高，容量最大

6.3.3.2 数据结构设计

Mooncake Store使用专门优化的数据结构：

cpp

// KV Cache 块描述符
struct KVCacheBlock {
    BlockId id; //块ID
    MemoryTier tier; // 当前存储层级
    void* addr; // 内存地址
    size_t size; // 块大小
    std::atomic<int> ref_count; // 引用计数
    std::chrono::time_point last_access; // 最后访问时间
    std::vector<BlockId>  dependencies; // 依赖块（用于前缀树）
};

// 前缀树索引
class PrefixTree {
public:
    //   插入token序列
    void insert(const std::vector<int64_t>& tokens, BlockId block_id);
      
    //   查找最长匹配前缀
    std::pair<size_t, BlockId> longestPrefixMatch(
            const std::vector<int64_t>& tokens
    ) const;

private:
    struct Node {
        int64_t token;
        BlockId block_id;
        std::unordered_map<int64_t, std::unique_ptr<Node>> children;
    };
    std::unique_ptr<Node> root_;
};

6.3.3.3 并发控制

Mooncake Store实现了高效的并发控制机制：

cpp

// 并发安全的KV Cache管理器
class ConcurrentKVCacheManager {
public:
    //   获取读锁
    std::shared_lock<std::shared_mutex> readLock(BlockId id);   
    
    //   获取写锁
    std::unique_lock<std::shared_mutex> writeLock(BlockId id);  
    
    //   无锁读取（用于热路径）
    const KVCacheBlock* getBlockUnsafe(BlockId id) const;

private:
    std::unordered_map<BlockId, std::shared_mutex> block_locks_;
    std::unordered_map<BlockId, KVCacheBlock> blocks_;
    mutable std::shared_mutex global_lock_;
};

6.3.4 Global Scheduler （全局调度器）

Conductor是Mooncake的全局调度器，负责协调整个集群的请求调度和资源分配。

6.3.4.1 以KV Cache为中心的调度

Conductor的调度决策基于KV Cache的分布和复用机会：

6.3.4.2 负载均衡策略

Conductor实现了多种负载均衡策略

Prefill阶段负载均衡

python

class CacheAwarePrefillScheduler:
     def select_prefill_node(self, request):
         candidates = self.get_candidate_nodes()

         best_node = None
         best_score = float('-inf')

         for node in candidates:
             #   计算KV Cache命中率
             cache_hit_len = self.prefix_match(request.tokens, node.cache)
             cache_hit_ratio = cache_hit_len / len(request.tokens)

             #   预估传输时间
             transfer_time = self.estimate_transfer_time(
                  node, cache_hit_len
             )

             #   预估计算时间
             compute_time = self.estimate_compute_time(
                  node, len(request.tokens) - cache_hit_len
             )

             #   预估队列等待时间
             queue_time = node.estimated_queue_time

             #   综合评分
             ttft_estimate = transfer_time + compute_time + queue_time
             score = self.compute_score(cache_hit_ratio, ttft_estimate)

             if score > best_score:
                  best_score = score
                  best_node = node

         return best_node

Decode阶段负载均衡

python

class LoadBalanceDecodeScheduler:
      def select_decode_node(self, request):
          candidates = self.get_candidate_nodes()

          best_node = None
          best_score = float('-inf')

          for node in candidates:
              #   当前batch大小
              current_batch = len(node.active_requests)

              #   预估TPOT
              tpot_estimate = self.estimate_tpot(node, request)

              #   内存使用情况
              memory_usage = node.memory_usage

              #   综合评分
              score = self.compute_score(current_batch, tpot_estimate, memory

              if score > best_score:
                   best_score = score
                   best_node = node

          return best_node

6.3.4.3 SLO保证

Conductor 通过预测模型保证SLO（Service Level Objective）：

python

class SLOEnforcer:
  def __init__(self):
     self.ttft_slo = 2.0      # 首token时间SLO（秒）
     self.tbt_slo = 0.1       # 每token时间SLO（秒）
     self.estimator = TTFTEstimator()


  def can_meet_slo(self, request, prefill_node, decode_node):
     #   预估TTFT
     estimated_ttft = self.estimator.estimate(
           request, prefill_node
     )

     #   预估TBT
     estimated_tbt = self.estimate_tbt(request, decode_node)

     #   检查是否满足SLO
     if estimated_ttft > self.ttft_slo:
           return False
     if estimated_tbt > self.tbt_slo:
           return False

     return True

  def early_rejection(self, request):
     """ 预测性早期拒绝"""
     best_prefill = self.select_best_prefill_node(request)
     best_decode = self.select_best_decode_node(request)

     if not self.can_meet_slo(request, best_prefill, best_decode):
           #   直接拒绝请求，返回429 Too Many Requests
           raise HTTPException(status_code=429, detail="System overloaded")

6.3.4.4 预测和预热

Conductor实现了智能的预测和预热机制

TTFT预测模型

python

class TTFTEstimator:
   def __init__(self):
       #   使用历史数据训练预测模型
       self.model = load_pretrained_model()

   def estimate(self, request, node):
       features = {
             'request_length': len(request.tokens),
             'cache_hit_length': self.get_cache_hit_length(request, node),
             'node_queue_length': node.queue_length,
             'node_gpu_utilization': node.gpu_utilization,
             'network_bandwidth': node.network_bandwidth,
       }

       return self.model.predict(features)

KV Cache 预热：

python

class KVCachePrefetcher:
   def prefetch_for_session(self, session_id, expected_tokens):
       """为会话预热KV Cache"""
       # 预测可能的后续token
       predicted_tokens = self.predict_next_tokens(expected_tokens)

       #   预计算KV Cache
       kv_cache = self.compute_kv_cache(predicted_tokens)

       #   存储到快速存储层
       self.store_in_l1_cache(session_id, kv_cache)

6.3.1 Transfer Engine （传输引擎） ​

6.3.1.1 基于ZeroMQ的通信 ​

6.3.1.2 RDMA 支持 ​

6.3.1.3序列化和反序列化 ​

6.3.1.4 性能优化 ​

6.3.2 P2P Store （点对点存储） ​

6.3.2.1 设计原理 ​

6.3.2.2 节点间数据共享 ​

6.3.2.3 一致性保证 ​

6.3.3 高性能内存存储 ​

6.3.3.1 内存池管理 ​

6.3.3.2 数据结构设计 ​

6.3.3.3 并发控制 ​

6.3.4 Global Scheduler （全局调度器） ​

6.3.4.1 以KV Cache为中心的调度 ​

6.3.4.2 负载均衡策略 ​

Prefill阶段负载均衡 ​

Decode阶段负载均衡 ​

6.3.4.3 SLO保证 ​

6.3.4.4 预测和预热 ​

TTFT预测模型 ​

KV Cache 预热： ​

6.3.1 Transfer Engine （传输引擎）

6.3.1.1 基于ZeroMQ的通信

6.3.1.2 RDMA 支持

6.3.1.3序列化和反序列化

6.3.1.4 性能优化

6.3.2 P2P Store （点对点存储）

6.3.2.1 设计原理

6.3.2.2 节点间数据共享

6.3.2.3 一致性保证

6.3.3 高性能内存存储

6.3.3.1 内存池管理

6.3.3.2 数据结构设计

6.3.3.3 并发控制

6.3.4 Global Scheduler （全局调度器）

6.3.4.1 以KV Cache为中心的调度

6.3.4.2 负载均衡策略

Prefill阶段负载均衡

Decode阶段负载均衡

6.3.4.3 SLO保证

6.3.4.4 预测和预热

TTFT预测模型

KV Cache 预热：