GPU 推理服务中的优先级调度与延迟预测

一句话总结

通过建模 PCIe 数据传输竞争和 kernel 执行干扰，优先级感知调度可将高优先级任务的 deadline 违例率降低最高 11 个百分点。

为什么需要这个？

现实中的推理服务困境

在企业私有化部署（on-premises）场景里，GPU 上通常同时运行多个模型的推理请求。你面对的是一个混乱的战场：

SLA 等级不同：实时语音识别（< 200ms）和离线批处理（< 5s）同时跑在一张 A100 上
负载波动剧烈：白天峰值时 GPU 利用率 > 90%，此时延迟预测完全失效
干扰来源复杂：不止 CUDA kernel 抢占 SM，PCIe 带宽竞争同样是延迟杀手

问题的核心：现有系统（Triton, TorchServe）的调度器假设每个请求的延迟是固定的，但并发执行时，一个 ResNet-50 的推理延迟可能因为另一个 BERT 的 H2D 传输而抖动 30%。

数据说话

在高 GPU 利用率（> 85%）场景下，主流推理服务框架的延迟预测误差：

框架	低利用率预测误差	高利用率预测误差
Triton Inference Server	~5%	~35%
固定 profile 方法	~8%	~42%
Strait 自适应预测	~6%	~12%

核心原理

直觉：把 GPU 推理想象成机场调度

想象一个繁忙机场：

跑道 = GPU SM（执行单元）
登机口到跑道的巴士 = PCIe 总线（数据传输）
头等舱乘客 = 高优先级推理请求（严格 deadline）
经济舱乘客 = 低优先级批处理请求

问题在于：巴士（PCIe）是共享的。头等舱乘客和经济舱乘客挤同一辆车，导致头等舱乘客也误机。

Strait 做的事情：

建模巴士拥堵（PCIe 竞争）
预测每条航班的实际延误（kernel 干扰模型）
给头等舱乘客安排专属车道（优先级调度）

硬件层面：两种干扰来源

干扰 1：PCIe 数据传输竞争

Host Memory → [PCIe x16] → GPU HBM → CUDA Kernel 执行
                 ↑
           多个任务共享这条通道
           理论带宽 64 GB/s (PCIe 4.0 x16)
           实际并发时每任务可能只有 20-30 GB/s

当两个模型同时做 H2D（Host to Device）传输时，PCIe 带宽被分割。一个 300MB 的模型在空闲时需要 5ms 传输，并发时可能需要 12ms——这直接让 deadline 预测失效。

干扰 2：Kernel 执行干扰

CUDA 的 MPS（Multi-Process Service）或 MIG 模式允许多 kernel 并发，但共享：

L2 Cache（竞争 cache line）
内存带宽（HBM 总线）
SM 资源（如果没有完全隔离）

干扰程度取决于任务的计算/访存比（arithmetic intensity），这是自适应预测模型要捕捉的关键特征。

自适应延迟预测模型

Strait 的核心公式：

\[T_{actual} = T_{base} + \Delta T_{PCIe}(B_{concurrent}) + \Delta T_{kernel}(I_{intensity}, N_{concurrent})\]

其中：

$T_{base}$：单独运行时的 profiling 延迟
$\Delta T_{PCIe}$：PCIe 竞争引入的额外传输延迟，是当前并发传输总带宽 $B_{concurrent}$ 的函数
$\Delta T_{kernel}$：kernel 干扰延迟，是任务的算术强度 $I_{intensity}$ 和并发任务数 $N_{concurrent}$ 的函数

代码实现

Baseline：朴素的 FIFO 调度器

大多数人会这样实现推理调度：

import queue
import threading
from dataclasses import dataclass
from typing import Optional

@dataclass
class InferenceRequest:
    task_id: str
    model_name: str
    deadline: float      # Unix timestamp
    priority: int        # 0=low, 1=high
    data_size_mb: float  # H2D 数据量
    arrival_time: float

class NaiveScheduler:
    """朴素 FIFO 调度器：不考虑优先级和延迟估计"""
    
    def __init__(self):
        self.queue = queue.Queue()
        # 从 profiling 得到的固定延迟表（问题所在！）
        self.latency_profile = {
            "resnet50": 3.2,   # ms，单独运行时测量
            "bert_base": 8.7,
            "yolov8": 5.1,
        }
    
    def submit(self, req: InferenceRequest):
        self.queue.put(req)
    
    def get_estimated_latency(self, req: InferenceRequest) -> float:
        # 直接查表，完全不考虑并发干扰
        return self.latency_profile.get(req.model_name, 10.0)
    
    def schedule_next(self) -> Optional[InferenceRequest]:
        # FIFO：先来先服务，不管 deadline
        if not self.queue.empty():
            return self.queue.get()
        return None

