当前位置：首页 > news >正文

异构计算集群上的推理调度器设计：GPU/CPU/NPU 混合调度的成本感知策略

news 2026/7/9 14:47:02

异构计算集群上的推理调度器设计：GPU/CPU/NPU 混合调度的成本感知策略

一、混合算力集群的调度困境：GPU 在排队，CPU 在空转

推理集群中常见这样的场景：8 张 A100 GPU 的任务队列已经排了 30 个请求，延迟持续攀升；而旁边的 64 核 CPU 节点却只有 15% 的利用率。NPU 加速卡更尴尬——由于算子兼容性问题，调度器默认将其排除在外，昂贵的硬件成了摆设。

问题的本质是：传统的 GPU-only 调度器忽略了异构算力的互补性。一个 13B 参数的模型，在 GPU 上推理延迟 50ms，在 CPU 上是 800ms。但很多场景下 800ms 的延迟完全可以接受（如批量离线处理、非实时对话）。将这些任务从 GPU 队列中卸载到 CPU/NPU 上执行，可以显著降低 GPU 的排队延迟。

异构调度的核心挑战不是"能不能调度"，而是"怎么定义最优分配策略"——需要同时考虑延迟 SLO、硬件成本（GPU 比 CPU 贵 10 倍）、以及模型对不同硬件的兼容性。

二、成本感知的多目标调度模型

graph TD A[推理请求到达] --> B{请求分类器} B -->|Latency-Sensitive| C[GPU 队列] B -->|Batch-OK| D{负载感知路由器} B -->|LoRA/小模型| E[NPU 队列] D -->|GPU 队列深度 < 阈值| C D -->|GPU 繁忙| F[CPU 队列] D -->|特定算子需求| E C --> G[GPU Executor] F --> H[CPU Executor] E --> I[NPU Executor] G --> J[结果聚合] H --> J I --> J J --> K[响应返回] style B fill:#1a1a2e,stroke:#e94560,color:#fff style D fill:#16213e,stroke:#0f3460,color:#fff

调度决策的核心是一组多目标优化方程：

Minimize: Cost = α × GPU_Cost + β × CPU_Cost + γ × NPU_Cost Subject to: Latency(p) ≤ SLO(p) for each request p

其中 α、β、γ 是硬件成本权重（反映单位算力的价格差异）。实际实现时，这个优化问题被简化为带优先级的贪心路由：先尝试最便宜的算力类型，被拒绝后再升级到更昂贵的算力。

三类请求的分流策略

延迟敏感请求（Latency-Sensitive）：由请求参数中的priority=realtime标记，直接路由到 GPU。这些请求通常来自用户交互场景，要求 P99 < 200ms。如果一个 GPU 节点无法满足延迟，可以使用模型并行（tensor parallelism）拆分到多 GPU。

批处理友好请求（Batch-OK）：离线评估、数据集推理等场景。当 GPU 队列深度超过阈值（如 ≥ 10）时，调度器自动将新请求路由到 CPU/NPU。延迟增加 5~20 倍，但 GPU 成本为零。

特殊算力请求：LoRA 微调模型的推理在小 NPU 上性能接近 GPU（因为 LoRA adapter 的矩阵乘在 NPU 的 systolic array 上效率较高）。这类请求被标记后优先调度到 NPU。

三、Rust 实现的多队列异构调度器

