模型服务的批处理调度攒批策略、超时控制与吞吐量优化一、为什么一次处理 1 个请求和一次处理 16 个请求总耗时差距能到 4 倍GPU 推理有一个独特的性能特征单次推理和批量推理的延迟差距不大但吞吐量差距巨大。一块 A100 GPU 处理一个 prompt 需要 2 秒但同时处理 16 个 prompt 可能只需要 3 秒。这是因为矩阵乘法等操作在大 batch 下可以充分利用 GPU 的并行计算单元和高带宽显存。batch size 太小GPU 的许多计算单元闲着batch size 增大到一定程度利用率饱和再增大吞吐提升不再明显。这个特性决定了模型服务的调度策略和传统的 CPU 服务截然不同。传统服务追求来一个处理一个越快越好而模型服务需要攒一批再处理吞吐优先。问题在于攒批就意味着等待。如果攒齐 16 个请求需要 3 秒那第一个请求的等待时间就变成了 3 秒——对实时判题场景来说这个延迟不可接受。这就是批处理调度的核心矛盾吞吐量和延迟的博弈。flowchart TD A[请求到达] -- B[加入攒批缓冲区] B -- C{调度决策点} C -- D{缓冲区请求数 最大批大小?} D --|是| E[立即发起批推理] D --|否| F{最早请求等待时间 超时阈值?} F --|是| E F --|否| G[继续等待更多请求] G -- B E -- H[组装 batch prompt] H -- I[调用模型批量推理] I -- J[拆分推理结果] J -- K[逐个返回给对应请求]二、动态批处理的两大调度策略策略一固定窗口攒批。设置一个固定的时间窗口如 100ms在这个窗口内到达的所有请求组成一个批次。窗口到期后不管批大小是多少立即发起推理。这个策略简单直接延迟上限可控最长等待 窗口大小但批大小不确定——低峰期可能只有一个请求批处理的效果体现不出来。策略二双条件触发。同时设置批大小上限和等待超时。缓冲区请求数达到上限时立即触发缓冲区等待时间超时时也立即触发。哪个条件先满足就用哪个条件执行。这个策略在高峰期自动变成大 batch 高吞吐在低峰期自动变成小 batch 低延迟适应性最好。无论哪种策略都有一个关键的设计第一个请求的等待时间和最后一个请求的等待时间差异很大。在使用双条件触发时超时阈值应该基于第一个请求的等待时间来计算确保每个请求的最大等待延迟都有上界。三、批处理调度器的生产实现 模型批处理调度器 设计要点 1. 双条件触发批满或超时以第一个请求的等待时间为准 2. 异步返回每个请求持有自己的 Future推理完成后 set_result 3. 背压控制缓冲区满时拒绝新请求防止内存溢出 import asyncio import time from dataclasses import dataclass from typing import List, Dict, Any dataclass class BatchRequest: 单个批处理请求 prompt: str future: asyncio.Future # 用于异步返回结果的 Future arrival_time: float # 到达时间用于超时判断 request_id: str # 唯一标识用于结果匹配 class ModelBatchScheduler: def __init__( self, max_batch_size: int 16, # 最大批大小 max_wait_ms: int 200, # 最长等待时间毫秒 max_buffer_size: int 1000 # 缓冲区最大容量背压控制 ): self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms / 1000.0 # 转换为秒 self.max_buffer_size max_buffer_size self.buffer: List[BatchRequest] [] self._loop_task: asyncio.Task None # 调度循环的任务引用 async def submit(self, prompt: str, request_id: str) - Dict[str, Any]: 提交一个推理请求 返回一个 awaitable调用方 await 它即可等待推理结果 如果缓冲区满直接抛异常拒绝 # 背压控制缓冲区满时拒绝新请求 # 这个保护很重要 —— 如果推理速度跟不上请求速度缓冲区会无限增长 # 最终导致 OOM。拒绝策略让上游感知到压力并做限流或降级。 if len(self.buffer) self.max_buffer_size: raise BufferFullError( f批处理缓冲区已满 (max{self.max_buffer_size}) ) # 创建 Future用于异步返回结果 future asyncio.get_event_loop().create_future() request BatchRequest( promptprompt, futurefuture, arrival_timetime.monotonic(), request_idrequest_id ) self.buffer.append(request) # 如果这是第一个进入空缓冲区的请求启动调度循环 if len(self.buffer) 1 and self._loop_task is None: self._loop_task asyncio.create_task(self._batch_loop()) # 等待推理结果 return await future async def _batch_loop(self): 批处理调度主循环 持续检查缓冲区满足条件时触发批推理 try: while self.buffer: # 等待批满或超时 await self._wait_for_batch_condition() if not self.buffer: break # 取出一批请求进行处理 batch self._pop_batch() # 异步执行批量推理不阻塞调度循环 asyncio.create_task(self._process_batch(batch)) finally: self._loop_task None async def _wait_for_batch_condition(self): 等待批处理条件满足 两个条件之一满足即可 - 缓冲区请求数达到 max_batch_size - 最早请求的等待时间超过 max_wait_ms while self.buffer: # 条件1批已满 if len(self.buffer) self.max_batch_size: return # 条件2最早请求等待超时 first_arrival self.buffer[0].arrival_time elapsed time.monotonic() - first_arrival if elapsed self.max_wait_ms: return # 还没满足条件短暂等待后重试 # sleep 时间要够短保证批满时能立刻响应 await asyncio.sleep(0.01) # 10ms 检查间隔 def _pop_batch(self) - List[BatchRequest]: 从缓冲区取出一批请求 batch_size min(self.max_batch_size, len(self.buffer)) batch self.buffer[:batch_size] self.buffer self.buffer[batch_size:] return batch async def _process_batch(self, batch: List[BatchRequest]): 执行批量推理 组装 prompt → 调用模型 → 拆分结果 → 逐个 set_result try: # 组装批量 prompt prompts [req.prompt for req in batch] # 调用模型批量推理 # 这里假设 model.infer_batch 是异步方法 results await model_service.infer_batch(prompts) # 拆分结果并返回假设结果顺序与输入顺序一致 for req, result in zip(batch, results): if not req.future.done(): req.future.set_result(result) except Exception as e: # 批量推理整体失败所有请求都标记为异常 # 生产环境可以更精细化处理如部分失败时重试 for req in batch: if not req.future.done(): req.future.set_exception(e) class BufferFullError(Exception): 缓冲区满异常 pass调度循环中的asyncio.sleep(0.01)看似粗暴实际上在 asyncio 中是高效的做法。10ms 的检查间隔在批大小未满的情况下引入了微小的延迟但在批满的情况下len(self.buffer) self.max_batch_size条件在每次循环都会被快速检查到。异步设计中_process_batch使用了asyncio.create_task而不是await。这个区别很关键如果 batch processing 本身需要 5 秒用await会让调度循环在 5 秒内无法接收新请求——这正好违背了批处理的本意——让后续到达的请求能进入下一个批次。四、批处理的边界场景冷启动批处理系统刚启动时只有零星的请求如果 max_wait_ms 设为 200ms这些请求都要等满 200ms 才能被处理。解决方案是设置动态等待时间——通过近期的请求到达率来预测再等多久可能凑够一个批次如果概率很低就直接发送。请求大小不均不同题目的 prompt 长度差异很大。简单的判断语法正确性的 prompt 可能只有 500 tokens复杂算法分析的 prompt 可能有 3000 tokens。GPU 的显存是按最长的 prompt * batch_size 来分配的如果批次中有特别长的 prompt会大量浪费显存。解决方案是按 prompt 长度做分组批处理相同长度区间的请求组成一批。批次取消用户可能在等待过程中取消了请求关闭页面。这时候对应的 prompt 应该从缓冲区中移除释放 Future 占用的资源。五、总结批处理调度是 GPU 推理场景下的核心优化手段。双条件触发批满 超时是最实用的调度策略它在高负载时自动利用批处理提高吞吐在低负载时自动缩短等待时间保证延迟。实现上需要注意背压保护、异步调度循环、失败时批量异常处理等细节。批大小和超时阈值这两个参数没有统一的最优值需要根据实际压测数据来调优但原则是一致的在延迟可接受的范围内批越大越好。