性能问题：在 GPU 利用率 85% 时，get_estimated_latency 误差高达 35%。高优先级任务排在低优先级后面，deadline 直接违例。

优化 v1：PCIe 竞争建模

import queue
from dataclasses import dataclass

@dataclass
class InferenceRequest:
    task_id: str
    model_name: str
    deadline: float
    priority: int
    # ... (其他字段省略)

class NaiveScheduler:
    """朴素 FIFO 调度器：不考虑优先级和延迟估计"""
    
    def __init__(self):
        self.queue = queue.Queue()
        self.latency_profile = {"resnet50": 3.2, "bert_base": 8.7, "yolov8": 5.1}
    
    def submit(self, req: InferenceRequest):
        self.queue.put(req)
    
    def get_estimated_latency(self, req: InferenceRequest) -> float:
        return self.latency_profile.get(req.model_name, 10.0)  # 直接查表，不考虑并发干扰
    
    def schedule_next(self):
        return self.queue.get() if not self.queue.empty() else None  # FIFO，不管 deadline

优化 v2：Kernel 干扰自适应预测

class PCIeContentionModel:
    """PCIe 带宽竞争建模：并发传输越多，每个任务分到的带宽越少。"""
    PCIE_TOTAL_BW_GBPS = 32.0

    def __init__(self):
        self.active_transfers = deque()  # (start_time, data_size_mb)
        # ... (threading.Lock 省略)

    def register_transfer(self, data_size_mb, start_time):
        self.active_transfers.append((start_time, data_size_mb))

    def _get_concurrent_bandwidth_demand(self, now):
        """计算 100ms 窗口内的并发带宽需求（GB/s）"""
        window = 0.1
        # ... (清理过期记录)
        total_mb = sum(mb for _, mb in self.active_transfers)
        return total_mb / 1024.0 / window

    def predict_transfer_time(self, data_size_mb):
        """预测竞争后的实际传输时间（ms）：带宽饱和时时间线性增长。"""
        concurrent_demand = self._get_concurrent_bandwidth_demand(time.time())

        if concurrent_demand > self.PCIE_TOTAL_BW_GBPS:
            # 带宽饱和：按需求比例分配
            available_bw = self.PCIE_TOTAL_BW_GBPS / (concurrent_demand / self.PCIE_TOTAL_BW_GBPS + 1)
        else:
            available_bw = self.PCIE_TOTAL_BW_GBPS

        return (data_size_mb / 1024.0) / available_bw * 1000.0

优化 v3：优先级感知调度器（完整实现）

class KernelInterferenceModel:
    """自适应 kernel 干扰预测：算术强度决定内存带宽竞争程度"""
    
    # (arithmetic_intensity: FLOP/byte, memory_bound: bool)
    MODEL_PROFILES = {
        "resnet50":    {"ai": 15.2, "memory_bound": False},  # compute bound
        "bert_base":   {"ai": 3.1,  "memory_bound": True},   # memory bound
        # ... (更多模型省略)
    }
    
    def __init__(self):
        self.interference_history = {}  # (model_a, model_b) -> 观测干扰系数列表
    
    def _compute_interference_factor(self, target_model: str, concurrent_models: list) -> float:
        target_profile = self.MODEL_PROFILES.get(target_model, {"ai": 5.0, "memory_bound": True})
        
        total_factor = 1.0
        for concurrent in concurrent_models:
            c_profile = self.MODEL_PROFILES.get(concurrent, {"ai": 5.0, "memory_bound": True})
            
            # 两者都是 memory-bound：HBM 带宽竞争，干扰大
            if target_profile["memory_bound"] and c_profile["memory_bound"]:
                interference = 1.25
            elif target_profile["memory_bound"] != c_profile["memory_bound"]:
                interference = 1.08  # 互补类型：干扰小
            else:
                interference = 1.15  # 都是 compute-bound：SM 竞争
            
            # 在线修正：历史 60% + 理论 40% 指数平滑
            key = (target_model, concurrent)
            if key in self.interference_history and len(self.interference_history[key]) >= 5:
                interference = 0.6 * np.mean(self.interference_history[key][-20:]) + 0.4 * interference
            
            total_factor *= interference
        
        return total_factor
    
    def predict_execution_time(self, model: str, base_latency_ms: float, concurrent_models: list) -> float:
        return base_latency_ms * self._compute_interference_factor(model, concurrent_models)
    
    def update_with_observation(self, target: str, concurrent: str, observed_factor: float):
        """在线学习：用实际观测修正模型"""
        self.interference_history.setdefault((target, concurrent), []).append(observed_factor)