use std::collections::{BinaryHeap, HashMap, VecDeque}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, RwLock, Semaphore}; use serde::{Deserialize, Serialize}; /// 算力类型 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum ComputeType { GPU, CPU, NPU, } /// 推理请求的优先级 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub enum Priority { /// 实时交互请求 Realtime = 0, /// 普通请求 Normal = 1, /// 批量离线处理 Batch = 2, } /// 推理请求 #[derive(Debug, Clone)] pub struct InferenceRequest { pub id: u64, pub model_name: String, pub priority: Priority, /// 延迟 SLO（Service Level Objective） /// None 表示无延迟要求（批处理场景） pub latency_slo: Option<Duration>, /// 预估的计算量（GFLOPS），用于算力分配 pub estimated_gflops: f64, /// 是否包含 LoRA adapter pub has_lora: bool, /// 请求到达时间 pub arrived_at: Instant, } /// 算力节点的运行时状态 #[derive(Debug, Clone)] pub struct ComputeNode { pub id: String, pub compute_type: ComputeType, /// 可用并发槽位（如 GPU 的 batch size 余量） pub available_slots: usize, /// 当前队列深度 pub queue_depth: usize, /// 平均处理延迟（滑动窗口统计） pub avg_latency: Duration, /// 单位成本系数 /// GPU: 1.0, CPU: 0.1, NPU: 0.3 pub cost_coefficient: f64, } /// 异构调度器 pub struct HeterogeneousScheduler { /// 按算力类型分组的节点池 nodes: HashMap<ComputeType, Vec<ComputeNode>>, /// 每种算力类型的请求队列 queues: HashMap<ComputeType, Arc<RwLock<VecDeque<InferenceRequest>>>>, /// GPU 队列深度阈值，超过后 overflow 到 CPU gpu_overflow_threshold: usize, /// 成本感知的权重参数 cost_weights: CostWeights, } #[derive(Debug, Clone)] struct CostWeights { gpu: f64, // 默认 1.0 cpu: f64, // 默认 0.1 npu: f64, // 默认 0.3 } impl HeterogeneousScheduler { /// 核心调度决策函数 /// /// 为什么用贪心而非全局最优解： /// 1. 推理请求数量动态变化，全局优化求解延迟太高（> 1ms） /// 2. 贪心算法的决策时间 < 10μs，对延迟尾部影响可忽略 /// 3. 在负载均匀分布的场景下，贪心 ≈ 全局最优 pub async fn schedule(&self, request: InferenceRequest) -> ComputeType { // 策略 1：实时请求直接走 GPU if request.priority == Priority::Realtime { return ComputeType::GPU; } // 策略 2：LoRA 请求优先 NPU // 为什么 LoRA 优先 NPU： // LoRA 的 adapter 矩阵乘在 NPU 的 systolic array 上有较好的加速效果 // 且 adapter 的尺寸通常匹配 NPU 的片上 SRAM 大小 if request.has_lora && self.has_available_slots(ComputeType::NPU) { return ComputeType::NPU; } // 策略 3：GPU 队列深度检查 let gpu_depth = self.get_queue_depth(ComputeType::GPU).await; if gpu_depth < self.gpu_overflow_threshold { return ComputeType::GPU; } // 策略 4：成本感知降级路由 // 尝试最便宜的算力，被拒绝后逐级升级 let candidates = [ (ComputeType::CPU, self.cost_weights.cpu), (ComputeType::NPU, self.cost_weights.npu), (ComputeType::GPU, self.cost_weights.gpu), ]; for (compute_type, _cost) in &candidates { if self.has_available_slots(*compute_type) { // SLO 检查：确保降级后的延迟仍满足 SLO if let Some(slo) = request.latency_slo { let avg_latency = self.get_avg_latency(*compute_type).await; if avg_latency > slo { continue; // 该算力无法满足延迟 SLO } } return *compute_type; } } // 所有算力都不可用，回退到 GPU（排队等待） ComputeType::GPU } async fn get_queue_depth(&self, ct: ComputeType) -> usize { self.queues.get(&ct) .map(|q| q.read().await.len()) .unwrap_or(0) } fn has_available_slots(&self, ct: ComputeType) -> bool { self.nodes.get(&ct) .map(|nodes| nodes.iter().any(|n| n.available_slots > 0)) .unwrap_or(false) } async fn get_avg_latency(&self, ct: ComputeType) -> Duration { self.nodes.get(&ct) .map(|nodes| { if nodes.is_empty() { return Duration::ZERO; } let total: Duration = nodes.iter().map(|n| n.avg_latency).sum(); total / nodes.len() as u32 }) .unwrap_or(Duration::ZERO) } /// 动态调整 overflow 阈值 /// /// 为什么需要动态调整： /// 固定阈值无法适应负载波动——高峰期需要更早 overflow， /// 低峰期可以容忍更深的 GPU 队列 pub async fn update_overflow_threshold(&mut self) { let gpu_depth = self.get_queue_depth(ComputeType::GPU).await; let cpu_depth = self.get_queue_depth(ComputeType::CPU).await; // 如果 CPU 空闲而 GPU 繁忙，降低 overflow 阈值 // 将更多请求引导到 CPU，平衡负载 if cpu_depth == 0 && gpu_depth > 5 { self.gpu_overflow_threshold = (self.gpu_overflow_threshold - 1).max(2); } else if cpu_depth > 10 && gpu_depth < 3 { self.gpu_overflow_threshold += 1; } } }