大模型推理队列优先级紧急请求不能被批量任务堵死一、实时对话请求等了 3 分钟因为批量摘要任务占满了 GPU我们的推理服务同时服务两类请求在线对话P99 延迟要求 5s和离线批量任务文档总结、数据分析允许排队。早期实现就是一个 FIFO 队列先进先出。结果批量任务一次性提交 200 个后后面的在线对话请求全部排在队尾——等 200 个每个耗时 10s 的批量任务跑完3 分钟后才轮到对话请求。这不是 GPU 不够的问题是调度策略的问题。推理队列必须引入优先级机制和高优先级抢占确保延迟敏感型请求不被吞吐敏感型请求堵死。二、优先级调度与抢占模型graph TD A[客户端请求] -- B{请求类型?} B --|在线对话| C[P0 紧急队列br/超时 30s] B --|API 调用| D[P1 高优队列br/超时 60s] B --|批量任务| E[P2 普通队列br/超时 300s] B --|离线分析| F[P3 低优队列br/无超时] C -- G{有空闲 GPU?} D -- G E -- G F -- G G --|是| H[直接执行] G --|否| I{高优请求br/可抢占?} I --|是| J[抢占低优任务] I --|否| K[排队等待] J -- L[Checkpoint 保存br/被占任务状态] L -- M[执行高优任务] M -- N[恢复被占任务] K -- O[超时?] O --|是| P[返回 503br/附预估等待时间] O --|否| G style C fill:#FF6B6B,color:#fff style D fill:#F5A623,color:#000 style F fill:#87CEEB,color:#000 style P fill:#FF4444,color:#fff核心调度策略四优先级队列P0在线对话→ P1API 调用→ P2批量任务→ P3离线分析。P0 有最高优先级和最短超时。高优抢占当 P0/P1 请求到达且没有空闲 GPU 时如果当前 GPU 执行的是 P2/P3 任务触发抢占Preemption。被抢占的任务保存 Checkpoint 后挂起等高优任务完成后再恢复。老化机制Aging长期排队的低优任务缓慢提升优先级防止低优任务永久饥饿。预估等待时间请求排队时根据当前队列长度和平均处理时间估算等待时长并返回客户端。客户端可据此决定是继续等还是降级。三、生产级优先级调度实现 大模型推理队列优先级调度器 支持多优先级、抢占、老化和等待时间预估 import asyncio import heapq import time import uuid from dataclasses import dataclass, field from enum import IntEnum from typing import Optional, Dict, Any, List class Priority(IntEnum): 优先级定义值越小优先级越高 P0_URGENT 0 # 紧急在线对话 P1_HIGH 1 # 高优API 调用 P2_NORMAL 2 # 普通批量任务 P3_LOW 3 # 低优离线分析 dataclass(orderTrue) class InferenceRequest: 推理请求 priority: int # 排序字段优先按 priority submit_time: float # 排序字段同优先级按 FIFO request_id: str field(compareFalse) model_name: str field(compareFalse) input_data: dict field(compareFalse) max_tokens: int field(compareFalse, default2048) timeout_sec: float field(compareFalse, default30) # 结果 result_future: asyncio.Future field(compareFalse) # 统计 created_at: float field(compareFalse, default_factorytime.time) started_at: Optional[float] field(compareFalse, defaultNone) checkpoint: Optional[Dict[str, Any]] field(compareFalse, defaultNone) is_preempted: bool field(compareFalse, defaultFalse) def __post_init__(self): if self.result_future is None: self.result_future asyncio.Future() property def wait_time_sec(self) - float: 当前已经等待的时间 return (self.started_at or time.time()) - self.created_at class PriorityInferenceScheduler: 优先级推理调度器 核心特性 1. 4 级优先级队列 2. 高优抢占低优任务 3. 老化机制防止饥饿 4. 等待时间预估 def __init__( self, max_concurrent: int 4, aging_interval_sec: float 60, aging_priority_boost: int -1, # 每轮老化降低 1 级 ): self.max_concurrent max_concurrent self.aging_interval_sec aging_interval_sec self.aging_priority_boost aging_priority_boost # 四组优先级队列每种优先级一个堆 self._queues: Dict[Priority, List[InferenceRequest]] { p: [] for p in Priority } self._queue_lock asyncio.Lock() # 当前运行的任务{request_id: request} self._running: Dict[str, InferenceRequest] {} # 统计信息 self._total_processed 0 self._total_preempted 0 self._recent_latencies: List[float] [] # 最近 100 次推理的耗时 # 启动后台任务 self._start_background_tasks() def _start_background_tasks(self): 启动后台维护任务 asyncio.create_task(self._aging_loop()) async def submit(self, request: InferenceRequest) - InferenceRequest: 提交推理请求 request.submit_time time.time() async with self._queue_lock: heapq.heappush(self._queues[Priority(request.priority)], request) # 触发调度 asyncio.create_task(self._dispatch()) return request async def _dispatch(self): 核心调度逻辑 策略 1. P0 请求到达时如果无空闲槽位抢占最低优先级的任务 2. P1 请求到达时如果无空闲槽位且当前运行 P3抢占 P3 3. P2/P3 有新请求时仅有空闲槽位时才分配 async with self._queue_lock: available_slots self.max_concurrent - len(self._running) # 按优先级从高到低处理 for priority in Priority: if available_slots 0: break queue self._queues[priority] while queue and available_slots 0: request heapq.heappop(queue) # 超时检查 if request.wait_time_sec request.timeout_sec: request.result_future.set_exception( TimeoutError( f排队超时 {request.wait_time_sec:.0f}s f限制 {request.timeout_sec}s ) ) continue self._running[request.request_id] request request.started_at time.time() asyncio.create_task(self._execute(request)) available_slots - 1 # 抢占逻辑P0/P1 到达且无空闲槽位 if available_slots 0: # 检查是否有 P0/P1 在等待 urgent_waiting len(self._queues[Priority.P0_URGENT]) high_waiting len(self._queues[Priority.P1_HIGH]) if urgent_waiting 0 or high_waiting 0: await self._try_preempt() async def _try_preempt(self): 尝试抢占低优先级的运行任务 抢占规则 - P0 可以抢占 P2/P3 - P1 可以抢占 P3 - 不能抢占 P1防止级联抢占 # 找到可以被抢占的最低优先级任务 preempt_target None min_preempt_priority Priority.P3_LOW for priority in Priority: if priority Priority.P1_HIGH: continue # P0/P1 不能被抢占 # 找到该优先级的一个运行任务 for req in self._running.values(): if req.priority priority and req.priority min_preempt_priority: preempt_target req min_preempt_priority Priority(req.priority) if preempt_target: await self._preempt_task(preempt_target) async def _preempt_task(self, request: InferenceRequest): 抢占一个低优任务 为什么需要 Checkpoint LLM 推理的结果是累积的已生成的 token 如果不保存状态被抢占后需要从头开始浪费 GPU 算力。 # 保存推理进度 checkpoint await self._save_checkpoint(request) request.checkpoint checkpoint request.is_preempted True self._total_preempted 1 # 从运行列表移除 self._running.pop(request.request_id, None) # 重新入队保持原优先级但插入队首 # 为什么回到原队列而非降低优先级 # 被占不是任务本身的错不应惩罚它 request.submit_time time.time() # 更新提交时间继续等待 async with self._queue_lock: heapq.heappush( self._queues[Priority(request.priority)], request ) # 触发调度 asyncio.create_task(self._dispatch()) async def _execute(self, request: InferenceRequest): 执行推理 try: # 从 Checkpoint 恢复如果有 if request.checkpoint: # 从上次中断的位置继续推理 pass # 调用模型推理 result await self._run_inference(request) request.result_future.set_result(result) # 更新统计数据 self._total_processed 1 if request.started_at: latency time.time() - request.started_at self._recent_latencies.append(latency) # 保留最近 100 次 if len(self._recent_latencies) 100: self._recent_latencies self._recent_latencies[-100:] except asyncio.CancelledError: # 被抢占 raise except Exception as e: request.result_future.set_exception(e) finally: self._running.pop(request.request_id, None) asyncio.create_task(self._dispatch()) async def _aging_loop(self): 老化循环定期提升长期等待的低优任务优先级 为什么需要老化 在持续高负载下低优任务可能被无限推迟饥饿。 老化让等待时间超过阈值的任务逐步提升优先级 最终被调度执行 while True: await asyncio.sleep(self.aging_interval_sec) async with self._queue_lock: for priority in range(Priority.P3_LOW.value, Priority.P0_URGENT.value - 1, -1): queue self._queues[Priority(priority)] new_queue [] while queue: request heapq.heappop(queue) # 等待超过老化间隔的任务提升一级优先级 if request.wait_time_sec self.aging_interval_sec: request.priority max( Priority.P0_URGENT, request.priority self.aging_priority_boost ) # 移到新队列 heapq.heappush( self._queues[Priority(request.priority)], request ) else: new_queue.append(request) # 重建堆 for req in new_queue: heapq.heappush(queue, req) def estimate_wait_time(self, priority: Priority) - float: 预估等待时间秒 async with self._queue_lock: ahead_count 0 for p in Priority: if p priority: # 更高优先级的所有排队请求 ahead_count len(self._queues[p]) elif p priority: # 同优先级中提交更早的 ahead_count len(self._queues[p]) # 加上正在运行但可被抢占的 for req in self._running.values(): if req.priority priority: ahead_count 1 # 使用平均推理延迟估算 avg_latency ( sum(self._recent_latencies[-20:]) / max(len(self._recent_latencies[-20:]), 1) ) if self._recent_latencies else 10.0 return ahead_count * avg_latency / self.max_concurrent async def _run_inference(self, request: InferenceRequest): 实际的模型推理简化实现 pass async def _save_checkpoint(self, request: InferenceRequest) - Dict: 保存推理检查点 return {} def get_stats(self) - Dict: 获取调度统计 return { queues: { p.name: len(self._queues[p]) for p in Priority }, running: len(self._running), total_processed: self._total_processed, total_preempted: self._total_preempted, avg_latency_ms: ( sum(self._recent_latencies) / len(self._recent_latencies) * 1000 if self._recent_latencies else 0 ), }四、优先级调度的陷阱缺点抢占的代价保存 Checkpoint 需要时间尤其是长序列的 KV Cache抢占本身可能耽误 2-5 秒。这段时间高优请求仍然在等待。P0 滥用风险如果所有调用方都把请求标为 P0优先级机制退化为 FIFO。必须严格控制 P0 配额如每用户每秒最多 2 个 P0 请求。Checkpoint 存储成本被抢占的任务需要保存状态。如果一次对话生成了 10000 token 的 KV CacheCheckpoint 可能达到几百 MB。禁用场景所有请求的延迟要求相同优先级机制 introduce overhead 无收益。单模型单实例部署没有并发就没有抢占场景。五、总结推理队列的优先级调度核心是三个机制多优先级队列P0-P3确保在线请求不被批量任务堵死高优抢占让紧急请求能抢走低优任务的 GPU 时间片老化机制防止低优任务永久饥饿。需要额外处理的两个工程问题抢占时的 Checkpoint 保存和恢复避免低优任务被抢后从头重算以及 P0 配额控制防止滥用。