常见错误

import heapq

class PriorityAwareScheduler:
    def __init__(self):
        # ... (PCIe/Kernel 预测模型初始化省略)
        self.high_priority_heap = []   # (urgency, arrival, req)
        self.low_priority_heap = []
        self.running_tasks = []
        self.LOW_PRIORITY_MAX_WAIT = 2.0  # 防饥饿阈值（秒）
    
    def submit(self, req):
        estimated_latency = self._estimate_total_latency(req)
        urgency = req.deadline - time.time() - estimated_latency  # 越小越紧迫
        entry = (urgency, req.arrival_time, req)
        if req.priority == 1:
            heapq.heappush(self.high_priority_heap, entry)
        else:
            heapq.heappush(self.low_priority_heap, entry)
    
    def _estimate_total_latency(self, req) -> float:
        """PCIe 传输时间 + Kernel 执行时间（含竞争干扰）"""
        transfer_time = self.pcie_model.predict_transfer_time(req.data_size_mb)
        exec_time = self.kernel_model.predict_execution_time(
            req.model_name, base_latency, [t.model_name for t in self.running_tasks]
        )
        return transfer_time + exec_time
    
    def schedule_next(self):
        now = time.time()
        self._promote_starving_low_priority(now)  # 防饥饿检查
        if self.high_priority_heap:
            _, _, req = heapq.heappop(self.high_priority_heap)
            return req
        elif self.low_priority_heap:
            _, _, req = heapq.heappop(self.low_priority_heap)
            return req
    
    def _promote_starving_low_priority(self, now: float):
        """等待超过阈值的低优先级任务提升至高优先级队列"""
        surviving = []
        for entry in self.low_priority_heap:
            urgency, arrival, req = entry
            if now - req.arrival_time > self.LOW_PRIORITY_MAX_WAIT:
                heapq.heappush(self.high_priority_heap, entry)
            else:
                surviving.append(entry)
        self.low_priority_heap = surviving
        heapq.heapify(self.low_priority_heap)

性能实测

测试环境：2× A100 80GB, PCIe 4.0, CUDA 12.1, Ubuntu 22.04

混合工作负载：ResNet-50（高优先级）+ BERT-base（低优先级），50/50 到达率，高优先级 deadline = 20ms

调度方案	HP Deadline 违例率	LP 违例率	延迟预测误差	GPU 利用率
FIFO (baseline)	18.3%	12.1%	35%	87%
Priority Queue（固定延迟）	14.7%	15.8%	35%	86%
+ PCIe 竞争建模	10.2%	16.1%	18%	87%
+ Kernel 干扰模型（Strait）	7.1%	17.3%	11%	88%

高优先级违例率从 18.3% 降至 7.1%，代价是低优先级违例率轻微上升（+5.2%），与论文数据吻合。

什么时候用 / 不用？

适用场景	不适用场景
GPU 利用率持续 > 70%	低利用率场景（< 50%），固定 profile 已够用
有明确 SLA 分级的混合负载	所有请求优先级相同
模型种类稳定（干扰历史可积累）	模型频繁更新，历史数据失效
on-premises 独占 GPU 集群	公有云（PCIe 拓扑不可预知）
任务有明确 deadline 约束	纯吞吐优化场景

调试技巧

如何验证 PCIe 竞争模型准确性：

# 用 nvtx + Nsight Systems 记录实际传输时间
# 对比预测值和实测值，如果误差 > 20% 说明带宽参数需要重新测量
nsys profile --trace=cuda,nvtx python inference_server.py

如何 profile 你的模型的 Arithmetic Intensity：

# Nsight Compute 直接给出 roofline 分析
ncu --metrics sm__sass_thread_inst_executed_op_fadd_pred_on.sum,\
l1tex__t_bytes.sum ./your_model_inference
# 目标：ai < 5 → memory bound，ai > 10 → compute bound

防饥饿阈值调整：LOW_PRIORITY_MAX_WAIT 应该设为低优先级任务 SLA 的 50%，不是固定 2 秒。

局限性

在线学习有冷启动问题：系统刚启动时干扰模型不准，需要约 200 个请求的 warm-up
MIG 模式下干扰模型需重新 profile：物理隔离改变了干扰的数学关系
防饥饿机制引入了优先级反转：极端情况下低优先级任务被提升后挤占高优先级 deadline
PCIe 竞争模型假设带宽均匀分配：实际 DMA 调度可能有其他策略，需要在目标硬件上验证